autogen_ext.agents.azure#

class AzureAIAgent(name: str, description: str, project_client: AIProjectClient, deployment_name: str, instructions: str, tools: Iterable[Literal['file_search', 'code_interpreter', 'bing_grounding', 'azure_ai_search', 'azure_function', 'sharepoint_grounding'] | BingGroundingToolDefinition | CodeInterpreterToolDefinition | SharepointToolDefinition | AzureAISearchToolDefinition | FileSearchToolDefinition | AzureFunctionToolDefinition | Tool | Callable[[...], Any] | Callable[[...], Awaitable[Any]]] | None = None, agent_id: str | None = None, thread_id: str | None = None, metadata: Dict[str, str] | None = None, response_format: _types.AgentsApiResponseFormatOption | None = None, temperature: float | None = None, tool_resources: models.ToolResources | None = None, top_p: float | None = None)[源代码]#

基类：BaseChatAgent

AutoGen 的 Azure AI 助手代理。

安装方法：

pip install "autogen-ext[azure]"  # 用于 Azure AI Foundry 代理服务

该代理利用 Azure AI 助手 API 创建具有以下功能的 AI 助手：

代码解释与执行
基于 Bing 搜索的落地
文件处理与搜索
自定义函数调用
多轮对话

该代理与 AutoGen 的消息系统集成，提供了一种在 AutoGen 框架内无缝使用 Azure AI 功能的方式。支持代码解释器、文件搜索等多种落地机制工具。

代理名称必须是有效的 Python 标识符：

必须以字母（A-Z，a-z）或下划线（_）开头
只能包含字母、数字（0-9）或下划线
不能是 Python 关键字
不能包含空格或特殊字符
不能以数字开头

查看如何创建具有用户托管身份的安全代理： https://learn.microsoft.com/en-us/azure/ai-services/agents/how-to/virtual-networks

示例：

使用 AzureAIAgent 创建基于 Bing 落地的代理：

import asyncio
import os

from autogen_agentchat.messages import TextMessage
from autogen_core import CancellationToken
from autogen_ext.agents.azure._azure_ai_agent import AzureAIAgent
from azure.ai.projects.aio import AIProjectClient
from azure.identity.aio import DefaultAzureCredential
import azure.ai.projects.models as models
import dotenv


async def bing_example():
    credential = DefaultAzureCredential()

    async with AIProjectClient.from_connection_string(  # type: ignore
        credential=credential, conn_str=os.getenv("AI_PROJECT_CONNECTION_STRING", "")
    ) as project_client:
        conn = await project_client.connections.get(connection_name=os.getenv("BING_CONNECTION_NAME", ""))

        bing_tool = models.BingGroundingTool(conn.id)
        agent_with_bing_grounding = AzureAIAgent(
            name="bing_agent",
            description="An AI assistant with Bing grounding",
            project_client=project_client,
            deployment_name="gpt-4o",
            instructions="You are a helpful assistant.",
            tools=bing_tool.definitions,
            metadata={"source": "AzureAIAgent"},
        )

        # 要使 Bing 落地工具返回引用，消息中必须包含让模型返回引用的指令
        # 例如："请为答案提供引用来源"

        result = await agent_with_bing_grounding.on_messages(
            messages=[
                TextMessage(
                    content="What is Microsoft's annual leave policy? Provide citations for your answers.",
                    source="user",
                )
            ],
            cancellation_token=CancellationToken(),
            message_limit=5,
        )
        print(result)


if __name__ == "__main__":
    dotenv.load_dotenv()
    asyncio.run(bing_example())

使用 AzureAIAgent 创建具有文件搜索功能的代理：

import asyncio
import os
import tempfile
import urllib.request

import dotenv
from autogen_agentchat.messages import TextMessage
from autogen_core import CancellationToken
from autogen_ext.agents.azure._azure_ai_agent import AzureAIAgent
from azure.ai.projects.aio import AIProjectClient
from azure.identity.aio import DefaultAzureCredential


async def file_search_example():
    # 从 GitHub 下载 README.md
    readme_url = "https://raw.githubusercontent.com/microsoft/autogen/refs/heads/main/README.md"
    temp_file = None

    try:
        # 创建临时文件存储下载的 README
        temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".md")
        urllib.request.urlretrieve(readme_url, temp_file.name)
        print(f"Downloaded README.md to {temp_file.name}")

        credential = DefaultAzureCredential()
        async with AIProjectClient.from_connection_string(  # type: ignore
            credential=credential, conn_str=os.getenv("AI_PROJECT_CONNECTION_STRING", "")
        ) as project_client:
            agent_with_file_search = AzureAIAgent(
                name="file_search_agent",
                description="An AI assistant with file search capabilities",
                project_client=project_client,
                deployment_name="gpt-4o",
                instructions="You are a helpful assistant.",
                tools=["file_search"],
                metadata={"source": "AzureAIAgent"},
            )

            ct: CancellationToken = CancellationToken()
            # 使用下载的 README 文件进行文件搜索
            await agent_with_file_search.on_upload_for_file_search(
                file_paths=[temp_file.name],
                vector_store_name="file_upload_index",
                vector_store_metadata={"source": "AzureAIAgent"},
                cancellation_token=ct,
            )
            result = await agent_with_file_search.on_messages(
                messages=[
                    TextMessage(content="Hello, what is AutoGen and what capabilities does it have?", source="user")
                ],
                cancellation_token=ct,
                message_limit=5,
            )
            print(result)
    finally:
        # 清理临时文件
        if temp_file and os.path.exists(temp_file.name):
            os.unlink(temp_file.name)
            print(f"Removed temporary file {temp_file.name}")


if __name__ == "__main__":
    dotenv.load_dotenv()
    asyncio.run(file_search_example())

使用 AzureAIAgent 创建具有代码解释器功能的代理：

import asyncio
import os

import dotenv
from autogen_agentchat.messages import TextMessage
from autogen_core import CancellationToken
from autogen_ext.agents.azure._azure_ai_agent import AzureAIAgent
from azure.ai.projects.aio import AIProjectClient
from azure.identity.aio import DefaultAzureCredential


async def code_interpreter_example():
    credential = DefaultAzureCredential()
    async with AIProjectClient.from_connection_string(  # type: ignore
        credential=credential, conn_str=os.getenv("AI_PROJECT_CONNECTION_STRING", "")
    ) as project_client:
        agent_with_code_interpreter = AzureAIAgent(
            name="code_interpreter_agent",
            description="An AI assistant with code interpreter capabilities",
            project_client=project_client,
            deployment_name="gpt-4o",
            instructions="You are a helpful assistant.",
            tools=["code_interpreter"],
            metadata={"source": "AzureAIAgent"},
        )

        await agent_with_code_interpreter.on_upload_for_code_interpreter(
            file_paths="/workspaces/autogen/python/packages/autogen-core/docs/src/user-guide/core-user-guide/cookbook/data/nifty_500_quarterly_results.csv",
            cancellation_token=CancellationToken(),
        )

        result = await agent_with_code_interpreter.on_messages(
            messages=[
                TextMessage(
                    content="Aggregate the number of stocks per industry and give me a markdown table as a result?",
                    source="user",
                )
            ],
            cancellation_token=CancellationToken(),
        )

        print(result)


if __name__ == "__main__":
    dotenv.load_dotenv()
    asyncio.run(code_interpreter_example())

property agent_id: str#

property deployment_name: str#

property description: str#: 代理的描述。团队使用此描述来决定使用哪些代理。描述应说明代理的能力以及如何与之交互。

async handle_text_message(content: str, cancellation_token: CancellationToken | None = None) → None[源代码]#

处理文本消息，将其添加到对话线程中。

参数:

content (str) -- 消息的文本内容
cancellation_token (CancellationToken) -- 用于取消处理的令牌

Returns:

None

property instructions: str#

async load_state(state: Mapping[str, Any]) → None[源代码]#

将先前保存的状态加载到当前代理中。

此方法反序列化并恢复先前保存的代理状态，使代理能够继续之前的对话或会话。

参数:: state (Mapping[str, Any]) -- 先前保存的状态字典

async on_messages(messages: Sequence[BaseChatMessage], cancellation_token: CancellationToken | None = None, message_limit: int = 1) → Response[源代码]#

处理传入消息并返回来自 Azure AI 代理的响应。

该方法是与代理交互的主要入口点。它委托给 on_messages_stream 并返回最终响应。

参数:

messages (Sequence[ChatMessage]) -- 要处理的消息序列
cancellation_token (CancellationToken) -- 用于取消操作的令牌
message_limit (int, optional) -- 从线程中检索的最大消息数量

Returns:

Response -- 代理的响应，包括聊天消息和任何内部事件

抛出:

AssertionError -- 如果流未返回最终结果

处理传入消息并生成来自 Azure AI 代理的流式响应。

该方法处理与 Azure AI 代理的完整交互流程： 1. 处理输入消息 2. 创建并监控运行 3. 处理工具调用及其结果 4. 检索并返回代理的最终响应

该方法在处理过程中会生成事件（如工具调用），最后生成包含代理消息的完整 Response。

参数:

messages (Sequence[ChatMessage]) -- 要处理的消息序列
cancellation_token (CancellationToken) -- 用于取消操作的令牌
message_limit (int, optional) -- 从线程中检索的最大消息数量
sleep_interval (float, optional) -- 轮询运行状态时的休眠间隔时间

生成器:

AgentEvent | ChatMessage | Response -- 处理过程中的事件及最终响应

抛出:

ValueError -- 如果运行失败或未收到来自助手的消息

async on_reset(cancellation_token: CancellationToken) → None[源代码]#

通过创建新线程重置代理的对话。

此方法允许在不丢失代理定义或功能的情况下重置对话。它会为新的对话创建一个新线程。

Note: 当前 Azure AI Agent API 不支持删除消息，因此改为创建新线程。

参数:: cancellation_token (CancellationToken) -- 用于取消处理的令牌

async on_upload_for_code_interpreter(file_paths: str | Iterable[str], cancellation_token: CancellationToken | None = None, sleep_interval: float = 0.5) → None[源代码]#

上传文件以供代码解释器工具使用。

此方法为代理的代码解释器工具上传文件，并更新线程的工具资源以包含这些文件。

参数:

file_paths (str | Iterable[str]) -- 要上传的文件路径(单个或多个)
cancellation_token (Optional[CancellationToken]) -- 用于取消处理的令牌
sleep_interval (float) -- 轮询文件状态之间的休眠时间

抛出:

ValueError -- 如果文件上传失败或代理不具备代码解释器能力

async on_upload_for_file_search(file_paths: str | Iterable[str], cancellation_token: CancellationToken, vector_store_name: str | None = None, data_sources: List[VectorStoreDataSource] | None = None, expires_after: VectorStoreExpirationPolicy | None = None, chunking_strategy: VectorStoreChunkingStrategyRequest | None = None, vector_store_metadata: Dict[str, str] | None = None, vector_store_polling_sleep_interval: float = 1) → None[源代码]#

上传文件以供文件搜索工具使用。

此方法处理文件搜索功能相关的文件上传，必要时创建向量存储，并更新代理配置以使用该向量存储。

参数:

file_paths (str | Iterable[str]) -- 要上传的文件路径(单个或多个)
cancellation_token (CancellationToken) -- 用于取消处理的令牌
vector_store_name (Optional[str]) -- 新建向量存储时指定的名称
data_sources (Optional[List[models.VectorStoreDataSource]]) -- 向量存储的附加数据源
expires_after (Optional[models.VectorStoreExpirationPolicy]) -- 向量存储内容的过期策略
chunking_strategy (Optional[models.VectorStoreChunkingStrategyRequest]) -- 文件内容分块策略
vector_store_metadata (Optional[Dict[str, str]]) -- 向量存储的附加元数据
vector_store_polling_sleep_interval (float) -- 轮询向量存储状态之间的休眠时间

抛出:

ValueError -- 如果此代理未启用文件搜索功能或文件上传失败

property produced_message_types: Sequence[type[Annotated[TextMessage | MultiModalMessage | StopMessage | ToolCallSummaryMessage | HandoffMessage, FieldInfo(annotation=NoneType, required=True, discriminator='type')]]]#: 助手代理产生的消息类型

async save_state() → Mapping[str, Any][源代码]#

保存代理的当前状态以便将来恢复。

此方法序列化代理的状态，包括代理ID、线程ID、消息ID以及相关资源如向量存储和上传的文件。

Returns:: Mapping[str, Any] -- 包含序列化状态数据的字典

property thread_id: str#

property tools: List[ToolDefinition]#

获取代理可用的工具列表。

Returns:: List[models.ToolDefinition] -- 工具定义列表