多智能体协作框架：用LLM构建自动化团队解决复杂任务

1. 项目概述当LLM学会“开会”一个多智能体协作框架的诞生如果你和我一样在尝试用大语言模型LLM解决稍微复杂一点的任务时总会遇到一个瓶颈单个模型的能力边界。让它写个邮件、总结个文档还行但一旦涉及到需要多步骤推理、跨领域知识整合或者需要“左手画圆、右手画方”的并行任务时单个LLM就显得力不从心了要么逻辑链条断裂要么输出质量不稳定。这时候一个自然而然的思路就冒出来了为什么不让多个LLM“智能体”一起工作呢就像我们人类团队一样有人负责规划有人负责执行有人负责审核通过协作来攻克复杂问题。AgnostiqHQ/multi-agent-llm这个开源项目正是为了解决这个问题而生。它不是一个简单的API包装器而是一个面向复杂任务的多智能体协作框架。你可以把它想象成一个虚拟的“会议室”或“项目组”你作为“项目经理”只需要下达一个高层级的指令比如“开发一个网页应用来展示实时天气数据”框架内的多个智能体就会自动分工、讨论、执行、迭代最终给你一个可运行的结果。这个框架的核心价值在于它将LLM从“一个强大的工具”提升为“一个可组织的生产力单元”。它不再局限于简单的问答或文本生成而是能够处理涉及代码生成、系统设计、文件操作、逻辑验证等一系列动作的复合型任务。对于开发者、产品经理、研究人员甚至是希望自动化工作流的非技术人员这都打开了一扇新的大门。接下来我将带你深入拆解这个框架的设计哲学、核心组件并分享如何从零开始搭建和定制属于你自己的多智能体团队。2. 框架核心设计角色、编排与通信要理解multi-agent-llm如何工作我们必须先抛开代码从它的设计理念入手。这个框架的架构非常清晰地反映了现实世界中的团队协作模式其核心可以概括为三个要素角色定义、工作流编排和智能体间通信。2.1 智能体角色系统不只是“另一个GPT”在这个框架里每个智能体都不是一个通用的“GPT”而是一个被赋予了特定角色、技能和上下文的“专家”。这是实现有效协作的第一步。框架通常预定义了一些经典角色你也可以轻松地自定义。1. 核心角色解析规划者/架构师这是团队的“大脑”。它的核心职责是任务分解。当你给出一个模糊的指令时规划者会首先分析需求将其拆解成一个有序的、可执行的子任务列表。例如对于“创建一个待办事项应用”的指令规划者可能会输出1. 设计数据库Schema2. 创建后端API用户认证、增删改查3. 开发前端界面列表展示、添加表单4. 编写部署脚本。它的提示词Prompt经过特殊设计强调逻辑性、结构化和无遗漏。执行者/开发者这是团队的“双手”。它接收来自规划者的具体子任务描述并生成可执行的代码或内容。执行者通常被赋予某个技术栈的上下文如“你是一名精通Python Flask和React的资深全栈工程师”以确保其输出的专业性。它的工作是从自然语言描述到具体产出的转换。审查者/测试者这是团队的“质量守门员”。它不会主动创造而是负责批判性验证。审查者会检查执行者生成的代码语法是否正确逻辑有无漏洞是否符合任务要求安全是否有隐患它基于一套预设的规则或通过运行静态分析工具来提供反馈。这个角色对于避免LLM的“幻觉”即生成看似合理实则错误的代码至关重要。协调者/管理者这是一个可选但强大的角色负责控制流程和解决冲突。当执行者和审查者就某个代码片段产生分歧时或者当任务执行偏离轨道时协调者会介入分析双方论点做出决策并指导下一步行动。它模拟了项目经理或技术负责人的职责。注意角色的有效性极度依赖于其“系统提示词”。一个精心设计的提示词应该明确定义角色、职责、输出格式、禁忌和思考框架。例如给审查者的提示词必须强调“找出潜在问题”而不是“赞美代码”。2. 角色定制化实践框架的强大之处在于角色的可塑性。假设你需要一个专注于数据可视化的智能体你可以这样定义# 示例自定义“数据可视化专家”智能体配置 agent: name: data_viz_specialist role: 你是一名数据可视化专家精通Matplotlib, Seaborn, Plotly和D3.js。你的目标是将枯燥的数据转化为直观、美观且信息丰富的图表。你特别注重图表类型的正确选择、颜色的可访问性以及图例的清晰度。 capabilities: - 根据数据特征推荐最佳图表类型 - 生成可直接运行的Python可视化代码 - 优化图表交互性和响应式设计 output_format: 请以代码块形式输出完整的Python脚本并附上简短说明解释你的设计选择。通过这样的定义你就可以将这个专家纳入你的智能体团队在处理数据分析任务时调用它。2.2 工作流编排引擎智能体协作的“剧本”定义了角色下一步就是规定他们如何互动。这就是工作流编排。multi-agent-llm框架通常采用一种基于状态或基于消息的编排模式。1. 顺序工作流最简单的模式如同流水线。规划者 → 执行者 → 审查者 → 结束。每个智能体完成后将结果传递给下一个。这种模式适合线性、依赖关系明确的任务。2. 循环迭代工作流这是更常见且强大的模式模拟了“开发-测试-修复”的敏捷循环。规划者制定计划 | v 执行者生成代码 —— 审查者检查代码 ^ | | v |——发现问题提供反馈——|执行者根据审查者的反馈修改代码然后再次提交审查。这个循环会持续进行直到审查者通过或者达到最大迭代次数。框架需要内置逻辑来判断何时退出循环例如审查者给出“通过”信号或连续三次修改仍未通过。3. 并行与分支工作流对于大型项目任务可以并行。例如规划者可能同时创建“设计前端”和“设计后端API”两个子任务。框架需要能够创建两个独立的执行者智能体或许一个专注前端一个专注后端并行工作并在它们都完成后由一个协调者智能体进行集成。这需要更复杂的编排逻辑可能涉及任务队列和依赖关系图。实操心得在初期建议从顺序循环迭代的混合模式开始。让规划者先分解任务然后对每个关键的子任务尤其是代码生成任务启动一个“执行-审查”循环。这样既能保证质量又不会让流程过于复杂。框架的配置文件中通常会有一个workflow或orchestration部分让你用YAML或JSON来定义这些步骤和跳转条件。2.3 通信与记忆机制让对话有“上下文”智能体之间不能各说各话它们需要共享上下文和历史。这就是通信总线和共享记忆池的作用。通信消息格式智能体间传递的不是原始字符串而是结构化的消息对象。一个典型的消息可能包含以下字段sender: 发送者角色如 “planner”recipient: 接收者角色如 “executor”content: 消息主体如具体的任务描述或代码type: 消息类型如 “task”, “code”, “feedback”, “approval”context_id: 关联的会话或任务ID共享记忆/状态管理所有智能体都能访问一个共享的“工作区”或“黑板”。这里存储了项目的核心资产原始任务描述规划者输出的完整任务分解清单每个子任务的当前状态待处理、执行中、审查中、已完成、失败生成的代码文件、文档等产物智能体间的完整对话历史这个共享记忆是智能体做出合理决策的基础。例如当审查者质疑某段代码时它可以回溯查看规划者对该子任务的最初要求从而判断执行者是否偏离了方向。常见问题记忆过长导致上下文窗口爆炸。LLM有上下文长度限制。当项目很大、对话历史很长时直接将所有历史塞进提示词是不可行的。解决方案是记忆摘要与关键信息提取。框架需要具备能力在每次调用智能体时不是传递全部历史而是传递当前任务相关的摘要、最近几条关键消息以及指向共享工作区中完整产物的引用。这本身就是一个值得设计的子模块。3. 从零搭建与核心环节实现理解了设计理念后我们动手搭建一个基础版本。这里我不会完全照搬某个特定项目的代码而是阐述实现的核心环节和关键代码片段你可以用任何熟悉的LLM SDK如OpenAI, Anthropic, 或本地部署的Ollama来实现。3.1 基础环境搭建与智能体类定义首先我们需要定义智能体这个核心类。一个智能体本质上是一个角色描述系统提示词 一个LLM调用客户端 消息处理逻辑。# core/agent.py import openai # 或 from anthropic import Anthropic from typing import Dict, Any, List class Agent: def __init__(self, name: str, role: str, model: str gpt-4, temperature: float 0.1): 初始化一个智能体。 :param name: 智能体名称如 planner :param role: 系统提示词定义智能体的角色和能力 :param model: 使用的LLM模型 :param temperature: 创造性对于执行类任务宜低0.1-0.3对于头脑风暴可稍高 self.name name self.role role self.model model self.temperature temperature self.client openai.OpenAI(api_keyyour_api_key) # 初始化客户端 self.conversation_history: List[Dict] [] # 该智能体的私有对话历史 def send_message(self, message: str, context: List[Dict] None) - str: 核心方法让智能体根据收到的消息和上下文进行思考并回复。 :param message: 接收到的消息内容 :param context: 额外的上下文消息列表例如从共享记忆来的相关历史 :return: 智能体的回复文本 # 1. 构建消息列表 messages [{role: system, content: self.role}] # 2. 添加上下文例如最近几轮相关对话 if context: # 这里可以实现一个简单的上下文窗口管理只保留最近N条或最相关的 filtered_context self._manage_context_window(context) messages.extend(filtered_context) # 3. 添加当前对话历史私有记忆 messages.extend(self.conversation_history[-6:]) # 保留最近3轮对话 # 4. 添加当前消息 messages.append({role: user, content: message}) # 5. 调用LLM try: response self.client.chat.completions.create( modelself.model, messagesmessages, temperatureself.temperature, streamFalse ) reply response.choices[0].message.content # 6. 更新私有历史 self.conversation_history.append({role: user, content: message}) self.conversation_history.append({role: assistant, content: reply}) return reply except Exception as e: return f[Agent {self.name} Error]: {str(e)} def _manage_context_window(self, context: List[Dict]) - List[Dict]: 简单的上下文管理限制长度或按重要性筛选。这是一个简化版。 # 更复杂的实现可以用Embedding计算相关性只保留最相关的几条 max_tokens 4000 current_tokens sum(len(str(msg)) for msg in context) # 粗略估算 if current_tokens max_tokens: # 策略保留开头任务定义和结尾最近对话 return context[:2] context[-4:] return context3.2 实现一个简单的顺序工作流引擎有了智能体我们需要一个“导演”来指挥他们。下面是一个极度简化的顺序工作流控制器。# core/orchestrator.py from .agent import Agent from .memory import SharedMemory # 假设有一个共享记忆类 class SequentialOrchestrator: def __init__(self, agents: Dict[str, Agent], shared_memory: SharedMemory): self.agents agents # 字典键为角色名值为Agent实例 self.memory shared_memory self.message_bus [] # 简单的消息总线记录所有消息 def run(self, initial_task: str): 运行一个简单的 规划 - 执行 - 审查 流程 print(f【开始任务】: {initial_task}) self.memory.set(initial_task, initial_task) # 步骤1: 规划 print(【阶段1】规划中...) planner: Agent self.agents[planner] plan planner.send_message( f请将以下任务分解为具体的、可执行的子任务步骤。任务{initial_task}。请以清晰的列表格式输出。 ) self.memory.set(plan, plan) self._log_message(planner, orchestrator, plan) print(f规划结果:\n{plan}) # 假设我们从规划结果中解析出第一个子任务实际中需要解析列表 # 这里简化处理假设规划者输出的第一段就是第一个子任务描述 sub_tasks [line.strip(- ).strip() for line in plan.split(\n) if line.strip().startswith(-)] if not sub_tasks: sub_tasks [plan] # 后备方案 for i, task_desc in enumerate(sub_tasks): print(f\n【处理子任务 {i1}】: {task_desc}) self.memory.set(fcurrent_task_{i}, task_desc) # 步骤2: 执行 print( - 执行中...) executor: Agent self.agents[executor] code_output executor.send_message( f请完成以下子任务{task_desc}。这是整体计划的一部分{plan}。请生成完整、可运行的代码。 ) self.memory.set(fcode_{i}, code_output) self._log_message(executor, orchestrator, code_output) print(f 生成的代码片段已保存。) # 步骤3: 审查 print( - 审查中...) reviewer: Agent self.agents[reviewer] feedback reviewer.send_message( f请审查以下代码它旨在实现{task_desc}。\n代码\n\n{code_output}\n\n f请检查语法错误、逻辑错误、安全漏洞和是否符合任务要求。如果通过请说‘审查通过’。如果有问题请具体指出。 ) self.memory.set(ffeedback_{i}, feedback) self._log_message(reviewer, orchestrator, feedback) if 审查通过 in feedback: print(f - 子任务 {i1} 审查通过) else: print(f - 审查发现需修改{feedback[:100]}...) # 打印前100字符 # 这里可以引入循环迭代将feedback再次发给executor修改 # 简化起见我们继续下一个任务 print(\n【所有子任务处理完毕】) final_output self.memory.get_all_outputs() # 从共享记忆中组装最终产物 return final_output def _log_message(self, sender, recipient, content): 记录消息到总线和日志 msg {sender: sender, recipient: recipient, content: content[:500]} # 截断以便日志 self.message_bus.append(msg) # 可以在这里实现更复杂的路由逻辑比如根据recipient将消息放入对应队列3.3 共享记忆与工作区实现共享记忆可以用一个简单的类来模拟它本质上是一个结构化的字典并持久化到文件或数据库。# core/memory.py import json import os from datetime import datetime class SharedMemory: def __init__(self, project_id: str): self.project_id project_id self.data { project_id: project_id, created_at: datetime.now().isoformat(), initial_task: , plan: , artifacts: {}, # 存储文件路径或代码内容 task_status: {}, # 记录每个子任务状态 message_history: [] # 完整的结构化消息历史 } self.workspace_dir f./workspace/{project_id} os.makedirs(self.workspace_dir, exist_okTrue) def set(self, key: str, value: Any): 设置一个值到内存中。对于‘artifacts’可以特殊处理为文件。 if key.startswith(artifact_): # 假设value是代码保存为文件 filename key.replace(artifact_, ) .py filepath os.path.join(self.workspace_dir, filename) with open(filepath, w, encodingutf-8) as f: f.write(str(value)) self.data[artifacts][filename] filepath else: # 使用点号表示法支持嵌套字典这里简化处理 keys key.split(.) d self.data for k in keys[:-1]: d d.setdefault(k, {}) d[keys[-1]] value def get(self, key: str, defaultNone): 从内存中获取值 keys key.split(.) d self.data for k in keys: if isinstance(d, dict) and k in d: d d[k] else: return default return d def add_message(self, message: Dict): 添加一条消息到历史 self.data[message_history].append({ **message, timestamp: datetime.now().isoformat() }) def save(self): 将内存状态保存到磁盘 state_file os.path.join(self.workspace_dir, memory_state.json) with open(state_file, w, encodingutf-8) as f: json.dump(self.data, f, indent2, ensure_asciiFalse) def get_all_outputs(self): 获取所有产物用于最终组装 outputs { plan: self.data.get(plan), artifacts: [] } for filename, path in self.data.get(artifacts, {}).items(): try: with open(path, r, encodingutf-8) as f: content f.read() outputs[artifacts].append({name: filename, content: content}) except: pass return outputs3.4 主程序入口与配置最后我们将所有部分组装起来。# main.py from core.agent import Agent from core.orchestrator import SequentialOrchestrator from core.memory import SharedMemory def main(): # 1. 定义智能体角色 agents_config { planner: Agent( nameplanner, role你是一个资深软件架构师和产品经理。擅长将模糊的需求分解为清晰、具体、可执行的技术子任务。你的输出必须是结构化的列表每个子任务应该独立且可被开发者直接理解。避免使用模糊的词汇。 ), executor: Agent( nameexecutor, role你是一名全栈软件开发工程师精通Python、JavaScript、HTML/CSS。你根据具体的任务描述编写简洁、高效、符合最佳实践的代码。你总是输出完整的、可运行的代码块并附上必要的注释。 ), reviewer: Agent( namereviewer, role你是一个严谨的代码审查员和安全专家。你的工作是找出代码中的bug、安全漏洞、性能问题以及是否符合任务要求。你的反馈必须具体、可操作明确指出问题所在和修改建议。如果代码完全符合要求请明确说‘审查通过’。 ) } # 2. 初始化共享记忆 import uuid project_id fproject_{uuid.uuid4().hex[:8]} shared_memory SharedMemory(project_id) # 3. 初始化编排器 orchestrator SequentialOrchestrator(agents_config, shared_memory) # 4. 运行任务 user_task 创建一个简单的网页上面有一个按钮点击后能显示当前日期和时间。 final_result orchestrator.run(user_task) # 5. 保存并展示结果 shared_memory.save() print(\n *50) print(任务完成输出如下) print(f项目ID: {project_id}) print(f工作区目录: {shared_memory.workspace_dir}) if final_result.get(artifacts): print(\n生成的代码文件) for artifact in final_result[artifacts]: print(f- {artifact[name]}) # 可以选择性地打印部分内容 # print(artifact[content][:200] ...) if __name__ __main__: main()运行这个程序你会看到控制台打印出规划、执行、审查的各个步骤并在./workspace/目录下生成一个包含所有代码文件和内存状态JSON的项目文件夹。这就是一个最基础的多智能体协作系统的雏形。4. 高级特性与扩展方向基础框架跑通后我们可以探索更高级的特性让这个系统更强大、更实用。4.1 动态智能体创建与工具调用一个强大的框架应该能根据任务需要动态创建具有特定技能的智能体而不是预先固定几个。例如当规划者发现任务需要“数据库设计”时它可以请求编排器“创建一个数据库专家智能体”。这需要框架具备智能体工厂模式。更关键的是智能体不能只“说”不“做”。它们需要能调用外部工具和API这是将LLM的认知能力转化为实际行动的关键。这通过“工具调用”或“函数调用”功能实现。实现思路为智能体类扩展一个tools属性这是一个工具函数列表。在调用LLM时将工具的描述名称、功能、参数作为一部分信息发送给模型。模型在回复中可能会指定它想调用哪个工具以及参数。框架截获这个请求在本地执行对应的工具函数如读写文件、执行Shell命令、调用Web API、查询数据库并将结果返回给模型让它基于结果继续思考。# 示例为执行者智能体添加文件读写工具 def write_file(filepath: str, content: str): 将内容写入指定文件路径 os.makedirs(os.path.dirname(filepath), exist_okTrue) with open(filepath, w, encodingutf-8) as f: f.write(content) return f文件已成功写入{filepath} def read_file(filepath: str): 读取指定文件路径的内容 try: with open(filepath, r, encodingutf-8) as f: return f.read() except FileNotFoundError: return f错误文件 {filepath} 不存在。 # 将这些工具绑定到智能体 executor.tools [ { type: function, function: { name: write_file, description: 将给定的内容写入指定路径的文件。如果目录不存在则创建。, parameters: {...} # 详细的参数JSON Schema } }, { type: function, function: { name: read_file, description: 读取指定路径的文件内容。, parameters: {...} } } ] # 对应的函数实现映射 executor.tool_functions { write_file: write_file, read_file: read_file }这样当执行者智能体需要创建一个index.html文件时它可以直接在回复中请求调用write_file工具框架会自动执行并返回结果智能体再基于“文件已创建”这个事实进行下一步操作。这极大地扩展了智能体的行动边界。4.2 人机交互与干预点全自动流程虽然酷但在实际应用中风险很高。一个成熟的框架必须设计人机交互节点允许人类在关键时刻进行监督、指导和纠正。关键的干预点包括任务规划确认在规划者输出分解方案后暂停流程将计划呈现给用户确认或修改。用户可以说“合并前两个任务”或“优先处理第三个”。关键决策点当智能体之间出现无法解决的分歧时例如审查者坚持认为有安全漏洞而执行者认为没有流程应暂停将双方论点和代码提交给用户仲裁。产物交付前在所有自动生成的代码合并后、执行部署或运行命令前应提供一个最终的审查和确认步骤。错误处理与超时当某个环节连续失败或超时时应自动暂停并告警等待人工介入。实现上可以在编排器中设置检查点。在这些检查点编排器不是调用下一个智能体而是将当前状态如规划方案、争议内容格式化后输出到控制台、日志文件或一个简单的Web界面并等待用户的输入通过命令行、API或界面。用户的输入会被作为特殊消息注入到流程中指导后续步骤。4.3 性能优化与成本控制使用商业LLM API如GPT-4成本不菲多智能体间频繁对话会迅速消耗token。优化成本至关重要。1. 模型分层使用规划/协调使用能力最强但最贵的模型如GPT-4因为需要深度理解和复杂规划。执行对于标准的代码生成可以使用能力稍弱但性价比高的模型如GPT-3.5-Turbo、Claude Haiku。审查审查逻辑相对固定可以尝试使用更小、更快的模型甚至结合静态分析工具如linters、安全扫描工具来部分替代LLM调用。2. 上下文精炼如前所述不是把所有历史都塞进提示词。需要开发一个“上下文精炼器”模块它的职责是针对当前要调用的智能体和当前任务从共享记忆和消息历史中提取出最相关、最精简的信息组装成提示词。这可以用向量数据库存储历史消息的Embedding结合相似度搜索来实现。3. 异步与并行执行对于独立的子任务编排器应该能够并行启动多个执行者智能体而不是傻傻地排队。这需要引入异步编程asyncio和任务队列。同时要小心管理并行的token消耗避免短时间内发起大量API请求导致速率限制或成本飙升。4. 缓存与记忆对于相似的子任务或重复性操作可以将LLM的响应缓存起来。例如如果多个项目都需要“创建基本的Flask应用结构”那么第一次生成的结果可以缓存后续直接复用或稍作修改避免重复调用API。5. 常见问题、排查技巧与避坑指南在实际搭建和运行多智能体系统时你会遇到各种各样的问题。以下是我从实践中总结的一些典型问题和解决方案。5.1 智能体“跑偏”或陷入循环问题现象智能体之间的对话开始偏离主题或者在一个“执行-审查”循环中来回修改永远无法通过。原因1提示词不够精确。规划者的任务分解太模糊导致执行者无所适从审查者的标准不明确导致它总是能挑出毛病。解决精细化提示词。给规划者提供“任务分解模板”给审查者提供明确的检查清单如“必须检查1. 语法错误2. 导入缺失3. 安全函数使用4. 是否符合任务描述第X条”。原因2缺乏全局上下文。执行者只看到了当前子任务忘记了整体目标导致代码实现孤立。解决在每次调用执行者时除了当前子任务描述还要附上“整体项目目标”和“已完成的相邻子任务”的摘要。原因3审查者过于严苛或存在“幻觉”。LLM有时会误报问题。解决引入“仲裁机制”。设置一个最大迭代次数如3次。如果超过次数仍未通过触发“升级”流程将代码、所有历史反馈和任务描述打包发送给一个更高级的“首席架构师”智能体或用GPT-4做最终裁定或者直接暂停等待人工介入。5.2 上下文长度爆炸与信息丢失问题现象随着对话进行提示词越来越长达到模型上限后被截断导致智能体“失忆”做出前后矛盾的决策。解决实施严格的上下文管理策略。摘要化每完成一个主要阶段如规划完成、一个子任务闭环让一个专门的“摘要智能体”或协调者将之前冗长的对话总结成一段精炼的“项目当前状态摘要”存入共享记忆。后续对话主要引用这个摘要而非全部历史。向量化检索将每条消息、每个生成的代码片段都生成向量嵌入Embedding存入向量数据库如Chroma, Weaviate。当需要为某个智能体构建上下文时根据当前任务描述去向量库中检索最相关的N条历史信息而不是按时间顺序取最近N条。分层记忆区分“工作记忆”当前活跃任务的相关信息和“长期记忆”已归档的任务结果、核心决策。定期将已完成部分从工作记忆转移到长期记忆。5.3 工具调用的安全性与可靠性问题现象智能体尝试执行危险命令如rm -rf /或调用不可靠的外部API。解决实现一个安全的“工具执行沙箱”。权限白名单严格定义智能体可以调用的工具列表。禁止直接执行任意Shell命令。如果必须则通过封装好的特定函数如run_safe_shell_command(cmd: str)在该函数内部对命令进行严格过滤和匹配只允许预定义的安全命令集。输入验证与净化对所有工具函数的输入参数进行严格的验证和净化防止注入攻击。资源隔离在Docker容器或沙箱环境中运行整个智能体系统限制其网络访问和文件系统权限。人工确认对于高风险操作如文件删除、安装系统包、访问生产数据库设计为必须经过人工确认环节。5.4 调试与监控困难问题现象流程复杂出错时难以定位是哪个智能体、哪条消息、哪个决策导致了问题。解决建立完善的日志和追踪系统。结构化日志记录每一次LLM调用输入提示词、输出结果、token消耗、耗时、每一次工具调用、每一次状态变更。为每个任务和会话分配唯一的trace_id将所有相关日志串联起来。可视化工作流将编排器的工作流状态实时输出到一个简单的仪表盘可以看到当前处于哪个阶段、哪个智能体在活动、子任务状态等。回放与复盘利用共享记忆中保存的完整消息历史可以实现任务执行的“全程回放”方便复盘问题根源。实操心得从小处着手逐步迭代。不要一开始就追求一个全自动、万能的多智能体系统。从一个非常具体的、边界清晰的简单任务开始比如“写一个Python脚本来重命名某个文件夹下的所有图片文件”实现两个智能体规划执行的协作。成功运行后再加入审查者。然后再尝试更复杂的任务逐步引入工具调用、循环迭代、人工干预等高级特性。每增加一个功能都进行充分的测试。这个领域目前仍在快速演进保持框架的模块化和可扩展性比追求一步到位的“完美设计”更重要。最终你会发现设计和管理好这些“数字员工”本身就是一个极其有趣且充满挑战的软件工程问题。