autogen_ext.code_executors.jupyter#

class JupyterCodeExecutor(kernel_name: str = 'python3', timeout: int = 60, output_dir: str | Path | None = None)[source]#

Bases: CodeExecutor, Component[JupyterCodeExecutorConfig]

一个代码执行器类,它使用 [nbclient](jupyter/nbclient) 有状态地执行代码。

危险

这将在本地计算机上执行代码。如果与 LLM 生成的代码一起使用,应谨慎。

直接使用的例子

import asyncio
from autogen_core import CancellationToken
from autogen_core.code_executor import CodeBlock
from autogen_ext.code_executors.jupyter import JupyterCodeExecutor


async def main() -> None:
    async with JupyterCodeExecutor() as executor:
        cancel_token = CancellationToken()
        code_blocks = [CodeBlock(code="print('hello world!')", language="python")]
        code_result = await executor.execute_code_blocks(code_blocks, cancel_token)
        print(code_result)


asyncio.run(main())

PythonCodeExecutionTool 一起使用的例子

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.code_executors.jupyter import JupyterCodeExecutor
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.code_execution import PythonCodeExecutionTool


async def main() -> None:
    async with JupyterCodeExecutor() as executor:
        tool = PythonCodeExecutionTool(executor)
        model_client = OpenAIChatCompletionClient(model="gpt-4o")
        agent = AssistantAgent("assistant", model_client=model_client, tools=[tool])
        result = await agent.run(task="What is the 10th Fibonacci number? Use Python to calculate it.")
        print(result)


asyncio.run(main())

CodeExecutorAgent 中使用的例子

import asyncio
from autogen_agentchat.agents import CodeExecutorAgent
from autogen_agentchat.messages import TextMessage
from autogen_ext.code_executors.jupyter import JupyterCodeExecutor
from autogen_core import CancellationToken


async def main() -> None:
    async with JupyterCodeExecutor() as executor:
        code_executor_agent = CodeExecutorAgent("code_executor", code_executor=executor)
        task = TextMessage(
            content='''Here is some code
    ```python
    print('Hello world')
    ```
    ''',
            source="user",
        )
        response = await code_executor_agent.on_messages([task], CancellationToken())
        print(response.chat_message)


asyncio.run(main())
参数:
  • kernel_name (str) – 要使用的内核名称。 默认为“python3”。

  • timeout (int) – 代码执行的超时时间,默认为 60。

  • output_dir (Path) – 用于保存输出文件的目录,默认为临时目录。

注意

不建议使用当前目录(“.”)作为输出目录。 使用它会引发弃用警告。

component_config_schema#

JupyterCodeExecutorConfig 的别名

component_provider_override: ClassVar[str | None] = 'autogen_ext.code_executors.jupyter.JupyterCodeExecutor'#

覆盖组件的 provider 字符串。 这应该用于防止内部模块名称成为模块名称的一部分。

async execute_code_blocks(code_blocks: list[CodeBlock], cancellation_token: CancellationToken) JupyterCodeResult[source]#

执行代码块并返回结果。

参数:

code_blocks (list[CodeBlock]) – 要执行的代码块列表。

返回值:

JupyterCodeResult – 代码执行的结果。

property output_dir: Path#
async restart() None[source]#

重启代码执行器。

async start() None[source]#

(实验性) 启动代码执行器。

通过创建一个新的 notebook 并使用指定的 Jupyter Kernel 对其进行设置,来初始化 Jupyter Notebook 执行环境。将执行器标记为已启动,允许执行代码。 在执行任何代码块之前,应调用此方法。

async stop() None[source]#

(实验性) 停止代码执行器。

通过退出 kernel 上下文并清理相关资源来终止 Jupyter Notebook 执行。

class JupyterCodeResult(exit_code: int, output: str, output_files: list[Path])[source]#

基类: CodeResult

用于 Jupyter 代码执行器的代码结果类。

output_files: list[Path]#