Microsoft Agent Framework 实战入门:从零构建多 Agent 协作工作流(Python 完整示例)
Microsoft Agent Framework 已达 Release Candidate 阶段(GA 预计 2026 年 3 月底),本文基于 Python RC 版本,手把手演示从单 Agent 创建到多 Agent 协作编排的完整工程流程,含可运行代码。
一、为什么要关注 Microsoft Agent Framework
2026 年 2 月 19 日,微软正式宣布 Microsoft Agent Framework 达到 Release Candidate(RC)状态(据 Microsoft Foundry 官方博客),这是继 Semantic Kernel 和 AutoGen 之后的统一继任框架。GA(正式版)预计在 3 月底发布(据 GitHub Discussion #4262 中维护者回复)。
核心定位
- 统一接口:合并 Semantic Kernel + AutoGen 的能力,提供 .NET 和 Python 双语言一致的编程模型
- 稳定 API:RC 阶段意味着 API 表面已锁定,不会有重大破坏性变更
- 生态整合:原生支持 MCP(Model Context Protocol)、A2A(Agent-to-Agent)、AG-UI 三大协议标准
- 多提供商:支持 Microsoft Foundry、Azure OpenAI、OpenAI、Anthropic Claude、AWS Bedrock、Ollama 等
与 Semantic Kernel / AutoGen 的关系
| 维度 | Semantic Kernel | AutoGen | Agent Framework |
|---|---|---|---|
| 状态 | 已合并,进入维护模式 | 已合并,仅修复 Bug | 活跃开发,即将 GA |
| Agent 创建 | Plugin + Planner | ConversableAgent | 统一 BaseAgent API |
| 编排模式 | 有限 | 多 Agent 对话 | Graph 工作流 + 多种编排模式 |
| 协议支持 | 无 | 有限 | MCP + A2A + AG-UI |
据 Microsoft 官方迁移指南(据 learn.microsoft.com),建议所有 Semantic Kernel 和 AutoGen 项目尽早迁移到 Agent Framework。
二、环境搭建
2.1 安装依赖
# 创建虚拟环境
python -m venv agent-fw-env
source agent-fw-env/bin/activate # Linux/macOS
# agent-fw-env\Scripts\activate # Windows
# 安装 RC 版本(截至 2026 年 3 月,最新为 rc6)
pip install agent-framework-core==1.0.0rc6
pip install agent-framework-azure-ai==1.0.0rc6
# 如果使用 OpenAI 直连
pip install agent-framework-openai==1.0.0rc6
# 如果需要 MCP 支持
pip install agent-framework-mcp==1.0.0rc6
2.2 配置凭据
import os
# 方式一:Azure OpenAI
os.environ["AZURE_OPENAI_ENDPOINT"] = "https://your-resource.openai.azure.com/"
os.environ["AZURE_OPENAI_API_KEY"] = "your-api-key"
os.environ["AZURE_OPENAI_DEPLOYMENT"] = "gpt-4o" # 你的部署名称
# 方式二:OpenAI 直连
os.environ["OPENAI_API_KEY"] = "sk-your-key"
三、Step 1:创建第一个 Agent
3.1 最小可用示例
from agent_framework import Agent, AgentOptions
from agent_framework.azure_ai import AzureAIChatClient
# 1. 创建模型客户端
client = AzureAIChatClient(
endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
api_key=os.environ["AZURE_OPENAI_API_KEY"],
deployment=os.environ["AZURE_OPENAI_DEPLOYMENT"],
)
# 2. 创建 Agent
agent = Agent(
options=AgentOptions(
name="CodeReviewer",
instructions="""你是一个专业的代码审查助手。
针对用户提交的代码,给出:
1. 潜在 Bug 和安全风险
2. 性能优化建议
3. 代码风格改进建议
请用中文回复,使用 Markdown 格式。""",
chat_client=client,
)
)
# 3. 调用 Agent
async def main():
response = await agent.invoke(
"请审查这段代码:\n"
"def get_user(id):\n"
" return db.execute(f'SELECT * FROM users WHERE id={id}')\n"
)
print(response.content)
import asyncio
asyncio.run(main())
仅需约 15 行核心代码即可创建一个可用的 Agent,这是 Agent Framework 强调的"简洁优先"设计哲学。
四、Step 2:添加 Function Tool
Agent 的核心能力来自工具调用(Tool Calling)。Agent Framework 提供类型安全的 Tool 定义:
from agent_framework import Agent, AgentOptions, tool
from agent_framework.azure_ai import AzureAIChatClient
import subprocess
import json
# 定义工具函数(使用 @tool 装饰器)
@tool
def run_lint(code: str, language: str = "python") -> str:
"""对代码运行 linter 检查,返回检查结果。
Args:
code: 待检查的源代码
language: 编程语言,默认 python
"""
# 将代码写入临时文件并运行 ruff
import tempfile
with tempfile.NamedTemporaryFile(
mode="w", suffix=".py", delete=False
) as f:
f.write(code)
f.flush()
result = subprocess.run(
["ruff", "check", f.name, "--output-format=json"],
capture_output=True, text=True
)
try:
issues = json.loads(result.stdout)
return json.dumps(issues, indent=2, ensure_ascii=False)
except json.JSONDecodeError:
return result.stdout or "检查通过,未发现问题"
@tool
def search_docs(query: str) -> str:
"""搜索 Python 官方文档中的相关内容。
Args:
query: 搜索关键词
"""
# 实际项目中对接搜索 API,此处返回示例
return f"搜索 '{query}' 的结果:建议参考 PEP 8 编码规范..."
# 创建带工具的 Agent
client = AzureAIChatClient(
endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
api_key=os.environ["AZURE_OPENAI_API_KEY"],
deployment=os.environ["AZURE_OPENAI_DEPLOYMENT"],
)
reviewer = Agent(
options=AgentOptions(
name="SmartReviewer",
instructions="你是一个代码审查助手。在审查前先用 run_lint 工具检查代码质量。",
chat_client=client,
tools=[run_lint, search_docs], # 注册工具
)
)
Agent 在推理时会根据 instructions 和用户输入自动决定是否调用工具,工具的参数类型由 Python 类型注解自动推断。
五、Step 3:多 Agent 编排(Graph 工作流)
这是 Agent Framework 最强大的特性之一。支持顺序、并发、Handoff、群聊等多种编排模式。
5.1 架构概览
5.2 实现多 Agent 协作
from agent_framework import Agent, AgentOptions
from agent_framework.workflows import (
GraphWorkflow,
SequentialWorkflow,
HandoffWorkflow,
)
from agent_framework.azure_ai import AzureAIChatClient
client = AzureAIChatClient(
endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
api_key=os.environ["AZURE_OPENAI_API_KEY"],
deployment=os.environ["AZURE_OPENAI_DEPLOYMENT"],
)
# 定义专业 Agent
planner = Agent(
options=AgentOptions(
name="Planner",
instructions="分析用户需求,拆解为具体的编码任务和调研任务。"
"输出 JSON 格式的任务列表。",
chat_client=client,
)
)
coder = Agent(
options=AgentOptions(
name="Coder",
instructions="根据任务描述编写高质量的 Python 代码。"
"包含类型注解、文档字符串和错误处理。",
chat_client=client,
)
)
researcher = Agent(
options=AgentOptions(
name="Researcher",
instructions="调研相关的技术方案、最佳实践和潜在风险。"
"输出结构化的调研报告。",
chat_client=client,
)
)
reviewer = Agent(
options=AgentOptions(
name="Reviewer",
instructions="综合代码实现和调研结果,给出最终的审查意见。"
"确保代码质量、安全性和可维护性。",
chat_client=client,
)
)
# 构建 Graph 工作流
workflow = GraphWorkflow()
# 添加节点
workflow.add_node("plan", planner)
workflow.add_node("code", coder)
workflow.add_node("research", researcher)
workflow.add_node("review", reviewer)
# 定义边(执行顺序)
workflow.add_edge("plan", "code") # plan 完成后触发 code
workflow.add_edge("plan", "research") # plan 完成后并行触发 research
workflow.add_edge("code", "review") # code 和 research 都完成后触发 review
workflow.add_edge("research", "review")
# 设置入口和出口
workflow.set_entry_point("plan")
workflow.set_finish_point("review")
# 执行工作流
async def run_workflow():
result = await workflow.invoke(
"帮我实现一个 FastAPI 中间件,用于限制 API 调用频率,"
"支持按 IP 和按用户两种维度的限流策略。"
)
print(result.content)
import asyncio
asyncio.run(run_workflow())
5.3 Handoff 模式(Agent 间接力)
当需要 Agent 根据对话上下文动态决定下一步由哪个 Agent 处理时,使用 Handoff:
from agent_framework.workflows import HandoffWorkflow
# Handoff 工作流:Agent 可以动态决定将任务交给谁
handoff_workflow = HandoffWorkflow(
agents=[planner, coder, researcher, reviewer],
default_agent=planner, # 默认入口 Agent
max_turns=10, # 防止无限循环
)
async def run_handoff():
result = await handoff_workflow.invoke(
"请帮我设计并实现一个 Redis 缓存层"
)
print(result.content)
六、Step 4:接入 MCP 工具服务
Agent Framework 原生支持 MCP(Model Context Protocol),可以让 Agent 直接调用外部 MCP 服务暴露的工具:
from agent_framework.mcp import McpToolProvider
# 连接到 MCP 服务器(例如本地文件系统 MCP 服务)
mcp_provider = McpToolProvider(
server_url="http://localhost:3001",
# 或使用 stdio 模式
# command=["npx", "-y", "@modelcontextprotocol/server-filesystem", "/tmp"]
)
# 获取 MCP 服务暴露的工具列表
mcp_tools = await mcp_provider.list_tools()
# 创建具备 MCP 工具能力的 Agent
file_agent = Agent(
options=AgentOptions(
name="FileAgent",
instructions="你可以通过 MCP 工具操作文件系统。",
chat_client=client,
tools=mcp_tools, # MCP 工具无缝集成
)
)
七、从 Semantic Kernel 迁移要点
据微软官方迁移指南(据 learn.microsoft.com),核心变更包括:
- 凭据处理:从
KernelBuilder迁移到ChatClient模式 - Plugin → Tool:
KernelFunction替换为@tool装饰器 - Planner → Workflow:从内置 Planner 迁移到显式 Graph/Handoff 工作流
- 会话管理:新增 Checkpointing 支持,可在工作流中保存和恢复状态
# Semantic Kernel(旧)
kernel = Kernel()
kernel.add_plugin(MyPlugin(), "my_plugin")
result = await kernel.invoke("my_plugin", "my_function", input="...")
# Agent Framework(新)
agent = Agent(
options=AgentOptions(
name="MyAgent",
chat_client=client,
tools=[my_tool_function], # 直接注册函数
)
)
result = await agent.invoke("...")
八、小结与下一步
Microsoft Agent Framework 通过统一 Semantic Kernel 和 AutoGen 的能力,提供了一套从单 Agent 到多 Agent 编排的完整开发栈。对于已有 Semantic Kernel/AutoGen 项目的团队,现在是迁移的最佳窗口期(RC API 已稳定,GA 即将到来)。
下一步学习路径:
- 官方文档快速上手:https://learn.microsoft.com/en-us/agent-framework/overview/
- GitHub 仓库关注 GA 发布:https://github.com/microsoft/agent-framework
- 尝试 Agent Skills 扩展(3 月 18 日发布):给 Agent 动态加载领域专业知识
- 探索 A2A 协议实现跨服务 Agent 协作
Agent Framework 把构建 AI Agent 的门槛从「框架拼装」降到了「几行代码」,GA 就在眼前,现在上手正是时候。 你的团队目前在用 Semantic Kernel、AutoGen 还是 LangGraph?迁移过程中踩过什么坑?欢迎评论区交流经验!
⭐ 建议收藏本文,GA 发布后对照实操
💬 评论区聊聊你的 Agent 开发技术栈选择
🔔 关注我,后续会更新 Agent Framework GA 正式版实测与迁移踩坑记录!
更多推荐


所有评论(0)