autogen_core.code_executor._base 源代码

# File based from: https://github.com/microsoft/autogen/blob/main/autogen/coding/base.py
# Credit to original authors

from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass
from types import TracebackType
from typing import List, Optional, Type

from pydantic import BaseModel
from typing_extensions import Self

from .._cancellation_token import CancellationToken
from .._component_config import ComponentBase




@dataclass
class CodeBlock:
    """从代理消息中提取的代码块。"""

    code: str
    language: str





@dataclass
class CodeResult:
    """代码执行的结果。"""

    exit_code: int
    output: str





class CodeExecutor(ABC, ComponentBase[BaseModel]):
    """执行代码块并返回结果。

    这是一个代码执行器的抽象基类。它定义了执行代码块并返回结果的接口。
    需要提供该类的具体实现以在特定环境中执行代码块。例如，:class:`~autogen_ext.code_executors.docker.DockerCommandLineCodeExecutor`
    在Docker容器的命令行环境中执行代码块。

    建议子类作为上下文管理器使用，以确保资源被正确清理。为此，需要实现
    :meth:`~autogen_core.code_executor.CodeExecutor.start`和
    :meth:`~autogen_core.code_executor.CodeExecutor.stop`方法，
    这些方法将在进入和退出上下文管理器时被调用。

    """

    component_type = "code_executor"



    @abstractmethod
    async def execute_code_blocks(
        self, code_blocks: List[CodeBlock], cancellation_token: CancellationToken
    ) -> CodeResult:
        """执行代码块并返回结果。

        该方法应由代码执行器实现。

        Args:
            code_blocks (List[CodeBlock]): 要执行的代码块。

        Returns:
            CodeResult: 代码执行结果。

        Raises:
            ValueError: 用户输入错误
            asyncio.TimeoutError: 代码执行超时
            asyncio.CancelledError: 执行期间触发取消令牌
        """
        ...




    @abstractmethod
    async def start(self) -> None:
        """启动代码执行器。"""
        ...




    @abstractmethod
    async def stop(self) -> None:
        """停止代码执行器并释放所有资源。"""
        ...




    @abstractmethod
    async def restart(self) -> None:
        """重启代码执行器。

        该方法应由代码执行器实现。

        当代理重置时会调用此方法。
        """
        ...


    async def __aenter__(self) -> Self:
        await self.start()
        return self

    async def __aexit__(
        self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType]
    ) -> Optional[bool]:
        await self.stop()
        return None