Azure AI Agent 框架 (Python)

Agent Framework Azure 托管智能体

使用 Microsoft Agent Framework Python SDK 在 Azure AI Foundry 上构建持久化智能体。

架构

用户查询 → AzureAIAgentsProvider → Azure AI Agent 服务 (持久化)
                    ↓
              Agent.run() / Agent.run_stream()
                    ↓
              工具：函数 | 托管工具 (代码/搜索/网页) | MCP
                    ↓
              AgentThread (对话持久化)

安装

# 完整框架 (推荐)
pip install agent-framework --pre
或仅安装 Azure 特定包
pip install agent-framework-azure-ai --pre

环境变量

export AZURE_AI_PROJECT_ENDPOINT="https://<project>.services.ai.azure.com/api/projects/<project-id>"
export AZURE_AI_MODEL_DEPLOYMENT_NAME="gpt-4o-mini"
export BING_CONNECTION_ID="your-bing-connection-id"  # 用于网页搜索

身份验证

from azure.identity.aio import AzureCliCredential, DefaultAzureCredential
开发环境
credential = AzureCliCredential()
生产环境
credential = DefaultAzureCredential()

核心工作流

基础智能体

import asyncio
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
async def main():
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentsProvider(credential=credential) as provider,
    ):
        agent = await provider.create_agent(
            name="MyAgent",
            instructions="你是一个得力的助手。",
        )
        
        result = await agent.run("你好！")
        print(result.text)
asyncio.run(main())

带有函数工具的智能体

from typing import Annotated
from pydantic import Field
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
def get_weather(
    location: Annotated[str, Field(description="需要查询天气的城市名称")],
) -> str:
    """获取指定地点的当前天气。"""
    return f"{location} 的天气：72°F，晴朗"
def get_current_time() -> str:
    """获取当前 UTC 时间。"""
    from datetime import datetime, timezone
    return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
async def main():
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentsProvider(credential=credential) as provider,
    ):
        agent = await provider.create_agent(
            name="WeatherAgent",
            instructions="你负责处理天气和时间相关的查询。",
            tools=[get_weather, get_current_time],  # 直接传递函数
        )
        
        result = await agent.run("西雅图的天气怎么样？")
        print(result.text)

带有托管工具的智能体

from agent_framework import (
    HostedCodeInterpreterTool,
    HostedFileSearchTool,
    HostedWebSearchTool,
)
from agent_framework.azure import AzureAIAgentsProvider
from azure.identity.aio import AzureCliCredential
async def main():
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentsProvider(credential=credential) as provider,
    ):
        agent = await provider.create_agent(
            name="MultiToolAgent",
            instructions="你可以执行",

### 流式响应

### 对话线程

async def main():
async with (
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="ChatAgent",
instructions="You are a helpful assistant.",
tools=[get_weather],
)

# 创建线程以持久化对话
thread = agent.get_new_thread()

# 第一轮对话
result1 = await agent.run("What's the weather in Seattle?", thread=thread)
print(f"Agent: {result1.text}")

# 第二轮对话 - 维持上下文
result2 = await agent.run("What about Portland?", thread=thread)
print(f"Agent: {result2.text}")

# 保存线程 ID 以便后续恢复
print(f"Conversation ID: {thread.conversation_id}")

### 结构化输出

class WeatherResponse(BaseModel):
model_config = ConfigDict(extra="forbid")

location: str
temperature: float
unit: str
conditions: str

async def main():
async with (
AzureCliCredential() as credential,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="StructuredAgent",
instructions="Provide weather information in structured format.",
response_format=WeatherResponse,
)

result = await agent.run("Weather in Seattle?")
weather = WeatherResponse.model_validate_json(result.text)
print(f"{weather.location}: {weather.temperature}°{weather.unit}")

## Provider 方法

| 方法 | 描述 |
|--------|-------------|
| create_agent() | 在 Azure AI 服务上创建新 Agent |
| get_agent(agent_id) | 通过 ID 获取现有 Agent |
| as_agent(sdk_agent) | 封装 SDK Agent 对象（无 HTTP 调用） |
托管工具快速参考
| 工具 | 导入路径 | 用途 |
|------|--------|---------|
| HostedCodeInterpreterTool | from agent_framework import HostedCodeInterpreterTool | 执行 Python 代码 |
| HostedFileSearchTool | from agent_framework import HostedFileSearchTool | 搜索向量存储 |
| HostedWebSearchTool | from agent_framework import HostedWebSearchTool | Bing 网页搜索 |
| HostedMCPTool | from agent_framework import H |
| HostedMCPTool | 服务托管 MCP |
| MCPStreamableHTTPTool | from agent_framework import MCPStreamableHTTPTool | 客户端托管 MCP |
完整示例

def get_weather(
location: Annotated[str, Field(description="City name")],
) -> str:
"""Get weather for a location."""
return f"Weather in {location}: 72°F, sunny"

class AnalysisResult(BaseModel):
summary: str
key_findings: list[str]
confidence: float

async def main():
async with (
AzureCliCredential() as credential,
MCPStreamableHTTPTool(
name="Docs MCP",
url="https://learn.microsoft.com/api/mcp",
) as mcp_tool,
AzureAIAgentsProvider(credential=credential) as provider,
):
agent = await provider.create_agent(
name="ResearchAssistant",
instructions="You are a research assistant with multiple capabilities.",
tools=[
get_weather,
HostedCodeInterpreterTool(),
HostedWebSearchTool(name="Bing"),
mcp_tool,
],
)

thread = agent.get_new_thread()

# 非流式输出
result = await agent.run(
"Search for Python best practices and summarize",
thread=thread,
)
print(f"Response: {result.text}")

# 流式输出
print("\nStreaming: ", end="")
async for chunk in agent.run_stream("Continue with examples", thread=thread):
if chunk.text:
print(chunk.text, end="", flush=True)
print()

# 结构化输出
result = await agent.run(
"Analyze findings",
thread=thread,
response_format=AnalysisResult,
)
analysis = AnalysisResult.model_validate_json(result.text)
print(f"\nConfidence: {analysis.confidence}")

if __name__ == "__main__":
asyncio.run(main())
``

• 始终使用异步上下文管理器：async with provider:

• 将函数直接传递给 tools= 参数（会自动转换为 AIFunction）

• 函数参数使用 Annotated[type, Field(description=...)]

• 多轮对话使用 get_new_thread()

• 服务托管 MCP 优先使用 HostedMCPTool，客户端托管 MCP 优先使用 MCPStreamableHTTPTool`

参考文件

• references/tools.md: 详细的托管工具模式

• references/mcp.md: MCP 集成（托管 + 本地）

• references/threads.md: 线程与对话管理

• references/advanced.md: OpenAPI、引用、结构化输出

适用场景

局限性

• 仅在任务明确符合上述范围时使用此技能。

• 不要将输出视为环境特定验证、测试或专家评审的替代方案。

• 如果缺少必要的输入、权限、安全边界或成功标准，请停止并请求澄清。