介绍
# Azure AI Agents Python SDK
使用 `azure-ai-agents` SDK 构建托管于 Azure AI Foundry 的代理。
## 安装
```bash pip install azure-ai-agents azure-identity # Or with azure-ai-projects for additional features pip install azure-ai-projects azure-identity ```
## 环境变量
```bash PROJECT_ENDPOINT="https://<resource>.services.ai.azure.com/api/projects/<project>" MODEL_DEPLOYMENT_NAME="gpt-4o-mini" ```
## 身份验证
```python from azure.identity import DefaultAzureCredential from azure.ai.agents import AgentsClient
credential = DefaultAzureCredential() client = AgentsClient( endpoint=os.environ["PROJECT_ENDPOINT"], credential=credential, ) ```
## 核心工作流
基本的代理生命周期：**创建代理 → 创建线程 → 创建消息 → 创建运行 → 获取响应**
### 最小示例
```python import os from azure.identity import DefaultAzureCredential from azure.ai.agents import AgentsClient
client = AgentsClient( endpoint=os.environ["PROJECT_ENDPOINT"], credential=DefaultAzureCredential(), )
# 1. Create agent agent = client.create_agent( model=os.environ["MODEL_DEPLOYMENT_NAME"], name="my-agent", instructions="You are a helpful assistant.", )
# 2. Create thread thread = client.threads.create()
# 3. Add message client.messages.create( thread_id=thread.id, role="user", content="Hello!", )
# 4. Create and process run run = client.runs.create_and_process(thread_id=thread.id, agent_id=agent.id)
# 5. Get response 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)
# Cleanup client.delete_agent(agent.id) ```
## 工具概览
| 工具 | 类 | 用例 | |------|-------|----------| | 代码解释器 | `CodeInterpreterTool` | 执行 Python，生成文件 | | 文件搜索 | `FileSearchTool` | 对上传的文档进行 RAG | | Bing 接地 | `BingGroundingTool` | Web 搜索 | | Azure AI Search | `AzureAISearchTool` | 搜索您的索引 | | 函数调用 | `FunctionTool` | 调用您的 Python 函数 | | OpenAPI | `OpenApiTool` | 调用 REST API | | MCP | `McpTool` | 模型上下文协议服务器 |
有关详细模式，请参阅 [references/tools.md](references/tools.md)。
## 添加工具
```python from azure.ai.agents import CodeInterpreterTool, FileSearchTool
agent = client.create_agent( model=os.environ["MODEL_DEPLOYMENT_NAME"], name="tool-agent", instructions="You can execute code and search files.", tools=[CodeInterpreterTool()], tool_resources={"code_interpreter": {"file_ids": [file.id]}}, ) ```
## 函数调用
```python from azure.ai.agents import FunctionTool, ToolSet
def get_weather(location: str) -> str: """Get weather for a location.""" return f"Weather in {location}: 72F, sunny"
functions = FunctionTool(functions=[get_weather]) toolset = ToolSet() toolset.add(functions)
agent = client.create_agent( model=os.environ["MODEL_DEPLOYMENT_NAME"], name="function-agent", instructions="Help with weather queries.", toolset=toolset, )
# Process run - toolset auto-executes functions run = client.runs.create_and_process( thread_id=thread.id, agent_id=agent.id, toolset=toolset, # Pass toolset for auto-execution ) ```
## 流式传输
```python 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"Error: {data}")
with client.runs.stream( thread_id=thread.id, agent_id=agent.id, event_handler=MyHandler(), ) as stream: stream.until_done() ```
有关高级模式，请参阅 [references/streaming.md](references/streaming.md)。
## 文件操作
### 上传文件
```python file = client.files.upload_and_poll( file_path="data.csv", purpose="assistants", ) ```
### 创建向量存储
```python 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]}}, ) ```
## 异步客户端
```python from azure.ai.agents.aio import AgentsClient
async with AgentsClient( endpoint=os.environ["PROJECT_ENDPOINT"], credential=DefaultAzureCredential(), ) as client: agent = await client.create_agent(...) # ... async operations ```
有关异步模式，请参阅 [references/async-patterns.md](references/async-patterns.md)。
## 响应格式
### JSON 模式
```python agent = client.create_agent( model=os.environ["MODEL_DEPLOYMENT_NAME"], response_format={"type": "json_object"}, ) ```
### JSON 架构
```python 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"], }, }, }, ) ```
## 线程管理
### 继续对话
```python # Save thread_id for later thread_id = thread.id
# Resume later client.messages.create( thread_id=thread_id, role="user", content="Follow-up question", ) run = client.runs.create_and_process(thread_id=thread_id, agent_id=agent.id) ```
### 列出消息
```python 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`**，**实时 UX 使用流式传输** 4. **将工具集传递给运行**以自动执行函数 5. **轮询操作**对长时间运行的操作使用 `*_and_poll` 方法
## 参考文件
- [references/tools.md](references/tools.md)：所有工具类型及详细示例 - [references/streaming.md](references/streaming.md)：事件处理程序和流式传输模式 - [references/async-patterns.md](references/async-patterns.md)：异步客户端使用