Skip to content

JupyterCodeExecutor

autogen.coding.jupyter.JupyterCodeExecutor #

JupyterCodeExecutor(jupyter_server, kernel_name='python3', timeout=60, output_dir=Path())

Bases: CodeExecutor

(Experimental) A code executor class that executes code statefully using a Jupyter server supplied to this class.

Each execution is stateful and can access variables created from previous executions in the same session.

PARAMETER DESCRIPTION
jupyter_server

The Jupyter server to use.

TYPE: Union[JupyterConnectable, JupyterConnectionInfo]

timeout

The timeout for code execution, by default 60.

TYPE: int DEFAULT: 60

kernel_name

The kernel name to use. Make sure it is installed. By default, it is "python3".

TYPE: str DEFAULT: 'python3'

output_dir

The directory to save output files, by default ".".

TYPE: str DEFAULT: Path()

Source code in autogen/coding/jupyter/jupyter_code_executor.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
def __init__(
    self,
    jupyter_server: Union[JupyterConnectable, JupyterConnectionInfo],
    kernel_name: str = "python3",
    timeout: int = 60,
    output_dir: Union[Path, str] = Path(),
):
    """(Experimental) A code executor class that executes code statefully using
    a Jupyter server supplied to this class.

    Each execution is stateful and can access variables created from previous
    executions in the same session.

    Args:
        jupyter_server (Union[JupyterConnectable, JupyterConnectionInfo]): The Jupyter server to use.
        timeout (int): The timeout for code execution, by default 60.
        kernel_name (str): The kernel name to use. Make sure it is installed.
            By default, it is "python3".
        output_dir (str): The directory to save output files, by default ".".
    """
    if timeout < 1:
        raise ValueError("Timeout must be greater than or equal to 1.")

    if isinstance(output_dir, str):
        output_dir = Path(output_dir)

    if not output_dir.exists():
        raise ValueError(f"Output directory {output_dir} does not exist.")

    if isinstance(jupyter_server, JupyterConnectable):
        self._connection_info = jupyter_server.connection_info
    elif isinstance(jupyter_server, JupyterConnectionInfo):
        self._connection_info = jupyter_server
    else:
        raise ValueError("jupyter_server must be a JupyterConnectable or JupyterConnectionInfo.")

    self._jupyter_client = JupyterClient(self._connection_info)
    available_kernels = self._jupyter_client.list_kernel_specs()
    if kernel_name not in available_kernels["kernelspecs"]:
        raise ValueError(f"Kernel {kernel_name} is not installed.")

    self._kernel_id = self._jupyter_client.start_kernel(kernel_name)
    self._kernel_name = kernel_name
    self._jupyter_kernel_client = self._jupyter_client.get_kernel_client(self._kernel_id)
    self._timeout = timeout
    self._output_dir = output_dir

code_extractor property #

code_extractor

(Experimental) Export a code extractor that can be used by an agent.

execute_code_blocks #

execute_code_blocks(code_blocks)

(Experimental) Execute a list of code blocks and return the result.

This method executes a list of code blocks as cells in the Jupyter kernel. See: https://jupyter-client.readthedocs.io/en/stable/messaging.html for the message protocol.

PARAMETER DESCRIPTION
code_blocks

A list of code blocks to execute.

TYPE: List[CodeBlock]

RETURNS DESCRIPTION
IPythonCodeResult

The result of the code execution.

TYPE: IPythonCodeResult

Source code in autogen/coding/jupyter/jupyter_code_executor.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def execute_code_blocks(self, code_blocks: list[CodeBlock]) -> IPythonCodeResult:
    """(Experimental) Execute a list of code blocks and return the result.

    This method executes a list of code blocks as cells in the Jupyter kernel.
    See: https://jupyter-client.readthedocs.io/en/stable/messaging.html
    for the message protocol.

    Args:
        code_blocks (List[CodeBlock]): A list of code blocks to execute.

    Returns:
        IPythonCodeResult: The result of the code execution.
    """
    self._jupyter_kernel_client.wait_for_ready()
    outputs = []
    output_files = []
    for code_block in code_blocks:
        code = silence_pip(code_block.code, code_block.language)
        result = self._jupyter_kernel_client.execute(code, timeout_seconds=self._timeout)
        if result.is_ok:
            outputs.append(result.output)
            for data in result.data_items:
                if data.mime_type == "image/png":
                    path = self._save_image(data.data)
                    outputs.append(f"Image data saved to {path}")
                    output_files.append(path)
                elif data.mime_type == "text/html":
                    path = self._save_html(data.data)
                    outputs.append(f"HTML data saved to {path}")
                    output_files.append(path)
                else:
                    outputs.append(json.dumps(data.data))
        else:
            return IPythonCodeResult(
                exit_code=1,
                output=f"ERROR: {result.output}",
            )

    return IPythonCodeResult(
        exit_code=0, output="\n".join([str(output) for output in outputs]), output_files=output_files
    )

restart #

restart()

(Experimental) Restart a new session.

Source code in autogen/coding/jupyter/jupyter_code_executor.py 
126
127
128
129
def restart(self) -> None:
    """(Experimental) Restart a new session."""
    self._jupyter_client.restart_kernel(self._kernel_id)
    self._jupyter_kernel_client = self._jupyter_client.get_kernel_client(self._kernel_id)

stop #

stop()

Stop the kernel.

Source code in autogen/coding/jupyter/jupyter_code_executor.py 
150
151
152
def stop(self) -> None:
    """Stop the kernel."""
    self._jupyter_client.delete_kernel(self._kernel_id)