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'#

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

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]) -- 要执行的代码块列表。

Returns:

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) -- 要启动的内核规范名称

Returns:

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[源代码]#