autogen_ext.code_executors.docker_jupyter

class DockerJupyterCodeExecutor(jupyter_server: JupyterConnectable | JupyterConnectionInfo, kernel_name: str = 'python3', timeout: int = 60, output_dir: Path | None = None)

基类: CodeExecutor, Component[DockerJupyterCodeExecutorConfig]

(实验性)一个代码执行器类，它使用提供给该类的 Jupyter 服务器来有状态地执行代码。

每次执行都是有状态的，并且可以访问在同一会话中从先前的执行创建的变量。

要使用它，您需要安装以下依赖项

pip install "autogen-ext[docker-jupyter-executor]"

参数:

• jupyter_server (Union[JupyterConnectable, JupyterConnectionInfo]) – 要使用的 Jupyter 服务器。

• kernel_name (str) – 要使用的内核名称。 确保已安装。 默认情况下，它是“python3”。

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

• output_dir (str) – 保存输出文件的目录，默认为 None。

直接使用它的示例

import asyncio
from autogen_core import CancellationToken
from autogen_core.code_executor import CodeBlock
from autogen_ext.code_executors.docker_jupyter import DockerJupyterCodeExecutor, DockerJupyterServer


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


asyncio.run(main())

使用您自己的 jupyter 镜像的示例

import asyncio
from autogen_core import CancellationToken
from autogen_core.code_executor import CodeBlock
from autogen_ext.code_executors.docker_jupyter import DockerJupyterCodeExecutor, DockerJupyterServer


async def main() -> None:
    async with DockerJupyterServer(custom_image_name="your_custom_images_name", expose_port=8888) as jupyter_server:
        async with DockerJupyterCodeExecutor(jupyter_server=jupyter_server) as executor:
            code_blocks = [CodeBlock(code="print('hello world!')", language="python")]
            code_result = await executor.execute_code_blocks(code_blocks, cancellation_token=CancellationToken())
            print(code_result)


asyncio.run(main())

与 PythonCodeExecutionTool 一起使用的示例

import asyncio
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.code_executors.docker_jupyter import DockerJupyterCodeExecutor, DockerJupyterServer
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.code_execution import PythonCodeExecutionTool


async def main() -> None:
    async with DockerJupyterServer() as jupyter_server:
        async with DockerJupyterCodeExecutor(jupyter_server=jupyter_server) 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.docker_jupyter import DockerJupyterCodeExecutor, DockerJupyterServer
from autogen_core import CancellationToken


async def main() -> None:
    async with DockerJupyterServer() as jupyter_server:
        async with DockerJupyterCodeExecutor(jupyter_server=jupyter_server) 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())

component_config_schema

别名 DockerJupyterCodeExecutorConfig

component_provider_override: ClassVar[str | None] = 'autogen_ext.code_executors.docker_jupyter.DockerJupyterCodeExecutor'

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

async execute_code_blocks(code_blocks: List[CodeBlock], cancellation_token: CancellationToken) → DockerJupyterCodeResult

(实验性)执行代码块列表并返回结果。

此方法将代码块列表作为 Jupyter 内核中的单元格执行。请参阅：https://jupyter-client.readthedocs.io/en/stable/messaging.html 获取消息协议。

参数:

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

返回:

DockerJupyterCodeResult – 代码执行的结果。

async restart() → None

(实验性)重新启动新会话。

async start() → None

(实验性)开始新会话。

async stop() → None

停止内核。

class DockerJupyterCodeResult(exit_code: int, output: str, output_files: list[Path])

基类： CodeResult

（实验性）IPython 代码执行器的代码结果类。

output_files: list[Path]

class DockerJupyterServer(*, custom_image_name: str | None = None, container_name: str | None = None, auto_remove: bool = True, stop_container: bool = True, docker_env: Dict[str, str] | None = None, expose_port: int = 8888, token: str | GenerateToken | None = None, work_dir: Path | str = '/workspace', bind_dir: Path | str | None = None)

基类： JupyterConnectable

DEFAULT_DOCKERFILE = 'FROM quay.io/jupyter/docker-stacks-foundation\n\n        SHELL ["/bin/bash", "-o", "pipefail", "-c"]\n\n        USER ${NB_UID}\n        RUN mamba install --yes jupyter_kernel_gateway ipykernel &&             mamba clean --all -f -y &&             fix-permissions "${CONDA_DIR}" &&             fix-permissions "/home/${NB_USER}"\n\n        ENV TOKEN="UNSET"\n        CMD python -m jupyter kernelgateway --KernelGatewayApp.ip=0.0.0.0             --KernelGatewayApp.port=8888             --KernelGatewayApp.auth_token="${TOKEN}"             --JupyterApp.answer_yes=true             --JupyterWebsocketPersonality.list_kernels=true\n\n        EXPOSE 8888\n\n        WORKDIR "${HOME}"\n        '

class GenerateToken

基类： object

property connection_info: JupyterConnectionInfo

返回此可连接对象的连接信息。

async get_client() → JupyterClient

async stop() → None

class JupyterClient(connection_info: JupyterConnectionInfo)

基类： object

async close() → None

关闭异步会话

async delete_kernel(kernel_id: str) → None

async get_kernel_client(kernel_id: str) → JupyterKernelClient

async list_kernel_specs() → Dict[str, Dict[str, str]]

async list_kernels() → List[Dict[str, str]]

async restart_kernel(kernel_id: str) → None

async start_kernel(kernel_spec_name: str) → str

异步启动一个新内核。

参数:

kernel_spec_name (str) – 要启动的内核规范的名称

返回:

str – 启动的内核的ID

class JupyterKernelClient(websocket: ClientConnection)

基类： object

用于与 Jupyter 内核通信的异步客户端。

async execute(code: str, timeout_seconds: float | None = None) → ExecutionResult

async stop() → None

async wait_for_ready(timeout_seconds: float | None = None) → bool