ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

多智能体协作实战:基于LangGraph的角色分工与协作机制

多智能体协作实战:基于LangGraph的角色分工与协作机制 1. 从单打独斗到团队作战为什么需要多智能体1.1 单智能体的天花板在哪里刚开始接触 Agent 开发的时候我也是从单智能体入手的。一个 LLM 加上几个工具套一个 ReAct 循环确实能跑通不少场景——查资料、写代码、做总结看起来什么都能干。但项目稍微复杂一点问题就全暴露出来了。最典型的症状是上下文爆炸。你让一个 Agent 同时负责需求分析、代码生成、测试验证它每一步都要把前面所有对话历史带上。跑到第十轮的时候Prompt 已经塞了几万 token模型开始忘事前面说过的约束后面就不遵守了。这不是模型不行是单智能体的架构本身就不适合长链路、多阶段的复杂任务。另一个问题是角色混淆。同一个 Agent 既要当产品经理理解需求又要当程序员写代码还要当测试挑毛病。你让它自己检查自己的输出它往往觉得我写得挺好的因为它的上下文里全是自己刚才的推理过程存在严重的自我确认偏差。这就像让一个人既当运动员又当裁判很难做到客观。还有一个隐性成本工具集膨胀。单智能体要处理所有任务就得挂载所有工具。工具一多模型选择工具的准确率就下降。实测下来当工具数量超过 15 个左右选错工具的概率明显上升。而且每个工具的描述都占 token进一步挤压了有效上下文。1.2 多智能体到底解决了什么问题多智能体的核心思路其实很朴素把一个大而全的 Agent 拆成几个小而专的 Agent每个只干一件事通过明确的协作机制串起来。这跟软件工程里的微服务拆分是一个道理——单体应用拆成微服务每个服务职责单一通过 API 通信。具体来说多智能体带来了几个实打实的好处。第一是上下文隔离。每个 Agent 只维护自己那一段上下文研究员 Agent 不需要知道代码生成的细节代码 Agent 也不需要背着研究阶段的全部资料。这样每个 Agent 的 Prompt 都能保持精简模型注意力集中输出质量自然更稳。第二是职责明确带来的可验证性。当审查者是一个独立的 Agent它的上下文里没有作者的推理过程它只能看到最终产出这时候它的批评就客观得多。这也是为什么很多代码生成系统会专门设一个 Review Agent效果比让生成者自检好很多。第三是可组合性和可扩展性。你想加一个新能力不用改现有 Agent 的逻辑直接加一个新角色的 Agent 挂到协作图里就行。这种松耦合让系统演进变得容易。1.3 什么场景适合上多智能体不是所有任务都值得上多智能体。我的经验是满足以下条件之一才考虑拆分任务有明显的阶段划分比如先研究、再规划、再执行、再验证任务需要多种专业视角比如既要技术可行性又要商业价值任务链路长到单智能体上下文扛不住或者任务对输出质量要求高需要独立的审查环节。反过来如果就是一个简单的问答或者单步工具调用硬拆成多智能体纯属给自己找麻烦——通信开销、调试复杂度、延迟都会上去。我见过有人把查天气拆成三个 Agent这就属于过度设计。2. 角色分工的设计方法论2.1 怎么划分角色才合理角色划分是多智能体设计里最关键的一步划得好系统顺畅划得不好就是互相甩锅。我的方法论是按认知职能划分而不是按业务模块划分。什么叫按认知职能划分就是看这个任务需要哪几种不同的思维方式。比如一个内容创作系统需要的认知职能是信息收集发散思维、结构规划收敛思维、内容生成创造思维、质量审查批判思维。这四种思维方式差异很大适合拆成四个 Agent。而按业务模块划分就容易出问题。比如你按前端后端数据库拆每个 Agent 都要同时做规划、写代码、自测那每个 Agent 内部还是单智能体的老问题等于没拆。一个实用的判断标准如果两个角色的 Prompt 里系统提示词高度重叠那它们可能不该拆开如果两个角色的系统提示词几乎完全不同那拆开就是对的。2.2 常见角色类型盘点在实际项目里我总结出几类高频出现的角色你可以根据任务直接套用或组合角色类型核心职责典型系统提示词要点规划者 Planner拆解任务、制定步骤输出结构化的任务列表不执行具体操作研究者 Researcher收集信息、检索资料只负责找信息不做判断和决策执行者 Executor调用工具、完成具体操作严格按规划执行不擅自改变方案审查者 Reviewer检查产出、挑错以批判视角审视必须给出具体问题协调者 Supervisor调度其他 Agent、汇总结果决定下一步交给谁不亲自干活反思者 Reflector复盘过程、提炼经验分析失败原因输出改进建议这里要特别说一下协调者和规划者的区别很多人会混淆。规划者是一次性的它在任务开始时制定一个计划就退场了协调者是持续性的它在整个执行过程中不断决定下一步该谁上。LangGraph 里的 Supervisor 模式就是典型的协调者角色。2.3 角色数量的取舍角色不是越多越好。我踩过的坑是一开始设计得很细搞了七八个 Agent结果调试的时候根本不知道是哪一环出的问题而且 Agent 之间的通信成本高得离谱一个简单任务要跑十几轮才结束。后来我总结出一个经验值中小型任务 2-4 个角色复杂任务 5-7 个角色超过 7 个就要考虑分层了。分层的意思是先分成几个大组每个组内部再有自己的小多智能体。比如一个软件开发系统顶层是需求组开发组测试组开发组内部再分前端 Agent后端 Agent。角色合并的原则是如果两个角色的交互极其频繁且合并后 Prompt 还能保持清晰那就合并。比如规划者和协调者在很多简单场景下可以合并成一个。3. 协作机制的核心实现3.1 三种主流协作拓扑多智能体的协作机制本质上就是定义谁把控制权交给谁。目前主流的有三种拓扑结构。第一种是流水线式Pipeline。Agent 按固定顺序执行A 做完交给 BB 做完交给 C。这种最简单可控性最强适合阶段划分明确的任务。缺点是灵活性差中间某一步出问题不好回退。第二种是监督者式Supervisor。有一个中心协调者它根据当前状态决定下一个该谁执行。这种灵活性好能动态调整是 LangGraph 里最常用的模式。缺点是协调者本身可能成为瓶颈而且协调者的决策质量直接影响全局。第三种是网络式Network。Agent 之间可以自由通信没有中心节点。这种最灵活但也最难控制容易出现无限循环或者通信风暴。实际项目里用得比较少除非是研究性质的多智能体强化学习场景。我的建议是新手从流水线式入手理解了再上监督者式网络式谨慎使用。大部分生产场景监督者式已经够用了。3.2 状态传递Agent 之间怎么对话多智能体协作的核心技术问题是Agent 之间怎么传递信息这里有两种主流做法。一种是消息传递式Agent 之间通过消息队列通信每个 Agent 有自己的收件箱。这种方式解耦彻底但实现复杂而且状态分散在各处调试困难。另一种是共享状态式所有 Agent 读写同一个状态对象StateLangGraph 就是这种模式。状态对象里定义了各种字段比如messages、plan、current_step、research_result等。每个 Agent 节点执行时读取需要的字段执行完把结果写回状态。共享状态式的好处是状态集中、可追溯、易调试。你可以在任何一步打印完整状态看清楚每个 Agent 到底读到了什么、写了什么。这也是我推荐 LangGraph 的主要原因之一。状态设计有个关键原则字段要按信息类型划分而不是按Agent划分。比如不要设计成agent_a_output、agent_b_output而应该设计成research_findings、code_artifacts、review_comments。这样即使以后换了 Agent 实现状态结构也不用改。3.3 用 LangGraph 搭建协作骨架LangGraph 的核心概念就三个State状态、Node节点、Edge边。State 是共享数据Node 是 Agent 执行单元Edge 定义流转关系。理解了这三个多智能体的骨架就搭起来了。先看一个最基础的状态定义from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages class TeamState(TypedDict): messages: Annotated[List, add_messages] task: str plan: List[str] current_step: int research_result: str final_output: str next_agent: str这里add_messages是 LangGraph 提供的一个 reducer它的作用是追加而不是覆盖。默认情况下节点返回的字段会覆盖旧值但 messages 这种需要累积的字段就要用 reducer 指定合并方式。这是新手最容易踩的坑之一——不加 reducer消息历史会被覆盖Agent 就失忆了。然后是节点定义。每个节点本质上就是一个函数输入是 State输出是要更新的字段def planner_node(state: TeamState): task state[task] # 调用 LLM 生成计划 plan llm.invoke(f把以下任务拆解成步骤{task}) return {plan: plan, next_agent: researcher} def researcher_node(state: TeamState): step state[plan][state[current_step]] result llm.invoke(f研究这个步骤所需的信息{step}) return {research_result: result, next_agent: executor}最后是图的组装graph StateGraph(TeamState) graph.add_node(planner, planner_node) graph.add_node(researcher, researcher_node) graph.add_node(executor, executor_node) graph.add_node(reviewer, reviewer_node) graph.set_entry_point(planner) graph.add_conditional_edges( researcher, lambda s: s[next_agent], {executor: executor, planner: planner} ) graph.add_edge(executor, reviewer) graph.add_edge(reviewer, END) app graph.compile()add_conditional_edges是实现动态路由的关键。它接收一个路由函数根据当前状态返回下一个节点的名字。这就是监督者模式的技术实现——路由函数扮演了协调者的角色。3.4 条件路由与循环控制多智能体系统里循环是常态。审查者发现问题要打回给执行者重做这就是一个循环。但循环必须要有终止条件否则就是死循环。我的做法是双重保险一是设置最大迭代次数二是设置质量阈值。比如审查者打分低于 7 分就打回重做但同时限制最多重做 3 次超过 3 次就强制通过并标记需人工介入。def route_after_review(state: TeamState): score state.get(review_score, 0) retry_count state.get(retry_count, 0) if score 7 or retry_count 3: return end return executor这里有个实操心得重试次数不要设太大。我一开始设了 5 次结果发现模型在同一个问题上反复犯同样的错重试 5 次和重试 2 次的效果差不多但成本翻了一倍多。后来改成 2-3 次性价比最高。如果 3 次还搞不定说明是任务本身或者 Prompt 有问题该人工介入了。4. 实操过程与核心环节实现4.1 环境准备与依赖安装先把环境搭起来。我用的是 Python 3.10LangGraph 对版本有一定要求太老的 Python 会有兼容问题。pip install langgraph langchain langchain-openai如果你要用 LangSmith 做可观测性强烈推荐多智能体调试没它基本没法干活再加一个pip install langsmith环境变量配置export OPENAI_API_KEYyour-key export LANGCHAIN_TRACING_V2true export LANGCHAIN_API_KEYyour-langsmith-key export LANGCHAIN_PROJECTmulti-agent-demo提示LangSmith 的 tracing 在多智能体场景下几乎是必需品。单智能体你还能靠 print 调试多智能体十几个节点来回跳没有可视化追踪根本理不清执行路径。4.2 完整的多智能体协作案例我以一个技术方案调研与撰写任务为例完整走一遍。这个任务需要调研技术现状、分析优缺点、撰写方案文档、审查质量。对应四个角色Researcher、Analyst、Writer、Reviewer外加一个 Supervisor 做调度。先定义完整状态from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages class DocState(TypedDict): messages: Annotated[List, add_messages] topic: str research_notes: str analysis: str draft: str review_feedback: str review_score: int retry_count: int next_agent: strSupervisor 节点的实现是核心它要根据当前进度决定下一步def supervisor_node(state: DocState): if not state.get(research_notes): return {next_agent: researcher} if not state.get(analysis): return {next_agent: analyst} if not state.get(draft): return {next_agent: writer} if state.get(review_score, 0) 7 and state.get(retry_count, 0) 3: return {next_agent: writer, retry_count: state.get(retry_count, 0) 1} return {next_agent: end}注意这里 Supervisor 没有调用 LLM而是用纯逻辑判断。这是个重要优化——能用代码逻辑判断的就不要浪费 LLM 调用。LLM 调用又慢又贵Supervisor 如果每轮都调 LLM 决策延迟会明显上去。只有当决策真的需要语义理解时才让 LLM 介入。Writer 节点要能接收审查反馈并修改def writer_node(state: DocState): feedback state.get(review_feedback, ) if feedback: prompt f根据以下反馈修改方案文档 反馈{feedback} 原稿{state[draft]} 调研{state[research_notes]} 分析{state[analysis]} else: prompt f基于以下材料撰写方案文档 调研{state[research_notes]} 分析{state[analysis]} draft llm.invoke(prompt) return {draft: draft, review_feedback: }Reviewer 节点要输出结构化评分def reviewer_node(state: DocState): prompt f审查以下文档从完整性、准确性、可读性三个维度打分1-10 并给出具体修改建议。输出格式 分数X 建议... 文档{state[draft]} result llm.invoke(prompt) # 解析分数 score parse_score(result) return {review_score: score, review_feedback: result}4.3 关键参数的选择与计算多智能体系统里有几个参数直接决定成本和效果我一个个说。最大迭代次数。这个前面提过建议 2-3 次。计算逻辑是假设单次执行成本为 C成功率为 p那么重试 n 次的期望成本是 C × (1 (1-p) (1-p)² ...)。当 p0.7 时重试 3 次的期望成本约 2.2C重试 5 次约 2.4C边际收益递减明显。上下文窗口分配。每个 Agent 的 Prompt 要控制长度。我的经验是系统提示词控制在 500 token 以内任务上下文控制在 2000 token 以内留给输出的空间至少 1000 token。如果某个 Agent 的输入经常超限说明状态设计有问题该拆字段了。温度参数。不同角色用不同温度。Researcher 和 Writer 可以高一点0.7-0.8鼓励发散Analyst 和 Reviewer 要低0.2-0.3保证严谨。这个细节很多人忽略但对输出质量影响不小。超时设置。每个节点都要设超时防止某个 Agent 卡死拖垮全局。一般 LLM 调用设 60 秒工具调用设 30 秒。4.4 执行现场记录与观察跑起来之后通过 LangSmith 能看到完整的执行轨迹。一个典型的成功执行是这样的第一轮Supervisor 判断缺 research_notes路由到 Researcher。Researcher 调用搜索工具返回约 800 字的调研笔记写入状态。第二轮Supervisor 判断缺 analysis路由到 Analyst。Analyst 读取 research_notes输出约 600 字的分析包含三个技术方案的对比。第三轮Supervisor 判断缺 draft路由到 Writer。Writer 综合前两者输出约 1500 字的方案文档。第四轮Supervisor 路由到 Reviewer。Reviewer 打分 6 分指出缺少成本分析和技术对比不够具体两个问题。第五轮Supervisor 发现分数低于 7 且 retry_count 为 0路由回 Writer。Writer 带着反馈重写补充了成本章节。第六轮Reviewer 重新打分 8 分通过。Supervisor 路由到 end。整个流程 6 轮耗时约 45 秒token 消耗约 12000。如果换成单智能体硬扛同样的任务大概要 3-4 万 token而且质量还不一定有这么好。5. 常见问题与排查技巧实录5.1 死循环与无限重试这是多智能体最常见的问题。症状是任务跑了几十轮还不结束token 哗哗地烧。排查思路分三步。第一步看是不是路由条件写错了。比如审查者永远返回低分或者 Supervisor 的判断逻辑有漏洞。我遇到过一次Reviewer 的分数解析函数有 bug永远返回 0导致无限重试。这种问题看 LangSmith 的轨迹一眼就能发现。第二步看是不是状态没更新。如果 Writer 修改后没有正确写回 draft 字段Reviewer 看到的还是旧版本自然一直打低分。这时候要检查节点的返回值确保字段名和状态定义一致。第三步加硬性熔断。不管逻辑对不对先设一个全局最大轮次比如 20 轮超过就强制结束并报警。这是保底措施生产环境必须有。5.2 Agent 之间信息丢失症状是后面的 Agent 抱怨没有收到前面的信息或者输出明显没用到前面的结果。根本原因通常是状态字段没接上。比如 Researcher 写的是research_result但 Analyst 读的是research_notes字段名不一致Analyst 读到的是空值。我的做法是给状态定义写单元测试。每个节点跑完后断言它该写的字段确实写进去了该读的字段确实有值。这个习惯帮我省了无数调试时间。另一个原因是reducer 配置错误。如果某个字段需要累积但没配 reducer后一个 Agent 的写入会覆盖前一个。messages 字段一定要配add_messages其他累积型字段也要配对应的 reducer。5.3 输出质量不稳定有时候同样的输入跑两次结果差异很大。这通常是温度参数和Prompt 稳定性的问题。先检查温度。Reviewer 和 Analyst 这类需要稳定输出的角色温度调到 0.2 以下。Writer 可以高一点但也不要超过 0.8。再检查 Prompt 里有没有引入不确定性。比如请给出一些建议就比请给出 3 条建议要飘。能约束格式的地方就约束格式让模型输出结构化内容稳定性会好很多。还有一个隐蔽原因是上下文顺序。同样的信息放在 Prompt 开头和放在结尾模型的使用效果不一样。关键约束建议放在系统提示词里具体任务信息放在用户消息里。5.4 常见问题速查表问题现象可能原因排查方法解决方案无限循环路由条件错误看 LangSmith 轨迹修正路由逻辑加熔断信息丢失字段名不一致打印状态对象统一字段命名消息被覆盖缺 reducer检查状态定义加 add_messages输出飘忽温度过高检查 LLM 配置降低温度某节点卡死无超时设置看节点耗时加超时和重试成本失控重试次数过多统计 token 消耗限制重试次数协调者决策差Supervisor 用 LLM看决策日志改用代码逻辑判断5.5 几个独家避坑技巧第一个技巧给每个 Agent 加自我介绍。在系统提示词里明确写你是团队中的 XX 角色你只负责 XX不要做 XX。这能有效防止 Agent 越界干活。我见过 Writer 自作主张去改调研结论的加了角色约束后就老实了。第二个技巧状态里加一个trace字段记录每个 Agent 的关键决策。调试的时候直接看 trace比翻 LangSmith 还快。生产环境可以关掉但开发阶段强烈建议开着。第三个技巧先用假数据跑通流程再接真 LLM。多智能体的骨架调试和 LLM 质量调试要分开。我一开始就是混在一起调出了问题根本分不清是流程 bug 还是模型不行。后来改成先用 mock 函数跑通所有路由再接真模型效率高了一倍不止。第四个技巧Reviewer 的 Prompt 要毒舌一点。默认的 LLM 倾向于说好话你要在系统提示词里明确要求以最严格的标准审查必须找出至少 3 个问题。不然 Reviewer 很容易变成橡皮图章起不到把关作用。6. 从能跑到好用进阶优化方向6.1 记忆机制的设计基础版的多智能体是无状态的每次任务都从零开始。但实际项目里很多信息需要跨任务保留比如用户的偏好、历史决策、常见错误模式。这就需要引入记忆机制。我的做法是分两层短期记忆用状态对象任务结束就清空长期记忆用向量库把重要的结论、偏好存起来下次任务开始时检索注入。LangGraph 本身不直接提供记忆但可以配合 LangChain 的 memory 模块或者自己接向量库。要注意的是长期记忆不能什么都存。我踩过的坑是存太多检索出来的都是噪音反而干扰了当前任务。后来改成只存经过验证的结论和用户明确表达的偏好效果就好多了。6.2 人工介入节点有些关键决策不能让 Agent 自己拍板需要人工确认。LangGraph 提供了interrupt机制可以在指定节点暂停等人工输入后再继续。from langgraph.checkpoint.memory import MemorySaver app graph.compile( checkpointerMemorySaver(), interrupt_before[executor] )这样在进入 executor 之前会暂停人工确认计划没问题后再放行。这个机制在涉及敏感操作比如发邮件、改数据库时特别有用。我的建议是凡是不可逆的操作前面都加一个人工确认节点。6.3 成本与延迟的平衡多智能体天然比单智能体贵、慢因为多了通信和协调开销。优化方向有几个。一是并行化。如果两个 Agent 之间没有依赖关系就让它们并行跑。比如 Researcher 和 Analyst 如果都只依赖原始任务可以同时启动。LangGraph 支持并行节点能显著降低延迟。二是模型分级。不是所有 Agent 都需要用最强的模型。Supervisor 的路由决策、简单的格式转换用小模型就够了。只有 Writer、Analyst 这种需要深度思考的角色才上大模型。实测下来这种分级能省 40% 左右的成本。三是缓存。相同的输入直接返回缓存结果。对于调研类任务很多查询是重复的缓存命中率能到 30% 以上。6.4 评估与持续迭代多智能体系统上线后怎么知道它好不好靠感觉不行得有量化指标。我一般关注四个指标任务成功率最终通过审查的比例、平均轮次越少越好、平均成本token 消耗、人工介入率需要人干预的比例。这四个指标每周统计一次看趋势。如果成功率下降先看是不是模型更新导致的再看是不是 Prompt 需要调整。如果轮次上升通常是某个 Agent 的输出质量下降了导致下游反复重试。定位到具体 Agent 后针对性优化它的 Prompt 或换模型。这套评估机制听起来麻烦但它是多智能体系统从能跑到好用的必经之路。没有度量就没有优化这是我做了多个项目后最深的体会。最后分享一个我个人的经验多智能体的复杂度是单智能体的好几倍所以能用单智能体解决的就别上多智能体。只有当单智能体确实扛不住的时候多智能体的价值才体现出来。我见过太多项目本来一个 Agent 加几个工具就能搞定非要拆成五个 Agent结果维护成本高得吓人效果还没单智能体好。架构选择要服务于问题而不是反过来。
返回列表