autogen_ext.code_executors.docker_jupyter#

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

Bases: CodeExecutor, Component[DockerJupyterCodeExecutorConfig]

(Experimentell) Eine Code-Executor-Klasse, die Code zustandsbehaftet über einen dieser Klasse bereitgestellten Jupyter-Server ausführt.

Jede Ausführung ist zustandsbehaftet und kann auf Variablen zugreifen, die aus früheren Ausführungen in derselben Sitzung erstellt wurden.

Um dies zu verwenden, müssen Sie die folgenden Abhängigkeiten installieren

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

Parameter:

jupyter_server (Union[JupyterConnectable, JupyterConnectionInfo]) – Der zu verwendende Jupyter-Server.
kernel_name (str) – Der zu verwendende Kernel-Name. Stellen Sie sicher, dass er installiert ist. Standardmäßig ist dies „python3“.
timeout (int) – Das Timeout für die Codeausführung, standardmäßig 60.
output_dir (str) – Das Verzeichnis zum Speichern von Ausgabedateien, standardmäßig None.

Beispiel für die direkte Verwendung

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())

Beispiel für die Verwendung mit Ihrem eigenen Jupyter-Image

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())

Beispiel für die Verwendung mit 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())

Beispiel für die Verwendung innerhalb eines 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#: Alias von DockerJupyterCodeExecutorConfig

component_provider_override: ClassVar[str | None] = 'autogen_ext.code_executors.docker_jupyter.DockerJupyterCodeExecutor'#: Überschreibe den Anbieter-String für die Komponente. Dies sollte verwendet werden, um zu verhindern, dass interne Modulnamen Teil des Modulnamens werden.

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

(Experimentell) Führt eine Liste von Codeblöcken aus und gibt das Ergebnis zurück.

Diese Methode führt eine Liste von Codeblöcken als Zellen im Jupyter-Kernel aus. Siehe: https://jupyter-client.readthedocs.io/en/stable/messaging.html für das Nachrichtenprotokoll.

Parameter:: code_blocks (List[CodeBlock]) – Eine Liste von auszuführenden Codeblöcken.
Gibt zurück:: DockerJupyterCodeResult – Das Ergebnis der Codeausführung.

async restart() → None[Quelle]#: (Experimentell) Startet eine neue Sitzung neu.

async start() → None[Quelle]#: (Experimentell) Startet eine neue Sitzung.

async stop() → None[Quelle]#: Beendet den Kernel.

Bases: 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[Quelle]#: Basiert auf: object

property connection_info: JupyterConnectionInfo#: Gibt die Verbindungsinformationen für diesen Connectable zurück.

async stop() → None[Quelle]#

async get_client() → JupyterClient[Quelle]#

class JupyterClient(connection_info: JupyterConnectionInfo)[Quelle]#

Basiert auf: object

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

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

async start_kernel(kernel_spec_name: str) → str[Quelle]#

Startet asynchron einen neuen Kernel.

Parameter:: kernel_spec_name (str) – Name der zu startenden Kernel-Spezifikation
Gibt zurück:: str – ID des gestarteten Kernels

async delete_kernel(kernel_id: str) → None[Quelle]#

async restart_kernel(kernel_id: str) → None[Quelle]#

async get_kernel_client(kernel_id: str) → JupyterKernelClient[Quelle]#

async close() → None[Quelle]#: Schließt die asynchrone Sitzung

class JupyterKernelClient(websocket: ClientConnection)[Quelle]#

Basiert auf: object

Ein asynchroner Client zur Kommunikation mit einem Jupyter-Kernel.

async stop() → None[Quelle]#

async wait_for_ready(timeout_seconds: float | None = None) → bool[Quelle]#

async execute(code: str, timeout_seconds: float | None = None) → ExecutionResult[Quelle]#

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

Bases: CodeResult

(Experimentell) Eine Code-Ergebnisklasse für den IPython-Code-Executor.

output_files: list[Path]#