引言：为什么需要多智能体？ 单一 AI Agent 的能力是有限的——一个 Agent 难以同时扮演产品经理、开发者和测试工程师的角色。微软 AutoGen 框架的核心思想是：让多个专业化 Agent 协作完成复杂任务，就像一个高效的团队一样。
AutoGen 核心架构 框架层次 ┌─────────────────────────────────┐ │ Application Layer │ │ (User Interface / API) │ ├─────────────────────────────────┤ │ Orchestration Layer │ │ ┌───────────┐ ┌────────────┐ │ │ │ Group Chat│ │ Workflow │ │ │ │ Manager │ │ Engine │ │ │ └───────────┘ └────────────┘ │ ├─────────────────────────────────┤ │ Agent Layer │ │ AssistantAgent│UserProxyAgent│ │ MultimodalAgent│CustomAgent │ ├─────────────────────────────────┤ │ Communication Layer │ │ Message Queue │ Event Bus │ ├─────────────────────────────────┤ │ Model Layer │ │ OpenAI │ Azure │ Local Models │ └─────────────────────────────────┘ Agent 类型 Agent 类型 职责 典型场景 AssistantAgent 执行任务、生成内容 编码、分析、写作 UserProxyAgent 代理用户行为 执行代码、提供反馈 GroupChatManager 管理多 Agent 对话 团队协作场景 MultimodalAgent 处理图像等多模态输入 视觉分析任务 CustomAgent 用户自定义 Agent 特殊业务逻辑 快速上手 安装与基础配置 # 安装 AutoGen # pip install autogen-agentchat autogen-ext import asyncio from autogen_agentchat.agents import AssistantAgent from autogen_agentchat.teams import RoundRobinGroupChat from autogen_agentchat.conditions import TextMentionTermination from autogen_ext.models.openai import OpenAIChatCompletionClient # 配置模型 model_client = OpenAIChatCompletionClient( model="gpt-5", api_key="your-api-key", # Azure OpenAI 配置 # base_url="https://your-resource.openai.azure.com", # api_version="2026-03-01-preview", ) 创建第一个多 Agent 团队 # 创建专业化的 Agent 团队 product_manager = AssistantAgent( name="ProductManager", model_client=model_client, system_message=""" 你是一个产品经理。你的职责： 1. 理解用户需求 2. 将需求拆解为具体的功能点 3. 定义验收标准 不要写代码，专注于需求分析。 当需求分析完成时，说"需求已明确，请开发人员开始实现"。 """, ) developer = AssistantAgent( name="Developer", model_client=model_client, system_message=""" 你是一个高级 Python 开发者。你的职责： 1. 根据产品经理的需求编写代码 2. 确保代码质量（类型标注、错误处理） 3. 编写单元测试 写完代码后，说"代码已完成，请测试人员审查"。 """, # 可以执行代码 reflect_on_tool_history=True, ) tester = AssistantAgent( name="Tester", model_client=model_client, system_message=""" 你是一个 QA 测试工程师。你的职责： 1. 审查代码的逻辑正确性 2. 检查边界条件和错误处理 3. 提出改进建议 审查完成后，说"APPROVED"表示通过，或指出需要修改的问题。 """, ) # 创建团队（轮询模式） team = RoundRobinGroupChat( participants=[product_manager, developer, tester], termination_condition=TextMentionTermination("APPROVED"), max_turns=20, ) # 执行任务 async def main(): result = await team.run( task="实现一个函数，接收日期字符串，返回该日期是星期几，" "支持多种日期格式输入" ) print(result) asyncio.run(main()) 高级：工作流编排 自定义工作流 from autogen_agentchat.teams import SelectorGroupChat from autogen_agentchat.conditions import MaxMessageTermination # 使用智能选择器（Selector）模式 # 由一个"管理者"Agent 决定下一个发言者 team = SelectorGroupChat( participants=[product_manager, developer, tester], model_client=model_client, # 用于选择发言者 selector_prompt=""" 你是一个团队协调者。根据当前对话状态， 选择下一个应该发言的团队成员。 可选成员：{participants} 对话历史：{history} 返回下一个发言者的名字。 """, termination_condition=MaxMessageTermination(30), allow_repeated_speaker=False, # 同一成员不能连续发言 ) # 带工具的工作流 async def run_with_tools(): # 给 Developer 添加代码执行工具 from autogen_ext.tools.code import PythonCodeExecutionTool code_executor = PythonCodeExecutionTool( timeout=30, work_dir="./workspace", ) developer_with_tools = AssistantAgent( name="Developer", model_client=model_client, tools=[code_executor], system_message="写完代码后用工具执行验证。", ) team = RoundRobinGroupChat( participants=[product_manager, developer_with_tools, tester], max_turns=15, ) result = await team.run(task="实现并验证一个快速排序算法") return result DAG 工作流 from autogen_agentchat.teams import GraphWorkflow from autogen_agentchat.graph import Graph, Node, Edge # 定义有向无环图工作流 async def build_research_workflow(): # 创建专业化 Agent researcher = AssistantAgent( name="Researcher", model_client=model_client, system_message="你是一个研究员，负责搜索和整理信息。", ) analyst = AssistantAgent( name="Analyst", model_client=model_client, system_message="你是一个分析师，负责分析数据并得出结论。", ) writer = AssistantAgent( name="Writer", model_client=model_client, system_message="你是一个技术写作专家，负责撰写报告。", ) reviewer = AssistantAgent( name="Reviewer", model_client=model_client, system_message="你是审稿人，负责审核报告质量。", ) # 构建工作流图 graph = Graph() # 添加节点 graph.add_node(Node("research", researcher)) graph.add_node(Node("analyze", analyst)) graph.add_node(Node("write", writer)) graph.add_node(Node("review", reviewer)) # 添加边（定义执行顺序） graph.add_edge(Edge("research", "analyze")) graph.add_edge(Edge("analyze", "write")) graph.add_edge(Edge("write", "review")) # 审核不通过可以回退 graph.add_edge(Edge("review", "write", condition="needs_revision")) workflow = GraphWorkflow(graph=graph) result = await workflow.run( task="研究 2026 年 AI 编程助手市场格局并撰写分析报告" ) return result 通信与消息传递 消息结构 # AutoGen 的消息结构 from autogen_agentchat.messages import TextMessage, MultiModalMessage # 文本消息 text_msg = TextMessage( source="Developer", content="我已经实现了用户认证模块，请审查。", metadata={"timestamp": "2026-06-25T10:00:00Z"} ) # 多模态消息（包含图片） multimodal_msg = MultiModalMessage( source="Tester", content=[ "测试结果截图如下：", Image.from_file("test-results.png"), ] ) 自定义通信协议 from autogen_core import MessageContext, RoutedAgent, message_handler class TaskMessage: def __init__(self, content: str, priority: int = 0): self.content = content self.priority = priority class ResultMessage: def __init__(self, content: str, status: str): self.content = content self.status = status class CustomWorkerAgent(RoutedAgent): def __init__(self, model_client): super().__init__("Worker") self.model_client = model_client self._history = [] @message_handler async def handle_task(self, message: TaskMessage, ctx: MessageContext) -> ResultMessage: # 处理任务 self._history.append(message.content) response = await self.model_client.create( messages=[{"role": "user", "content": message.content}] ) return ResultMessage( content=response.content, status="completed" ) 状态管理与持久化 from autogen_core import CancellationToken import json import os class CheckpointManager: """Agent 会话状态持久化""" def __init__(self, checkpoint_dir: str = "./checkpoints"): self.checkpoint_dir = checkpoint_dir os.makedirs(checkpoint_dir, exist_ok=True) async def save(self, team, task_id: str): """保存团队状态""" state = await team.save_state() path = os.path.join(self.checkpoint_dir, f"{task_id}.json") with open(path, "w") as f: json.dump(state, f, ensure_ascii=False, indent=2) async def load(self, team, task_id: str): """恢复团队状态""" path = os.path.join(self.checkpoint_dir, f"{task_id}.json") if os.path.exists(path): with open(path) as f: state = json.load(f) await team.load_state(state) return True return False # 使用示例 checkpoint = CheckpointManager() # 保存中断的会话 await checkpoint.save(team, "task-001") # 恢复会话 await checkpoint.load(team, "task-001") result = await team.run(task="继续之前的任务") 性能优化 并行 Agent 执行 from autogen_agentchat.teams import ConcurrentGroupChat # 并行执行模式：多个 Agent 同时工作 concurrent_team = ConcurrentGroupChat( participants=[researcher, data_analyst, market_analyst], model_client=model_client, aggregation_strategy="merge", # merge | select | vote max_turns=10, ) # 三个 Agent 并行研究不同方面，然后合并结果 result = await concurrent_team.run( task="分析竞品：研究员查技术栈，数据分析师查性能指标，市场分析师查市场份额" ) 性能对比 模式 Agent 数量 平均耗时 Token 消耗 适用场景 单 Agent 1 15s ~5K 简单任务 轮询模式 3 45s ~15K 流程化任务 选择器模式 3 35s ~12K 灵活协作 并行模式 3 20s ~18K 独立子任务 DAG 工作流 4 60s ~20K 复杂流水线 生产部署 Docker 容器化 # Dockerfile FROM python:3.12-slim WORKDIR /app # 安装依赖 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制代码 COPY . . # 启动 Agent 服务 CMD ["python", "-m", "autogen_server", "--host", "0.0.0.0", "--port", "8080"] # requirements.txt autogen-agentchat>=0.5.0 autogen-ext[openai,azure]>=0.5.0 fastapi>=0.115.0 uvicorn>=0.30.0 API 服务封装 from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel app = FastAPI(title="AutoGen Agent Service") class TaskRequest(BaseModel): task: str team_config: dict = {} max_turns: int = 20 class TaskResponse(BaseModel): task_id: str status: str result: str | None = None @app.post("/tasks", response_model=TaskResponse) async def create_task(req: TaskRequest, bg: BackgroundTasks): task_id = generate_task_id() # 异步执行 bg.add_task(run_agent_team, task_id, req.task, req.team_config) return TaskResponse(task_id=task_id, status="running") @app.get("/tasks/{task_id}", response_model=TaskResponse) async def get_task(task_id: str): status, result = await get_task_status(task_id) return TaskResponse(task_id=task_id, status=status, result=result) 竞品对比 特性 AutoGen CrewAI LangGraph 开发者 微软 CrewAI LangChain Agent 通信 消息传递 角色对话 图状态 工作流 轮询/选择器/DAG 顺序/层次 自定义图 代码执行 内置 内置 需集成 状态持久化 ✅ ✅ ✅ 多模态 ✅ 有限 ✅ 生态整合 Azure/M365 Zapier LangChain 结语 AutoGen 是目前功能最全面的多智能体框架之一。它提供了从简单对话到复杂工作流的完整工具链，适合构建生产级的多 Agent 系统。其微软生态整合使其在企业场景中具有天然优势，但学习曲线相对陡峭。
...
