名称： azure-ai-agents-py
描述： 使用 Azure AI Agents Python SDK (azure-ai-agents) 构建 AI 智能体。适用于在 Azure AI Foundry 上创建搭载工具（文件搜索、代码解释器、必应联网搜索、Azure AI 搜索、函数调用、OpenAPI、MCP）的智能体、管理会话与消息、实现流式响应或操作向量存储。此为底层 SDK，如需更高级的抽象，请改用 agent-framework 技能。
package: azure-ai-agents

Azure AI Agents Python SDK

使用 azure-ai-agents SDK 构建托管在 Azure AI Foundry 上的智能体。

安装

pip install azure-ai-agents azure-identity
# 或安装 azure-ai-projects 以获得更多功能
pip install azure-ai-projects azure-identity

环境变量

PROJECT_ENDPOINT="https://<resource>.services.ai.azure.com/api/projects/<project>"
MODEL_DEPLOYMENT_NAME="gpt-4o-mini"

身份验证

from azure.identity import DefaultAzureCredential
from azure.ai.agents import AgentsClient

credential = DefaultAzureCredential()
client = AgentsClient(
    endpoint=os.environ["PROJECT_ENDPOINT"],
    credential=credential,
)

核心工作流

智能体的基本生命周期：创建智能体 → 创建会话 → 创建消息 → 创建运行 → 获取响应

最小示例

import os
from azure.identity import DefaultAzureCredential
from azure.ai.agents import AgentsClient

client = AgentsClient(
    endpoint=os.environ["PROJECT_ENDPOINT"],
    credential=DefaultAzureCredential(),
)

# 1. 创建智能体
agent = client.create_agent(
    model=os.environ["MODEL_DEPLOYMENT_NAME"],
    name="my-agent",
    instructions="你是一个乐于助人的助手。",
)

# 2. 创建会话
thread = client.threads.create()

# 3. 添加消息
client.messages.create(
    thread_id=thread.id,
    role="user",
    content="你好！",
)

# 4. 创建并处理运行
run = client.runs.create_and_process(thread_id=thread.id, agent_id=agent.id)

# 5. 获取响应
if run.status == "completed":
    messages = client.messages.list(thread_id=thread.id)
    for msg in messages:
        if msg.role == "assistant":
            print(msg.content[0].text.value)

# 清理
client.delete_agent(agent.id)

工具概览

详细用法请参阅 references/tools.md。

添加工具

from azure.ai.agents import CodeInterpreterTool, FileSearchTool

agent = client.create_agent(
    model=os.environ["MODEL_DEPLOYMENT_NAME"],
    name="tool-agent",
    instructions="你可以执行代码并搜索文件。",
    tools=[CodeInterpreterTool()],
    tool_resources={"code_interpreter": {"file_ids": [file.id]}},
)

函数调用

from azure.ai.agents import FunctionTool, ToolSet

def get_weather(location: str) -> str:
    """获取指定地点的天气。"""
    return f"{location} 的天气：72华氏度，晴朗"

functions = FunctionTool(functions=[get_weather])
toolset = ToolSet()
toolset.add(functions)

agent = client.create_agent(
    model=os.environ["MODEL_DEPLOYMENT_NAME"],
    name="function-agent",
    instructions="帮助处理天气查询。",
    toolset=toolset,
)

# 处理运行 - toolset 自动执行函数
run = client.runs.create_and_process(
    thread_id=thread.id,
    agent_id=agent.id,
    toolset=toolset,  # 传递 toolset 以实现自动执行
)

流式传输

from azure.ai.agents import AgentEventHandler

class MyHandler(AgentEventHandler):
    def on_message_delta(self, delta):
        if delta.text:
            print(delta.text.value, end="", flush=True)

    def on_error(self, data):
        print(f"错误：{data}")

with client.runs.stream(
    thread_id=thread.id,
    agent_id=agent.id,
    event_handler=MyHandler(),
) as stream:
    stream.until_done()

高级模式请参阅 references/streaming.md。

文件操作

上传文件

file = client.files.upload_and_poll(
    file_path="data.csv",
    purpose="assistants",
)

创建向量存储

vector_store = client.vector_stores.create_and_poll(
    file_ids=[file.id],
    name="my-store",
)

agent = client.create_agent(
    model=os.environ["MODEL_DEPLOYMENT_NAME"],
    tools=[FileSearchTool()],
    tool_resources={"file_search": {"vector_store_ids": [vector_store.id]}},
)

异步客户端

from azure.ai.agents.aio import AgentsClient

async with AgentsClient(
    endpoint=os.environ["PROJECT_ENDPOINT"],
    credential=DefaultAzureCredential(),
) as client:
    agent = await client.create_agent(...)
    # ... 异步操作

异步模式请参阅 references/async-patterns.md。

响应格式

JSON 模式

agent = client.create_agent(
    model=os.environ["MODEL_DEPLOYMENT_NAME"],
    response_format={"type": "json_object"},
)

JSON 架构

agent = client.create_agent(
    model=os.environ["MODEL_DEPLOYMENT_NAME"],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "weather_response",
            "schema": {
                "type": "object",
                "properties": {
                    "temperature": {"type": "number"},
                    "conditions": {"type": "string"},
                },
                "required": ["temperature", "conditions"],
            },
        },
    },
)

会话管理

继续对话

# 保存 thread_id 供后续使用
thread_id = thread.id

# 稍后恢复
client.messages.create(
    thread_id=thread_id,
    role="user",
    content="后续问题",
)
run = client.runs.create_and_process(thread_id=thread_id, agent_id=agent.id)

列出消息

messages = client.messages.list(thread_id=thread.id, order="asc")
for msg in messages:
    role = msg.role
    content = msg.content[0].text.value
    print(f"{role}: {content}")

最佳实践

1. 使用上下文管理器 处理异步客户端
2. 完成后清理智能体：client.delete_agent(agent.id)
3. 简单场景使用 create_and_process，实时用户体验使用流式传输
4. 将 toolset 传递给运行 以实现自动函数执行
5. 轮询操作 使用 *_and_poll 方法处理长时间运行的操作

参考文件

• references/tools.md：所有工具类型及详细示例
• references/streaming.md：事件处理器与流式传输模式
• references/async-patterns.md：异步客户端用法