autogen_ext.code_executors.azure#

class ACADynamicSessionsCodeExecutor(pool_management_endpoint: str, credential: TokenProvider, timeout: int = 60, work_dir: 路径 | str = Path('.'), functions: Sequence[FunctionWithRequirements[任何, A] | Callable[[...], 任何] | FunctionWithRequirementsStr] = [], functions_module: str = 'functions')[源代码]#

基类：CodeExecutor

（实验性质）一个通过Azure容器应用动态会话实例执行代码的代码执行器类。

注意

该类需要为autogen-ext包安装azure额外组件：

pip install "autogen-ext[azure]"

注意

这将在Azure动态代码容器上执行LLM生成的代码。

执行环境类似于jupyter笔记本，允许增量代码执行。参数函数在每次会话开始时按顺序执行一次。然后每个代码块按接收顺序依次执行。每个环境都有静态定义的一组可用包，这些包无法更改。 目前，尝试使用环境中不可用的包将会导致错误。要获取支持的包列表，请调用get_available_packages函数。 目前唯一支持的语言是Python。 对于Python代码，请在代码块中使用“python”作为语言。

Parameters:

  • pool_management_endpoint (str) – Azure 容器应用动态会话的端点。

  • credential (TokenProvider) – 一个实现了get_token函数的对象。

  • timeout (int) – 任何单个代码块的执行超时时间。默认值为60。

  • work_dir (str) – 代码执行的工作目录。如果为 None，将使用默认的工作目录。默认工作目录是当前目录“.”。

  • functions (列表[联合[FunctionWithRequirements[任何, A], 可调用[..., 任何]]]) – 可供代码执行器使用的函数列表。默认值为空列表。

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: 列表[路径 | str], cancellation_token: CancellationToken) → 列表[str][源代码]#
async execute_code_blocks(code_blocks: 列表[CodeBlock], cancellation_token: CancellationToken) → CodeResult[源代码]#

（实验性）执行代码块并返回结果。

Parameters:

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

  • cancellation_token (CancellationToken) – 用于取消操作的令牌

  • input_files (Optional[Union[Path, str]]) – 代码块需要访问的任何文件

Returns:

CodeResult – 代码执行的结果。

format_functions_for_prompt(prompt_template: str = FUNCTION_PROMPT_TEMPLATE) → str[源代码]#

（实验性）格式化提示的函数。

模板包含一个变量： - $functions: 以stubs格式化的函数，每个函数之间有两个换行符。

Parameters:

prompt_template (str) – 提示模板。默认为类的默认值。

Returns:

str – 格式化的提示信息。

property functions: 列表[str]#
property functions_module: str#

（实验性）函数的模块名称。

async get_available_packages(cancellation_token: CancellationToken) → set[str][源代码]#
async get_file_list(cancellation_token: CancellationToken) → 列表[str][源代码]#
async restart() → 无[源代码]#

（实验性）重启代码执行器。

property timeout: int#

（实验性的）代码执行的超时时间。

async upload_files(files: 列表[路径 | str], cancellation_token: CancellationToken) → 无[源代码]#
property work_dir: 路径#

（实验性）代码执行的工作目录。

class TokenProvider(*args, **kwargs)[源代码]#

基础类: Protocol

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