autogen_ext.code_executors.azure

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

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:

pip install 'autogen-ext[azure]==0.4.0.dev11'

Caution

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]

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

(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

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

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

async restart() → None

(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

property work_dir: Path

(Experimental) The working directory for the code execution.

class TokenProvider(*args, **kwargs)

Bases: Protocol

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