autogen_ext.code_executors#

class ACADynamicSessionsCodeExecutor(pool_management_endpoint: str, credential: TokenProvider, timeout: int = 60, work_dir: Path | str = PosixPath('.'), functions: Sequence[FunctionWithRequirements[Any, A] | Callable[[...], Any] | FunctionWithRequirementsStr] = [], functions_module: str = 'functions')[source]#

Bases: CodeExecutor

(Experimental) A code executor class that executes code through a an Azure Container Apps Dynamic Sessions instance.

Note

This class requires the azure extra for the autogen-ext package.

This will execute LLM generated code on an Azure dynamic code container.

The execution environment is similar to that of a jupyter notebook which allows for incremental code execution. The parameter functions are executed in order once at the beginning of each session. Each code block is then executed serially and in the order they are received. Each environment has a statically defined set of available packages which cannot be changed. Currently, attempting to use packages beyond what is available on the environment will result in an error. To get the list of supported packages, call the get_available_packages function. Currently the only supported language is Python. For Python code, use the language “python” for the code block.

Parameters:

pool_management_endpoint (str) – The azure container apps dynamic sessions endpoint.
credential (TokenProvider) – An object that implements the get_token function.
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.

FUNCTION_PROMPT_TEMPLATE: ClassVar[str] = 'You have access to the following user defined functions.\n\n$functions'#

SUPPORTED_LANGUAGES: ClassVar[List[str]] = ['python']#

async download_files(files: List[Path | str], cancellation_token: CancellationToken) → List[str][source]#

async execute_code_blocks(code_blocks: List[CodeBlock], cancellation_token: CancellationToken) → CodeResult[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
input_files (Optional[Union[Path, str]]) – Any files the code blocks will need to access

Returns:

CodeResult – The result of the code execution.

format_functions_for_prompt(prompt_template: str = 'You have access to the following user defined functions.\n\n$functions') → str[source]#

(Experimental) Format the functions for a prompt.

The template includes one variable: - $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 get_available_packages(cancellation_token: CancellationToken) → set[str][source]#

async get_file_list(cancellation_token: CancellationToken) → List[str][source]#

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

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

async upload_files(files: List[Path | str], cancellation_token: CancellationToken) → None[source]#

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

class DockerCommandLineCodeExecutor(image: str = 'python:3-slim', container_name: str | None = None, *, timeout: int = 60, work_dir: Path | str = PosixPath('.'), bind_dir: str | Path | None = None, auto_remove: bool = True, stop_container: bool = True, functions: Sequence[FunctionWithRequirements[Any, A] | Callable[[...], Any] | FunctionWithRequirementsStr] = [], functions_module: str = 'functions')[source]#

Bases: CodeExecutor

Executes code through a command line environment in a Docker container.

Note

This class requires the docker extra for the autogen-ext package.

The executor first saves each code block in a file in the working directory, and then executes the code file in the container. The executor executes the code blocks in the order they are received. Currently, the executor only supports 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:

image (_type_, optional) – Docker image to use for code execution. Defaults to “python:3-slim”.
container_name (Optional[str], optional) – Name of the Docker container which is created. If None, will autogenerate a name. Defaults to None.
timeout (int, optional) – The timeout for code execution. Defaults to 60.
work_dir (Union[Path, str], optional) – The working directory for the code execution. Defaults to Path(“.”).
bind_dir (Union[Path, str], optional) – The directory that will be bound
spawn (to the code executor container. Useful for cases where you want to)
work_dir. (the container from within a container. Defaults to)
auto_remove (bool, optional) – If true, will automatically remove the Docker container when it is stopped. Defaults to True.
stop_container (bool, optional) – If true, will automatically stop the container when stop is called, when the context manager exits or when the Python process exits with atext. Defaults to True.
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”.

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']#

property bind_dir: Path#: (Experimental) The binding directory for the code execution container.

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.
Returns:: CommandlineCodeResult – The result of the code 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.

async start() → None[source]#

async stop() → None[source]#: (Experimental) Stop the code executor.

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

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

class TokenProvider(*args, **kwargs)[source]#

Bases: Protocol

get_token(*scopes: str, claims: str | None = None, tenant_id: str | None = None, **kwargs: Any) → AccessToken[source]#