AutoGen智能体开发：autogen_core.code_executor包

autogen_core.code_executor

王国平

455人浏览 · 2025-12-04 07:43:34

王国平 · 2025-12-04 07:43:34 发布

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

基类: object

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

code: str

language: str

class CodeExecutor[源代码]

基类: ABC, ComponentBase[BaseModel]

执行代码块并返回结果。

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

建议将子类用作上下文管理器，以确保资源得到正确清理。为此，请实现start()和stop()方法，这些方法将在进入和退出上下文管理器时调用。

component_type: ClassVar[ComponentType] = 'code_executor'

组件的逻辑类型。

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

执行代码块并返回结果。

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

参数:

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

CodeResult – 代码执行的结果。

抛出:

ValueError – 用户输入中的错误
TimeoutError – 代码执行超时
CancelledError – 执行期间CancellationToken被撤销

abstract async start() → None [源代码]

启动代码执行器。

abstract async stop() → None [源代码]

停止代码执行器并释放任何资源。

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

重新启动代码执行器。

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

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

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

基类: object

代码执行的结果。

exit_code: int

output: str

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

基类: object

name: str

alias: str

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

基类: object

module: str

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

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

基类: Generic[T, P]

func: Callable[[P], T]

python_packages: Sequence[str]

global_imports: Sequence[str | ImportFromModule | Alias]

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

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

基类: object

func: str

python_packages: Sequence[str]

global_imports: Sequence[str | ImportFromModule | Alias]

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

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 对象。默认为 []。

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:
    """Load some sample data.

    Returns:
        pandas.DataFrame: A DataFrame with sample data
    """
    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():
    # The decorated function can be used in executed code
    with tempfile.TemporaryDirectory() as temp_dir:
        executor = LocalCommandLineCodeExecutor(work_dir=temp_dir, functions=[load_data])
        code = f"""from {executor.functions_module} import load_data

        # Use the imported function
        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)  # Output: John

# Run the async example
asyncio.run(run_example())

《AI提示工程必知必会》为读者提供了丰富的AI提示工程知识与实战技能。《AI提示工程必知必会》主要内容包括各类提示词的应用，如问答式、指令式、状态类、建议式、安全类和感谢类提示词，以及如何通过实战演练掌握提示词的使用技巧；使用提示词进行文本摘要、改写重述、语法纠错、机器翻译等语言处理任务，以及在数据挖掘、程序开发等领域的应用；AI在绘画创作上的应用，百度文心一言和阿里通义大模型这两大智能平台的特性与功能，以及市场调研中提示词的实战应用。通过阅读《AI提示工程必知必会》，读者可掌握如何有效利用AI提示工程提升工作效率，创新工作流程，并在职场中脱颖而出。