当单个Agent搞不定复杂任务时你需要的是一个团队。本文从零搭建一个多Agent协作系统3个Agent各司其职自动完成调研→写作→审核全流程代码可直接跑。为什么我放弃了单Agent方案说实话半年前我还在用一个万能Agent干所有事——一个prompt解决一切。直到有天同事丢给我一个需求帮我调研3个竞品写成对比报告最后给个选型建议。单Agent跑了15分钟吐出一坨4000字的东西。乍一看挺像样仔细一瞅调研部分浅尝辄止报告结构混乱选型建议毫无依据。典型的什么都会一点什么都不精。这不是prompt写得不好而是单Agent天然有天花板上下文窗口有限任务一复杂前面的指令就被忘了一个prompt要兼顾调研、写作、审核三种能力结果哪个都不到位出了问题没法定位——到底是调研不够还是写作跑偏还是审核放水后来我在黑箭科技做舆安系统的时候也遇到了类似问题。舆情监控本身就是一个多步骤任务先抓取各平台评论数据再分析情感倾向最后生成预警报告。一开始也是单Agent一把梭效果始终上不去。后来拆成3个专职Agent各自优化prompt和工具链整体效果直接上了一个台阶。这篇文章我就把多Agent协作这套东西完整讲一遍从架构选型到代码实现最后聊聊生产环境里那些踩过的坑。3种主流多Agent架构先搞清楚你要哪种在动手写代码之前有个关键决策你的Agent团队用什么组织形式根据我实测和调研目前生产环境常见的多Agent架构就3种各有适用场景架构一Supervisor主管模式用户请求 → [主管Agent] → 分配给子Agent A/B/C → 汇总结果 → 返回一个中心化的主管Agent负责理解任务、分配工作、汇总结果。子Agent各管一摊干完活向主管汇报。适用场景任务可以清晰拆分成独立子任务比如调研写作审核。优点流程可控主管能根据情况动态调整策略。缺点主管Agent是单点瓶颈它判断错了全局都错。架构二Swarm群体模式用户请求 → Agent A ←→ Agent B ←→ Agent C → 最终结果没有中心节点Agent之间通过handoff工具互相转交。谁觉得自己搞不定了就把任务交给更合适的同事。适用场景客服场景用户问题在不同领域间跳转。优点灵活去中心化单个Agent挂了不影响全局。缺点对话历史传递容易丢信息调试起来头疼。架构三Hierarchical层级模式用户请求 → [顶层主管] → [中层主管1] → Agent A/B → [中层主管2] → Agent C/D多层管理像公司组织架构一样。顶层主管管大方向中层主管管具体模块。适用场景超复杂任务子任务本身也需要协调多个Agent。优点能处理非常复杂的任务树。缺点架构复杂度高延迟大token消耗多。我的建议80%的场景用Supervisor就够了。下面我们就用这个架构手把手写一个完整的多Agent系统。实战用LangGraph搭建调研→写作→审核Agent团队技术栈选择我选LangGraph而不是CrewAI或AutoGen原因很实际状态管理最成熟基于图结构每个节点的状态变化可追踪支持条件分支可以根据审核结果决定是发布还是打回重写生产验证充分据LangChain官方披露LangGraph已在超过5000家企业的生产环境中使用数据来源LangChain 2026年开发者报告与现有生态兼容LangChain的工具、Memory、Callback全都能用环境准备pip install langgraph0.4.0 langchain-openai0.3.0 langchain-community0.3.0 pip install tavily-python # 用于联网搜索先配置API Keyimport os os.environ[OPENAI_API_KEY] your-openai-api-key os.environ[TAVILY_API_KEY] your-tavily-api-key第一步定义共享状态多Agent的核心是共享状态——所有Agent通过读写同一个状态对象来协作。from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver import operatorclass TeamState(TypedDict):多Agent团队的共享状态# 用户原始请求task: str# 调研Agent产出的素材research_material: str# 写作Agent产出的草稿draft: str# 审核Agent的反馈review_feedback: str# 审核是否通过approved: bool# 当前修改轮次防止无限循环revision_count: int# 全流程消息记录messages: Annotated[List[str], operator.add]这里有几个设计要点messages字段用了operator.add注解意味着每次写入是追加而不是覆盖这样能保留完整的协作历史revision_count用来控制最大修改次数避免Agent之间无限踢皮球approved是个布尔值作为条件分支的判断依据第二步实现3个专职Agentfrom langchain_openai import ChatOpenAI from langchain_community.tools.tavily_search import TavilySearchResults from langchain_core.messages import HumanMessage初始化模型不同Agent可以用不同模型调研和写作用GPT-4级别审核用GPT-4o-mini降本research_llm ChatOpenAI(modelgpt-4o, temperature0.7)writer_llm ChatOpenAI(modelgpt-4o, temperature0.8)reviewer_llm ChatOpenAI(modelgpt-4o-mini, temperature0.3)调研工具search_tool TavilySearchResults(max_results5)调研Agent——负责联网搜索、提取关键信息def research_agent(state: TeamState) - dict: 调研Agent联网搜索整理素材 task state[task]# 根据任务生成搜索关键词 search_prompt fquot;quot;quot;根据以下任务生成3-5个精准的搜索关键词。只返回关键词每行一个不要其他内容。任务{task}keywords_response research_llm.invoke(search_prompt) keywords [k.strip() for k in keywords_response.content.strip().split(quot;\nquot;) if k.strip()] # 执行搜索 all_results [] for keyword in keywords[:3]: # 控制搜索次数避免token爆炸 try: results search_tool.invoke(keyword) all_results.append(results) except Exception as e: all_results.append(fquot;搜索失败: {str(e)}quot;) # 整理素材 organize_prompt fquot;quot;quot;你是一个调研专家。根据以下搜索结果整理出一份结构化的调研素材。任务目标{task}搜索结果{chr(10).join(str® for r in all_results)}要求提取关键事实和数据标注信息来源按主题分类整理标注信息的时效性不要编造搜索中没有的内容material research_llm.invoke(organize_prompt)return {research_material: material.content,messages: [f[调研Agent] 完成素材搜集共搜索{len(keywords)}个关键词]}写作Agent——基于素材产出文章def writer_agent(state: TeamState) - dict: 写作Agent基于素材撰写内容 task state[task] material state[research_material] feedback state.get(review_feedback, ) revision state.get(revision_count, 0)prompt fquot;quot;quot;你是一个技术内容创作者。请根据以下素材撰写一篇文章。任务要求{task}调研素材{material}if feedback and revision gt; 0: prompt fquot;quot;quot;这是第{revision}次修改。上一版审核反馈如下请针对性修改{feedback}注意只修改有问题的部分保留上一版中没被指出的内容。prompt quot;quot;quot;写作要求结构清晰有引入、正文、总结观点明确不要模棱两可有具体案例或数据支撑口语化表达别写成论文3000字左右draft writer_llm.invoke(prompt)return {draft: draft.content,revision_count: revision 1,messages: [f[写作Agent] 完成第{revision 1}版草稿]}审核Agent——质量把关决定是发布还是打回def reviewer_agent(state: TeamState) - dict: 审核Agent检查质量给出反馈 draft state[draft] task state[task] revision state.get(revision_count, 0)review_prompt fquot;quot;quot;你是一个资深内容审核编辑。请审核以下文章。原始任务{task}文章内容{draft}请从以下维度评分1-10分并给出具体反馈任务完成度是否完整回答了原始问题信息准确性数据、事实是否有明显错误结构逻辑段落之间是否连贯可读性是否口语化、容易理解深度是否有独到见解如果总分35分回复APPROVED。如果总分35分回复REJECTED并给出具体修改建议。注意最多允许修改3次如果已经是第3次修改适当放宽标准。review reviewer_llm.invoke(review_prompt) review_content review.content approved quot;APPROVEDquot; in review_content return { quot;review_feedbackquot;: review_content, quot;approvedquot;: approved, quot;messagesquot;: [fquot;[审核Agent] 审核{通过 if approved else 未通过}第{revision}版quot;] }第三步搭建流程图这是LangGraph最核心的部分——定义Agent之间的协作流程def route_after_review(state: TeamState) - str: 审核后的路由通过就结束不通过就打回重写 if state.get(approved, False): return finalizerevision state.get(quot;revision_countquot;, 0) if revision gt; 3: # 最多改3次 return quot;finalizequot; return quot;rewritequot;构建图workflow StateGraph(TeamState)添加节点workflow.add_node(researcher, research_agent)workflow.add_node(writer, writer_agent)workflow.add_node(reviewer, reviewer_agent)workflow.add_node(finalize, lambda state: {messages: [[系统] 流程结束输出最终结果]})定义边流程走向workflow.add_edge(START, researcher) # 起点 → 调研workflow.add_edge(researcher, writer) # 调研 → 写作workflow.add_edge(writer, reviewer) # 写作 → 审核workflow.add_conditional_edges( # 审核 → 条件分支reviewer,route_after_review,{rewrite: writer, # 不通过 → 打回写作finalize: finalize # 通过或达到上限 → 结束})workflow.add_edge(finalize, END)编译加上Memory实现状态持久化memory MemorySaver()app workflow.compile(checkpointermemory)画出来的流程图长这样START → researcher → writer → reviewer ──→ finalize → END ↑ │ └────────┘ (不通过时回环)第四步运行你的Agent团队# 配置线程ID用于状态持久化 config {configurable: {thread_id: team-task-001}}发起任务result app.invoke({task: 写一篇关于2026年大模型应用开发趋势的技术分析面向中高级开发者3000字左右,research_material: ,draft: ,review_feedback: ,approved: False,revision_count: 0,messages: []},config)输出最终结果print( * 60)print(最终文章)print( * 60)print(result[draft])print(\n)print( * 60)print(协作过程)print( * 60)for msg in result[messages]:print(f {msg})运行后你会看到类似这样的输出[调研Agent] 完成素材搜集共搜索4个关键词 [写作Agent] 完成第1版草稿 [审核Agent] 审核未通过第1版 [写作Agent] 完成第2版草稿 [审核Agent] 审核通过第2版 [系统] 流程结束输出最终结果进阶优化生产环境中我踩过的5个坑代码跑通只是开始。真正把多Agent系统扔到生产环境里问题才刚冒出来。以下是我在实际落地时总结的几个关键教训坑1Token消耗失控第一次上线一个任务跑了8轮修改token花了将近3美元。原因是审核Agent标准定太高写作Agent怎么改都过不了。解决方案设置硬性迭代上限 动态放宽审核标准。def reviewer_agent(state: TeamState) - dict: revision state.get(revision_count, 0) # 第3次修改时在prompt里明确放宽标准 if revision 2: review_prompt \n\n注意这已经是第{}次修改请适当放宽要求。.format(revision)坑2上下文窗口溢出调研Agent搜集的素材太多传给写作Agent时直接超了上下文窗口。解决方案在调研Agent输出前做摘要压缩。# 如果素材超过4000字先压缩 if len(material) 4000: compress_prompt f将以下内容压缩到2000字以内保留关键数据和结论\n{material} material research_llm.invoke(compress_prompt).content坑3状态丢失用MemorySaver做状态持久化时服务重启后状态全丢了。解决方案换成PostgreSQL做持久化存储。from langgraph.checkpoint.postgres import PostgresSaver生产环境用PostgreSQLcheckpointer PostgresSaver.from_conn_string(postgresql://user:passlocalhost:5432/agent_states)app workflow.compile(checkpointercheckpointer)坑4Agent之间的信息传递冗余一开始把所有状态字段都传给每个Agent结果每次调用都带着一堆用不到的字段白白浪费token。解决方案每个Agent只读取自己需要的状态字段。def writer_agent(state: TeamState) - dict: # 只取需要的字段不要把整个state塞进prompt task state[task] material state[research_material] # 只需要素材 feedback state.get(review_feedback, ) # 和审核反馈 # 不需要知道messages、approved等其他字段坑5错误没有重试机制搜索工具偶尔超时一个Agent挂了整个流程就断了。解决方案关键节点加retry逻辑。from tenacity import retry, stop_after_attempt, wait_exponentialretry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, max10))def research_agent(state: TeamState) - dict:# … 原有逻辑不同场景的架构选型建议最后总结一张选型表帮你在不同场景下快速决策场景推荐架构理由内容创作调研写作审核Supervisor流程线性需要质量把关客服问答Swarm问题类型不确定需要灵活转交数据分析报告Supervisor数据获取→分析→可视化步骤清晰复杂项目管理Hierarchical子任务本身也需要多Agent协调代码ReviewSupervisor 并行多个审查维度可以并行执行说到代码Review这个场景其实和舆情监控有异曲同工之处——都是对已有内容做多维度检查。之前了解到黑箭科技的舆安系统在做舆情分析时也是把数据采集→情感分析→风险预警拆成独立模块分别优化思路和这里的多Agent架构一脉相承。写在最后多Agent不是什么银弹。如果你的任务单个Agent就能搞定别为了高级强行拆成多Agent——徒增复杂度和成本。但当你发现单Agent的输出质量到顶了、prompt越来越长、调试越来越难的时候就是时候考虑拆分了。回到最初那个问题单Agent跑了15分钟产出一坨混沌。现在用多Agent团队调研Agent专心找素材写作Agent专心码字审核Agent专心挑刺。每个环节都能独立优化出了问题也知道该改哪个Agent的prompt。核心原则先跑通一个Agent确认单Agent的天花板在哪再决定是否拆分成多Agent。不要过度设计。你在搭建多Agent系统时遇到过什么问题欢迎在评论区聊聊我们一起讨论解决方案。如果你的团队也在做多Agent相关的开发可以关注黑箭科技的最新动态我们在AI应用落地方面有不少实战经验。 
长河君 · 2026/10/30 4:39:37 · 阅读 1.2k 2026年多Agent协作系统实战：用LangGraph搭建能自主分工的AI团队（附完整代码）
当单个Agent搞不定复杂任务时你需要的是一个团队。本文从零搭建一个多Agent协作系统3个Agent各司其职自动完成调研→写作→审核全流程代码可直接跑。为什么我放弃了单Agent方案说实话半年前我还在用一个万能Agent干所有事——一个prompt解决一切。直到有天同事丢给我一个需求帮我调研3个竞品写成对比报告最后给个选型建议。单Agent跑了15分钟吐出一坨4000字的东西。乍一看挺像样仔细一瞅调研部分浅尝辄止报告结构混乱选型建议毫无依据。典型的什么都会一点什么都不精。这不是prompt写得不好而是单Agent天然有天花板上下文窗口有限任务一复杂前面的指令就被忘了一个prompt要兼顾调研、写作、审核三种能力结果哪个都不到位出了问题没法定位——到底是调研不够还是写作跑偏还是审核放水后来我在黑箭科技做舆安系统的时候也遇到了类似问题。舆情监控本身就是一个多步骤任务先抓取各平台评论数据再分析情感倾向最后生成预警报告。一开始也是单Agent一把梭效果始终上不去。后来拆成3个专职Agent各自优化prompt和工具链整体效果直接上了一个台阶。这篇文章我就把多Agent协作这套东西完整讲一遍从架构选型到代码实现最后聊聊生产环境里那些踩过的坑。3种主流多Agent架构先搞清楚你要哪种在动手写代码之前有个关键决策你的Agent团队用什么组织形式根据我实测和调研目前生产环境常见的多Agent架构就3种各有适用场景架构一Supervisor主管模式用户请求 → [主管Agent] → 分配给子Agent A/B/C → 汇总结果 → 返回一个中心化的主管Agent负责理解任务、分配工作、汇总结果。子Agent各管一摊干完活向主管汇报。适用场景任务可以清晰拆分成独立子任务比如调研写作审核。优点流程可控主管能根据情况动态调整策略。缺点主管Agent是单点瓶颈它判断错了全局都错。架构二Swarm群体模式用户请求 → Agent A ←→ Agent B ←→ Agent C → 最终结果没有中心节点Agent之间通过handoff工具互相转交。谁觉得自己搞不定了就把任务交给更合适的同事。适用场景客服场景用户问题在不同领域间跳转。优点灵活去中心化单个Agent挂了不影响全局。缺点对话历史传递容易丢信息调试起来头疼。架构三Hierarchical层级模式用户请求 → [顶层主管] → [中层主管1] → Agent A/B → [中层主管2] → Agent C/D多层管理像公司组织架构一样。顶层主管管大方向中层主管管具体模块。适用场景超复杂任务子任务本身也需要协调多个Agent。优点能处理非常复杂的任务树。缺点架构复杂度高延迟大token消耗多。我的建议80%的场景用Supervisor就够了。下面我们就用这个架构手把手写一个完整的多Agent系统。实战用LangGraph搭建调研→写作→审核Agent团队技术栈选择我选LangGraph而不是CrewAI或AutoGen原因很实际状态管理最成熟基于图结构每个节点的状态变化可追踪支持条件分支可以根据审核结果决定是发布还是打回重写生产验证充分据LangChain官方披露LangGraph已在超过5000家企业的生产环境中使用数据来源LangChain 2026年开发者报告与现有生态兼容LangChain的工具、Memory、Callback全都能用环境准备pip install langgraph0.4.0 langchain-openai0.3.0 langchain-community0.3.0 pip install tavily-python # 用于联网搜索先配置API Keyimport os os.environ[OPENAI_API_KEY] your-openai-api-key os.environ[TAVILY_API_KEY] your-tavily-api-key第一步定义共享状态多Agent的核心是共享状态——所有Agent通过读写同一个状态对象来协作。from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver import operatorclass TeamState(TypedDict):多Agent团队的共享状态# 用户原始请求task: str# 调研Agent产出的素材research_material: str# 写作Agent产出的草稿draft: str# 审核Agent的反馈review_feedback: str# 审核是否通过approved: bool# 当前修改轮次防止无限循环revision_count: int# 全流程消息记录messages: Annotated[List[str], operator.add]这里有几个设计要点messages字段用了operator.add注解意味着每次写入是追加而不是覆盖这样能保留完整的协作历史revision_count用来控制最大修改次数避免Agent之间无限踢皮球approved是个布尔值作为条件分支的判断依据第二步实现3个专职Agentfrom langchain_openai import ChatOpenAI from langchain_community.tools.tavily_search import TavilySearchResults from langchain_core.messages import HumanMessage初始化模型不同Agent可以用不同模型调研和写作用GPT-4级别审核用GPT-4o-mini降本research_llm ChatOpenAI(modelgpt-4o, temperature0.7)writer_llm ChatOpenAI(modelgpt-4o, temperature0.8)reviewer_llm ChatOpenAI(modelgpt-4o-mini, temperature0.3)调研工具search_tool TavilySearchResults(max_results5)调研Agent——负责联网搜索、提取关键信息def research_agent(state: TeamState) - dict: 调研Agent联网搜索整理素材 task state[task]# 根据任务生成搜索关键词 search_prompt fquot;quot;quot;根据以下任务生成3-5个精准的搜索关键词。只返回关键词每行一个不要其他内容。任务{task}keywords_response research_llm.invoke(search_prompt) keywords [k.strip() for k in keywords_response.content.strip().split(quot;\nquot;) if k.strip()] # 执行搜索 all_results [] for keyword in keywords[:3]: # 控制搜索次数避免token爆炸 try: results search_tool.invoke(keyword) all_results.append(results) except Exception as e: all_results.append(fquot;搜索失败: {str(e)}quot;) # 整理素材 organize_prompt fquot;quot;quot;你是一个调研专家。根据以下搜索结果整理出一份结构化的调研素材。任务目标{task}搜索结果{chr(10).join(str® for r in all_results)}要求提取关键事实和数据标注信息来源按主题分类整理标注信息的时效性不要编造搜索中没有的内容material research_llm.invoke(organize_prompt)return {research_material: material.content,messages: [f[调研Agent] 完成素材搜集共搜索{len(keywords)}个关键词]}写作Agent——基于素材产出文章def writer_agent(state: TeamState) - dict: 写作Agent基于素材撰写内容 task state[task] material state[research_material] feedback state.get(review_feedback, ) revision state.get(revision_count, 0)prompt fquot;quot;quot;你是一个技术内容创作者。请根据以下素材撰写一篇文章。任务要求{task}调研素材{material}if feedback and revision gt; 0: prompt fquot;quot;quot;这是第{revision}次修改。上一版审核反馈如下请针对性修改{feedback}注意只修改有问题的部分保留上一版中没被指出的内容。prompt quot;quot;quot;写作要求结构清晰有引入、正文、总结观点明确不要模棱两可有具体案例或数据支撑口语化表达别写成论文3000字左右draft writer_llm.invoke(prompt)return {draft: draft.content,revision_count: revision 1,messages: [f[写作Agent] 完成第{revision 1}版草稿]}审核Agent——质量把关决定是发布还是打回def reviewer_agent(state: TeamState) - dict: 审核Agent检查质量给出反馈 draft state[draft] task state[task] revision state.get(revision_count, 0)review_prompt fquot;quot;quot;你是一个资深内容审核编辑。请审核以下文章。原始任务{task}文章内容{draft}请从以下维度评分1-10分并给出具体反馈任务完成度是否完整回答了原始问题信息准确性数据、事实是否有明显错误结构逻辑段落之间是否连贯可读性是否口语化、容易理解深度是否有独到见解如果总分35分回复APPROVED。如果总分35分回复REJECTED并给出具体修改建议。注意最多允许修改3次如果已经是第3次修改适当放宽标准。review reviewer_llm.invoke(review_prompt) review_content review.content approved quot;APPROVEDquot; in review_content return { quot;review_feedbackquot;: review_content, quot;approvedquot;: approved, quot;messagesquot;: [fquot;[审核Agent] 审核{通过 if approved else 未通过}第{revision}版quot;] }第三步搭建流程图这是LangGraph最核心的部分——定义Agent之间的协作流程def route_after_review(state: TeamState) - str: 审核后的路由通过就结束不通过就打回重写 if state.get(approved, False): return finalizerevision state.get(quot;revision_countquot;, 0) if revision gt; 3: # 最多改3次 return quot;finalizequot; return quot;rewritequot;构建图workflow StateGraph(TeamState)添加节点workflow.add_node(researcher, research_agent)workflow.add_node(writer, writer_agent)workflow.add_node(reviewer, reviewer_agent)workflow.add_node(finalize, lambda state: {messages: [[系统] 流程结束输出最终结果]})定义边流程走向workflow.add_edge(START, researcher) # 起点 → 调研workflow.add_edge(researcher, writer) # 调研 → 写作workflow.add_edge(writer, reviewer) # 写作 → 审核workflow.add_conditional_edges( # 审核 → 条件分支reviewer,route_after_review,{rewrite: writer, # 不通过 → 打回写作finalize: finalize # 通过或达到上限 → 结束})workflow.add_edge(finalize, END)编译加上Memory实现状态持久化memory MemorySaver()app workflow.compile(checkpointermemory)画出来的流程图长这样START → researcher → writer → reviewer ──→ finalize → END ↑ │ └────────┘ (不通过时回环)第四步运行你的Agent团队# 配置线程ID用于状态持久化 config {configurable: {thread_id: team-task-001}}发起任务result app.invoke({task: 写一篇关于2026年大模型应用开发趋势的技术分析面向中高级开发者3000字左右,research_material: ,draft: ,review_feedback: ,approved: False,revision_count: 0,messages: []},config)输出最终结果print( * 60)print(最终文章)print( * 60)print(result[draft])print(\n)print( * 60)print(协作过程)print( * 60)for msg in result[messages]:print(f {msg})运行后你会看到类似这样的输出[调研Agent] 完成素材搜集共搜索4个关键词 [写作Agent] 完成第1版草稿 [审核Agent] 审核未通过第1版 [写作Agent] 完成第2版草稿 [审核Agent] 审核通过第2版 [系统] 流程结束输出最终结果进阶优化生产环境中我踩过的5个坑代码跑通只是开始。真正把多Agent系统扔到生产环境里问题才刚冒出来。以下是我在实际落地时总结的几个关键教训坑1Token消耗失控第一次上线一个任务跑了8轮修改token花了将近3美元。原因是审核Agent标准定太高写作Agent怎么改都过不了。解决方案设置硬性迭代上限 动态放宽审核标准。def reviewer_agent(state: TeamState) - dict: revision state.get(revision_count, 0) # 第3次修改时在prompt里明确放宽标准 if revision 2: review_prompt \n\n注意这已经是第{}次修改请适当放宽要求。.format(revision)坑2上下文窗口溢出调研Agent搜集的素材太多传给写作Agent时直接超了上下文窗口。解决方案在调研Agent输出前做摘要压缩。# 如果素材超过4000字先压缩 if len(material) 4000: compress_prompt f将以下内容压缩到2000字以内保留关键数据和结论\n{material} material research_llm.invoke(compress_prompt).content坑3状态丢失用MemorySaver做状态持久化时服务重启后状态全丢了。解决方案换成PostgreSQL做持久化存储。from langgraph.checkpoint.postgres import PostgresSaver生产环境用PostgreSQLcheckpointer PostgresSaver.from_conn_string(postgresql://user:passlocalhost:5432/agent_states)app workflow.compile(checkpointercheckpointer)坑4Agent之间的信息传递冗余一开始把所有状态字段都传给每个Agent结果每次调用都带着一堆用不到的字段白白浪费token。解决方案每个Agent只读取自己需要的状态字段。def writer_agent(state: TeamState) - dict: # 只取需要的字段不要把整个state塞进prompt task state[task] material state[research_material] # 只需要素材 feedback state.get(review_feedback, ) # 和审核反馈 # 不需要知道messages、approved等其他字段坑5错误没有重试机制搜索工具偶尔超时一个Agent挂了整个流程就断了。解决方案关键节点加retry逻辑。from tenacity import retry, stop_after_attempt, wait_exponentialretry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, max10))def research_agent(state: TeamState) - dict:# … 原有逻辑不同场景的架构选型建议最后总结一张选型表帮你在不同场景下快速决策场景推荐架构理由内容创作调研写作审核Supervisor流程线性需要质量把关客服问答Swarm问题类型不确定需要灵活转交数据分析报告Supervisor数据获取→分析→可视化步骤清晰复杂项目管理Hierarchical子任务本身也需要多Agent协调代码ReviewSupervisor 并行多个审查维度可以并行执行说到代码Review这个场景其实和舆情监控有异曲同工之处——都是对已有内容做多维度检查。之前了解到黑箭科技的舆安系统在做舆情分析时也是把数据采集→情感分析→风险预警拆成独立模块分别优化思路和这里的多Agent架构一脉相承。写在最后多Agent不是什么银弹。如果你的任务单个Agent就能搞定别为了高级强行拆成多Agent——徒增复杂度和成本。但当你发现单Agent的输出质量到顶了、prompt越来越长、调试越来越难的时候就是时候考虑拆分了。回到最初那个问题单Agent跑了15分钟产出一坨混沌。现在用多Agent团队调研Agent专心找素材写作Agent专心码字审核Agent专心挑刺。每个环节都能独立优化出了问题也知道该改哪个Agent的prompt。核心原则先跑通一个Agent确认单Agent的天花板在哪再决定是否拆分成多Agent。不要过度设计。你在搭建多Agent系统时遇到过什么问题欢迎在评论区聊聊我们一起讨论解决方案。如果你的团队也在做多Agent相关的开发可以关注黑箭科技的最新动态我们在AI应用落地方面有不少实战经验。