autogen_core.code_executor#

class Alias(name: 'str', alias: 'str')[源代码]#

基类：object

alias: str#

name: str#

class CodeBlock(code: str, language: str)[源代码]#

基类：object

从代理消息中提取的代码块。

code: str#

language: str#

class CodeExecutor[源代码]#

基类：ABC, ComponentBase[BaseModel]

执行代码块并返回结果。

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

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

component_type: ClassVar[ComponentType] = 'code_executor'#: 组件的逻辑类型。

abstract async execute_code_blocks(code_blocks: List[CodeBlock], cancellation_token: CancellationToken) → CodeResult[源代码]#

执行代码块并返回结果。

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

参数:

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

Returns:

CodeResult -- 代码执行结果。

抛出:

ValueError -- 用户输入错误
TimeoutError -- 代码执行超时
CancelledError -- 执行期间触发取消令牌

abstract async restart() → None[源代码]#

重启代码执行器。

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

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

abstract async start() → None[源代码]#: 启动代码执行器。

abstract async stop() → None[源代码]#: 停止代码执行器并释放所有资源。

class CodeResult(exit_code: int, output: str)[源代码]#

基类：object

代码执行的结果。

exit_code: int#

output: str#

class FunctionWithRequirements(func: 'Callable[P, T]', python_packages: 'Sequence[str]' = <factory>, global_imports: 'Sequence[Import]' = <factory>)[源代码]#

基类：Generic[T, P]

classmethod from_callable(func: Callable[[P], T], python_packages: Sequence[str] = [], global_imports: Sequence[str | ImportFromModule | Alias] = []) → FunctionWithRequirements[T, P][源代码]#

static from_str(func: str, python_packages: Sequence[str] = [], global_imports: Sequence[str | ImportFromModule | Alias] = []) → FunctionWithRequirementsStr[源代码]#

func: Callable[[P], T]#

global_imports: Sequence[str | ImportFromModule | Alias]#

python_packages: Sequence[str]#

class FunctionWithRequirementsStr(func: 'str', python_packages: 'Sequence[str]' = [], global_imports: 'Sequence[Import]' = [])[源代码]#

基类：object

compiled_func: Callable[[...], Any]#

func: str#

global_imports: Sequence[str | ImportFromModule | Alias]#

python_packages: Sequence[str]#

class ImportFromModule(module: 'str', imports: 'Union[Tuple[Union[str, Alias], ...], List[Union[str, Alias]]]')[源代码]#

基类：object

imports: Tuple[str | Alias, ...]#

module: str#

with_requirements(python_packages: Sequence[str] = [], global_imports: Sequence[str | ImportFromModule | Alias] = []) → Callable[[Callable[[P], T]], FunctionWithRequirements[T, P]][源代码]#

为函数添加代码执行环境所需的包和导入依赖装饰器。

该装饰器通过将函数包装在`FunctionWithRequirements`对象中，使其可在动态执行的代码块中被引用。该对象会跟踪函数的依赖项。当被装饰的函数传递给代码执行器时，它可以在执行的代码中按名称导入，所有依赖项会自动处理。

参数:

python_packages (Sequence[str], optional) -- 函数所需的Python包。可包含版本说明(如["pandas>=1.0.0"])。默认为[]。
global_imports (Sequence[Import], optional) -- 函数所需的导入语句。可以是字符串("numpy")、ImportFromModule对象或Alias对象。默认为[]。

Returns:

Callable[[Callable[P, T]], FunctionWithRequirements[T, P]] -- 一个装饰器，包装目标函数的同时保留其功能并注册其依赖项。

示例

import tempfile
import asyncio
from autogen_core import CancellationToken
from autogen_core.code_executor import with_requirements, CodeBlock
from autogen_ext.code_executors.local import LocalCommandLineCodeExecutor
import pandas

@with_requirements(python_packages=["pandas"], global_imports=["pandas"])
def load_data() -> pandas.DataFrame:
    """加载示例数据。

    Returns:
        pandas.DataFrame: 包含示例数据的DataFrame
    """
    data = {
        "name": ["John", "Anna", "Peter", "Linda"],
        "location": ["New York", "Paris", "Berlin", "London"],
        "age": [24, 13, 53, 33],
    }
    return pandas.DataFrame(data)

async def run_example():
    # 被装饰的函数可在执行代码中使用
    with tempfile.TemporaryDirectory() as temp_dir:
        executor = LocalCommandLineCodeExecutor(work_dir=temp_dir, functions=[load_data])
        code = f"""from {executor.functions_module} import load_data

        # 使用导入的函数
        data = load_data()
        print(data['name'][0])"""

        result = await executor.execute_code_blocks(
            code_blocks=[CodeBlock(language="python", code=code)],
            cancellation_token=CancellationToken(),
        )
        print(result.output)  # 输出: John

# 运行异步示例
asyncio.run(run_example())