1. 从单兵作战到团队协作为什么你的Agent需要队友如果你已经跟着这个系列从零开始一步步搭建了自己的智能体那你现在手里应该有一个能独立完成特定任务的“单兵”了。它能帮你写邮件、查资料、甚至写点简单的代码感觉还不错对吧但不知道你有没有遇到过这样的场景你想让Agent帮你规划一次旅行它需要先查机票酒店、再根据你的预算和偏好做行程、最后还得生成一份图文并茂的攻略。你会发现让一个Agent干所有这些事要么它做得马马虎虎要么你得把指令拆得极其细致过程繁琐得还不如自己动手。这就是单智能体的瓶颈。一个Agent再强大其能力也是聚焦的。就像你不能指望一个顶级的数据库专家同时也是一个顶尖的UI设计师。Agent Team智能体团队协作要解决的就是这个“全才”难题。它的核心思想是“专业的人做专业的事”通过创建多个具备不同专长的智能体并设计一套高效的协作机制让它们像一支训练有素的团队一样共同完成复杂的、多步骤的任务。这不仅仅是“多开几个窗口”那么简单。真正的团队协作涉及到角色定义、任务分解、沟通协议、冲突解决和结果聚合等一系列复杂问题。想象一下你作为项目经理手下有程序员、测试员和产品经理你需要清晰地传达需求、协调他们的工作顺序、处理他们之间的意见分歧并最终整合出一份可交付的成果。构建一个Agent Team你就是在扮演这个“项目经理”的角色只不过你的“员工”是代码和模型。在接下来的内容里我不会空谈概念而是会带你从零开始设计并实现一个具备实用价值的Agent Team。我们会聚焦于一个经典且能体现协作价值的场景内容创作团队。这个团队将由三个智能体组成一个“策划编辑”、一个“文案写手”、一个“校对审核”。通过这个具体案例你将彻底理解团队协作中的每一个技术细节和设计考量。2. 团队蓝图定义角色、能力与协作流程在写第一行代码之前我们必须先把团队的设计图纸画清楚。盲目地创建几个Agent然后让它们互相喊话只会得到一堆混乱的日志和错误的结果。2.1 核心角色设计与能力边界我们的内容创作团队由三个角色构成每个角色都有明确的职责和“技能树”策划编辑 (Planner Agent)核心职责理解用户最原始、可能模糊的需求将其转化为清晰、可执行的内容大纲和创作指令。它是团队的“大脑”和“项目经理”。能力定义需求分析能通过多轮对话澄清模糊需求例如用户说“写一篇关于Python的文章”策划编辑会追问“面向新手还是进阶”、“侧重理论还是实战”、“需要包含代码示例吗”。大纲生成根据明确的需求生成结构化内容大纲包括标题、各级小标题、每个部分的核心论点、建议的数据或案例来源。任务分发将大纲拆解成具体的写作任务并附上详细的写作要求风格、语气、字数、关键词等交给文案写手。技术实现倾向它需要较强的逻辑推理和结构化思维能力。在底层模型选择上可以优先考虑那些在“指令遵循”和“逻辑链”表现较好的模型。文案写手 (Writer Agent)核心职责接收策划编辑的详细指令负责具体段落的撰写将大纲填充为生动、流畅的正文内容。它是团队的“双手”。能力定义风格化写作能根据指令灵活切换文风如科技博客的严谨、营销文案的活泼、教程的亲切。信息整合能够基于策划编辑提供的要点或自行搜索如果团队具备此功能的信息组织成连贯的文字。初稿生成产出高质量的文本初稿。技术实现倾向它需要强大的文本生成和风格模仿能力。可以选择在创意写作或特定领域文本生成上微调过的模型。校对审核 (Reviewer Agent)核心职责对文案写手产出的初稿进行审核确保内容质量。它是团队的“质量守门员”。能力定义事实核查检查文中的关键数据、引用、技术细节是否准确这可能需要接入检索或知识库。逻辑与流畅度检查检查文章逻辑是否自洽段落衔接是否自然有无语病或啰嗦之处。风格与规范符合度检查确保文章符合策划编辑最初设定的风格和格式要求。修改建议不仅指出问题还能提供具体的修改建议或直接输出修改后的段落。技术实现倾向它需要敏锐的“找茬”能力和一定的批判性思维。可以选用在文本理解、对比分析上表现更强的模型。注意这里有一个关键设计原则——能力隔离。策划编辑不应该去干写稿的活校对审核也不应该擅自重写大纲。清晰的能力边界是减少团队内部混乱和冲突的基础。在实际编码中我们会通过系统提示词System Prompt来严格约束每个Agent的行为范围。2.2 设计团队协作的工作流角色定义好了它们该如何配合我们设计一个单向流水线式工作流这是最简单也是最稳定的协作模式。用户输入 - [策划编辑] - (大纲写作指令) - [文案写手] - (内容初稿) - [校对审核] - (最终稿) - 输出给用户这个流程看似简单但暗含了几个必须处理的技术环节信息传递格式Agent之间不能靠“心领神会”。我们必须定义一套结构化的数据格式来传递任务和结果。例如策划编辑传递给文案写手的不应该是一段自然语言描述而是一个JSON对象{ task_id: write_section_1, section_title: Python列表解析式的优势, key_points: [语法简洁, 执行效率通常比for循环高, 可读性强的条件筛选], writing_style: 技术教程亲切易懂带代码示例, word_count_limit: 300, keywords: [列表解析, for循环, 性能对比] }结构化数据能极大减少后续Agent的解析歧义。状态管理与回调当文案写手完成自己的部分后如何通知校对审核我们需要一个“协调者”Orchestrator或“工作流引擎”来管理整个流程的状态。这个协调者负责按顺序触发Agent并将上一个Agent的输出作为下一个Agent的输入。它可以是另一个简单的控制程序也可以将这个逻辑嵌入到主调用流程中。错误处理与超时如果文案写手生成的内容完全跑题了怎么办如果某个Agent处理超时了怎么办工作流中必须设计错误处理机制。例如校对审核如果发现初稿质量极差无法修改应该有能力将任务“打回”给文案写手或甚至策划编辑并附上错误原因。这涉及到工作流中的“循环”或“条件分支”设计。在初版实现中我们可以先实现最基础的线性流程确保它能跑通。在后续迭代中再加入错误处理和更复杂的路由逻辑。3. 搭建舞台实现多智能体系统的通信骨架现在我们进入实战环节。假设我们已经有了三个独立的、基于大语言模型API例如 OpenAI GPT, Claude, 或本地部署的 Llama 等的智能体类。每个类都能通过generate_response(prompt)方法完成任务。现在的核心问题是如何让它们“对话”3.1 实现一个简单的消息总线Message Bus我们不需要一开始就引入复杂的消息队列如RabbitMQ、Kafka。一个轻量级的、内存中的消息总线足以支撑我们初版的团队协作。它的核心功能是接收来自某个Agent的消息并根据预定的规则将其投递给下一个Agent。我们可以用一个Python类来实现class SimpleMessageBus: def __init__(self): self.agents {} # 存储注册的Agentkey为agent_namevalue为agent实例 self.workflow_definitions {} # 存储工作流定义 def register_agent(self, agent_name, agent_instance): 注册一个智能体到总线上 self.agents[agent_name] agent_instance print(f[Bus] Agent {agent_name} 已注册。) def define_workflow(self, workflow_name, steps): 定义一个工作流。 steps: 列表例如 [(planner, writer), (writer, reviewer)] 表示 planner 的输出给 writerwriter 的输出给 reviewer。 self.workflow_definitions[workflow_name] steps print(f[Bus] 工作流 {workflow_name} 已定义: {steps}) def start_workflow(self, workflow_name, initial_input, initial_senderuser): 启动一个工作流。 initial_input: 初始输入用户需求。 initial_sender: 初始发送者用于找到工作流第一步的接收者。 if workflow_name not in self.workflow_definitions: raise ValueError(f未定义的工作流: {workflow_name}) workflow_steps self.workflow_definitions[workflow_name] current_output initial_input current_sender initial_sender for step in workflow_steps: sender, receiver step # 只有当上一步的发送者匹配当前步骤的发送者时才执行这一步 # 这是一种简单的线性驱动方式。更复杂的需要状态机。 if sender current_sender: print(f[Bus] 将任务从 {sender} 传递给 {receiver}。) if receiver not in self.agents: raise ValueError(f未注册的Agent: {receiver}) receiver_agent self.agents[receiver] # 这里可以设计更复杂的消息封装这里简单传递文本 # 实际应用中current_output 应该是一个包含元数据的消息对象 try: current_output receiver_agent.generate_response(current_output) current_sender receiver print(f[Bus] {receiver} 处理完成。) except Exception as e: print(f[Bus] Agent {receiver} 处理失败: {e}) # 错误处理可以终止工作流或进入错误处理流程 break else: # 如果发送者不匹配说明工作流定义或执行顺序有误 print(f[Bus] 警告期望的发送者 {sender} 与当前发送者 {current_sender} 不匹配。跳过此步骤。) # 可以根据策略决定是否继续或终止 continue print(f[Bus] 工作流 {workflow_name} 执行完毕。) return current_output # 返回最终输出这个SimpleMessageBus虽然简陋但它明确了几个关键概念Agent注册每个Agent需要在总线上“挂牌”总线才知道它的存在。工作流定义协作的路径是预先定义好的define_workflow。这给了我们控制权。顺序执行总线按定义好的步骤依次调用Agent并传递数据。3.2 封装智能体赋予角色与记忆之前我们可能只有一个通用的Agent类。现在我们需要为每个角色创建特定的子类或者至少用不同的系统提示词System Prompt来初始化它们。这是实现“能力隔离”的关键。以策划编辑PlannerAgent为例class PlannerAgent: def __init__(self, name, llm_client): self.name name self.llm llm_client # 核心定义角色的系统提示词 self.system_prompt 你是一个专业的策划编辑。你的任务是将用户模糊的内容需求转化为清晰、可执行的内容大纲和写作指令。 请遵循以下步骤 1. **需求澄清**如果用户需求不够具体提出最多3个关键问题来澄清例如目标读者、文章深度、风格、字数、核心关键词。 2. **大纲生成**基于明确的需求生成一份详细的内容大纲。大纲应包含文章标题、导语、至少3个主要章节每个章节需有标题和2-4个核心论点。 3. **任务拆分**将大纲拆分成独立的写作任务。为每个任务创建一个JSON对象包含以下字段 - task_id: 唯一任务标识如 section_1 - section_title: 章节标题 - key_points: 本章节必须涵盖的核心论点列表 - writing_style: 要求的写作风格 - word_count_limit: 字数建议 - notes: 给写手的其他特别说明 你的输出必须是纯JSON格式是一个包含clarification_questions如果需要、outline和writing_tasks三个键的字典。不要输出任何额外的解释或Markdown格式。 def generate_response(self, user_input): # 构建对话历史这里简单处理实际可能需要维护更长的记忆 messages [ {role: system, content: self.system_prompt}, {role: user, content: user_input} ] response self.llm.chat_completion(messages) # 这里需要解析响应确保它是有效的JSON。实际应用中需要健壮的解析和错误处理。 import json try: # 假设LLM返回的是纯JSON文本 plan json.loads(response) return plan except json.JSONDecodeError: # 如果解析失败可以记录日志并返回一个错误结构或者尝试修复 print(f[{self.name}] 响应不是有效JSON返回原始文本。) return {error: Invalid JSON response, raw_output: response}文案写手和校对审核的类结构类似但拥有完全不同的system_prompt用于约束其行为。例如校对审核的提示词会强调“你是一个严格的校对员你的工作是找出文本中的事实错误、逻辑矛盾、语病和风格不一致并提供具体的修改建议或修改后的文本。你的输出应该是一个包含issues问题列表和revised_text修订后的全文的JSON对象。”3.3 串联一切主程序与工作流执行最后我们写一个主程序把角色、总线和工作流串联起来def main(): # 1. 初始化LLM客户端这里用伪代码实际需替换成OpenAI、Anthropic等SDK调用 llm_client get_llm_client(modelgpt-4) # 示例 # 2. 创建三个具有不同角色的智能体 planner PlannerAgent(nameplanner, llm_clientllm_client) writer WriterAgent(namewriter, llm_clientllm_client) # 假设已定义 reviewer ReviewerAgent(namereviewer, llm_clientllm_client) # 假设已定义 # 3. 创建消息总线并注册智能体 bus SimpleMessageBus() bus.register_agent(planner, planner) bus.register_agent(writer, writer) bus.register_agent(reviewer, reviewer) # 4. 定义“内容创作”工作流 bus.define_workflow(content_creation, [(user, planner), (planner, writer), (writer, reviewer)]) # 5. 获取用户输入并启动工作流 user_request input(请输入你的内容需求) final_result bus.start_workflow(workflow_namecontent_creation, initial_inputuser_request, initial_senderuser) # 6. 输出最终结果 print(\n *50) print(【最终生成内容】) # final_result 可能是校对审核返回的JSON从中提取修订后的文本 if isinstance(final_result, dict) and revised_text in final_result: print(final_result[revised_text]) else: print(final_result) # 或其他处理方式运行这个程序你就看到了一个最基础的、三个智能体线性协作完成内容创作的全过程。每个Agent各司其职通过消息总线进行数据交接。4. 超越流水线处理冲突、循环与动态路由基础流水线跑通了但现实世界的团队协作远非一帆风顺。文案写手可能对某个论点理解有偏差校对审核可能认为某个部分需要重写而非修改。这就需要我们为团队引入更高级的“管理”机制。4.1 引入评审与驳回机制在校对审核Agent的逻辑里我们不能只让它直接修改。它应该先评估初稿的质量。我们可以修改其system_prompt和输出格式你的输出必须是以下两种之一 1. **审核通过**如果文章质量合格仅需少量修改。输出格式{status: approved_with_minor_changes, revised_text: 这里是修改后的全文} 2. **需要重大修改**如果文章存在严重问题如核心论点错误、结构混乱、严重偏离要求。输出格式{status: needs_major_revision, feedback: 详细的修改意见指出具体问题和修改方向, problematic_section_task_id: section_2}然后我们的消息总线和工作流逻辑就需要升级。当校对审核返回needs_major_revision时总线不能直接结束。它需要根据problematic_section_task_id将feedback和原始任务信息重新发送给文案写手进行重写。这就形成了一个循环。实现这个功能我们需要将简单的步骤列表升级为一个状态机State Machine。每个任务如“撰写section_2”都有一个状态如“待处理”、“写作中”、“待审核”、“审核通过”、“需重写”。总线根据当前任务的状态和Agent返回的结果来决定下一个状态是什么以及下一个执行者是谁。4.2 实现一个简单的任务状态机我们可以设计一个Task类和一个WorkflowEngine类来管理class Task: def __init__(self, task_id, description, assigned_agentNone): self.task_id task_id self.description description # 来自策划编辑的JSON self.status pending # pending, executing, waiting_review, approved, needs_rework self.assigned_agent assigned_agent self.result None self.feedback None class WorkflowEngine: def __init__(self, message_bus): self.bus message_bus self.tasks {} self.task_history [] # 记录任务状态变迁 def create_tasks_from_plan(self, plan_data): 从策划编辑的规划中创建一系列写作任务 writing_tasks plan_data.get(writing_tasks, []) for task_desc in writing_tasks: task_id task_desc[task_id] self.tasks[task_id] Task(task_idtask_id, descriptiontask_desc, assigned_agentwriter) self.task_history.append((task_id, created, pending)) def process(self): 主处理循环直到所有任务达到终态approved或手动终止 while not self.all_tasks_finalized(): for task_id, task in self.tasks.items(): if task.status pending: # 分配给写手 self._assign_task(task, writer) elif task.status waiting_review: # 分配给审核员 self._assign_task(task, reviewer) elif task.status needs_rework: # 重新分配给写手并附上反馈 self._assign_task(task, writer, feedbacktask.feedback) # ... 处理其他状态 def _assign_task(self, task, agent_name, feedbackNone): 将任务分配给指定Agent并更新状态 print(f[Engine] 将任务 {task.task_id} 分配给 {agent_name}。) task.status executing task.assigned_agent agent_name # 构建给Agent的输入信息 input_data task.description if feedback: input_data[feedback_from_reviewer] feedback # 通过消息总线发送这里简化实际总线需支持定向发送 result self.bus.send_to_agent(agent_name, input_data) # 处理Agent返回的结果 self._handle_agent_result(task, agent_name, result) def _handle_agent_result(self, task, agent_name, result): if agent_name writer: task.result result task.status waiting_review elif agent_name reviewer: if result.get(status) approved_with_minor_changes: task.status approved task.result result.get(revised_text, task.result) # 更新为修订稿 elif result.get(status) needs_major_revision: task.status needs_rework task.feedback result.get(feedback) # 这里可以设置一个重试计数器避免无限循环 # ... 记录历史 self.task_history.append((task.task_id, agent_name, task.status))这个引擎比简单的线性总线复杂得多但它能处理任务依赖、状态流转和条件分支如驳回重写。这是构建健壮Agent Team的核心。4.3 动态路由与管理者智能体更进一步我们甚至可以引入一个管理者ManagerAgent。它的职责不是完成具体工作而是监督和协调。动态任务分配策划编辑生成了10个写作任务但其中3个是技术性极强的。管理者可以识别这一点并将这3个任务路由给一个专门的“技术写手”Agent而不是通用的“文案写手”。冲突仲裁文案写手和校对审核对某处修改争执不下。管理者可以介入根据预设的规则如“事实准确性优先于文采”或自行判断做出最终决策。流程优化管理者可以监控任务执行时间如果发现某个Agent总是超时它可以决定将任务重新分配给另一个同类型的备用Agent。实现管理者Agent需要赋予它更高的权限和更全面的上下文信息它的提示词可能像这样“你是团队的项目经理。你负责接收所有任务和中间结果并根据每个成员的专长和当前负载动态分配任务。当成员间产出冲突时你负责做出最终裁定。你的目标是保证项目质量和效率。”5. 实战中的挑战与优化策略当你真正运行起一个多智能体系统时一系列在单智能体场景下不明显的问题会集中爆发。这里分享几个我踩过的坑和对应的解决思路。5.1 上下文长度与信息衰减这是最棘手的问题之一。策划编辑生成的大纲和指令可能很长文案写手生成的初稿更长当这些内容全部作为历史对话传递给校对审核时很容易超过模型的最大上下文长度。解决方案摘要与提炼不要传递完整的原始对话历史。在每个交接点强制要求输出Agent对当前结果生成一个结构化摘要。例如文案写手在提交初稿时必须同时提交一个summary字段用三五句话概括本部分核心内容。下游Agent如校对审核在需要全局判断时优先参考摘要。向量化存储与检索将所有中间产出大纲、初稿、修改意见存入向量数据库。当任何一个Agent需要历史上下文时不传递全文而是根据当前任务内容从向量库中检索最相关的几个片段。这能极大节省上下文窗口并让信息获取更精准。分层处理对于校对审核可以设计两阶段。第一阶段只让它基于“章节摘要”和“写作要求”进行高阶评审结构、核心论点。第二阶段对于需要细改的章节再传入该章节的完整文本进行逐字校对。5.2 一致性维护多个智能体各自为政很容易导致最终成果风格不一、术语前后矛盾。例如策划编辑要求用“人工智能”文案写手可能写成“AI”校对审核又可能改回“人工智能”。解决方案创建共享知识库风格指南在项目启动时由策划编辑或一个专门的“初始化Agent”生成一份本项目专用的风格指南。这份指南以系统提示词或独立文件的形式分发给团队每一个成员。内容应包括核心术语表如统一使用“Python”而非“python”、目标读者定位、文章整体语气、禁止使用的词汇等。在消息中嵌入一致性指令在任务传递的JSON中除了具体内容永远包含一个style_guide或consistency_rules字段里面是本次任务需要特别注意的一致性要点。最终一致性检查在团队输出最终结果前增加一个专门的“一致性检查”Agent。它的任务不是修改内容而是通读全文标记出所有风格、术语、格式不一致的地方生成一个报告交由管理者Agent或文案写手进行最终统一修订。5.3 错误传播与系统韧性在流水线中上游的错误会被放大并传递给下游。如果策划编辑错误理解了需求那么后续所有工作都是徒劳。解决方案关键节点人工确认Human-in-the-loop在战略性的节点设置“检查点”。例如在策划编辑输出大纲后不自动执行而是将大纲呈现给用户或一个监督员确认。确认无误后再启动后续的自动化流程。这牺牲了一点全自动的效率但极大地提升了结果的可靠性。多方案生成与选择对于关键环节如大纲生成可以让Agent生成2-3个备选方案然后由一个“评估者”Agent或人工选择最优的一个再进入下一环节。这增加了计算成本但降低了因单次生成不佳而导致全盘失败的风险。完备的日志与回滚系统必须记录每个Agent的完整输入和输出。当最终结果不满意时可以回溯整个决策链精准定位是哪个环节出了问题。基于日志可以实现部分流程的回滚和重试。构建一个高效、稳定的Agent Team其复杂性远超单个智能体。它更像是在设计一个微型的软件系统或组织架构。你需要考虑通信协议、状态管理、错误处理、资源分配等一系列问题。但一旦成功其所能释放的生产力和所能解决的复杂问题将是单智能体无法比拟的。从今天设计的这个内容创作小团队出发你可以将这套模式扩展到客服系统、代码开发、数据分析等无限场景。记住好的团队协作始于清晰的角色定义成于稳健的通信机制终于对一致性和错误的周密管理。 从单智能体到多智能体协作：构建高效Agent Team的实战指南
1. 从单兵作战到团队协作为什么你的Agent需要队友如果你已经跟着这个系列从零开始一步步搭建了自己的智能体那你现在手里应该有一个能独立完成特定任务的“单兵”了。它能帮你写邮件、查资料、甚至写点简单的代码感觉还不错对吧但不知道你有没有遇到过这样的场景你想让Agent帮你规划一次旅行它需要先查机票酒店、再根据你的预算和偏好做行程、最后还得生成一份图文并茂的攻略。你会发现让一个Agent干所有这些事要么它做得马马虎虎要么你得把指令拆得极其细致过程繁琐得还不如自己动手。这就是单智能体的瓶颈。一个Agent再强大其能力也是聚焦的。就像你不能指望一个顶级的数据库专家同时也是一个顶尖的UI设计师。Agent Team智能体团队协作要解决的就是这个“全才”难题。它的核心思想是“专业的人做专业的事”通过创建多个具备不同专长的智能体并设计一套高效的协作机制让它们像一支训练有素的团队一样共同完成复杂的、多步骤的任务。这不仅仅是“多开几个窗口”那么简单。真正的团队协作涉及到角色定义、任务分解、沟通协议、冲突解决和结果聚合等一系列复杂问题。想象一下你作为项目经理手下有程序员、测试员和产品经理你需要清晰地传达需求、协调他们的工作顺序、处理他们之间的意见分歧并最终整合出一份可交付的成果。构建一个Agent Team你就是在扮演这个“项目经理”的角色只不过你的“员工”是代码和模型。在接下来的内容里我不会空谈概念而是会带你从零开始设计并实现一个具备实用价值的Agent Team。我们会聚焦于一个经典且能体现协作价值的场景内容创作团队。这个团队将由三个智能体组成一个“策划编辑”、一个“文案写手”、一个“校对审核”。通过这个具体案例你将彻底理解团队协作中的每一个技术细节和设计考量。2. 团队蓝图定义角色、能力与协作流程在写第一行代码之前我们必须先把团队的设计图纸画清楚。盲目地创建几个Agent然后让它们互相喊话只会得到一堆混乱的日志和错误的结果。2.1 核心角色设计与能力边界我们的内容创作团队由三个角色构成每个角色都有明确的职责和“技能树”策划编辑 (Planner Agent)核心职责理解用户最原始、可能模糊的需求将其转化为清晰、可执行的内容大纲和创作指令。它是团队的“大脑”和“项目经理”。能力定义需求分析能通过多轮对话澄清模糊需求例如用户说“写一篇关于Python的文章”策划编辑会追问“面向新手还是进阶”、“侧重理论还是实战”、“需要包含代码示例吗”。大纲生成根据明确的需求生成结构化内容大纲包括标题、各级小标题、每个部分的核心论点、建议的数据或案例来源。任务分发将大纲拆解成具体的写作任务并附上详细的写作要求风格、语气、字数、关键词等交给文案写手。技术实现倾向它需要较强的逻辑推理和结构化思维能力。在底层模型选择上可以优先考虑那些在“指令遵循”和“逻辑链”表现较好的模型。文案写手 (Writer Agent)核心职责接收策划编辑的详细指令负责具体段落的撰写将大纲填充为生动、流畅的正文内容。它是团队的“双手”。能力定义风格化写作能根据指令灵活切换文风如科技博客的严谨、营销文案的活泼、教程的亲切。信息整合能够基于策划编辑提供的要点或自行搜索如果团队具备此功能的信息组织成连贯的文字。初稿生成产出高质量的文本初稿。技术实现倾向它需要强大的文本生成和风格模仿能力。可以选择在创意写作或特定领域文本生成上微调过的模型。校对审核 (Reviewer Agent)核心职责对文案写手产出的初稿进行审核确保内容质量。它是团队的“质量守门员”。能力定义事实核查检查文中的关键数据、引用、技术细节是否准确这可能需要接入检索或知识库。逻辑与流畅度检查检查文章逻辑是否自洽段落衔接是否自然有无语病或啰嗦之处。风格与规范符合度检查确保文章符合策划编辑最初设定的风格和格式要求。修改建议不仅指出问题还能提供具体的修改建议或直接输出修改后的段落。技术实现倾向它需要敏锐的“找茬”能力和一定的批判性思维。可以选用在文本理解、对比分析上表现更强的模型。注意这里有一个关键设计原则——能力隔离。策划编辑不应该去干写稿的活校对审核也不应该擅自重写大纲。清晰的能力边界是减少团队内部混乱和冲突的基础。在实际编码中我们会通过系统提示词System Prompt来严格约束每个Agent的行为范围。2.2 设计团队协作的工作流角色定义好了它们该如何配合我们设计一个单向流水线式工作流这是最简单也是最稳定的协作模式。用户输入 - [策划编辑] - (大纲写作指令) - [文案写手] - (内容初稿) - [校对审核] - (最终稿) - 输出给用户这个流程看似简单但暗含了几个必须处理的技术环节信息传递格式Agent之间不能靠“心领神会”。我们必须定义一套结构化的数据格式来传递任务和结果。例如策划编辑传递给文案写手的不应该是一段自然语言描述而是一个JSON对象{ task_id: write_section_1, section_title: Python列表解析式的优势, key_points: [语法简洁, 执行效率通常比for循环高, 可读性强的条件筛选], writing_style: 技术教程亲切易懂带代码示例, word_count_limit: 300, keywords: [列表解析, for循环, 性能对比] }结构化数据能极大减少后续Agent的解析歧义。状态管理与回调当文案写手完成自己的部分后如何通知校对审核我们需要一个“协调者”Orchestrator或“工作流引擎”来管理整个流程的状态。这个协调者负责按顺序触发Agent并将上一个Agent的输出作为下一个Agent的输入。它可以是另一个简单的控制程序也可以将这个逻辑嵌入到主调用流程中。错误处理与超时如果文案写手生成的内容完全跑题了怎么办如果某个Agent处理超时了怎么办工作流中必须设计错误处理机制。例如校对审核如果发现初稿质量极差无法修改应该有能力将任务“打回”给文案写手或甚至策划编辑并附上错误原因。这涉及到工作流中的“循环”或“条件分支”设计。在初版实现中我们可以先实现最基础的线性流程确保它能跑通。在后续迭代中再加入错误处理和更复杂的路由逻辑。3. 搭建舞台实现多智能体系统的通信骨架现在我们进入实战环节。假设我们已经有了三个独立的、基于大语言模型API例如 OpenAI GPT, Claude, 或本地部署的 Llama 等的智能体类。每个类都能通过generate_response(prompt)方法完成任务。现在的核心问题是如何让它们“对话”3.1 实现一个简单的消息总线Message Bus我们不需要一开始就引入复杂的消息队列如RabbitMQ、Kafka。一个轻量级的、内存中的消息总线足以支撑我们初版的团队协作。它的核心功能是接收来自某个Agent的消息并根据预定的规则将其投递给下一个Agent。我们可以用一个Python类来实现class SimpleMessageBus: def __init__(self): self.agents {} # 存储注册的Agentkey为agent_namevalue为agent实例 self.workflow_definitions {} # 存储工作流定义 def register_agent(self, agent_name, agent_instance): 注册一个智能体到总线上 self.agents[agent_name] agent_instance print(f[Bus] Agent {agent_name} 已注册。) def define_workflow(self, workflow_name, steps): 定义一个工作流。 steps: 列表例如 [(planner, writer), (writer, reviewer)] 表示 planner 的输出给 writerwriter 的输出给 reviewer。 self.workflow_definitions[workflow_name] steps print(f[Bus] 工作流 {workflow_name} 已定义: {steps}) def start_workflow(self, workflow_name, initial_input, initial_senderuser): 启动一个工作流。 initial_input: 初始输入用户需求。 initial_sender: 初始发送者用于找到工作流第一步的接收者。 if workflow_name not in self.workflow_definitions: raise ValueError(f未定义的工作流: {workflow_name}) workflow_steps self.workflow_definitions[workflow_name] current_output initial_input current_sender initial_sender for step in workflow_steps: sender, receiver step # 只有当上一步的发送者匹配当前步骤的发送者时才执行这一步 # 这是一种简单的线性驱动方式。更复杂的需要状态机。 if sender current_sender: print(f[Bus] 将任务从 {sender} 传递给 {receiver}。) if receiver not in self.agents: raise ValueError(f未注册的Agent: {receiver}) receiver_agent self.agents[receiver] # 这里可以设计更复杂的消息封装这里简单传递文本 # 实际应用中current_output 应该是一个包含元数据的消息对象 try: current_output receiver_agent.generate_response(current_output) current_sender receiver print(f[Bus] {receiver} 处理完成。) except Exception as e: print(f[Bus] Agent {receiver} 处理失败: {e}) # 错误处理可以终止工作流或进入错误处理流程 break else: # 如果发送者不匹配说明工作流定义或执行顺序有误 print(f[Bus] 警告期望的发送者 {sender} 与当前发送者 {current_sender} 不匹配。跳过此步骤。) # 可以根据策略决定是否继续或终止 continue print(f[Bus] 工作流 {workflow_name} 执行完毕。) return current_output # 返回最终输出这个SimpleMessageBus虽然简陋但它明确了几个关键概念Agent注册每个Agent需要在总线上“挂牌”总线才知道它的存在。工作流定义协作的路径是预先定义好的define_workflow。这给了我们控制权。顺序执行总线按定义好的步骤依次调用Agent并传递数据。3.2 封装智能体赋予角色与记忆之前我们可能只有一个通用的Agent类。现在我们需要为每个角色创建特定的子类或者至少用不同的系统提示词System Prompt来初始化它们。这是实现“能力隔离”的关键。以策划编辑PlannerAgent为例class PlannerAgent: def __init__(self, name, llm_client): self.name name self.llm llm_client # 核心定义角色的系统提示词 self.system_prompt 你是一个专业的策划编辑。你的任务是将用户模糊的内容需求转化为清晰、可执行的内容大纲和写作指令。 请遵循以下步骤 1. **需求澄清**如果用户需求不够具体提出最多3个关键问题来澄清例如目标读者、文章深度、风格、字数、核心关键词。 2. **大纲生成**基于明确的需求生成一份详细的内容大纲。大纲应包含文章标题、导语、至少3个主要章节每个章节需有标题和2-4个核心论点。 3. **任务拆分**将大纲拆分成独立的写作任务。为每个任务创建一个JSON对象包含以下字段 - task_id: 唯一任务标识如 section_1 - section_title: 章节标题 - key_points: 本章节必须涵盖的核心论点列表 - writing_style: 要求的写作风格 - word_count_limit: 字数建议 - notes: 给写手的其他特别说明 你的输出必须是纯JSON格式是一个包含clarification_questions如果需要、outline和writing_tasks三个键的字典。不要输出任何额外的解释或Markdown格式。 def generate_response(self, user_input): # 构建对话历史这里简单处理实际可能需要维护更长的记忆 messages [ {role: system, content: self.system_prompt}, {role: user, content: user_input} ] response self.llm.chat_completion(messages) # 这里需要解析响应确保它是有效的JSON。实际应用中需要健壮的解析和错误处理。 import json try: # 假设LLM返回的是纯JSON文本 plan json.loads(response) return plan except json.JSONDecodeError: # 如果解析失败可以记录日志并返回一个错误结构或者尝试修复 print(f[{self.name}] 响应不是有效JSON返回原始文本。) return {error: Invalid JSON response, raw_output: response}文案写手和校对审核的类结构类似但拥有完全不同的system_prompt用于约束其行为。例如校对审核的提示词会强调“你是一个严格的校对员你的工作是找出文本中的事实错误、逻辑矛盾、语病和风格不一致并提供具体的修改建议或修改后的文本。你的输出应该是一个包含issues问题列表和revised_text修订后的全文的JSON对象。”3.3 串联一切主程序与工作流执行最后我们写一个主程序把角色、总线和工作流串联起来def main(): # 1. 初始化LLM客户端这里用伪代码实际需替换成OpenAI、Anthropic等SDK调用 llm_client get_llm_client(modelgpt-4) # 示例 # 2. 创建三个具有不同角色的智能体 planner PlannerAgent(nameplanner, llm_clientllm_client) writer WriterAgent(namewriter, llm_clientllm_client) # 假设已定义 reviewer ReviewerAgent(namereviewer, llm_clientllm_client) # 假设已定义 # 3. 创建消息总线并注册智能体 bus SimpleMessageBus() bus.register_agent(planner, planner) bus.register_agent(writer, writer) bus.register_agent(reviewer, reviewer) # 4. 定义“内容创作”工作流 bus.define_workflow(content_creation, [(user, planner), (planner, writer), (writer, reviewer)]) # 5. 获取用户输入并启动工作流 user_request input(请输入你的内容需求) final_result bus.start_workflow(workflow_namecontent_creation, initial_inputuser_request, initial_senderuser) # 6. 输出最终结果 print(\n *50) print(【最终生成内容】) # final_result 可能是校对审核返回的JSON从中提取修订后的文本 if isinstance(final_result, dict) and revised_text in final_result: print(final_result[revised_text]) else: print(final_result) # 或其他处理方式运行这个程序你就看到了一个最基础的、三个智能体线性协作完成内容创作的全过程。每个Agent各司其职通过消息总线进行数据交接。4. 超越流水线处理冲突、循环与动态路由基础流水线跑通了但现实世界的团队协作远非一帆风顺。文案写手可能对某个论点理解有偏差校对审核可能认为某个部分需要重写而非修改。这就需要我们为团队引入更高级的“管理”机制。4.1 引入评审与驳回机制在校对审核Agent的逻辑里我们不能只让它直接修改。它应该先评估初稿的质量。我们可以修改其system_prompt和输出格式你的输出必须是以下两种之一 1. **审核通过**如果文章质量合格仅需少量修改。输出格式{status: approved_with_minor_changes, revised_text: 这里是修改后的全文} 2. **需要重大修改**如果文章存在严重问题如核心论点错误、结构混乱、严重偏离要求。输出格式{status: needs_major_revision, feedback: 详细的修改意见指出具体问题和修改方向, problematic_section_task_id: section_2}然后我们的消息总线和工作流逻辑就需要升级。当校对审核返回needs_major_revision时总线不能直接结束。它需要根据problematic_section_task_id将feedback和原始任务信息重新发送给文案写手进行重写。这就形成了一个循环。实现这个功能我们需要将简单的步骤列表升级为一个状态机State Machine。每个任务如“撰写section_2”都有一个状态如“待处理”、“写作中”、“待审核”、“审核通过”、“需重写”。总线根据当前任务的状态和Agent返回的结果来决定下一个状态是什么以及下一个执行者是谁。4.2 实现一个简单的任务状态机我们可以设计一个Task类和一个WorkflowEngine类来管理class Task: def __init__(self, task_id, description, assigned_agentNone): self.task_id task_id self.description description # 来自策划编辑的JSON self.status pending # pending, executing, waiting_review, approved, needs_rework self.assigned_agent assigned_agent self.result None self.feedback None class WorkflowEngine: def __init__(self, message_bus): self.bus message_bus self.tasks {} self.task_history [] # 记录任务状态变迁 def create_tasks_from_plan(self, plan_data): 从策划编辑的规划中创建一系列写作任务 writing_tasks plan_data.get(writing_tasks, []) for task_desc in writing_tasks: task_id task_desc[task_id] self.tasks[task_id] Task(task_idtask_id, descriptiontask_desc, assigned_agentwriter) self.task_history.append((task_id, created, pending)) def process(self): 主处理循环直到所有任务达到终态approved或手动终止 while not self.all_tasks_finalized(): for task_id, task in self.tasks.items(): if task.status pending: # 分配给写手 self._assign_task(task, writer) elif task.status waiting_review: # 分配给审核员 self._assign_task(task, reviewer) elif task.status needs_rework: # 重新分配给写手并附上反馈 self._assign_task(task, writer, feedbacktask.feedback) # ... 处理其他状态 def _assign_task(self, task, agent_name, feedbackNone): 将任务分配给指定Agent并更新状态 print(f[Engine] 将任务 {task.task_id} 分配给 {agent_name}。) task.status executing task.assigned_agent agent_name # 构建给Agent的输入信息 input_data task.description if feedback: input_data[feedback_from_reviewer] feedback # 通过消息总线发送这里简化实际总线需支持定向发送 result self.bus.send_to_agent(agent_name, input_data) # 处理Agent返回的结果 self._handle_agent_result(task, agent_name, result) def _handle_agent_result(self, task, agent_name, result): if agent_name writer: task.result result task.status waiting_review elif agent_name reviewer: if result.get(status) approved_with_minor_changes: task.status approved task.result result.get(revised_text, task.result) # 更新为修订稿 elif result.get(status) needs_major_revision: task.status needs_rework task.feedback result.get(feedback) # 这里可以设置一个重试计数器避免无限循环 # ... 记录历史 self.task_history.append((task.task_id, agent_name, task.status))这个引擎比简单的线性总线复杂得多但它能处理任务依赖、状态流转和条件分支如驳回重写。这是构建健壮Agent Team的核心。4.3 动态路由与管理者智能体更进一步我们甚至可以引入一个管理者ManagerAgent。它的职责不是完成具体工作而是监督和协调。动态任务分配策划编辑生成了10个写作任务但其中3个是技术性极强的。管理者可以识别这一点并将这3个任务路由给一个专门的“技术写手”Agent而不是通用的“文案写手”。冲突仲裁文案写手和校对审核对某处修改争执不下。管理者可以介入根据预设的规则如“事实准确性优先于文采”或自行判断做出最终决策。流程优化管理者可以监控任务执行时间如果发现某个Agent总是超时它可以决定将任务重新分配给另一个同类型的备用Agent。实现管理者Agent需要赋予它更高的权限和更全面的上下文信息它的提示词可能像这样“你是团队的项目经理。你负责接收所有任务和中间结果并根据每个成员的专长和当前负载动态分配任务。当成员间产出冲突时你负责做出最终裁定。你的目标是保证项目质量和效率。”5. 实战中的挑战与优化策略当你真正运行起一个多智能体系统时一系列在单智能体场景下不明显的问题会集中爆发。这里分享几个我踩过的坑和对应的解决思路。5.1 上下文长度与信息衰减这是最棘手的问题之一。策划编辑生成的大纲和指令可能很长文案写手生成的初稿更长当这些内容全部作为历史对话传递给校对审核时很容易超过模型的最大上下文长度。解决方案摘要与提炼不要传递完整的原始对话历史。在每个交接点强制要求输出Agent对当前结果生成一个结构化摘要。例如文案写手在提交初稿时必须同时提交一个summary字段用三五句话概括本部分核心内容。下游Agent如校对审核在需要全局判断时优先参考摘要。向量化存储与检索将所有中间产出大纲、初稿、修改意见存入向量数据库。当任何一个Agent需要历史上下文时不传递全文而是根据当前任务内容从向量库中检索最相关的几个片段。这能极大节省上下文窗口并让信息获取更精准。分层处理对于校对审核可以设计两阶段。第一阶段只让它基于“章节摘要”和“写作要求”进行高阶评审结构、核心论点。第二阶段对于需要细改的章节再传入该章节的完整文本进行逐字校对。5.2 一致性维护多个智能体各自为政很容易导致最终成果风格不一、术语前后矛盾。例如策划编辑要求用“人工智能”文案写手可能写成“AI”校对审核又可能改回“人工智能”。解决方案创建共享知识库风格指南在项目启动时由策划编辑或一个专门的“初始化Agent”生成一份本项目专用的风格指南。这份指南以系统提示词或独立文件的形式分发给团队每一个成员。内容应包括核心术语表如统一使用“Python”而非“python”、目标读者定位、文章整体语气、禁止使用的词汇等。在消息中嵌入一致性指令在任务传递的JSON中除了具体内容永远包含一个style_guide或consistency_rules字段里面是本次任务需要特别注意的一致性要点。最终一致性检查在团队输出最终结果前增加一个专门的“一致性检查”Agent。它的任务不是修改内容而是通读全文标记出所有风格、术语、格式不一致的地方生成一个报告交由管理者Agent或文案写手进行最终统一修订。5.3 错误传播与系统韧性在流水线中上游的错误会被放大并传递给下游。如果策划编辑错误理解了需求那么后续所有工作都是徒劳。解决方案关键节点人工确认Human-in-the-loop在战略性的节点设置“检查点”。例如在策划编辑输出大纲后不自动执行而是将大纲呈现给用户或一个监督员确认。确认无误后再启动后续的自动化流程。这牺牲了一点全自动的效率但极大地提升了结果的可靠性。多方案生成与选择对于关键环节如大纲生成可以让Agent生成2-3个备选方案然后由一个“评估者”Agent或人工选择最优的一个再进入下一环节。这增加了计算成本但降低了因单次生成不佳而导致全盘失败的风险。完备的日志与回滚系统必须记录每个Agent的完整输入和输出。当最终结果不满意时可以回溯整个决策链精准定位是哪个环节出了问题。基于日志可以实现部分流程的回滚和重试。构建一个高效、稳定的Agent Team其复杂性远超单个智能体。它更像是在设计一个微型的软件系统或组织架构。你需要考虑通信协议、状态管理、错误处理、资源分配等一系列问题。但一旦成功其所能释放的生产力和所能解决的复杂问题将是单智能体无法比拟的。从今天设计的这个内容创作小团队出发你可以将这套模式扩展到客服系统、代码开发、数据分析等无限场景。记住好的团队协作始于清晰的角色定义成于稳健的通信机制终于对一致性和错误的周密管理。