autogen_core.components.code_executor#

class autogen_core.components.code_executor.Alias(name: 'str', alias: 'str')[source]#

Bases: object

alias: str#

name: str#

class autogen_core.components.code_executor.CodeBlock(code: str, language: str)[source]#

Bases: object

A code block extracted fromm an agent message.

code: str#

language: str#

class autogen_core.components.code_executor.CodeExecutor(*args, **kwargs)[source]#

Bases: Protocol

Executes code blocks and returns the result.

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
asyncio.TimeoutError – Code execution timeouts
asyncio.CancelledError – CancellationToken evoked during execution

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 autogen_core.components.code_executor.CodeResult(exit_code: int, output: str)[source]#

Bases: object

Result of a code execution.

exit_code: int#

output: str#

class autogen_core.components.code_executor.CommandLineCodeResult(exit_code: int, output: str, code_file: str | None)[source]#

Bases: CodeResult

A code result class for command line code executor.

code_file: str | None#

class autogen_core.components.code_executor.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 autogen_core.components.code_executor.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 autogen_core.components.code_executor.ImportFromModule(module: 'str', imports: 'List[Union[str, Alias]]')[source]#

Bases: object

imports: List[str | Alias]#

module: str#

class autogen_core.components.code_executor.LocalCommandLineCodeExecutor(timeout: int = 60, work_dir: Path | str = PosixPath('.'), functions: Sequence[FunctionWithRequirements[Any, A] | Callable[[...], Any] | FunctionWithRequirementsStr] = [], functions_module: str = 'functions', virtual_env_context: SimpleNamespace | None = None)[source]#

Bases: CodeExecutor

A code executor class that executes code through a local command line environment.

Danger

This will execute code on the local machine. If being used with LLM generated code, caution should be used.

Each code block is saved as a file and executed in a separate process in the working directory, and a unique file is generated and saved in the working directory for each code block. The code blocks are executed in the order they are received. Command line code is sanitized using regular expression match against a list of dangerous commands in order to prevent self-destructive commands from being executed which may potentially affect the users environment. Currently the only supported languages is Python and shell scripts. For Python code, use the language “python” for the code block. For shell scripts, use the language “bash”, “shell”, or “sh” for the code block.

Parameters:

timeout (int) – The timeout for the execution of any single code block. Default is 60.
work_dir (str) – The working directory for the code execution. If None, a default working directory will be used. The default working directory is the current directory “.”.
functions (List[Union[FunctionWithRequirements[Any, A], Callable[..., Any]]]) – A list of functions that are available to the code executor. Default is an empty list.
functions_module (str, optional) – The name of the module that will be created to store the functions. Defaults to “functions”.
virtual_env_context (Optional[SimpleNamespace], optional) – The virtual environment context. Defaults to None.

Example:

How to use LocalCommandLineCodeExecutor with a virtual environment different from the one used to run the autogen application: Set up a virtual environment using the venv module, and pass its context to the initializer of LocalCommandLineCodeExecutor. This way, the executor will run code within the new environment.

import venv
from pathlib import Path

from autogen_core.base import CancellationToken
from autogen_core.components.code_executor import CodeBlock, LocalCommandLineCodeExecutor

work_dir = Path("coding")
work_dir.mkdir(exist_ok=True)

venv_dir = work_dir / ".venv"
venv_builder = venv.EnvBuilder(with_pip=True)
venv_builder.create(venv_dir)
venv_context = venv_builder.ensure_directories(venv_dir)

local_executor = LocalCommandLineCodeExecutor(work_dir=work_dir, virtual_env_context=venv_context)
await local_executor.execute_code_blocks(
    code_blocks=[
        CodeBlock(language="bash", code="pip install matplotlib"),
    ],
    cancellation_token=CancellationToken(),
)

FUNCTION_PROMPT_TEMPLATE: ClassVar[str] = 'You have access to the following user defined functions. They can be accessed from the module called `$module_name` by their function names.\n\nFor example, if there was a function called `foo` you could import it by writing `from $module_name import foo`\n\n$functions'#

SUPPORTED_LANGUAGES: ClassVar[List[str]] = ['bash', 'shell', 'sh', 'pwsh', 'powershell', 'ps1', 'python']#

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

(Experimental) Execute the code blocks and return the result.

Parameters:

code_blocks (List[CodeBlock]) – The code blocks to execute.
cancellation_token (CancellationToken) – a token to cancel the operation

Returns:

CommandLineCodeResult – The result of the code execution.

format_functions_for_prompt(prompt_template: str = 'You have access to the following user defined functions. They can be accessed from the module called `$module_name` by their function names.\n\nFor example, if there was a function called `foo` you could import it by writing `from $module_name import foo`\n\n$functions') → str[source]#

(Experimental) Format the functions for a prompt.

The template includes two variables: - $module_name: The module name. - $functions: The functions formatted as stubs with two newlines between each function.

Parameters:: prompt_template (str) – The prompt template. Default is the class default.
Returns:: str – The formatted prompt.

property functions: List[str]#

property functions_module: str#: (Experimental) The module name for the functions.

async restart() → None[source]#: (Experimental) Restart the code executor.

property timeout: int#: (Experimental) The timeout for code execution.

property work_dir: Path#: (Experimental) The working directory for the code execution.

autogen_core.components.code_executor.build_python_functions_file(funcs: Sequence[FunctionWithRequirements[Any, P] | Callable[[...], Any] | FunctionWithRequirementsStr]) → str[source]#

autogen_core.components.code_executor.extract_markdown_code_blocks(markdown_text: str) → List[CodeBlock][source]#

autogen_core.components.code_executor.get_file_name_from_content(code: str, workspace_path: Path) → str | None[source]#

autogen_core.components.code_executor.get_required_packages(code: str, lang: str) → set[str][source]#

autogen_core.components.code_executor.lang_to_cmd(lang: str) → str[source]#

autogen_core.components.code_executor.silence_pip(code: str, lang: str) → str[source]#: Apply -qqq flag to pip install commands.

autogen_core.components.code_executor.to_stub(func: Callable[[...], Any] | FunctionWithRequirementsStr) → str[source]#

Generate a stub for a function as a string

Parameters:: func (Callable[..., Any]) – The function to generate a stub for
Returns:: str – The stub for the function

autogen_core.components.code_executor.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

Parameters:

python_packages (List[str], optional) – Packages required to function. Can include version info.. Defaults to [].
global_imports (List[Import], optional) – Required imports. Defaults to [].

Returns:

Callable[[Callable[P, T]], FunctionWithRequirements[T, P]] – The decorated function