LangGraph多智能体实战：角色分工与协作机制详解

1. 从一个Agent打天下到一群Agent各司其职的认知转变如果你最近在折腾Agent开发大概率经历过这样一个阶段一开始觉得单个Agent挺能打给它一个系统提示词挂上几个工具就能完成搜索、总结、写代码这些任务。但任务一复杂问题就来了——上下文越堆越长工具调用开始互相干扰模型在该搜索还是该计算之间反复横跳最后输出一堆看似合理但经不起推敲的东西。这不是模型不行而是架构选错了。单Agent的本质是一个大脑同时干所有事它的瓶颈不在模型能力而在上下文窗口的注意力分配和职责边界模糊。多智能体Multi-Agent要解决的核心问题就是把一个全能选手拆成一支分工明确的团队。我最初接触多智能体是从LangGraph开始的。当时看到角色分工和协作机制这两个词第一反应是这不就是给每个Agent写不同的提示词然后让它们轮流发言吗实测下来完全不是这么回事。真正的难点在于——谁来决定下一个该谁说话、信息怎么在Agent之间传递、出现分歧时怎么收敛。这三个问题不解决多智能体系统就是一堆各说各话的复读机。这篇文章适合两类人一是已经用LangChain或类似框架跑通过单Agent、想往多Agent架构迁移的开发者二是正在做复杂任务编排、发现单Agent上下文爆炸或工具冲突的工程师。我会从角色分工的设计逻辑讲起拆解协作机制的几种典型模式再结合LangGraph的实际节点编排给出可复现的配置思路。不堆概念只讲我踩过的坑和验证过的方案。2. 角色分工不是给每个Agent起个名字那么简单2.1 分工的本质是上下文隔离与职责收敛很多人做多智能体的第一步是创建三个Agent分别叫研究员分析师写手然后给每个写一段提示词。跑起来发现研究员把分析师的活也干了写手开始质疑研究员的数据来源整个流程乱成一锅粥。问题出在分工的粒度上。角色分工的核心不是命名而是上下文隔离。每个Agent应该只看到它完成自己任务所需的最小信息集而不是把全局上下文一股脑塞给所有人。这背后的逻辑是模型的注意力是有限资源无关信息越多关键信息的权重就越低。举个具体例子。假设你要做一个竞品分析报告生成的多Agent系统。合理的分工是这样的搜索Agent只负责根据关键词检索信息输出原始素材列表。它的上下文里只有搜索工具和查询语句不需要知道最终报告长什么样。分析Agent接收搜索Agent的输出负责提取关键维度、对比差异。它的上下文里只有结构化的素材和对比框架不需要知道搜索是怎么执行的。撰写Agent接收分析结果负责组织成文。它的上下文里只有分析结论和写作规范不需要看到原始搜索结果。这样做的直接好处是每个Agent的提示词可以写得非常聚焦工具集也可以收窄。搜索Agent只挂搜索工具分析Agent只挂数据处理工具撰写Agent只挂文本生成能力。工具冲突的问题自然消失了。2.2 角色定义的三要素目标、输入契约、输出契约我在实际项目中总结出一个角色定义的模板包含三个必须明确的要素目标这个Agent存在的唯一目的是什么用一句话说清楚不要超过20个字。比如从给定文本中提取所有日期和金额比负责信息提取工作要有效得多。输入契约它期望接收什么格式的数据是纯文本、JSON对象、还是消息列表字段有哪些必填还是可选这个契约必须在系统设计阶段就定死不能靠Agent自己猜。输出契约它必须产出什么格式的数据同样要精确到字段级别。比如分析Agent的输出必须是{dimensions: [...], comparison: {...}, confidence: 0.85}这样的结构而不是一段自由文本。下面是一个用LangGraph定义Agent节点的简化示例展示了契约如何落地from typing import TypedDict, List from langgraph.graph import StateGraph, END class SearchInput(TypedDict): query: str max_results: int class SearchOutput(TypedDict): raw_results: List[dict] source_count: int class AnalysisInput(TypedDict): raw_results: List[dict] dimensions: List[str] class AnalysisOutput(TypedDict): comparison: dict key_findings: List[str] confidence: float class AgentState(TypedDict): query: str search_output: SearchOutput analysis_output: AnalysisOutput final_report: str这个State定义就是整个系统的数据总线。每个节点只读写自己关心的字段不碰其他部分。LangGraph的StateGraph机制天然支持这种隔离——节点函数接收完整State但只返回自己更新的字段框架会自动合并。2.3 什么时候该拆什么时候不该拆不是所有任务都适合多Agent。我见过有人把写一封邮件拆成主题生成Agent正文撰写Agent语气润色Agent这就过度设计了。拆分的判断标准很简单上下文是否超过单模型有效处理范围如果任务需要的背景信息超过8000 token且这些信息可以按职责切分就值得拆。工具集是否冲突如果不同任务需要调用互斥的工具比如一个要查数据库、一个要调外部API拆开更安全。是否有明确的阶段划分如果任务天然分为收集→处理→输出这样的阶段每个阶段可以独立验证就适合拆。反过来如果任务步骤少于3步、上下文很短、工具集统一单Agent加一个好的提示词就够了。多Agent带来的编排复杂度、调试难度和延迟增加是实实在在的成本。3. 协作机制从轮流发言到有向图编排3.1 三种主流协作模式的适用场景多智能体系统的协作机制本质上是在回答控制流怎么走。目前实践中比较成熟的有三种模式顺序流水线Sequential PipelineAgent按固定顺序执行前一个的输出是后一个的输入。适合阶段明确的线性任务比如搜索→分析→撰写。优点是可控性强、调试简单缺点是缺乏反馈循环前一步错了后面全错。** supervisor 模式主管调度**有一个专门的调度Agent根据当前状态决定下一个该哪个Agent执行。适合任务路径不固定、需要动态决策的场景。比如客服系统里主管Agent根据用户问题类型决定转给技术Agent还是账单Agent。LangGraph官方文档里有完整的supervisor实现示例。辩论/协商模式Debate/Negotiation多个Agent对同一问题给出方案通过多轮讨论收敛。适合需要多视角验证的高风险决策比如代码审查、方案评估。这种模式成本最高但能有效降低单点错误。我用得最多的是supervisor模式因为它在灵活性和可控性之间平衡得最好。下面重点讲这个。3.2 Supervisor模式的节点设计与路由逻辑Supervisor模式的核心是一个路由函数它根据当前State决定下一个执行的节点。在LangGraph里这通过add_conditional_edges实现。from langgraph.graph import StateGraph, END def supervisor_router(state: AgentState) - str: 根据当前状态决定下一步 if not state.get(search_output): return search_agent if not state.get(analysis_output): return analysis_agent if not state.get(final_report): return writer_agent return END workflow StateGraph(AgentState) workflow.add_node(search_agent, search_node) workflow.add_node(analysis_agent, analysis_node) workflow.add_node(writer_agent, writer_node) workflow.set_entry_point(supervisor) workflow.add_conditional_edges( supervisor, supervisor_router, { search_agent: search_agent, analysis_agent: analysis_agent, writer_agent: writer_agent, END: END } )这个路由逻辑看起来简单但有几个关键决策点路由依据是什么我见过有人用LLM来做路由决策让模型判断下一步该谁。这在任务类型不确定时有用但代价是每次路由都要调一次模型延迟和成本都上去了。更稳的做法是用状态字段的存在性来判断——如果search_output为空就去搜索如果analysis_output为空就去分析。这种确定性路由在90%的场景下够用而且完全可预测。循环怎么处理如果分析Agent发现搜索结果不够需要重新搜索怎么办这时候路由函数要能识别需要回退的状态。我的做法是在State里加一个retry_count字段分析Agent如果判定素材不足就把search_output清空并递增retry_count路由函数看到retry_count超过阈值就强制进入撰写阶段避免死循环。并行分支怎么合流有些任务可以并行执行比如同时搜索三个不同维度。LangGraph支持并行节点但合流时需要确保所有分支都完成。这个用add_edge把多个节点指向同一个下游节点即可框架会等待所有入边完成。3.3 消息传递共享State还是显式消息队列多Agent之间的信息传递有两种设计哲学共享State所有Agent读写同一个全局状态对象。LangGraph默认就是这个模式。优点是简单直接Agent之间不需要知道彼此的存在缺点是状态膨胀快而且容易出现一个Agent改了字段另一个Agent读到脏数据的问题。显式消息队列每个Agent有独立的输入输出队列通过消息传递通信。这更接近传统分布式系统的设计隔离性好但实现复杂度高调试时需要追踪消息链路。我的建议是在LangGraph里用共享State但通过字段命名空间做逻辑隔离。比如搜索相关的字段统一加search_前缀分析相关的加analysis_前缀。这样既享受了框架的便利又避免了字段冲突。同时在State定义里用TypedDict的totalFalse标记可选字段让类型检查帮你发现遗漏。4. 用LangGraph落地多Agent系统的关键细节4.1 环境准备与依赖版本锁定LangGraph的迭代速度很快不同版本之间的API有差异。我踩过最坑的一次是本地用0.0.x版本跑通的代码部署到服务器上因为自动安装了0.1.xStateGraph的导入路径变了直接报错。所以第一件事是锁版本。我的requirements.txt里通常这样写langgraph0.2.60 langchain-core0.3.25 langchain-openai0.2.14如果你用的是其他模型提供商把langchain-openai换成对应的包即可。LangGraph本身不绑定模型它只负责编排模型调用通过LangChain的接口层完成。安装完成后用下面这段代码验证环境from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage import langgraph print(langgraph.__version__)版本号能正常打印说明基础环境没问题。4.2 State设计中的字段粒度与更新策略State的字段粒度直接决定了系统的可维护性。我见过有人把整个对话历史塞进一个messages字段所有Agent都往里面追加。结果跑到后面每个Agent的上下文里都堆了几十条无关消息模型完全抓不住重点。正确的做法是按职责分字段并且明确每个字段的更新策略字段名类型更新者更新策略说明querystr入口只写一次用户原始查询search_resultsList[dict]搜索Agent覆盖写每次搜索替换旧结果analysisdict分析Agent覆盖写分析结论draftstr撰写Agent覆盖写当前草稿retry_countint任意Agent递增重试计数errorsList[str]任意Agent追加错误日志关键点是除了errors和retry_count其他字段都是覆盖写。这意味着每个Agent在更新自己的输出字段时不需要关心之前的值是什么直接替换即可。这大大简化了状态管理逻辑。LangGraph的State合并机制默认是后写覆盖但如果你需要追加而不是覆盖可以用Annotated标记from typing import Annotated from operator import add class AgentState(TypedDict): errors: Annotated[List[str], add] retry_count: int # ... 其他字段Annotated[List[str], add]告诉LangGraph这个字段的更新方式是把新值加到旧值上而不是替换。这个细节在官方文档里藏得比较深但非常实用。4.3 节点函数的编写规范与错误处理每个节点函数本质上是一个(State) - dict的映射。它接收完整State返回要更新的字段。这里有几个我总结的规范永远不要修改传入的State。LangGraph的State在节点之间传递时框架会做浅拷贝但如果你直接修改嵌套对象比如往search_results列表里append可能会影响到其他节点。正确做法是创建新对象def search_node(state: AgentState) - dict: query state[query] try: results perform_search(query) return {search_results: results, retry_count: state.get(retry_count, 0)} except Exception as e: return { errors: [fSearch failed: {str(e)}], retry_count: state.get(retry_count, 0) 1 }错误要显式返回不要抛异常。LangGraph的节点如果抛异常整个图会中断。更好的做法是把错误写进State的errors字段让路由函数决定是重试还是跳过。这样系统有自愈能力。节点函数保持幂等。同一个节点可能因为重试被多次调用如果它有副作用比如写数据库要确保重复执行不会产生脏数据。我的做法是把副作用操作放到图的最后或者用retry_count做去重判断。4.4 调试多Agent系统的实用技巧多Agent系统的调试比单Agent难得多因为问题可能出在任何一个节点而且节点之间的交互是动态的。我常用的几个手段给每个节点加日志。在节点函数入口和出口打印State的关键字段用logging模块而不是print方便控制级别。日志格式建议包含节点名、时间戳、关键字段值。用LangGraph的可视化功能。workflow.compile().get_graph().draw_mermaid()可以生成图结构虽然这里不能用mermaid渲染但你可以把输出贴到支持mermaid的编辑器里查看。这能帮你快速发现路由配置错误。构造最小复现用例。当系统行为异常时不要在全量数据上调试。构造一个只有2-3条输入的测试用例手动追踪每个节点的输入输出通常能快速定位问题。检查State的字段类型。我遇到过因为retry_count被某个节点返回成字符串导致路由判断失效的情况。用TypedDict加运行时类型检查比如pydantic可以避免这类问题。5. 多Agent系统的性能与成本控制5.1 模型调用的次数优化多Agent系统最直接的成本来源是模型调用次数。一个三节点的流水线如果每个节点调一次模型就是3次调用。如果加上supervisor路由也用模型就是4次。如果还有重试循环可能到6-8次。优化方向有几个路由用规则不用模型。前面提到的状态字段存在性判断就是零成本路由。只有在任务类型真正不确定时才用模型做路由。合并可并行的节点。如果两个节点的输入相同、输出可以合并考虑用一个节点完成。比如提取日期和提取金额可以合并成提取结构化字段。缓存中间结果。如果同一个查询在短时间内重复出现搜索结果可以缓存。LangGraph本身不提供缓存但可以在节点函数里接一个简单的内存缓存或Redis。用小模型做简单任务。搜索Agent的查询改写、分析Agent的字段提取这些任务用7B级别的小模型就够不需要上GPT-4级别。LangGraph支持在节点级别指定不同的模型。5.2 延迟与吞吐的权衡多Agent系统的延迟是累加的搜索2秒 分析3秒 撰写5秒 10秒。如果用户期望3秒内响应这个架构就不合适。降低延迟的手段并行化独立节点如果搜索和分析之间没有依赖可以并行执行。LangGraph的并行边支持这个。流式输出撰写Agent可以用流式生成用户看到第一个字的时间会大幅提前。预计算对于可预测的查询提前把搜索结果缓存好。吞吐方面多Agent系统天然比单Agent消耗更多资源。如果要做高并发需要考虑每个Agent节点的并发控制。LangGraph本身是同步执行的高并发场景需要配合异步框架或消息队列。5.3 什么场景下多Agent反而不如单Agent我踩过的一个坑做一个新闻摘要功能本来单Agent跑得好好的我非要拆成抓取Agent摘要Agent格式化Agent。结果延迟从2秒涨到6秒成本翻了三倍输出质量没有明显提升。复盘下来判断标准是如果任务的总上下文不超过4000 token且工具集不超过3个单Agent通常是更优解。多Agent的价值在于处理单Agent搞不定的复杂度而不是单Agent能搞定但我想炫技。另一个反直觉的结论多Agent系统的输出质量不一定比单Agent高。因为信息在Agent之间传递时会有损耗每个Agent的理解偏差会累积。如果单Agent的提示词工程做得足够好它的端到端表现可能更稳定。6. 从原型到生产多Agent系统的工程化考量6.1 可观测性建设生产环境的多Agent系统没有可观测性就是盲人摸象。我建议至少记录三类数据每次运行的完整State快照。包括每个节点的输入输出、耗时、模型调用次数。这些数据可以存到结构化日志或数据库中用于事后分析。关键指标监控每个节点的平均耗时、错误率、重试率。如果某个节点的错误率突然上升说明上游数据格式变了或者模型行为漂移了。成本追踪按运行维度统计token消耗和模型调用次数。多Agent系统的成本容易失控必须有实时监控。LangGraph生态里有LangSmith这样的工具可以做追踪但如果你不想引入外部依赖自己用logging加一个简单的数据库表也能实现核心功能。6.2 版本管理与回滚策略多Agent系统的变更点很多提示词、路由逻辑、State结构、模型版本。任何一个变更都可能影响整体行为。我的做法是提示词和路由逻辑纳入版本控制每次变更记录变更原因和预期影响。State结构变更要向后兼容新增字段用可选类型删除字段先标记废弃再移除。保留最近三个版本的完整配置出问题时可以快速回滚。6.3 安全边界与输入校验多Agent系统里每个Agent都可能成为攻击入口。特别是当Agent可以调用外部工具时恶意输入可能导致工具被滥用。基本的防护措施输入校验在入口节点对用户输入做长度限制、字符过滤、注入检测。工具权限最小化每个Agent只挂它必需的工具不要图省事给所有Agent挂全套工具。输出审查在最终输出前加一个审查节点检查是否包含敏感信息或不当内容。速率限制对模型调用和工具调用做频率限制防止被刷。这些措施在单Agent系统里也重要但在多Agent系统里更关键因为攻击面更大、链路更长。6.4 测试策略单元测试、集成测试与端到端测试多Agent系统的测试要分层单元测试每个节点函数独立测试用mock数据验证输入输出契约。这层测试不调真实模型速度快覆盖率高。集成测试测试节点之间的连接和路由逻辑。用固定的State输入验证路由函数返回正确的下一个节点。端到端测试用真实模型跑完整流程验证最终输出质量。这层测试成本高不需要每次提交都跑可以在发布前执行。我通常维护一个黄金测试集包含10-20个典型输入和预期输出特征。每次修改提示词或路由逻辑后跑一遍黄金测试集对比输出变化。这能有效捕捉回归问题。7. 我踩过的三个典型坑与修复过程7.1 坑一State字段被意外覆盖导致数据丢失现象分析Agent的输出偶尔会变成空字典导致撰写Agent拿到空数据。排查过程加日志后发现搜索Agent在重试时返回了{search_results: new_results}但LangGraph的State合并机制把整个State替换了而不是只更新search_results字段。原因是搜索Agent的返回字典里包含了其他字段的默认值。根因LangGraph的State更新是浅合并但如果你返回的字典里包含了某个字段的None值它会覆盖掉之前的值。修复节点函数只返回自己负责的字段不要返回其他字段的默认值。如果某个字段不需要更新就不要出现在返回字典里。# 错误做法 return {search_results: results, analysis: None, draft: None} # 正确做法 return {search_results: results}7.2 坑二路由函数死循环导致请求超时现象某些查询下系统在搜索和分析之间无限循环直到超时。排查过程路由函数判断如果analysis为空就去分析但分析Agent在素材不足时返回了空analysis导致路由又回到搜索。搜索Agent返回的结果格式不对分析Agent再次判定素材不足形成死循环。根因缺少循环终止条件。修复在State里加retry_count路由函数检查retry_count是否超过阈值。同时分析Agent在判定素材不足时要明确设置一个analysis_insufficient标志而不是返回空analysis。def supervisor_router(state: AgentState) - str: if state.get(retry_count, 0) 3: return writer_agent # 强制进入撰写用现有素材 if not state.get(search_results): return search_agent if not state.get(analysis): return analysis_agent return writer_agent7.3 坑三模型输出格式不稳定导致解析失败现象分析Agent要求输出JSON但模型偶尔返回带markdown代码块的JSON导致解析失败。排查过程日志显示模型返回了json\n{...}\n这样的格式。虽然提示词里明确说了只输出JSON但模型并不总是遵守。根因对模型输出格式的假设过于乐观。修复在解析前做预处理剥离markdown代码块标记。同时用pydantic做输出校验解析失败时触发重试。import json import re from pydantic import BaseModel, ValidationError class AnalysisResult(BaseModel): comparison: dict key_findings: list[str] confidence: float def parse_analysis_output(raw: str) - AnalysisResult: # 剥离markdown代码块 cleaned re.sub(r^(?:json)?\s*|\s*$, , raw.strip()) try: data json.loads(cleaned) return AnalysisResult(**data) except (json.JSONDecodeError, ValidationError) as e: raise ValueError(fFailed to parse analysis output: {e})这三个坑的共同教训是多Agent系统的健壮性取决于最薄弱的那个环节。每个节点的输出都要当作不可信输入来处理做好校验和兜底。8. 多Agent架构的扩展方向与个人体会多Agent系统跑通之后扩展方向主要有几个。一是引入记忆机制让Agent能跨会话记住之前的交互这在客服、助手类场景里很有价值。LangGraph支持通过checkpointer做状态持久化配合外部向量库可以实现长期记忆。二是动态角色生成根据任务类型自动决定需要哪些Agent而不是预先写死。这需要更强的路由逻辑和角色模板管理。三是人机协同在关键节点插入人工审核LangGraph的interrupt机制可以支持这个。我个人在实际操作中的体会是多Agent系统的复杂度增长不是线性的而是指数级的。每增加一个Agent就多一层交互可能性和故障点。所以我的原则是能用两个Agent解决就不上三个能用规则路由就不上模型路由。真正需要多Agent的场景是那些单Agent确实因为上下文或工具限制搞不定的任务而不是为了架构好看。最后分享一个小技巧在开发阶段给每个Agent节点加一个debug模式开启后节点会打印完整的输入输出和耗时。这个开关通过环境变量控制生产环境关掉。这比事后翻日志效率高得多。