autogen_ext.code_executors.azure#

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 容器应用动态会话实例执行代码。

注意

这个类需要 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 时,此值将被重置

注意

使用当前目录 (“.”) 作为工作目录已弃用。 使用它将引发弃用警告。

FUNCTION_PROMPT_TEMPLATE: ClassVar[str] = '您可以使用以下用户定义的函数。\n\n$functions'#
SUPPORTED_LANGUAGES: ClassVar[List[str]] = ['python']#
异步 download_files(files: List[Path | str], cancellation_token: CancellationToken) List[str][source]#
异步 execute_code_blocks(code_blocks: List[CodeBlock], cancellation_token: CancellationToken) CodeResult[source]#

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

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

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

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

返回值:

CodeResult – 代码执行的结果。

format_functions_for_prompt(prompt_template: str = FUNCTION_PROMPT_TEMPLATE) str[source]#

(实验性)格式化提示的函数。

该模板包含一个变量: - $functions:格式化为存根的函数,每个函数之间有两个换行符。

参数:

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

返回值:

str – 格式化的提示。

属性 functions: List[str]#
属性 functions_module: str#

(实验性)函数的模块名称。

异步 get_available_packages(cancellation_token: CancellationToken) set[str][source]#
异步 get_file_list(cancellation_token: CancellationToken) List[str][source]#
异步 restart() None[source]#

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

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

异步 start() None[source]#

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

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

async stop() None[source]#

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

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

property timeout: int#

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

async upload_files(files: List[Path | str], cancellation_token: CancellationToken) None[source]#
property work_dir: Path#
class TokenProvider(*args, **kwargs)[source]#

基类:Protocol

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