开源智能体框架AutoGen深度解析：多Agent协作的工程实践
AutoGen：对话驱动的多Agent框架 微软研究院的AutoGen开创了"对话即协作"的Agent范式。与LangGraph的图驱动不同，AutoGen将多Agent协作建模为一组Agent之间的对话，每个Agent有独立的角色和能力。 核心架构 Agent类型体系 from autogen import ( ConversableAgent, AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager ) # AssistantAgent: AI助手，有系统消息定义角色 researcher = AssistantAgent( name="Researcher", system_message="""你是一位AI研究分析师。 职责： 1. 分析用户需求 2. 搜索和整理相关信息 3. 提供结构化的分析报告 约束： - 基于事实，不编造 - 注明信息来源 - 区分事实和推测 """, llm_config={"model": "gpt-4o", "temperature": 0.3}, tools=[web_search, knowledge_base_search] ) # UserProxyAgent: 用户代理，可以执行代码 user_proxy = UserProxyAgent( name="User", human_input_mode="NEVER", # 不等待人类输入 code_execution_config={ "work_dir": "workspace", "use_docker": True, # 安全执行环境 "timeout": 60 } ) 对话管理 class ConversationManager: def __init__(self): self.conversations = {} self.termination_conditions = [] def add_termination(self, condition): """添加对话终止条件""" self.termination_conditions.append(condition) def check_termination(self, messages): for condition in self.termination_conditions: if condition.check(messages): return True return False # 常见终止条件 class MaxRoundsTermination: def __init__(self, max_rounds=10): self.max_rounds = max_rounds def check(self, messages): return len(messages) >= self.max_rounds class KeywordTermination: def __init__(self, keywords): self.keywords = keywords def check(self, messages): if messages: return any(kw in messages[-1]["content"] for kw in self.keywords) return False 多Agent协作模式 模式1：顺序对话 def sequential_conversation(task): """Agent按顺序处理任务""" # Agent 1: 分析需求 analysis = analyst.generate(f"分析以下任务：{task}") # Agent 2: 编写代码 code = coder.generate(f"基于以下分析编写代码：{analysis}") # Agent 3: 审查代码 review = reviewer.generate(f"审查以下代码：{code}") # Agent 4: 优化代码 if "问题" in review: final_code = coder.generate(f"根据审查意见优化代码：{review}") else: final_code = code return final_code 模式2：群聊协作 def group_chat_collaboration(task): """多Agent群聊协作""" agents = [ UserProxyAgent(name="User", human_input_mode="NEVER"), AssistantAgent(name="Planner", system_message="负责制定计划"), AssistantAgent(name="Coder", system_message="负责编写代码"), AssistantAgent(name="Tester", system_message="负责测试"), AssistantAgent(name="Reviewer", system_message="负责审查") ] group_chat = GroupChat( agents=agents, messages=[], max_round=20, speaker_selection_method="auto" # 自动选择下一个发言者 ) manager = GroupChatManager( groupchat=group_chat, llm_config={"model": "gpt-4o"} ) agents[0].initiate_chat(manager, message=task) 模式3：嵌套对话 def nested_conversation(task): """Agent内部发起子对话""" # 主Agent处理任务 main_agent = AssistantAgent( name="Main", system_message="你是项目经理，可以委托子任务给其他Agent" ) # 当主Agent遇到需要深入研究的子问题时 # 它可以发起一个子对话 def research_subtask(subtask): researcher = AssistantAgent( name="Researcher", system_message="你是研究员，擅长信息检索" ) result = researcher.generate(f"研究：{subtask}") return result # 主Agent可以在处理过程中调用子对话 main_agent.register_function( function_map={"research": research_subtask} ) 代码执行环境 安全执行 class SafeCodeExecutor: def __init__(self): self.docker_config = { "image": "python:3.11-slim", "timeout": 60, "memory_limit": "512m", "cpu_limit": 1.0, "network": "none", # 禁止网络访问 } self.allowed_packages = [ "numpy", "pandas", "matplotlib", "scipy", "scikit-learn" ] def execute(self, code): # 1. 静态检查 issues = self._static_check(code) if issues: return {"error": "代码检查未通过", "issues": issues} # 2. Docker执行 result = self._docker_exec(code) return result def _static_check(self, code): """静态安全检查""" forbidden = [ "import os", "import subprocess", "import socket", "open(", "__import__", "eval(", "exec(" ] issues = [] for pattern in forbidden: if pattern in code: issues.append(f"禁止使用: {pattern}") return issues 代码执行反馈循环 def code_feedback_loop(agent, task, max_attempts=3): """代码编写-执行-修正的反馈循环""" for attempt in range(max_attempts): # Agent生成代码 code = agent.generate(f"任务：{task}\n尝试：{attempt+1}") # 执行代码 result = executor.execute(code) if result["success"]: return code, result["output"] # 反馈错误，让Agent修正 feedback = f""" 代码执行失败： 错误信息：{result['error']} 请修正代码。 """ task = task + "\n\n" + feedback return None, "达到最大尝试次数" 高级特性 Agent可序列化 def save_agent_state(agent, path): """保存Agent状态，支持恢复""" state = { "name": agent.name, "system_message": agent.system_message, "llm_config": agent.llm_config, "chat_history": agent.chat_messages, "registered_tools": list(agent.tools.keys()) } with open(path, 'w') as f: json.dump(state, f) def load_agent_state(path): """从文件恢复Agent""" with open(path) as f: state = json.load(f) agent = AssistantAgent( name=state["name"], system_message=state["system_message"], llm_config=state["llm_config"] ) agent.chat_messages = state["chat_history"] return agent 自定义Agent行为 class CustomAgent(ConversableAgent): def __init__(self, name, **kwargs): super().__init__(name, **kwargs) self.register_hook("process_message_before_send", self._preprocess) self.register_hook("process_message_after_receive", self._postprocess) def _preprocess(self, message): """发送前预处理""" # 添加时间戳 message["content"] = f"[{datetime.now()}] {message['content']}" return message def _postprocess(self, message): """接收后处理""" # 记录消息日志 self._log(message) return message def _log(self, message): """消息日志""" with open("agent_log.jsonl", 'a') as f: f.write(json.dumps({ "timestamp": datetime.now().isoformat(), "sender": message.get("from"), "content": message["content"][:200] }) + "\n") 实际案例：数据分析Agent def build_data_analysis_agent(): """构建数据分析Agent系统""" # 数据科学家Agent data_scientist = AssistantAgent( name="DataScientist", system_message="""你是数据科学家，负责： 1. 理解用户的数据分析需求 2. 编写Python代码进行数据分析 3. 解释分析结果 使用pandas, numpy, matplotlib进行数据分析。 确保代码包含异常处理和数据验证。 """, llm_config={"model": "gpt-4o"} ) # 代码执行Agent code_runner = UserProxyAgent( name="CodeRunner", human_input_mode="NEVER", code_execution_config={ "work_dir": "analysis_workspace", "use_docker": "python:3.11-slim" }, system_message="你负责执行代码并返回结果。不生成代码，只执行。" ) # 启动分析 code_runner.initiate_chat( data_scientist, message="""分析销售数据： 1. 读取 /data/sales.csv 2. 按月统计销售趋势 3. 找出Top 10产品 4. 生成可视化图表 5. 输出分析报告 """ ) AutoGen vs 其他框架 维度 AutoGen LangGraph CrewAI 核心范式 对话驱动 图驱动 角色驱动 代码执行 内置Docker 需自定义 需自定义 多Agent 原生支持 需手动编排 支持 状态管理 对话历史 检查点 任务上下文 适合场景 研究探索 生产系统 快速原型 结语 AutoGen将多Agent协作还原为最自然的形式——对话。它的优势在于代码执行能力和灵活的对话管理。劣势在于控制精度不如图驱动框架。对于需要多专家协作探索的研究型任务，AutoGen是最佳选择。对于需要精确控制执行流程的生产系统，LangGraph更合适。选择框架的关键是匹配你的任务特性。 ...
