autogen_ext.code_executors.azure#

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

基类：Protocol

get_token(*scopes: str, claims: str | None = None, tenant_id: str | None = None, **kwargs: Any) → AccessToken[source]#
class ACADynamicSessionsCodeExecutor(pool_management_endpoint: str, credential: TokenProvider, timeout: int = 60, work_dir: Path | str | None = None, functions: Sequence[FunctionWithRequirements[Any, A] | Callable[[...], Any] | FunctionWithRequirementsStr] = [], functions_module: str = 'functions', suppress_result_output: bool = False, session_id: str | None = None)[source]#

基类： CodeExecutor

（实验性）一个代码执行器类，通过 Azure Container Apps 动态会话实例执行代码。

注意

此类别需要 `autogen-ext` 包的 `azure` 额外功能

pip install "autogen-ext[azure]"

注意

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

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

参数:

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

  • credential (TokenProvider) – 实现 `get_token` 函数的对象。

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

  • work_dir (str) – 代码执行的工作目录。如果为 None，将使用默认工作目录。默认工作目录是临时目录。

  • functions (List[Union[FunctionWithRequirements[Any, A], Callable[..., Any]]]) – 可供代码执行器使用的函数列表。默认为空列表。

  • bool (suppress_result_output) – 默认情况下，执行器会将执行响应中的任何结果信息附加到结果输出中。将其设置为 True 可防止此操作。

  • session_id (str) – 代码执行的会话 ID（传递给动态会话）。如果为 None，将生成一个新的会话 ID。默认为 None。请注意，调用 `restart` 时此值将被重置

注意

将当前目录（“.”）用作工作目录已被弃用。使用它将引发弃用警告。

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

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

模板包含一个变量：- `functions`：函数以存根形式格式化，每个函数之间有两行新行。

参数:

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

返回:

str – 格式化的提示。

property functions_module: str#

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

property functions: List[str]#
property timeout: int#

(实验性) 代码执行的超时时间。

property work_dir: Path#
async get_available_packages(cancellation_token: CancellationToken) → set[str]#
async get_file_list(cancellation_token: CancellationToken) → List[str]#
async upload_files(files: List[Path | str], cancellation_token: CancellationToken) → None#
async download_files(files: List[Path | str], cancellation_token: CancellationToken) → List[str]#
async execute_code_blocks(code_blocks: List[CodeBlock], cancellation_token: CancellationToken) → CodeResult#

(实验性) 执行代码块并返回结果。

参数:

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

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

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

返回:

CodeResult – 代码执行的结果。

async restart() → None#

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

通过生成新的会话 ID 并重置设置变量来重置执行器的内部状态。这会导致下一次代码执行重新初始化环境并重新运行任何设置代码。

async start() → None#

(实验性) 启动代码执行器。

将代码执行器标记为已启动。

async stop() → None#

(实验性) 停止代码执行器。

在清理临时工作目录（如果已创建）后停止代码执行器。