Bases: BaseModel
(Experimental) A code executor class that executes code statefully using an embedded IPython kernel managed by this class.
This will execute LLM generated code on the local machine.
Each execution is stateful and can access variables created from previous executions in the same session. The kernel must be installed before using this class. The kernel can be installed using the following command: python -m ipykernel install --user --name {kernel_name} where kernel_name is the name of the kernel to install.
Source code in autogen/coding/jupyter/embedded_ipython_code_executor.py
| def __init__(self, **kwargs: Any):
super().__init__(**kwargs)
# Check if the kernel is installed.
if self.kernel_name not in KernelSpecManager().find_kernel_specs():
raise ValueError(
f"Kernel {self.kernel_name} is not installed. "
"Please first install it with "
f"`python -m ipykernel install --user --name {self.kernel_name}`."
)
self._kernel_manager = KernelManager(kernel_name=self.kernel_name)
self._kernel_manager.start_kernel()
self._kernel_client = self._kernel_manager.client()
self._kernel_client.start_channels()
self._timeout = self.timeout
self._kernel_name = self.kernel_name
self._output_dir = Path(self.output_dir)
|
class-attribute instance-attribute
timeout = Field(default=60, ge=1, description='The timeout for code execution.')
class-attribute instance-attribute
kernel_name = Field(default='python3', description='The kernel name to use. Make sure it is installed.')
class-attribute instance-attribute
output_dir = Field(default='.', description='The directory to save output files.')
(Experimental) Export a code extractor that can be used by an agent.
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 an IPython kernel managed by this class. 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/embedded_ipython_code_executor.py
| 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 an IPython kernel
managed by this class.
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._kernel_client.wait_for_ready()
outputs = []
output_files = []
for code_block in code_blocks:
code = self._process_code(code_block.code)
self._kernel_client.execute(code, store_history=True)
while True:
try:
msg = self._kernel_client.get_iopub_msg(timeout=self._timeout)
msg_type = msg["msg_type"]
content = msg["content"]
if msg_type in ["execute_result", "display_data"]:
for data_type, data in content["data"].items():
if data_type == "text/plain":
# Output is a text.
outputs.append(data)
elif data_type.startswith("image/"):
# Output is an image.
path = self._save_image(data)
outputs.append(f"Image data saved to {path}")
output_files.append(path)
elif data_type == "text/html":
# Output is an html.
path = self._save_html(data)
outputs.append(f"HTML data saved to {path}")
output_files.append(path)
else:
# Output raw data.
outputs.append(json.dumps(data))
elif msg_type == "stream":
# Output is a text.
outputs.append(content["text"])
elif msg_type == "error":
# Output is an error.
return IPythonCodeResult(
exit_code=1,
output=f"ERROR: {content['ename']}: {content['evalue']}\n{content['traceback']}",
)
if msg_type == "status" and content["execution_state"] == "idle":
break
# handle time outs.
except Empty:
return IPythonCodeResult(
exit_code=1,
output=f"ERROR: Timeout waiting for output from code block: {code_block.code}",
)
# We return the full output.
return IPythonCodeResult(
exit_code=0, output="\n".join([str(output) for output in outputs]), output_files=output_files
)
|
(Experimental) Restart a new session.
Source code in autogen/coding/jupyter/embedded_ipython_code_executor.py
| def restart(self) -> None:
"""(Experimental) Restart a new session."""
self._kernel_client.stop_channels()
self._kernel_manager.shutdown_kernel()
self._kernel_manager = KernelManager(kernel_name=self.kernel_name)
self._kernel_manager.start_kernel()
self._kernel_client = self._kernel_manager.client()
self._kernel_client.start_channels()
|