ARTICLE DETAIL

资讯详情

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

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

多智能体系统实战:基于LangGraph的角色分工与协作机制 1. 从单兵作战到团队协同为什么需要多智能体1.1 单智能体的天花板在哪里刚开始接触 Agent 开发的时候我也是从单智能体入手的。一个 LLM 加上几个工具套一个 ReAct 循环就能做出挺像样的东西——查资料、写代码、调 API一个人全包。但项目稍微复杂一点问题就全冒出来了。最典型的场景你让一个 Agent 同时负责需求分析、代码生成和测试验证。它会在一个上下文窗口里塞进三种完全不同的思维模式结果就是——需求分析做得不够深入代码生成时忘了前面的约束测试环节又因为上下文太长开始丢信息。这不是模型能力的问题是架构的问题。单智能体的核心瓶颈有三个。第一是上下文污染不同任务的信息互相干扰前面聊的需求细节会干扰后面写代码的判断。第二是工具过载当一个 Agent 挂载了十几个工具它在选择工具时的准确率会明显下降实测下来工具超过 8 个之后选错率就开始飙升。第三是无法并行所有事情串行处理一个环节卡住整个流程就停了。打个比方单智能体就像一个全能但只有一个人的创业公司什么都要自己干看着灵活实际上每件事都做不到极致。1.2 多智能体到底解决了什么问题多智能体的思路很直接既然一个人干不好所有事那就拆成几个人每人负责自己擅长的部分通过协作完成整体目标。具体来说多智能体架构带来三个核心收益。专业化分工让每个 Agent 的系统提示词更聚焦工具集更精简在自己领域内的表现明显优于通才 Agent。上下文隔离意味着每个 Agent 维护自己的对话历史不会互相污染代码 Agent 不需要知道需求讨论过程中的所有细节只需要拿到最终的需求文档就行。并行执行则让没有依赖关系的任务可以同时跑比如前端代码和后端代码可以两个 Agent 同时写。但这里有个关键认知多智能体不是银弹。它引入了新的复杂度——通信开销、状态同步、错误传播。如果任务本身很简单硬上多智能体就是过度设计。我的经验判断标准是当你的单 Agent 系统提示词超过 2000 字、工具超过 6 个、或者流程中有明显的阶段划分时就该考虑拆成多智能体了。1.3 什么样的项目适合多智能体架构不是所有项目都需要多智能体。我踩过的坑是早期有个简单的文档问答项目硬拆成了检索 Agent、总结 Agent、格式化 Agent 三个角色结果延迟翻了三倍效果还不如一个 Agent 加好一点的提示词。适合多智能体的场景通常有这几个特征任务可以自然分解为多个专业领域比如软件开发中的需求、设计、编码、测试不同阶段需要不同的工具集比如调研阶段需要搜索工具编码阶段需要代码执行工具流程中有明确的检查点或审核环节比如代码写完需要 review 才能合并。反过来如果你的任务是一个连续的、不可分割的推理过程比如数学证明或者单轮问答那单智能体反而是更好的选择。多智能体的价值在于协作没有协作需求就不要硬造协作。2. 角色分工的设计方法论2.1 怎么拆角色才合理角色拆分是多智能体设计中最关键的一步拆得好事半功倍拆得不好就是灾难。我总结了一个原则按职责边界拆不按功能点拆。什么意思举个例子你要做一个竞品分析系统。按功能点拆可能是搜索 Agent、爬取 Agent、分析 Agent、写报告 Agent。但这样拆的问题是搜索和爬取本质上是一件事的两个步骤硬拆开反而增加了通信成本。按职责边界拆应该是信息收集 Agent负责搜索和爬取、数据分析 Agent负责整理和分析、报告撰写 Agent负责输出。判断职责边界是否清晰的三个标准每个 Agent 的输入输出是否明确、Agent 之间是否需要频繁来回沟通、一个 Agent 的失败是否会影响其他 Agent 的核心逻辑。如果两个 Agent 之间需要来回沟通超过三次才能完成一个任务那大概率它们应该合并。2.2 常见角色类型与职责定义在实际项目中我遇到过和设计过的角色类型大致可以归为几类。规划者负责拆解任务、制定执行计划它通常不直接干活而是输出一个结构化的任务列表。执行者是真正干活的角色比如代码生成、文档撰写、数据查询。审核者负责检查执行者的输出质量发现问题后打回重做。协调者在多 Agent 需要并行工作时负责汇总结果、解决冲突。以 LangGraph 为例一个典型的软件开发多智能体系统可能包含这些角色角色职责核心工具输出物需求分析师理解用户需求输出需求文档无特殊工具结构化需求文档架构师根据需求设计技术方案知识库检索技术方案文档编码 Agent按方案写代码代码执行、文件读写代码文件测试 Agent验证代码正确性代码执行、测试框架测试报告审核 Agent最终质量把关无特殊工具审核意见这张表的关键在于每个角色的输出物是下一个角色的输入形成清晰的流水线。工具分配也遵循最小权限原则需求分析师不需要代码执行工具编码 Agent 不需要搜索工具。2.3 角色提示词的设计要点每个 Agent 的系统提示词决定了它的行为边界。我见过最常见的错误是提示词写得太泛比如“你是一个有帮助的助手”——这种提示词在多智能体系统里等于没写。好的角色提示词应该包含四个要素。身份定义要具体到专业领域不是“你是程序员”而是“你是一个专注于 Python 后端开发的工程师擅长 FastAPI 和 PostgreSQL”。职责边界要明确说什么该做什么不该做“你只负责生成代码不负责运行测试测试由测试 Agent 完成”。输出格式要严格约定因为下游 Agent 需要解析你的输出“以 JSON 格式返回包含 code 和 explanation 两个字段”。协作协议要说明遇到问题怎么办“如果需求不明确在输出中标记 NEED_CLARIFICATION 并说明具体问题”。这里有个实操技巧提示词里的输出格式约定最好用具体的 schema 而不是自然语言描述。比如用 Pydantic 模型定义输出结构然后在提示词里直接引用这个 schema这样下游解析的成功率会高很多。3. 协作机制的核心实现3.1 通信模式消息传递还是共享状态多智能体之间的通信有两种基本模式。消息传递是 Agent A 直接把结果发给 Agent B像打电话。共享状态是所有 Agent 读写同一个状态对象像白板。LangGraph 采用的是共享状态模式所有 Agent 节点读写同一个 State。这种模式的好处是解耦——Agent A 不需要知道谁会消费它的输出只需要把结果写到 State 的对应字段就行。坏处是 State 的结构设计变得很关键字段太多会导致每个 Agent 都要处理大量无关信息。我的经验是流程固定的用共享状态流程动态的用消息传递。比如软件开发流水线需求→设计→编码→测试流程是确定的用共享状态最清晰。但如果是客服系统用户可能问任何问题需要动态路由到不同 Agent那消息传递加路由更合适。在 LangGraph 里State 通常用 TypedDict 定义from typing import TypedDict, Annotated from langgraph.graph import add_messages class DevState(TypedDict): messages: Annotated[list, add_messages] requirements: str design_doc: str code: str test_report: str current_phase: str注意Annotated[list, add_messages]这个写法它告诉 LangGraph 这个字段是追加而不是覆盖。这个细节很关键我当初没注意导致消息历史一直被覆盖调试了半天才发现。3.2 编排模式流水线、路由还是层级协作的编排方式决定了 Agent 之间的调用关系。常见的三种模式各有适用场景。流水线模式最简单Agent 按固定顺序依次执行A 的输出是 B 的输入。适合流程确定的场景比如内容生产选题→写作→编辑→发布。实现上用 LangGraph 的线性图就行每个节点是一个 Agent。路由模式需要一个路由 Agent 根据当前状态决定下一步调用哪个 Agent。适合分支逻辑多的场景比如客服系统根据用户意图路由到技术支持、退款处理或投诉建议。路由 Agent 本身通常不干活只做决策。层级模式是树状结构顶层 Agent 负责拆解任务把子任务分配给下层 Agent下层 Agent 还可以继续往下分。适合复杂项目比如一个完整的软件项目顶层是项目经理 Agent下面是前端、后端、测试三个组长 Agent每个组长再管理具体的执行 Agent。实际项目中这三种模式经常混用。我做过的一个系统就是顶层用层级模式做任务拆解每个子团队内部用流水线模式执行团队之间用路由模式做协调。3.3 状态管理与上下文传递多智能体系统里状态管理是最容易出 bug 的地方。核心问题是每个 Agent 需要看到多少上下文。全量共享是最简单的做法所有 Agent 都能看到完整的 State。但这样会导致两个问题上下文窗口浪费每个 Agent 都要处理大量无关信息信息泄露比如审核 Agent 看到了编码 Agent 的中间思考过程可能产生偏见。我的做法是按需裁剪。每个 Agent 节点在执行前从全局 State 中提取自己需要的字段组装成自己的上下文。比如编码 Agent 只需要 requirements 和 design_doc不需要看到需求分析阶段的原始对话记录。def coding_agent(state: DevState): # 只提取需要的字段 context { requirements: state[requirements], design: state[design_doc] } # 执行编码逻辑 result llm.invoke(build_prompt(context)) # 只写回自己负责的字段 return {code: result, current_phase: coding_done}这种做法的另一个好处是当某个 Agent 需要升级或替换时只要保持输入输出接口不变内部实现随便改。4. 基于 LangGraph 的实操落地4.1 环境搭建与项目结构先说环境。LangGraph 的依赖很轻量核心就是langgraph和langchain-core但如果要用到具体的模型和工具还需要装对应的包。pip install langgraph langchain-core langchain-openai项目结构我建议这样组织这是踩过坑之后总结的project/ ├── agents/ │ ├── __init__.py │ ├── planner.py # 规划 Agent │ ├── coder.py # 编码 Agent │ ├── tester.py # 测试 Agent │ └── reviewer.py # 审核 Agent ├── graph/ │ ├── __init__.py │ ├── state.py # State 定义 │ └── workflow.py # 图编排 ├── tools/ │ └── __init__.py ├── config.py └── main.py每个 Agent 一个文件State 定义单独放图编排单独放。这样当 Agent 数量增多时不会乱。我见过把所有东西塞一个文件的超过三个 Agent 之后基本没法维护。4.2 定义 State 与 Agent 节点State 的设计原则是字段对应流程中的关键产物而不是 Agent 的内部状态。比如 requirements、design_doc、code 这些是产物而“编码 Agent 当前在第几轮迭代”这种是内部状态不应该放进全局 State。from typing import TypedDict, Annotated, Literal from langgraph.graph import add_messages class TeamState(TypedDict): messages: Annotated[list, add_messages] task: str plan: str code: str test_result: str review_status: Literal[approved, rejected, pending] iteration: int每个 Agent 节点本质上就是一个函数接收 State 返回 State 的更新部分from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o, temperature0) def planner_node(state: TeamState): prompt f你是一个技术规划师。根据以下任务制定执行计划 任务{state[task]} 输出格式 1. 任务拆解 2. 技术选型 3. 关键风险 response llm.invoke(prompt) return {plan: response.content, iteration: 0}这里有个细节temperature0对于需要稳定输出的 Agent 很重要特别是规划者和审核者。编码 Agent 可以稍微高一点0.2 左右让它有一点灵活性。4.3 构建协作图与条件路由图编排是 LangGraph 的核心。先定义节点再定义边最后编译成可执行图。from langgraph.graph import StateGraph, END workflow StateGraph(TeamState) # 添加节点 workflow.add_node(planner, planner_node) workflow.add_node(coder, coder_node) workflow.add_node(tester, tester_node) workflow.add_node(reviewer, reviewer_node) # 设置入口 workflow.set_entry_point(planner) # 添加边 workflow.add_edge(planner, coder) workflow.add_edge(coder, tester) # 条件边测试通过去审核不通过回编码 def test_router(state: TeamState): if PASS in state[test_result]: return reviewer return coder workflow.add_conditional_edges( tester, test_router, {reviewer: reviewer, coder: coder} ) # 审核条件边 def review_router(state: TeamState): if state[review_status] approved: return END return coder workflow.add_conditional_edges( reviewer, review_router, {coder: coder, END: END} ) app workflow.compile()这段代码里最关键的是两个条件路由。测试不通过回编码审核不通过也回编码形成一个带反馈的循环。但这里有个坑必须设置最大迭代次数否则可能无限循环。我一般会在 State 里加 iteration 字段每次回退加一超过阈值就强制结束并报错。4.4 加入 Human-in-the-Loop实际生产环境里完全自动化的多智能体系统风险很高。LangGraph 提供了 interrupt 机制可以在关键节点暂停等人工确认后再继续。from langgraph.checkpoint.memory import MemorySaver # 编译时加入 checkpointer memory MemorySaver() app workflow.compile( checkpointermemory, interrupt_before[reviewer] # 审核前暂停 ) # 执行 config {configurable: {thread_id: task-001}} result app.invoke({task: 写一个用户登录接口}, config) # 人工检查后继续 app.invoke(None, config)interrupt_before指定在哪些节点前暂停。我通常会在两个地方加人工介入一是规划完成后确认计划没问题再往下走二是最终审核前人工看一眼代码质量。这样既保留了自动化的效率又避免了完全失控的风险。5. 常见问题与排查技巧实录5.1 Agent 之间信息传递丢失这是最高频的问题。表现是下游 Agent 拿不到上游的输出或者拿到的是空值。排查思路按这个顺序来。先检查 State 字段名是否一致。上游写的是design_doc下游读的是design这种拼写不一致在 TypedDict 里不会报错但运行时就是拿不到数据。我建议用常量定义字段名避免手写字符串。再检查返回值格式。LangGraph 的节点函数返回的字典会被合并到 State 里但如果你返回的是{design_doc: None}它会把原来的值覆盖成 None。所以返回前要确认值不为空。最后检查条件路由。如果路由函数返回了未在映射中定义的 keyLangGraph 会静默失败后续节点不会执行。这个坑我踩过调试了半小时才发现是路由返回值写错了。5.2 无限循环与死锁多智能体系统里Agent A 等 Agent B 的输出Agent B 又等 Agent A 的输出就死锁了。或者测试不通过回编码编码改了测试还是不通过来回循环。防死锁的核心是设置迭代上限。在 State 里加一个计数器每次循环加一超过阈值就强制走异常出口。def test_router(state: TeamState): if state[iteration] 5: return force_end if PASS in state[test_result]: return reviewer return coder另一个技巧是给回退加约束。不要让编码 Agent 无限制地重写而是在回退时把测试报告作为反馈传给它让它有针对性地修改。我实测下来带反馈的回退比盲目重写平均迭代次数从 4.2 次降到 1.8 次。5.3 输出格式不一致导致解析失败下游 Agent 需要解析上游的输出但 LLM 的输出格式经常不稳定。今天返回 JSON明天返回 Markdown后天加了一段解释文字。解决方案是结构化输出。LangChain 支持用 Pydantic 模型约束输出格式from pydantic import BaseModel, Field class CodeOutput(BaseModel): code: str Field(description生成的代码) explanation: str Field(description代码说明) dependencies: list[str] Field(description依赖列表) structured_llm llm.with_structured_output(CodeOutput) result structured_llm.invoke(prompt) # result 一定是 CodeOutput 类型字段一定存在用了结构化输出之后解析失败率从大概 15% 降到了接近 0。代价是稍微增加了一点延迟但完全值得。5.4 常见问题速查表问题现象可能原因排查方法解决方案下游拿不到数据字段名不一致打印 State 内容统一用常量定义字段名无限循环缺少迭代上限看 iteration 值加计数器强制退出解析失败输出格式不稳定检查原始输出用结构化输出Agent 选错工具工具描述不清看工具调用日志优化工具描述减少工具数响应太慢串行执行看各节点耗时无依赖的节点并行化结果质量差提示词太泛检查系统提示词加具体身份和输出约束6. 多智能体系统的进阶优化6.1 记忆机制的设计多智能体系统里记忆分两层。短期记忆是当前任务的上下文存在 State 里任务结束就清空。长期记忆是跨任务的知识积累比如用户偏好、历史决策、常见错误模式。LangGraph 的 checkpointer 天然支持短期记忆每个 thread_id 对应一个独立的会话。长期记忆需要自己实现我通常用向量数据库存历史任务的摘要在新任务开始时检索相关经验注入到规划 Agent 的提示词里。这里有个经验长期记忆不要存原始对话要存结构化的经验条目。比如“用户 X 偏好用 TypeScript 而不是 JavaScript”、“上次类似任务在数据库连接池配置上出过问题”。原始对话太长检索效率低而且噪音大。6.2 错误处理与降级策略多智能体系统里一个 Agent 失败不应该导致整个系统崩溃。我的做法是每个 Agent 节点都有 try-catch失败时返回一个标记了错误的状态由路由决定是重试、跳过还是终止。def coder_node(state: TeamState): try: result structured_llm.invoke(prompt) return {code: result.code, error: None} except Exception as e: return {code: , error: str(e)}然后在路由里判断def after_coder(state: TeamState): if state.get(error): if state[iteration] 3: return retry_coder return fallback return tester降级策略也很重要。如果编码 Agent 连续失败可以降级到用一个更简单的模板生成代码或者直接转人工处理。关键是不要让系统卡死。6.3 性能优化并行与缓存多智能体系统天然比单 Agent 慢因为多了通信和编排开销。优化手段主要有两个。并行执行没有依赖关系的 Agent 可以同时跑。LangGraph 支持并行节点比如前端代码和后端代码可以两个 Agent 同时生成。实测下来并行化能把总耗时降低 40% 左右。缓存相同的输入不需要重复调用 LLM。我通常用 LangChain 的缓存机制对规划、审核这类确定性高的节点开启缓存。编码节点因为需要多样性一般不缓存。from langchain_core.caches import InMemoryCache from langchain_core.globals import set_llm_cache set_llm_cache(InMemoryCache())这个缓存对开发调试特别有用改下游逻辑的时候不用每次都重新跑上游的 LLM 调用省时省钱。6.4 评估与迭代多智能体系统上线后怎么知道它好不好我一般从三个维度评估。任务完成率是最基本的多少任务最终成功完成。平均迭代次数反映效率迭代次数越多说明 Agent 之间的配合越差。人工介入率反映自动化程度需要人工干预的比例越低越好。评估数据从 LangGraph 的 trace 里拿每个节点的输入输出、耗时、状态变化都有记录。我习惯每周跑一次评估看指标趋势如果某个指标恶化就针对性优化对应的 Agent。迭代的时候一次只改一个 Agent改完跑评估对比。同时改多个 Agent 会导致无法定位是哪个改动带来的效果变化这是血泪教训。7. 我踩过的坑与实战心得7.1 不要过早引入多智能体这是我最大的教训。有个项目一开始就设计了五个 Agent结果开发了两周发现其中三个 Agent 的职责用提示词工程就能在一个 Agent 里解决。多智能体带来的通信开销和调试复杂度远远超过了它带来的收益。正确的做法是先用单 Agent 跑通遇到明确的瓶颈再拆。拆的时候也要一个一个拆拆一个验证一个不要一次性全拆。7.2 提示词比架构更重要我见过太多人花大量时间设计复杂的 Agent 拓扑却用着敷衍的提示词。实际上一个提示词写得好的单 Agent效果往往超过提示词写得差的多 Agent 系统。每个 Agent 的提示词都值得反复打磨。我的习惯是先写一版跑 20 个测试用例看失败案例针对性修改提示词再跑。通常迭代 3-5 轮才能稳定。7.3 日志和可观测性是生命线多智能体系统的调试难度是单 Agent 的好几倍。没有完善的日志出了问题根本不知道是哪个 Agent 的哪一步出了错。我的做法是每个 Agent 节点的输入输出都打日志包括完整的提示词和原始响应。用 LangSmith 或者自己搭一个简单的日志系统都行。关键是出问题时能快速定位。7.4 从小处着手逐步扩展最后一个心得不要一上来就搞复杂的层级编排和动态路由。先从最简单的流水线开始两个 AgentA 的输出给 B。跑通了再加第三个。加条件路由加循环加人工介入。每一步都验证稳定了再往下走。我现在的习惯是新项目先用一个周末搭一个两 Agent 的最小原型验证核心流程能跑通再花时间扩展。这样即使方向错了损失也有限。多智能体系统本质上是在用工程手段弥补单个 LLM 的能力边界。它不是越复杂越好而是在合适的地方用合适的复杂度。理解每个 Agent 的能力边界设计清晰的协作协议做好错误处理和可观测性剩下的就是迭代打磨了。
返回列表