1. 项目概述为什么我们需要一个多 Agent 团队最近在折腾 AI 应用开发我发现单打独斗的 Claude 虽然能力很强但面对复杂任务时总有些力不从心。比如我想让它帮我分析一份数据报告再根据分析结果写个总结最后生成一个可视化图表。这个过程涉及数据分析、文案撰写和代码生成让一个“人”来回切换角色不仅效率低还容易出错。这让我开始思考能不能像组建一个项目团队一样为不同的任务分配不同的“专家”AI呢这就是多 Agent 系统的核心价值。简单来说一个多 Agent 系统就是由多个具备特定技能和角色的 AI 智能体Agent组成的协作网络。在这个项目中我们的目标是搭建一个由 4 个 Agent 组成的微型团队。这个团队能像一支训练有素的队伍各司其职通过沟通和协作共同完成一个用户提出的复杂指令。这不再是简单的“一问一答”而是实现了任务分解、规划、执行和校验的自动化工作流。对于开发者、数据分析师、内容创作者甚至个人效率爱好者而言这意味着你可以拥有一个永不疲倦、高度专业化的虚拟助手团队7x24小时待命处理那些繁琐或需要多领域知识的工作。2. 核心设计构建一个高效协作的 Agent 团队架构设计一个能真正干活的多 Agent 系统关键在于清晰的职责划分和高效的通信机制。我们不能简单地把四个 Claude 实例扔在一起就让它们开会那只会导致混乱和资源浪费。经过多次尝试和调整我设计了一套比较稳定高效的 4-Agent 团队架构它模拟了一个小型项目组的运作模式。2.1 角色定义与职责划分我的团队由四个核心角色构成每个角色都有明确的技能栈和职责边界项目经理Project Manager Agent这是团队的大脑和指挥中心。它的核心职责是理解用户意图、拆解任务、制定计划并分配工作。当用户提出一个模糊或复杂的请求时例如“帮我分析一下公司上个季度的销售数据并写一份给管理层的报告”项目经理 Agent 需要将其分解为具体的、可执行的任务比如任务一数据清洗与整理任务二趋势分析与洞察提取任务三报告大纲与文案撰写任务四关键图表生成。它不负责具体执行而是负责规划和调度。技术专家Technical Expert Agent这是团队的“工匠”。它擅长代码编写、数据处理、工具调用和解决具体的技术问题。当项目经理分配来“计算各地区销售额同比增长率”或“生成一个销售额月度趋势的折线图”这样的任务时就由技术专家 Agent 接手。它熟悉 Python、SQL、各种数据分析库和可视化工具能够产出可运行、可验证的代码或数据结果。文案专员Copywriter Agent这是团队的“笔杆子”。它专注于文本内容的创作、润色、格式化和风格统一。技术专家产出了一堆数据和图表但如何将它们组织成一份逻辑清晰、语言得体、符合商务场景的报告这就是文案专员的工作。它负责撰写报告正文、提炼核心结论、编写邮件正文、润色用户界面文案等所有与文字相关的工作。质量审核员Quality Reviewer Agent这是团队的“守门人”。它的职责是对前三个 Agent 的产出进行交叉检查、逻辑验证和最终整合。例如检查技术专家生成的图表数据是否与文案专员报告中的描述一致检查报告的逻辑是否自洽有没有遗漏关键结论检查最终输出的格式是否符合要求。它不直接创造内容而是确保最终交付物的准确性和完整性。注意角色定义不是一成不变的。你可以根据你的常用场景调整。比如如果你主要做创意设计可以加入一个“设计师 Agent”如果主要做学术研究可以加入一个“文献调研 Agent”。关键是让每个 Agent 的职责单一且明确避免功能重叠导致的指令冲突。2.2 通信与协作流程设计定义了角色接下来就要设计它们如何“开会”。我采用的是基于“共享工作区”和“顺序工作流”的混合模式这比让所有 Agent 同时发言要可控得多。整个系统的运行始于用户的一条自然语言指令。这条指令首先被发送给项目经理 Agent。项目经理 Agent 的工作是进行任务解析与规划。它不会立即开始执行而是先思考。这个过程包括理解用户的深层需求是什么要完成这个需求需要哪些步骤每个步骤由哪个角色的 Agent 来执行最合适预期的产出物是什么思考完成后它会生成一份详细的任务规划书。这份规划书通常包括任务列表、每个任务的负责人即对应的 Agent、任务的输入和预期输出。这份规划书就是整个团队的“项目章程”。接下来系统进入执行阶段。项目经理 Agent 会根据规划书将第一个任务连同所有必要的上下文包括原始用户指令和规划书本身传递给指定的执行 Agent比如技术专家 Agent。技术专家 Agent 收到任务后开始独立工作。它可能会进行代码编写、数据查询或调用某个 API。工作完成后它会将产出物可能是一段代码、一个数据处理结果、一个图表文件提交到团队的共享工作区。这个“共享工作区”可以是一个简单的文本缓冲区、一个共享的变量字典或者一个更结构化的内存对象所有 Agent 都能读取其中的内容。然后项目经理 Agent会监控工作区的状态。一旦技术专家 Agent 的任务标记为完成项目经理就会根据规划书将下一个任务分配给文案专员 Agent并将技术专家的产出物作为输入传递给它。文案专员 Agent 开始撰写报告完成后同样将草稿提交到共享工作区。当所有执行类任务都完成后项目经理 Agent会触发最终的质量检查环节。它将共享工作区中所有的中间产出物和最终草稿一并交给质量审核员 Agent。审核员 Agent 会进行一致性检查、逻辑审查和格式校对。如果发现问题它会将修改意见反馈回共享工作区甚至可能要求某个环节的 Agent 重新处理。如果没有问题它会对所有内容进行最终整合与格式化生成一个完整的、用户友好的最终答案。这个流程的关键在于项目经理 Agent 始终是流程的驱动者和协调者它确保了任务按计划推进避免了多个 Agent 无秩序地同时修改同一份内容。而共享工作区则充当了信息的枢纽保证了上下文在不同执行阶段的无损传递。3. 技术实现从零开始搭建你的 Agent 团队理论讲完了我们来点实在的。下面我将以目前比较流行的Claude Code或 Claude Desktop API 作为每个 Agent 的“大脑”用 Python 来构建这个系统的骨架。这里不依赖复杂的框架我们从最核心的模块自己搭这样你才能彻底理解其运作原理。3.1 环境准备与核心工具选型首先你需要一个能稳定调用 Claude 模型的环境。我推荐使用Anthropic 官方提供的 Claude API。相比于在网页界面操作API 调用更稳定、可编程是构建自动化系统的基石。获取 API 密钥前往 Anthropic 官网注册并创建一个项目获取你的ANTHROPIC_API_KEY。妥善保管不要泄露。安装 Python 库我们将主要使用anthropic官方库和python-dotenv来管理密钥。pip install anthropic python-dotenv创建环境变量文件在项目根目录创建一个.env文件写入你的密钥ANTHROPIC_API_KEYyour_api_key_here模型选择Anthropic 提供了多个模型如claude-3-5-sonnet-20241022能力强速度适中、claude-3-haiku-20240307速度快成本低。对于 Agent 系统我建议对“项目经理”和“质量审核员”使用能力更强的 Sonnet 模型以确保任务规划和最终审核的质量对“技术专家”和“文案专员”可以使用 Haiku 模型以节约成本。你可以在代码中灵活配置。3.2 构建基础 Agent 类所有 Agent 都有一些共同的行为接收消息、调用 Claude API、返回结果。我们先抽象一个基础类。import os from anthropic import Anthropic from dotenv import load_dotenv load_dotenv() # 加载环境变量 class BaseAgent: def __init__(self, name, role, modelclaude-3-5-sonnet-20241022, system_prompt): 初始化一个基础 Agent。 :param name: Agent 名称如 “ProjectManager” :param role: Agent 角色描述用于构建系统提示词 :param model: 使用的 Claude 模型 :param system_prompt: 自定义的系统提示词如果为空则根据 role 生成 self.name name self.role role self.model model self.client Anthropic(api_keyos.getenv(ANTHROPIC_API_KEY)) # 系统提示词是 Agent 的“人格”和“职责说明书”至关重要 if system_prompt: self.system_prompt system_prompt else: self.system_prompt f你是一个专业的{role}。请严格遵守你的角色定位只完成你职责范围内的任务。 你的思考过程应清晰、有条理。最终请将你的输出放在 response 标签内。 def invoke(self, user_message, contextNone): 调用 Agent 处理任务。 :param user_message: 具体的任务指令 :param context: 可选的上下文信息如之前任务的产出 :return: Agent 的回复文本 # 构建完整的消息历史。在实际系统中这里会更复杂可能包含多轮对话。 messages [] if context: # 将上下文作为系统提示的一部分或单独的消息传入这里简单拼接 full_prompt f上下文信息\n{context}\n\n当前任务{user_message} else: full_prompt user_message messages.append({role: user, content: full_prompt}) try: response self.client.messages.create( modelself.model, max_tokens4096, # 根据任务调整 systemself.system_prompt, messagesmessages ) # 提取响应内容 response_text response.content[0].text # 一个简单的解析尝试提取 response 标签内的内容如果没有标签则返回全部 if response in response_text and /response in response_text: start response_text.find(response) len(response) end response_text.find(/response) return response_text[start:end].strip() else: return response_text.strip() except Exception as e: return fAgent {self.name} 调用失败: {str(e)}这个BaseAgent类封装了与 Claude API 交互的基本逻辑。每个具体的 Agent如项目经理、技术专家都将继承这个类并通过定制system_prompt来赋予其独特的“人格”。3.3 实现四个具体的 Agent现在我们来创建那四个核心角色。关键在于为它们编写高度专业化、指令清晰的系统提示词。class ProjectManagerAgent(BaseAgent): def __init__(self, modelclaude-3-5-sonnet-20241022): system_prompt 你是经验丰富的项目经理。你的核心职责是理解复杂需求并将其分解为具体的、可顺序执行的任务。 请按以下步骤工作 1. **理解需求**仔细分析用户的原始指令明确最终目标。 2. **任务分解**将目标分解为多个子任务。每个子任务应该 - 由单个角色技术专家、文案专员独立完成。 - 有明确的输入和输出描述。 - 输出物能被下一个任务直接使用。 3. **制定计划**为子任务排序并指定执行者。通常顺序是数据/技术任务 - 文案创作 - 审核整合。 4. **输出规划**用清晰的结构化格式输出你的计划。例如 ## 任务规划 **总目标:** [用户目标] **任务1 (技术专家):** [具体任务描述]输入[...]输出[...] **任务2 (文案专员):** [具体任务描述]输入[任务1的输出]输出[...] **任务3 (质量审核员):** 审核最终报告确保数据与文案一致格式正确。 请将最终计划放在 response 标签内。不要执行具体任务。 super().__init__(nameProjectManager, role项目经理, modelmodel, system_promptsystem_prompt) class TechnicalExpertAgent(BaseAgent): def __init__(self, modelclaude-3-haiku-20240307): system_prompt 你是资深技术专家精通Python、数据分析、可视化及各类工具API调用。 你的职责是接收具体的、可操作的技术任务如“计算某列平均值”、“生成折线图”、“调用某API获取数据”并输出可验证的成果。 你的输出必须是 1. **代码/命令**如果任务是编码提供完整、可运行的代码片段并附上简要说明。 2. **数据结果**如果任务是计算提供清晰的结果如表格、数值。 3. **文件/链接**如果生成了图表文件或调用了外部服务说明产出物的位置或内容。 请确保你的输出准确、简洁且只包含与当前技术任务直接相关的内容。将最终产出放在 response 标签内。 super().__init__(nameTechnicalExpert, role技术专家, modelmodel, system_promptsystem_prompt) class CopywriterAgent(BaseAgent): def __init__(self, modelclaude-3-haiku-20240307): system_prompt 你是专业的文案专员擅长撰写、润色和格式化各类文档报告、邮件、总结等。 你的职责是根据提供的材料如数据、图表、要点创作或整理出逻辑清晰、语言流畅、风格得体的文本内容。 请注意 1. 严格基于输入的材料进行创作不要虚构信息。 2. 注意文档的结构如标题、段落、列表。 3. 语言风格需符合任务要求如商务正式、简洁明了。 将你撰写的完整文案放在 response 标签内。 super().__init__(nameCopywriter, role文案专员, modelmodel, system_promptsystem_prompt) class QualityReviewerAgent(BaseAgent): def __init__(self, modelclaude-3-5-sonnet-20241022): system_prompt 你是严格的质量审核员。你的任务是对最终交付物进行最终检查。 你需要检查 1. **一致性**文案中引用的数据、结论是否与技术专家提供的原始数据完全一致 2. **完整性**报告是否涵盖了任务规划中的所有要求有无遗漏关键点 3. **逻辑性**论述是否逻辑自洽有无矛盾或跳跃 4. **格式与语言**文档格式是否规范有无语法错误或不当表述 如果发现问题请明确指出问题所在并提供具体的修改建议。 如果一切正常请对内容进行最终整合与美化输出一个可以直接交付给用户的最终版本。 将你的审核结论或最终成品放在 response 标签内。 super().__init__(nameQualityReviewer, role质量审核员, modelmodel, system_promptsystem_prompt)3.4 构建团队协调器与工作流引擎有了四个 Agent我们需要一个“导演”来指挥它们按流程工作。这个协调器负责维护共享工作区并按照项目经理生成的计划来驱动任务。class TeamOrchestrator: def __init__(self): # 初始化四个 Agent self.pm ProjectManagerAgent() self.tech TechnicalExpertAgent() self.writer CopywriterAgent() self.reviewer QualityReviewerAgent() # 共享工作区用一个字典来存储各个阶段的产出 self.workspace { original_query: , project_plan: , tech_output: , writer_output: , final_output: } def run(self, user_query): 执行完整的工作流 print(f用户指令: {user_query}) self.workspace[original_query] user_query # 阶段 1: 规划 print(\n--- 阶段1: 项目经理制定计划 ---) plan_prompt f请为以下用户需求制定详细的任务执行计划\n{user_query} project_plan self.pm.invoke(plan_prompt) print(f生成计划:\n{project_plan}) self.workspace[project_plan] project_plan # 解析计划这里简化处理实际可能需要用LLM或规则来解析计划文本提取任务列表 # 我们假设计划中明确提到了“任务1技术专家...”这样的格式。 # 为了演示我们手动驱动流程。 # 阶段 2: 技术执行 (假设第一个任务是技术任务) print(\n--- 阶段2: 技术专家执行任务 ---) # 这里需要从 project_plan 中提取出给技术专家的具体指令。实际应用中这部分需要解析。 # 我们简化直接让技术专家根据用户指令和计划处理可能的技术部分。 tech_prompt f根据以下项目计划和原始需求执行你需要负责的技术部分。 项目计划 {project_plan} 原始需求 {user_query} 请开始你的工作。 tech_output self.tech.invoke(tech_prompt) print(f技术专家产出:\n{tech_output}) self.workspace[tech_output] tech_output # 阶段 3: 文案撰写 print(\n--- 阶段3: 文案专员撰写报告 ---) writer_prompt f请根据以下材料撰写一份完整的报告/总结。 原始需求{user_query} 技术分析结果{tech_output} 请整合信息形成逻辑清晰、语言专业的文档。 writer_output self.writer.invoke(writer_prompt) print(f文案专员产出:\n{writer_output}) self.workspace[writer_output] writer_output # 阶段 4: 质量审核与整合 print(\n--- 阶段4: 质量审核员进行最终检查 ---) review_prompt f请审核以下最终报告草稿并检查其与技术数据的一致性。 原始需求{user_query} 项目计划{project_plan} 技术原始数据{tech_output} 报告草稿{writer_output} 请执行审核如果发现问题请指出如果无误请输出最终定稿。 final_output self.reviewer.invoke(review_prompt) print(f最终产出:\n{final_output}) self.workspace[final_output] final_output return final_output # 运行示例 if __name__ __main__: team TeamOrchestrator() # 模拟一个用户查询 user_query 分析过去一年每月‘产品A’的销售额计算月均销售额和同比增长最快的月份并撰写一份简要分析报告。 final_result team.run(user_query) print(\n *50) print(流程结束最终结果已保存在工作区。)这个TeamOrchestrator类实现了一个最简单的顺序工作流。在实际应用中project_plan的解析会是一个难点你可能需要让项目经理 Agent 以严格的 JSON 或 YAML 格式输出计划以便程序能自动提取任务列表、执行者和输入输出关系实现更动态的流程控制。4. 高级技巧与实战优化搭建出基础框架只是第一步。要让这个多 Agent 系统真正可靠、高效地运行起来还需要解决一系列工程化问题。下面分享我在实践中总结的几个关键优化点。4.1 提升系统稳定性的关键提示词工程Agent 的行为完全由系统提示词System Prompt决定。写得模糊Agent 就会行为错乱。我的经验是提示词必须包含以下几个部分身份与职责锚定开头必须用最强烈的语气明确“你是谁”、“你只做什么”。例如“你必须且仅作为一名技术专家行事你的唯一职责是编写和验证代码。你绝不撰写报告正文或进行最终总结。”输入输出格式规范强制规定 Agent 的响应格式。使用 XML 标签如task_result.../task_result是很好的方法便于后续程序化解析。例如要求技术专家将代码放在code标签内将执行结果放在result标签内。思考链Chain-of-Thought要求对于复杂任务要求 Agent 先输出它的思考过程放在thinking标签里再输出最终答案。这不仅能提升结果质量也便于调试。你可以在提示词中写明“请逐步推理并将推理过程放在thinking标签内将最终答案放在response标签内。”错误处理与边界声明告诉 Agent 当遇到无法处理的情况时该如何应对。例如“如果你认为分配给你的任务超出你的角色范围或输入信息不足请输出error任务描述不清需要更多信息XXX/error。”4.2 实现动态工作流与错误处理基础版本是硬编码的顺序流程。一个成熟的系统需要能根据项目经理的计划动态创建流程。结构化输出解析改造你的项目经理 Agent要求它必须以 JSON 格式输出计划。# 在 ProjectManagerAgent 的系统提示词中增加 “请将任务计划以如下 JSON 格式输出且只输出此 JSON { goal: 总体目标描述, tasks: [ { id: 1, assignee: TechnicalExpert, // 或 Copywriter description: 具体的任务描述, inputs: [original_query], // 依赖的输入来源 expected_output: 产出描述 }, ... ] }”这样TeamOrchestrator就可以解析这个 JSON动态地按顺序执行tasks列表。错误处理与重试机制任何一个 Agent 调用 API 都可能失败网络、限流、内容过滤。必须在invoke方法外围添加重试逻辑如tenacity库。更重要的是当某个 Agent 的输出不符合预期如没有按格式输出时系统应该能检测到并触发一个“错误处理 Agent”或反馈给项目经理重新规划该任务。上下文管理随着任务链变长如何将之前所有步骤的产出有效地传递给后续 Agent 是个挑战。简单的字符串拼接会很快超出模型的上下文窗口。解决方案是摘要化。可以引入一个额外的“上下文管理 Agent”它的职责是将冗长的中间结果如大段代码、数据表格提炼成简洁的摘要只将摘要和最关键的原数据传递给下一步。或者使用向量数据库存储中间产物在需要时进行检索。4.3 成本控制与性能优化同时运行多个 AgentAPI 调用成本会成倍增加。必须精打细算模型分级使用如前所述对规划、审核等需要深度思考的环节使用昂贵但能力强的模型如 Claude 3.5 Sonnet对执行、写作等相对标准的任务使用更经济的模型如 Claude 3 Haiku。这能在保证质量的同时大幅降低成本。缓存策略对于相似的、重复性的用户查询例如每天生成格式类似的销售报告可以将项目经理生成的“任务计划”缓存起来。下次遇到类似请求时可以直接复用计划跳过规划步骤节省一次 LLM 调用。异步并行执行如果任务计划中的某些任务之间没有依赖关系可以让对应的 Agent 并行执行。例如技术专家在处理数据的同时文案专员可以提前撰写报告的模板和固定部分。这需要更复杂的工作流引擎如使用asyncio和依赖关系分析但对缩短整体响应时间效果显著。5. 常见问题与避坑指南在实际搭建和运行过程中我踩过不少坑。这里把最常见的问题和解决方案列出来希望能帮你省点时间。5.1 Agent 角色混淆或越界这是初期最常见的问题。比如你让技术专家分析数据它却在回复的最后加上了“综上所述我认为...”这样的总结性文案侵占了文案专员的职责。根因系统提示词不够强硬和具体或者不同 Agent 的提示词存在重叠领域。解决方案强化角色边界在每个 Agent 的提示词开头用大写、加粗在提示词中用符号表示等方式强调“你必须是...你绝不能...”。负面示例在提示词中给出反面例子。例如在技术专家的提示词里加上“错误示例用户问‘分析数据并总结’你输出‘数据表明...因此我们得出结论...’。正确做法你只应输出‘数据分析结果...’不要包含‘结论’、‘总结’等文案性词汇。”后处理过滤在接收到 Agent 响应后增加一个简单的规则检查如果发现包含特定越界关键词如“报告显示”、“综上所述”可以自动截断或触发重生成。5.2 任务规划不切实际或无法解析项目经理 Agent 有时会制定出逻辑混乱、无法执行的计划或者它的输出是自然语言段落程序无法自动提取任务列表。根因给项目经理的指令不够清晰或者没有强制要求结构化输出。解决方案提供规划范例在项目经理的系统提示词中直接给一个完美的 JSON 格式计划示例。LLM 的“模仿”能力很强看到范例后输出质量会大幅提升。分步引导在用户指令中可以引导用户稍微结构化地提问或者先让一个“需求澄清 Agent”与用户对话将模糊需求转化为清晰的需求说明书再交给项目经理。容错与人工干预系统应能检测到无法解析的计划并触发降级策略例如将计划文本直接展示给用户确认或转入一个更简单的单 Agent 处理流程。5.3 上下文丢失与信息衰减在长链条任务中后续 Agent 可能忘记了最初的目标或者曲解了前序步骤的产出。根因工作流中信息传递是“接力棒”式的每一步都可能丢失细节。解决方案维护核心上下文将最原始的用户指令和最终目标作为“全局上下文”传递给每一个执行步骤的 Agent。在调用每个 Agent 时除了它直接的上游输入永远附带上original_query。使用检查点在关键步骤完成后如技术分析完成、报告草稿完成让质量审核员提前介入做一个中期检查确保方向没有跑偏而不是等到最后。结构化中间产物要求每个 Agent 的产出不仅是结果还要包含“任务摘要”和“给下一环节的说明”。例如技术专家输出{raw_data: [...], summary: 发现3月销售额环比增长50%, note_for_writer: 请在报告中重点强调3月的增长表现。}5.4 API 调用失败与速率限制根因网络不稳定、Anthropic API 有速率限制RPM/TPM。解决方案指数退避重试对所有 API 调用封装重试逻辑在遇到 429过多请求或 5xx 错误时等待一段时间后重试。等待时间应逐渐增加如 1s, 2s, 4s...。设置合理的超时为每个请求设置超时时间如 30 秒避免因单个请求卡死整个流程。监控与告警记录每次调用的状态、耗时和 token 使用量。当失败率异常或成本激增时能及时收到通知。搭建一个多 Agent 系统就像组建和管理一个真实的团队初期需要反复调试角色职责和协作流程。一旦调通其带来的效率提升和解决问题的能力是单 Agent 模式无法比拟的。从今天这个简单的 4-Agent 框架开始你可以逐步扩展加入更多专业角色设计更复杂的协作网络甚至让 Agent 们能够使用工具、访问互联网构建真正强大的 AI 助理团队。