autogen_core.code_executor#

class Alias(name: 'str', alias: 'str')[source]#

Bases: object

alias: str#

name: str#

class CodeBlock(code: str, language: str)[source]#

Bases: object

A code block extracted fromm an agent message.

code: str#

language: str#

class CodeExecutor[source]#

Bases: ABC, ComponentBase[BaseModel]

Executes code blocks and returns the result.

component_type: ClassVar[ComponentType] = 'code_executor'#

The logical type of the component.

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

Execute code blocks and return the result.

This method should be implemented by the code executor.

Parameters:

code_blocks (List[CodeBlock]) – The code blocks to execute.

Returns:

CodeResult – The result of the code execution.

Raises:

• ValueError – Errors in user inputs

• TimeoutError – Code execution timeouts

• CancelledError – CancellationToken evoked during execution

abstract async restart() → None[source]#

Restart the code executor.

This method should be implemented by the code executor.

This method is called when the agent is reset.

class CodeResult(exit_code: int, output: str)[source]#

Bases: object

Result of a code execution.

exit_code: int#

output: str#

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

Bases: Generic[T, P]

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

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

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]' = [])[source]#

Bases: 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]]]')[source]#

Bases: 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]][source]#

Decorate a function with package and import requirements for code execution environments.

This decorator makes a function available for reference in dynamically executed code blocks by wrapping it in a FunctionWithRequirements object that tracks its dependencies. When the decorated function is passed to a code executor, it can be imported by name in the executed code, with all dependencies automatically handled.

Parameters:

• python_packages (Sequence[str], optional) – Python packages required by the function. Can include version specifications (e.g., [“pandas>=1.0.0”]). Defaults to [].

• global_imports (Sequence[Import], optional) – Import statements required by the function. Can be strings (“numpy”), ImportFromModule objects, or Alias objects. Defaults to [].

Returns:

Callable[[Callable[P, T]], FunctionWithRequirements[T, P]] – A decorator that wraps the target function, preserving its functionality while registering its dependencies.

Example

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