1. 项目概述从单兵作战到群体智能的跃迁最近在折腾一个挺有意思的东西我把它叫做“让AI自己协作”。听起来有点科幻但核心就是设计并实现一个多智能体系统或者更时髦一点叫它“Swarm”蜂群系统。这玩意儿不是简单地让几个ChatGPT对话机器人拉个群聊而是构建一个具备明确分工、能自主沟通、甚至能动态调整策略来完成复杂任务的AI团队。想想看我们平时让AI干点活无论是写代码、分析数据还是生成报告本质上还是“一对一”的指令-响应模式。我们得像项目经理一样把一个大任务拆解成无数个小步骤然后一步步喂给AI中间还得不断检查、纠偏、给新指令。效率低不说一旦任务稍微复杂点比如“帮我开发一个带用户管理和支付功能的小程序”光是把需求描述清楚就得费老大劲更别提过程中各种技术细节的决策了。多智能体系统要解决的就是这个“项目经理”的瓶颈。它的目标是创建一个数字世界的“特种部队”里面有指挥官Orchestrator、有侦察兵Researcher、有工程师Coder、有质检员Reviewer。你只需要下达一个最高指令比如“攻克那个山头完成某个项目”它们自己就会开会讨论战术、分配任务、同步进度、互相查漏补缺最终把结果呈报给你。这不仅仅是自动化这是赋予AI一种组织行为和协同智能。我之所以投入时间研究这个是因为在实际工作中越来越频繁地遇到那些“知道目标但路径极其复杂”的问题。单靠一个AI哪怕是最强大的模型其知识广度、思维深度和持续专注力都是有限的。而一个设计良好的多智能体系统能够通过角色 specialization专业化分工和 inter-agent communication智能体间通信实现“112”的群体智慧涌现。接下来我就把自己在设计和实现这套系统过程中的核心思路、关键设计、踩过的坑以及一些实用的心得完整地分享出来。2. 核心设计思路如何构建一个“活”起来的AI团队设计多智能体系统最忌讳的就是拍脑袋定几个角色然后就让它们开始对话。那结果大概率会变成一场混乱的、离题的、甚至无限循环的“废话大会”。要让AI团队高效协作必须像设计一个真正的组织架构和工作流程一样进行顶层设计。2.1 智能体角色定义不是越多越好而是越准越好角色的定义是整个系统的基石。我的经验是初期千万不要追求“大而全”的角色矩阵而是从最核心的职能开始。通常一个能处理通用任务的最小可行团队MVT包含以下四个角色任务规划与分解者Planner/Orchestrator这是团队的“大脑”。它的核心能力是理解最终目标并进行结构化分解。它不负责具体执行而是产出像“项目计划书”一样的东西一个任务列表Task List每个任务都有明确的描述、输入、期望输出和成功标准。它还需要决定哪个任务由哪个角色执行最合适。信息搜集与研究者Researcher这是团队的“眼睛和耳朵”。当任务需要外部知识、最新数据或特定领域信息时就由它出动。它需要擅长使用搜索工具、阅读文档、总结信息并能判断信息的可靠性和相关性。执行与构建者Executor/Builder这是团队的“双手”。通常是程序员、写作者、数据分析师等具体技能的化身。它接收明确的任务指令和所需上下文然后产出具体的成果如一段代码、一份文档、一个数据分析图表。评审与质检者Reviewer/Evaluator这是团队的“守门员”。它的工作是挑剔和找茬。检查执行者的产出是否符合要求、有无逻辑错误、风格是否一致、安全性如何。它不能只说“不好”必须给出具体的修改建议和理由。注意初期不建议设置“管理者”或“仲裁者”角色。因为这容易让系统陷入关于“管理权”的无效讨论中。我们应该通过设计清晰的通信协议和决策流程来让协作自然发生而不是增加一个发号施令的层级。2.2 通信协议设计让对话有焦点、有结果智能体之间不能漫无目的地聊天。我们需要设计一套“通信协议”规定它们说什么、怎么说、何时说。这直接决定了协作效率。基于共享工作区的异步通信我放弃了让智能体直接“对话”的模式而是引入了一个“共享工作区”Shared Workspace的概念。这可以是一个文本文件、一个数据库表或一个项目管理看板如Kanban。所有任务状态、产出物、评审意见都更新在这里。智能体通过“读取工作区最新状态”和“向工作区提交更新”来间接通信。这避免了信息在对话中被淹没也提供了完整的审计日志。结构化消息模板智能体发出的每条消息都不是自由文本而是结构化的数据。例如执行者完成任务后提交的消息格式可能是{ agent: Executor, task_id: TASK_001, action: SUBMIT_OUTPUT, content: 这里是生成的代码..., status: COMPLETED, metadata: {language: python, files: [main.py]} }评审者发出的评审意见格式可能是{ agent: Reviewer, target_task_id: TASK_001, action: REQUEST_REVISION, feedback: 第30行存在潜在的空指针异常建议增加判空逻辑。, severity: MEDIUM, suggested_change: if data is not None: ... }这种结构化极大地降低了其他智能体理解消息的难度方便自动化处理。触发与响应机制系统是事件驱动的。工作区状态的特定变化会触发特定智能体的行动。例如“当某个任务状态被规划者标记为ASSIGNED_TO_EXECUTOR时执行者智能体被唤醒去领取该任务。” 或者“当某个任务产出被更新且状态为PENDING_REVIEW时评审者智能体被唤醒。”2.3 工作流与状态机控制协作的生命周期整个系统的运行本质上是一个预定义的工作流Workflow或状态机State Machine。一个经典的任务生命周期如下待规划IDLE用户输入一个目标Goal。规划中PLANNING规划者智能体被触发分析目标分解任务将任务列表和初始状态写入工作区。状态变为PLANNED。待执行READY执行者智能体查看工作区领取状态为READY且与自己技能匹配的任务将状态更新为IN_PROGRESS。执行中IN_PROGRESS执行者进行工作期间可能调用工具如代码解释器、浏览器搜索完成后提交产出将状态更新为PENDING_REVIEW。待评审PENDING_REVIEW评审者智能体被触发检查产出给出通过或修改意见。若通过状态变为COMPLETED若需修改状态变回IN_PROGRESS并附上反馈。已完成COMPLETED任务完成。规划者会检查所有子任务是否都已完成若是则汇总最终结果向用户报告整个流程结束。这个状态机保证了流程的有序性防止多个智能体同时修改同一个任务造成混乱。3. 关键技术实现从理论到可运行的代码有了设计思路接下来就是如何用代码实现。这里我选择 Python 作为主要语言因为它有丰富的AI生态和异步编程支持。核心框架不绑定特定模型但以 OpenAI GPT 系列为例进行说明。3.1 智能体基类与角色实例化首先定义一个所有智能体的基类封装共同的行为。import abc import json from typing import Dict, Any, Optional class Agent(abc.ABC): def __init__(self, name: str, role: str, model: str gpt-4): self.name name self.role role self.model model # 系统提示词System Prompt是定义角色的灵魂 self.system_prompt self._define_system_prompt() abc.abstractmethod def _define_system_prompt(self) - str: 定义该角色的系统提示词。必须由子类实现。 pass async def perceive(self, workspace_state: Dict[str, Any]) - Optional[Dict[str, Any]]: 观察共享工作区决定是否需要行动及提取相关上下文。 # 这里实现具体逻辑例如检查是否有分配给自己的新任务 relevant_info self._extract_relevant_info(workspace_state) if self._should_act(relevant_info): return relevant_info return None abc.abstractmethod async def act(self, context: Dict[str, Any]) - Dict[str, Any]: 基于上下文执行行动并返回行动结果结构化消息。 pass def _should_act(self, info: Dict[str, Any]) - bool: # 简单的触发逻辑可根据任务状态、指派对象等判断 return bool(info) def _extract_relevant_info(self, state: Dict[str, Any]) - Dict[str, Any]: # 从全局状态中提取与本智能体相关的部分 # 例如提取所有状态为READY且技能要求匹配的任务 my_tasks [t for t in state.get(tasks, []) if t.get(status) READY and self.role in t.get(required_roles, [])] return {tasks: my_tasks} if my_tasks else {}然后实例化具体的角色。以“执行者”为例class ExecutorAgent(Agent): def _define_system_prompt(self) - str: return f你是一个专业的{self.role}例如Python后端工程师。你的职责是高效、准确地完成分配给你的具体任务。 你接收到的信息将包含任务的详细描述、输入数据以及期望的输出格式。 你必须严格遵守以下规则 1. 只做任务描述范围内的事情不要画蛇添足。 2. 你的输出必须是可执行、可验证的如代码、数据、文本。 3. 如果任务描述不清或缺少关键信息你必须明确提出问题而不是猜测。 4. 完成后严格按照指定的格式JSON提交你的产出。 你的输出格式必须是 {{ output: 你的主要产出内容, explanation: 对产出的简要说明可选, status: COMPLETED, metadata: {{}} // 任何额外信息如使用的库版本 }} async def act(self, context: Dict[str, Any]) - Dict[str, Any]: # context 包含从工作区提取的相关任务信息 task context[tasks][0] # 假设领取第一个相关任务 task_description task[description] # 调用大模型API生成任务产出 # 这里需要将任务描述、上下文和系统提示词组合成对话 messages [ {role: system, content: self.system_prompt}, {role: user, content: f请完成以下任务\n{task_description}\n\n请直接给出最终输出并确保格式正确。} ] # 假设有一个 call_llm 函数来调用大模型 llm_response await call_llm(messages, modelself.model) # 解析LLM的响应期望它返回一个JSON字符串 try: result json.loads(llm_response) except json.JSONDecodeError: # 如果模型没有返回标准JSON尝试提取或包装 result {output: llm_response, explanation: Model returned non-JSON response., status: COMPLETED} # 构造发送到工作区的结构化消息 action_message { agent: self.name, task_id: task[id], action: SUBMIT_OUTPUT, content: result, status: COMPLETED } return action_message规划者、评审者等角色的实现类似主要区别在于_define_system_prompt和act方法中的逻辑。规划者的提示词要强调分解和结构化能力评审者的提示词要强调批判性思维和细节检查。3.2 共享工作区与状态管理工作区可以用一个简单的类来模拟核心是维护一个任务列表和消息日志。class SharedWorkspace: def __init__(self): self.tasks [] # 所有任务对象列表 self.message_log [] # 所有智能体交换的消息 self.final_result None def add_task(self, task: Dict): self.tasks.append(task) def update_task(self, task_id: str, updates: Dict): for task in self.tasks: if task[id] task_id: task.update(updates) break def get_task(self, task_id: str) - Optional[Dict]: for task in self.tasks: if task[id] task_id: return task return None def get_tasks_by_status(self, status: str) - List[Dict]: return [t for t in self.tasks if t.get(status) status] def post_message(self, message: Dict): self.message_log.append({ **message, timestamp: time.time() # 添加时间戳 }) # 根据消息类型可能触发任务状态更新 self._process_message(message) def _process_message(self, message: Dict): action message.get(action) task_id message.get(task_id) if action SUBMIT_OUTPUT: self.update_task(task_id, {status: PENDING_REVIEW, output: message.get(content)}) elif action APPROVE: self.update_task(task_id, {status: COMPLETED}) elif action REQUEST_REVISION: self.update_task(task_id, {status: IN_PROGRESS, feedback: message.get(feedback)}) # ... 处理其他 action在实际项目中这个工作区可以替换为数据库如SQLite、PostgreSQL或更专业的任务队列如Celery Redis以实现持久化和更好的并发控制。3.3 协调器引擎驱动整个系统运转协调器Orchestrator是系统的调度中心。它不参与具体任务而是负责初始化工作区、管理智能体列表、并驱动整个事件循环。import asyncio class SwarmOrchestrator: def __init__(self, workspace: SharedWorkspace): self.workspace workspace self.agents [] # 注册的智能体列表 def register_agent(self, agent: Agent): self.agents.append(agent) async def run_cycle(self): 运行一个协作周期所有智能体感知、决策、行动一次。 workspace_snapshot { tasks: self.workspace.tasks.copy(), messages: self.workspace.message_log[-10:] # 最近10条消息作为上下文 } agent_actions [] # 并行让所有智能体感知环境 perception_tasks [agent.perceive(workspace_snapshot) for agent in self.agents] perceptions await asyncio.gather(*perception_tasks) # 收集需要行动的智能体及其上下文 for agent, context in zip(self.agents, perceptions): if context is not None: agent_actions.append((agent, context)) # 串行执行行动避免对同一资源的并发写冲突 for agent, context in agent_actions: action_result await agent.act(context) self.workspace.post_message(action_result) # 短暂延迟模拟处理时间也让日志更清晰 await asyncio.sleep(0.5) async def run_until_completion(self, initial_goal: str, max_cycles: int 20): 从初始目标开始运行直到所有任务完成或达到最大周期数。 # 1. 初始化创建初始的“规划”任务 initial_task { id: GOAL_001, description: initial_goal, status: IDLE, required_roles: [planner] } self.workspace.add_task(initial_task) cycle_count 0 while cycle_count max_cycles: cycle_count 1 print(f\n 协作周期 {cycle_count} 开始 ) # 检查是否所有任务都已完成 all_tasks self.workspace.tasks completed all(t.get(status) COMPLETED for t in all_tasks) if completed and len(all_tasks) 1: # 至少有一个非初始任务 print(所有任务已完成) # 汇总最终结果 self._compile_final_result() break await self.run_cycle() # 打印当前状态便于调试 self._print_workspace_status() if cycle_count max_cycles: print(f达到最大循环次数 {max_cycles}流程终止。) def _compile_final_result(self): # 从所有已完成的任务中提取产出组合成最终报告 outputs [] for task in self.workspace.tasks: if task.get(status) COMPLETED and task.get(id) ! GOAL_001: outputs.append(task.get(output)) self.workspace.final_result \n---\n.join([str(o) for o in outputs]) print(f最终结果\n{self.workspace.final_result}) def _print_workspace_status(self): for task in self.workspace.tasks: print(f Task [{task[id]}]: {task.get(status)} - {task.get(description, )[:50]}...)这个协调器实现了最核心的“感知-行动”循环。在每一轮循环中所有智能体并行地“看”一眼工作区perceive判断自己是否需要干活。需要干活的智能体再依次串行执行自己的行动act并将结果以结构化消息的形式发布到工作区触发状态更新。循环持续进行直到所有任务标记为完成。3.4 工具调用集成扩展智能体的能力边界智能体不能只靠“想”还需要能“做”。这就需要集成工具调用Tool Calling / Function Calling。例如让研究员智能体能调用搜索引擎API让执行者智能体能运行代码片段。 我们可以在智能体的act方法中集成这一功能。以研究员为例class ResearcherAgent(Agent): async def act(self, context: Dict[str, Any]) - Dict[str, Any]: task context[tasks][0] query task[description] # 假设任务描述就是搜索查询 # 构建给LLM的对话其中包含可用的工具描述 messages [ {role: system, content: self.system_prompt}, {role: user, content: f请搜索并总结关于以下主题的信息{query}} ] # 假设 call_llm_with_tools 支持工具调用 llm_response await call_llm_with_tools( messages, tools[{ type: function, function: { name: web_search, description: 执行一次网络搜索, parameters: { type: object, properties: {query: {type: string}}, required: [query] } } }], modelself.model ) # 检查LLM是否决定调用工具 if llm_response.get(tool_calls): for tool_call in llm_response[tool_calls]: if tool_call[function][name] web_search: search_query json.loads(tool_call[function][arguments])[query] # 实际调用搜索函数这里用模拟 search_results await self._perform_web_search(search_query) # 将搜索结果作为上下文再次发送给LLM让它总结 messages.append({ role: tool, tool_call_id: tool_call[id], content: json.dumps(search_results) }) final_summary await call_llm_with_tools(messages, tools[], modelself.model) result_content final_summary[content] else: result_content llm_response[content] action_message { agent: self.name, task_id: task[id], action: SUBMIT_RESEARCH, content: result_content, status: COMPLETED } return action_message async def _perform_web_search(self, query: str) - List[Dict]: # 这里集成真实的搜索API如SerperAPI、Google Custom Search等 # 返回结构化的搜索结果列表 return [{title: 模拟结果1, snippet: f关于{query}的模拟信息摘要..., link: #}]通过工具调用智能体从“思考者”变成了“行动者”能力范围得到了质的提升。4. 实战演练构建一个自动化数据分析报告生成系统理论说再多不如跑一个实例。假设我们的目标是“分析过去一个月某电商平台的销售数据找出销售额下降最多的三个商品类别并给出可能的原因和改进建议。”4.1 系统初始化与角色配置首先我们初始化工作区、协调器并组建我们的AI团队。async def main(): workspace SharedWorkspace() orchestrator SwarmOrchestrator(workspace) # 组建团队 planner PlannerAgent(name首席策划师, roleplanner) researcher ResearcherAgent(name市场研究员, roleresearcher) analyst ExecutorAgent(name数据分析师, roleanalyst) # 一个专门的数据分析执行者 reviewer ReviewerAgent(name质量审核员, rolereviewer) for agent in [planner, researcher, analyst, reviewer]: orchestrator.register_agent(agent) # 设定目标并运行 goal 分析过去一个月某电商平台的销售数据找出销售额下降最多的三个商品类别并给出可能的原因和改进建议。假设销售数据已以CSV格式提供。 await orchestrator.run_until_completion(goal, max_cycles15) if __name__ __main__: asyncio.run(main())4.2 协作过程推演与日志分析运行上述代码系统会开始自动协作。我们可以通过打印的日志来观察其内部运作以下为模拟输出 协作周期 1 开始 Task [GOAL_001]: IDLE - 分析过去一个月某电商平台的销售数据... 首席策划师 感知到新目标开始规划。 首席策划师 发布规划结果创建了5个子任务。 - TASK_001 [READY]: 理解数据结构和字段含义 (角色: analyst) - TASK_002 [READY]: 进行数据清洗与预处理 (角色: analyst) - TASK_003 [READY]: 计算各商品类别过去一个月的销售额环比变化 (角色: analyst) - TASK_004 [READY]: 调研可能导致销售额下降的常见市场原因 (角色: researcher) - TASK_005 [READY]: 综合分析与报告撰写 (角色: analyst) 协作周期 2 开始 数据分析师 领取任务 TASK_001。 数据分析师 提交了 TASK_001 的输出识别出数据包含字段日期、商品ID、类别、销售额、销量... 质量审核员 审核 TASK_001 通过。 数据分析师 领取任务 TASK_002。 市场研究员 领取任务 TASK_004。 协作周期 3 开始 数据分析师 提交了 TASK_002 的输出已完成数据清洗处理了缺失值与异常值。 市场研究员 调用工具 web_search查询“电商销售额下降原因”。 市场研究员 提交了 TASK_004 的输出总结了5点常见原因竞争加剧、季节性波动、负面评价、供应链问题、营销活动减少... 质量审核员 审核 TASK_004 通过并建议在最终报告中引用具体数据佐证。 协作周期 4-7 ... 分析师完成TASK_003计算出下降最多的三个类别是“消费电子”、“家居装饰”、“运动户外”环比下降分别为-15%、-12%、-8%。 研究员根据分析师的初步结果进行了更深入的定向搜索例如“消费电子 近期市场疲软原因”。 分析师综合所有信息完成TASK_005生成包含数据图表、原因分析和具体建议的完整报告草案。 协作周期 8 开始 质量审核员 审核 TASK_005报告草案发现一处数据引用格式不统一并提出修改意见。 数据分析师 根据反馈修改报告。 质量审核员 再次审核通过。 所有任务已完成 最终结果 一份结构清晰、数据翔实、分析到位的自动化分析报告通过这个推演可以看到智能体们像一支训练有素的团队一样工作规划者拆解任务执行者分析师和研究者并行工作评审者确保质量。整个过程无需人工干预自动推进直至生成最终报告。4.3 核心参数调优与性能考量在实现过程中有几个关键参数和设计选择直接影响系统表现系统提示词System Prompt的精度这是智能体的“人格”和“职责说明书”。提示词必须清晰、无歧义并强制要求结构化输出。模糊的提示词会导致智能体行为不可预测。我的经验是在提示词中明确“禁止做什么”和“必须怎么做”比只描述“要做什么”更有效。工作区状态的设计粒度任务状态不是越细越好。我最初设计了十几种状态导致状态转换逻辑极其复杂智能体也容易困惑。后来精简为IDLEREADYIN_PROGRESSPENDING_REVIEWCOMPLETED五个核心状态逻辑清晰了很多。循环触发与超时机制协调器的run_cycle是定时轮询还是事件驱动我采用了简单的轮询每个周期所有智能体感知一次实现简单但可能有延迟。在生产环境可以考虑基于工作区变更事件Event来触发特定智能体效率更高。同时必须为每个任务设置超时例如IN_PROGRESS状态超过30分钟无更新则重置为READY防止某个智能体“卡住”导致流程停滞。大模型API的成本与延迟这是实际运行中最现实的约束。每个智能体的每次act都是一次API调用。一个复杂任务可能涉及数十次调用成本不菲。优化策略包括缓存Caching对相似的感知信息或任务缓存LLM的响应。任务合并让规划者一次性生成更粗粒度的任务减少交互轮次。使用性价比更高的模型对评审、简单分类等任务使用gpt-3.5-turbo对需要深度思考的规划、创作任务再用gpt-4。异步与并发如代码所示perceive阶段可以并行但act阶段我选择了串行以避免写冲突。如果工作区设计得好例如任务间无依赖act也可以并行以提升速度。5. 常见问题、踩坑实录与进阶技巧在实际搭建和测试过程中我遇到了各种各样的问题这里总结几个最有代表性的。5.1 智能体陷入循环或讨论离题这是初期最常见的问题。表现就是两个智能体针对一个细节来回争论或者开始讨论与任务无关的哲学问题。根因系统提示词约束力不足或者智能体在act方法中接收的上下文过于宽泛例如把整个对话历史都塞进去。解决方案强化提示词约束在系统提示词开头用醒目的方式强调例如“【绝对指令】你是一名数据分析师。你只关心与当前任务ID{task_id} 相关的数据和指令。禁止讨论任务范围外的任何话题禁止与其它智能体进行开放式对话。你的输出必须是纯JSON格式。”严格过滤上下文在perceive和act方法中只提供与当前智能体本次行动绝对必要的信息。例如执行者只需要看到自己领取的那个任务的描述和输入不需要看到整个工作区的所有消息。设置回合限制在协调器中对同一个任务在PENDING_REVIEW和IN_PROGRESS之间的状态转换设置最大次数如3次。超过次数则触发“升级”机制例如将任务标记为NEEDS_HUMAN_INTERVENTION并通知用户。5.2 任务分解不合理或过于琐碎规划者智能体有时会把一个简单任务分解成几十个毫无意义的微步骤或者分解出的任务彼此存在循环依赖。根因给规划者的提示词或示例不够好或者它缺乏对任务复杂度的判断力。解决方案提供高质量示例Few-Shot Learning在规划者的系统提示词中不仅告诉它“要分解任务”更给它2-3个不同领域如数据分析、代码开发、文案撰写的优秀分解案例。案例中要展示如何把握任务粒度。引入验证步骤规划者产出任务列表后可以设计一个简单的“任务清单验证器”可以是另一个轻量级LLM调用或规则引擎检查任务数量是否在合理范围如3-10个检查是否有循环依赖。如果验证不通过则要求规划者重新规划。人工设定任务模板对于非常垂直的领域如每周生成固定格式的周报可以不完全依赖AI规划而是预定义好任务流程模板规划者只负责填充模板中的具体参数。5.3 错误处理与系统鲁棒性网络错误、API限额、LLM输出格式错误等都会导致系统崩溃。根因代码假设一切都会顺利进行缺乏异常处理。解决方案为每个LLM调用添加重试机制async def call_llm_with_retry(messages, max_retries3): for i in range(max_retries): try: return await call_llm(messages) except (APITimeoutError, RateLimitError) as e: if i max_retries - 1: raise wait_time (i 1) * 5 # 指数退避 print(fAPI调用失败{wait_time}秒后重试... 错误: {e}) await asyncio.sleep(wait_time) except InvalidResponseError as e: # 处理返回格式错误可以尝试修复或使用默认值 print(fLLM返回格式异常: {e}) return {content: {}, error: str(e)} # 返回一个安全的默认响应设计降级策略当某个智能体如研究员因工具调用失败而无法完成任务时不应让整个流程卡死。可以更新任务状态为BLOCKED并让规划者重新评估看是否能用其他方式如跳过该步骤或基于已有知识进行推理继续流程。实现完备的日志记录每个智能体的每次感知、决策、行动以及完整的LLM请求与响应。这是事后排查问题的唯一依据。日志应输出到文件并包含时间戳、智能体名、任务ID等关键信息。5.4 进阶技巧让系统更智能、更高效在基础版本跑通后可以尝试以下优化动态角色分配不要让智能体与角色固定绑定。可以设计一个“资源管理器”根据任务的技能需求如required_skills: [‘python’, ‘data_visualization’]从智能体池中动态选择最合适的“空闲智能体”来执行。这更接近真实的资源调度。引入反思Reflection智能体在主要流程之外增加一个“反思者”角色。它的任务不是处理具体工作而是定期如每5个周期阅读工作区日志评估整体协作效率并提出改进建议例如“我发现‘数据分析师’和‘研究员’在等待彼此产出建议规划者下次将任务TASK_004和TASK_003的顺序调整一下。” 这能让系统具备一定的自我优化能力。人类在环Human-in-the-loop在关键节点设置“人工审批”任务。例如当规划者生成的任务清单总预估成本超过某个阈值时自动创建一个状态为AWAITING_APPROVAL的任务并通知真实用户。用户批准或修改后流程才继续。这平衡了自动化与可控性。设计和实现一个多智能体系统就像在数字世界里组建和训练一支团队。初期你会被各种意外情况和低效协作搞得焦头烂额但一旦理顺了角色职责、通信协议和工作流程它所带来的自动化潜力是巨大的。从自动生成周报、处理客服工单到辅助软件设计、进行竞品分析应用场景非常广泛。最关键的是这套框架是模型无关的你可以随时将背后的GPT-4换成Claude、Gemini或任何其他大模型甚至混合使用不同模型来完成不同任务。
1. 项目概述从单兵作战到群体智能的跃迁最近在折腾一个挺有意思的东西我把它叫做“让AI自己协作”。听起来有点科幻但核心就是设计并实现一个多智能体系统或者更时髦一点叫它“Swarm”蜂群系统。这玩意儿不是简单地让几个ChatGPT对话机器人拉个群聊而是构建一个具备明确分工、能自主沟通、甚至能动态调整策略来完成复杂任务的AI团队。想想看我们平时让AI干点活无论是写代码、分析数据还是生成报告本质上还是“一对一”的指令-响应模式。我们得像项目经理一样把一个大任务拆解成无数个小步骤然后一步步喂给AI中间还得不断检查、纠偏、给新指令。效率低不说一旦任务稍微复杂点比如“帮我开发一个带用户管理和支付功能的小程序”光是把需求描述清楚就得费老大劲更别提过程中各种技术细节的决策了。多智能体系统要解决的就是这个“项目经理”的瓶颈。它的目标是创建一个数字世界的“特种部队”里面有指挥官Orchestrator、有侦察兵Researcher、有工程师Coder、有质检员Reviewer。你只需要下达一个最高指令比如“攻克那个山头完成某个项目”它们自己就会开会讨论战术、分配任务、同步进度、互相查漏补缺最终把结果呈报给你。这不仅仅是自动化这是赋予AI一种组织行为和协同智能。我之所以投入时间研究这个是因为在实际工作中越来越频繁地遇到那些“知道目标但路径极其复杂”的问题。单靠一个AI哪怕是最强大的模型其知识广度、思维深度和持续专注力都是有限的。而一个设计良好的多智能体系统能够通过角色 specialization专业化分工和 inter-agent communication智能体间通信实现“112”的群体智慧涌现。接下来我就把自己在设计和实现这套系统过程中的核心思路、关键设计、踩过的坑以及一些实用的心得完整地分享出来。2. 核心设计思路如何构建一个“活”起来的AI团队设计多智能体系统最忌讳的就是拍脑袋定几个角色然后就让它们开始对话。那结果大概率会变成一场混乱的、离题的、甚至无限循环的“废话大会”。要让AI团队高效协作必须像设计一个真正的组织架构和工作流程一样进行顶层设计。2.1 智能体角色定义不是越多越好而是越准越好角色的定义是整个系统的基石。我的经验是初期千万不要追求“大而全”的角色矩阵而是从最核心的职能开始。通常一个能处理通用任务的最小可行团队MVT包含以下四个角色任务规划与分解者Planner/Orchestrator这是团队的“大脑”。它的核心能力是理解最终目标并进行结构化分解。它不负责具体执行而是产出像“项目计划书”一样的东西一个任务列表Task List每个任务都有明确的描述、输入、期望输出和成功标准。它还需要决定哪个任务由哪个角色执行最合适。信息搜集与研究者Researcher这是团队的“眼睛和耳朵”。当任务需要外部知识、最新数据或特定领域信息时就由它出动。它需要擅长使用搜索工具、阅读文档、总结信息并能判断信息的可靠性和相关性。执行与构建者Executor/Builder这是团队的“双手”。通常是程序员、写作者、数据分析师等具体技能的化身。它接收明确的任务指令和所需上下文然后产出具体的成果如一段代码、一份文档、一个数据分析图表。评审与质检者Reviewer/Evaluator这是团队的“守门员”。它的工作是挑剔和找茬。检查执行者的产出是否符合要求、有无逻辑错误、风格是否一致、安全性如何。它不能只说“不好”必须给出具体的修改建议和理由。注意初期不建议设置“管理者”或“仲裁者”角色。因为这容易让系统陷入关于“管理权”的无效讨论中。我们应该通过设计清晰的通信协议和决策流程来让协作自然发生而不是增加一个发号施令的层级。2.2 通信协议设计让对话有焦点、有结果智能体之间不能漫无目的地聊天。我们需要设计一套“通信协议”规定它们说什么、怎么说、何时说。这直接决定了协作效率。基于共享工作区的异步通信我放弃了让智能体直接“对话”的模式而是引入了一个“共享工作区”Shared Workspace的概念。这可以是一个文本文件、一个数据库表或一个项目管理看板如Kanban。所有任务状态、产出物、评审意见都更新在这里。智能体通过“读取工作区最新状态”和“向工作区提交更新”来间接通信。这避免了信息在对话中被淹没也提供了完整的审计日志。结构化消息模板智能体发出的每条消息都不是自由文本而是结构化的数据。例如执行者完成任务后提交的消息格式可能是{ agent: Executor, task_id: TASK_001, action: SUBMIT_OUTPUT, content: 这里是生成的代码..., status: COMPLETED, metadata: {language: python, files: [main.py]} }评审者发出的评审意见格式可能是{ agent: Reviewer, target_task_id: TASK_001, action: REQUEST_REVISION, feedback: 第30行存在潜在的空指针异常建议增加判空逻辑。, severity: MEDIUM, suggested_change: if data is not None: ... }这种结构化极大地降低了其他智能体理解消息的难度方便自动化处理。触发与响应机制系统是事件驱动的。工作区状态的特定变化会触发特定智能体的行动。例如“当某个任务状态被规划者标记为ASSIGNED_TO_EXECUTOR时执行者智能体被唤醒去领取该任务。” 或者“当某个任务产出被更新且状态为PENDING_REVIEW时评审者智能体被唤醒。”2.3 工作流与状态机控制协作的生命周期整个系统的运行本质上是一个预定义的工作流Workflow或状态机State Machine。一个经典的任务生命周期如下待规划IDLE用户输入一个目标Goal。规划中PLANNING规划者智能体被触发分析目标分解任务将任务列表和初始状态写入工作区。状态变为PLANNED。待执行READY执行者智能体查看工作区领取状态为READY且与自己技能匹配的任务将状态更新为IN_PROGRESS。执行中IN_PROGRESS执行者进行工作期间可能调用工具如代码解释器、浏览器搜索完成后提交产出将状态更新为PENDING_REVIEW。待评审PENDING_REVIEW评审者智能体被触发检查产出给出通过或修改意见。若通过状态变为COMPLETED若需修改状态变回IN_PROGRESS并附上反馈。已完成COMPLETED任务完成。规划者会检查所有子任务是否都已完成若是则汇总最终结果向用户报告整个流程结束。这个状态机保证了流程的有序性防止多个智能体同时修改同一个任务造成混乱。3. 关键技术实现从理论到可运行的代码有了设计思路接下来就是如何用代码实现。这里我选择 Python 作为主要语言因为它有丰富的AI生态和异步编程支持。核心框架不绑定特定模型但以 OpenAI GPT 系列为例进行说明。3.1 智能体基类与角色实例化首先定义一个所有智能体的基类封装共同的行为。import abc import json from typing import Dict, Any, Optional class Agent(abc.ABC): def __init__(self, name: str, role: str, model: str gpt-4): self.name name self.role role self.model model # 系统提示词System Prompt是定义角色的灵魂 self.system_prompt self._define_system_prompt() abc.abstractmethod def _define_system_prompt(self) - str: 定义该角色的系统提示词。必须由子类实现。 pass async def perceive(self, workspace_state: Dict[str, Any]) - Optional[Dict[str, Any]]: 观察共享工作区决定是否需要行动及提取相关上下文。 # 这里实现具体逻辑例如检查是否有分配给自己的新任务 relevant_info self._extract_relevant_info(workspace_state) if self._should_act(relevant_info): return relevant_info return None abc.abstractmethod async def act(self, context: Dict[str, Any]) - Dict[str, Any]: 基于上下文执行行动并返回行动结果结构化消息。 pass def _should_act(self, info: Dict[str, Any]) - bool: # 简单的触发逻辑可根据任务状态、指派对象等判断 return bool(info) def _extract_relevant_info(self, state: Dict[str, Any]) - Dict[str, Any]: # 从全局状态中提取与本智能体相关的部分 # 例如提取所有状态为READY且技能要求匹配的任务 my_tasks [t for t in state.get(tasks, []) if t.get(status) READY and self.role in t.get(required_roles, [])] return {tasks: my_tasks} if my_tasks else {}然后实例化具体的角色。以“执行者”为例class ExecutorAgent(Agent): def _define_system_prompt(self) - str: return f你是一个专业的{self.role}例如Python后端工程师。你的职责是高效、准确地完成分配给你的具体任务。 你接收到的信息将包含任务的详细描述、输入数据以及期望的输出格式。 你必须严格遵守以下规则 1. 只做任务描述范围内的事情不要画蛇添足。 2. 你的输出必须是可执行、可验证的如代码、数据、文本。 3. 如果任务描述不清或缺少关键信息你必须明确提出问题而不是猜测。 4. 完成后严格按照指定的格式JSON提交你的产出。 你的输出格式必须是 {{ output: 你的主要产出内容, explanation: 对产出的简要说明可选, status: COMPLETED, metadata: {{}} // 任何额外信息如使用的库版本 }} async def act(self, context: Dict[str, Any]) - Dict[str, Any]: # context 包含从工作区提取的相关任务信息 task context[tasks][0] # 假设领取第一个相关任务 task_description task[description] # 调用大模型API生成任务产出 # 这里需要将任务描述、上下文和系统提示词组合成对话 messages [ {role: system, content: self.system_prompt}, {role: user, content: f请完成以下任务\n{task_description}\n\n请直接给出最终输出并确保格式正确。} ] # 假设有一个 call_llm 函数来调用大模型 llm_response await call_llm(messages, modelself.model) # 解析LLM的响应期望它返回一个JSON字符串 try: result json.loads(llm_response) except json.JSONDecodeError: # 如果模型没有返回标准JSON尝试提取或包装 result {output: llm_response, explanation: Model returned non-JSON response., status: COMPLETED} # 构造发送到工作区的结构化消息 action_message { agent: self.name, task_id: task[id], action: SUBMIT_OUTPUT, content: result, status: COMPLETED } return action_message规划者、评审者等角色的实现类似主要区别在于_define_system_prompt和act方法中的逻辑。规划者的提示词要强调分解和结构化能力评审者的提示词要强调批判性思维和细节检查。3.2 共享工作区与状态管理工作区可以用一个简单的类来模拟核心是维护一个任务列表和消息日志。class SharedWorkspace: def __init__(self): self.tasks [] # 所有任务对象列表 self.message_log [] # 所有智能体交换的消息 self.final_result None def add_task(self, task: Dict): self.tasks.append(task) def update_task(self, task_id: str, updates: Dict): for task in self.tasks: if task[id] task_id: task.update(updates) break def get_task(self, task_id: str) - Optional[Dict]: for task in self.tasks: if task[id] task_id: return task return None def get_tasks_by_status(self, status: str) - List[Dict]: return [t for t in self.tasks if t.get(status) status] def post_message(self, message: Dict): self.message_log.append({ **message, timestamp: time.time() # 添加时间戳 }) # 根据消息类型可能触发任务状态更新 self._process_message(message) def _process_message(self, message: Dict): action message.get(action) task_id message.get(task_id) if action SUBMIT_OUTPUT: self.update_task(task_id, {status: PENDING_REVIEW, output: message.get(content)}) elif action APPROVE: self.update_task(task_id, {status: COMPLETED}) elif action REQUEST_REVISION: self.update_task(task_id, {status: IN_PROGRESS, feedback: message.get(feedback)}) # ... 处理其他 action在实际项目中这个工作区可以替换为数据库如SQLite、PostgreSQL或更专业的任务队列如Celery Redis以实现持久化和更好的并发控制。3.3 协调器引擎驱动整个系统运转协调器Orchestrator是系统的调度中心。它不参与具体任务而是负责初始化工作区、管理智能体列表、并驱动整个事件循环。import asyncio class SwarmOrchestrator: def __init__(self, workspace: SharedWorkspace): self.workspace workspace self.agents [] # 注册的智能体列表 def register_agent(self, agent: Agent): self.agents.append(agent) async def run_cycle(self): 运行一个协作周期所有智能体感知、决策、行动一次。 workspace_snapshot { tasks: self.workspace.tasks.copy(), messages: self.workspace.message_log[-10:] # 最近10条消息作为上下文 } agent_actions [] # 并行让所有智能体感知环境 perception_tasks [agent.perceive(workspace_snapshot) for agent in self.agents] perceptions await asyncio.gather(*perception_tasks) # 收集需要行动的智能体及其上下文 for agent, context in zip(self.agents, perceptions): if context is not None: agent_actions.append((agent, context)) # 串行执行行动避免对同一资源的并发写冲突 for agent, context in agent_actions: action_result await agent.act(context) self.workspace.post_message(action_result) # 短暂延迟模拟处理时间也让日志更清晰 await asyncio.sleep(0.5) async def run_until_completion(self, initial_goal: str, max_cycles: int 20): 从初始目标开始运行直到所有任务完成或达到最大周期数。 # 1. 初始化创建初始的“规划”任务 initial_task { id: GOAL_001, description: initial_goal, status: IDLE, required_roles: [planner] } self.workspace.add_task(initial_task) cycle_count 0 while cycle_count max_cycles: cycle_count 1 print(f\n 协作周期 {cycle_count} 开始 ) # 检查是否所有任务都已完成 all_tasks self.workspace.tasks completed all(t.get(status) COMPLETED for t in all_tasks) if completed and len(all_tasks) 1: # 至少有一个非初始任务 print(所有任务已完成) # 汇总最终结果 self._compile_final_result() break await self.run_cycle() # 打印当前状态便于调试 self._print_workspace_status() if cycle_count max_cycles: print(f达到最大循环次数 {max_cycles}流程终止。) def _compile_final_result(self): # 从所有已完成的任务中提取产出组合成最终报告 outputs [] for task in self.workspace.tasks: if task.get(status) COMPLETED and task.get(id) ! GOAL_001: outputs.append(task.get(output)) self.workspace.final_result \n---\n.join([str(o) for o in outputs]) print(f最终结果\n{self.workspace.final_result}) def _print_workspace_status(self): for task in self.workspace.tasks: print(f Task [{task[id]}]: {task.get(status)} - {task.get(description, )[:50]}...)这个协调器实现了最核心的“感知-行动”循环。在每一轮循环中所有智能体并行地“看”一眼工作区perceive判断自己是否需要干活。需要干活的智能体再依次串行执行自己的行动act并将结果以结构化消息的形式发布到工作区触发状态更新。循环持续进行直到所有任务标记为完成。3.4 工具调用集成扩展智能体的能力边界智能体不能只靠“想”还需要能“做”。这就需要集成工具调用Tool Calling / Function Calling。例如让研究员智能体能调用搜索引擎API让执行者智能体能运行代码片段。 我们可以在智能体的act方法中集成这一功能。以研究员为例class ResearcherAgent(Agent): async def act(self, context: Dict[str, Any]) - Dict[str, Any]: task context[tasks][0] query task[description] # 假设任务描述就是搜索查询 # 构建给LLM的对话其中包含可用的工具描述 messages [ {role: system, content: self.system_prompt}, {role: user, content: f请搜索并总结关于以下主题的信息{query}} ] # 假设 call_llm_with_tools 支持工具调用 llm_response await call_llm_with_tools( messages, tools[{ type: function, function: { name: web_search, description: 执行一次网络搜索, parameters: { type: object, properties: {query: {type: string}}, required: [query] } } }], modelself.model ) # 检查LLM是否决定调用工具 if llm_response.get(tool_calls): for tool_call in llm_response[tool_calls]: if tool_call[function][name] web_search: search_query json.loads(tool_call[function][arguments])[query] # 实际调用搜索函数这里用模拟 search_results await self._perform_web_search(search_query) # 将搜索结果作为上下文再次发送给LLM让它总结 messages.append({ role: tool, tool_call_id: tool_call[id], content: json.dumps(search_results) }) final_summary await call_llm_with_tools(messages, tools[], modelself.model) result_content final_summary[content] else: result_content llm_response[content] action_message { agent: self.name, task_id: task[id], action: SUBMIT_RESEARCH, content: result_content, status: COMPLETED } return action_message async def _perform_web_search(self, query: str) - List[Dict]: # 这里集成真实的搜索API如SerperAPI、Google Custom Search等 # 返回结构化的搜索结果列表 return [{title: 模拟结果1, snippet: f关于{query}的模拟信息摘要..., link: #}]通过工具调用智能体从“思考者”变成了“行动者”能力范围得到了质的提升。4. 实战演练构建一个自动化数据分析报告生成系统理论说再多不如跑一个实例。假设我们的目标是“分析过去一个月某电商平台的销售数据找出销售额下降最多的三个商品类别并给出可能的原因和改进建议。”4.1 系统初始化与角色配置首先我们初始化工作区、协调器并组建我们的AI团队。async def main(): workspace SharedWorkspace() orchestrator SwarmOrchestrator(workspace) # 组建团队 planner PlannerAgent(name首席策划师, roleplanner) researcher ResearcherAgent(name市场研究员, roleresearcher) analyst ExecutorAgent(name数据分析师, roleanalyst) # 一个专门的数据分析执行者 reviewer ReviewerAgent(name质量审核员, rolereviewer) for agent in [planner, researcher, analyst, reviewer]: orchestrator.register_agent(agent) # 设定目标并运行 goal 分析过去一个月某电商平台的销售数据找出销售额下降最多的三个商品类别并给出可能的原因和改进建议。假设销售数据已以CSV格式提供。 await orchestrator.run_until_completion(goal, max_cycles15) if __name__ __main__: asyncio.run(main())4.2 协作过程推演与日志分析运行上述代码系统会开始自动协作。我们可以通过打印的日志来观察其内部运作以下为模拟输出 协作周期 1 开始 Task [GOAL_001]: IDLE - 分析过去一个月某电商平台的销售数据... 首席策划师 感知到新目标开始规划。 首席策划师 发布规划结果创建了5个子任务。 - TASK_001 [READY]: 理解数据结构和字段含义 (角色: analyst) - TASK_002 [READY]: 进行数据清洗与预处理 (角色: analyst) - TASK_003 [READY]: 计算各商品类别过去一个月的销售额环比变化 (角色: analyst) - TASK_004 [READY]: 调研可能导致销售额下降的常见市场原因 (角色: researcher) - TASK_005 [READY]: 综合分析与报告撰写 (角色: analyst) 协作周期 2 开始 数据分析师 领取任务 TASK_001。 数据分析师 提交了 TASK_001 的输出识别出数据包含字段日期、商品ID、类别、销售额、销量... 质量审核员 审核 TASK_001 通过。 数据分析师 领取任务 TASK_002。 市场研究员 领取任务 TASK_004。 协作周期 3 开始 数据分析师 提交了 TASK_002 的输出已完成数据清洗处理了缺失值与异常值。 市场研究员 调用工具 web_search查询“电商销售额下降原因”。 市场研究员 提交了 TASK_004 的输出总结了5点常见原因竞争加剧、季节性波动、负面评价、供应链问题、营销活动减少... 质量审核员 审核 TASK_004 通过并建议在最终报告中引用具体数据佐证。 协作周期 4-7 ... 分析师完成TASK_003计算出下降最多的三个类别是“消费电子”、“家居装饰”、“运动户外”环比下降分别为-15%、-12%、-8%。 研究员根据分析师的初步结果进行了更深入的定向搜索例如“消费电子 近期市场疲软原因”。 分析师综合所有信息完成TASK_005生成包含数据图表、原因分析和具体建议的完整报告草案。 协作周期 8 开始 质量审核员 审核 TASK_005报告草案发现一处数据引用格式不统一并提出修改意见。 数据分析师 根据反馈修改报告。 质量审核员 再次审核通过。 所有任务已完成 最终结果 一份结构清晰、数据翔实、分析到位的自动化分析报告通过这个推演可以看到智能体们像一支训练有素的团队一样工作规划者拆解任务执行者分析师和研究者并行工作评审者确保质量。整个过程无需人工干预自动推进直至生成最终报告。4.3 核心参数调优与性能考量在实现过程中有几个关键参数和设计选择直接影响系统表现系统提示词System Prompt的精度这是智能体的“人格”和“职责说明书”。提示词必须清晰、无歧义并强制要求结构化输出。模糊的提示词会导致智能体行为不可预测。我的经验是在提示词中明确“禁止做什么”和“必须怎么做”比只描述“要做什么”更有效。工作区状态的设计粒度任务状态不是越细越好。我最初设计了十几种状态导致状态转换逻辑极其复杂智能体也容易困惑。后来精简为IDLEREADYIN_PROGRESSPENDING_REVIEWCOMPLETED五个核心状态逻辑清晰了很多。循环触发与超时机制协调器的run_cycle是定时轮询还是事件驱动我采用了简单的轮询每个周期所有智能体感知一次实现简单但可能有延迟。在生产环境可以考虑基于工作区变更事件Event来触发特定智能体效率更高。同时必须为每个任务设置超时例如IN_PROGRESS状态超过30分钟无更新则重置为READY防止某个智能体“卡住”导致流程停滞。大模型API的成本与延迟这是实际运行中最现实的约束。每个智能体的每次act都是一次API调用。一个复杂任务可能涉及数十次调用成本不菲。优化策略包括缓存Caching对相似的感知信息或任务缓存LLM的响应。任务合并让规划者一次性生成更粗粒度的任务减少交互轮次。使用性价比更高的模型对评审、简单分类等任务使用gpt-3.5-turbo对需要深度思考的规划、创作任务再用gpt-4。异步与并发如代码所示perceive阶段可以并行但act阶段我选择了串行以避免写冲突。如果工作区设计得好例如任务间无依赖act也可以并行以提升速度。5. 常见问题、踩坑实录与进阶技巧在实际搭建和测试过程中我遇到了各种各样的问题这里总结几个最有代表性的。5.1 智能体陷入循环或讨论离题这是初期最常见的问题。表现就是两个智能体针对一个细节来回争论或者开始讨论与任务无关的哲学问题。根因系统提示词约束力不足或者智能体在act方法中接收的上下文过于宽泛例如把整个对话历史都塞进去。解决方案强化提示词约束在系统提示词开头用醒目的方式强调例如“【绝对指令】你是一名数据分析师。你只关心与当前任务ID{task_id} 相关的数据和指令。禁止讨论任务范围外的任何话题禁止与其它智能体进行开放式对话。你的输出必须是纯JSON格式。”严格过滤上下文在perceive和act方法中只提供与当前智能体本次行动绝对必要的信息。例如执行者只需要看到自己领取的那个任务的描述和输入不需要看到整个工作区的所有消息。设置回合限制在协调器中对同一个任务在PENDING_REVIEW和IN_PROGRESS之间的状态转换设置最大次数如3次。超过次数则触发“升级”机制例如将任务标记为NEEDS_HUMAN_INTERVENTION并通知用户。5.2 任务分解不合理或过于琐碎规划者智能体有时会把一个简单任务分解成几十个毫无意义的微步骤或者分解出的任务彼此存在循环依赖。根因给规划者的提示词或示例不够好或者它缺乏对任务复杂度的判断力。解决方案提供高质量示例Few-Shot Learning在规划者的系统提示词中不仅告诉它“要分解任务”更给它2-3个不同领域如数据分析、代码开发、文案撰写的优秀分解案例。案例中要展示如何把握任务粒度。引入验证步骤规划者产出任务列表后可以设计一个简单的“任务清单验证器”可以是另一个轻量级LLM调用或规则引擎检查任务数量是否在合理范围如3-10个检查是否有循环依赖。如果验证不通过则要求规划者重新规划。人工设定任务模板对于非常垂直的领域如每周生成固定格式的周报可以不完全依赖AI规划而是预定义好任务流程模板规划者只负责填充模板中的具体参数。5.3 错误处理与系统鲁棒性网络错误、API限额、LLM输出格式错误等都会导致系统崩溃。根因代码假设一切都会顺利进行缺乏异常处理。解决方案为每个LLM调用添加重试机制async def call_llm_with_retry(messages, max_retries3): for i in range(max_retries): try: return await call_llm(messages) except (APITimeoutError, RateLimitError) as e: if i max_retries - 1: raise wait_time (i 1) * 5 # 指数退避 print(fAPI调用失败{wait_time}秒后重试... 错误: {e}) await asyncio.sleep(wait_time) except InvalidResponseError as e: # 处理返回格式错误可以尝试修复或使用默认值 print(fLLM返回格式异常: {e}) return {content: {}, error: str(e)} # 返回一个安全的默认响应设计降级策略当某个智能体如研究员因工具调用失败而无法完成任务时不应让整个流程卡死。可以更新任务状态为BLOCKED并让规划者重新评估看是否能用其他方式如跳过该步骤或基于已有知识进行推理继续流程。实现完备的日志记录每个智能体的每次感知、决策、行动以及完整的LLM请求与响应。这是事后排查问题的唯一依据。日志应输出到文件并包含时间戳、智能体名、任务ID等关键信息。5.4 进阶技巧让系统更智能、更高效在基础版本跑通后可以尝试以下优化动态角色分配不要让智能体与角色固定绑定。可以设计一个“资源管理器”根据任务的技能需求如required_skills: [‘python’, ‘data_visualization’]从智能体池中动态选择最合适的“空闲智能体”来执行。这更接近真实的资源调度。引入反思Reflection智能体在主要流程之外增加一个“反思者”角色。它的任务不是处理具体工作而是定期如每5个周期阅读工作区日志评估整体协作效率并提出改进建议例如“我发现‘数据分析师’和‘研究员’在等待彼此产出建议规划者下次将任务TASK_004和TASK_003的顺序调整一下。” 这能让系统具备一定的自我优化能力。人类在环Human-in-the-loop在关键节点设置“人工审批”任务。例如当规划者生成的任务清单总预估成本超过某个阈值时自动创建一个状态为AWAITING_APPROVAL的任务并通知真实用户。用户批准或修改后流程才继续。这平衡了自动化与可控性。设计和实现一个多智能体系统就像在数字世界里组建和训练一支团队。初期你会被各种意外情况和低效协作搞得焦头烂额但一旦理顺了角色职责、通信协议和工作流程它所带来的自动化潜力是巨大的。从自动生成周报、处理客服工单到辅助软件设计、进行竞品分析应用场景非常广泛。最关键的是这套框架是模型无关的你可以随时将背后的GPT-4换成Claude、Gemini或任何其他大模型甚至混合使用不同模型来完成不同任务。
用法大不同，错误原因竟在这儿)
