news 2026/10/21 8:09:29

LangGraph多代理实战:代码评审系统的架构设计与状态管理

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
LangGraph多代理实战:代码评审系统的架构设计与状态管理

最近不少做应用侧的开发者拿着自己的Agent项目来找我对方案,问题几乎都长一个样:单代理越写越复杂,效果反而越来越差。上下文一乱,改了提示词,绕回来还是乱;工具调用一多,同一个模型上挂几十个函数,参数错配频繁到让人怀疑人生。聊到最后我发现,大多数人的问题不在于“会不会写Agent”,而在于把全部职能塞进了同一个代理里。这类场景最直接的解法,就是用LangGraph把单代理拆成多个各司其职的子代理,也就是Multi-Agent架构。这篇文章我会用一个完整的代码评审案例,把LangGraph Multi-Agent从设计、编码到问题排查的全流程讲清楚。

不管你现在是在做信息检索类应用,还是准备把大模型接进内部系统,只要一个任务能拆成多个独立环节,这篇文章的思路都能直接套用。我会少讲空泛概念,多给能直接落地的代码、图形设计和避坑经验。特别是那些被单代理的上下文冲突、工具调度混乱折磨过的同学,看完这篇应该会有一种“终于可以把锅甩给架构”的解脱感。

1. 单代理的“失控现场”与多代理的核心思路

1.1 那些年我们踩过的单代理深坑

单代理失控不是偶发问题,而是规模效应必然导致的。我先说一个最常见的悲剧:你要做一个带代码评审能力的助手,一开始给它定义了一个角色,让它既要读代码,又要检查安全漏洞,还要顺带生成文档。刚开始demo跑得很顺,但用户一旦把真实仓库的diff丢进来,问题马上集中爆发。

第一个坑是上下文冲突。代码评审需要的是“挑刺”视角,安全检查需要的是“漏洞”视角,文档生成需要的是“总结”视角。这三个视角在同一个system prompt里会互相污染。模型一会儿觉得自己是架构师,一会儿又切换成安全审计员,最终输出变成一个四不像。第二个坑是指令膨胀。为了覆盖所有职责,你不停往prompt里加规则,到最后几百行约束混在一起,模型真正记住的只剩最后几条。第三个坑是工具爆炸。代码检索、漏洞库查询、文档更新、仓库操作全部绑在同一个LLM上,工具数量一大,模型经常把参数张冠李戴,明明要查CVE编号,它给你把关键词塞进了文件路径参数里。

第四个坑最隐蔽:调试无从下手。出错了以后,你根本说不清是角色定义的问题、指令冲突的问题,还是工具调用的问题。改prompt跟抽卡一样,改一次祈祷一次。

这些坑的本质,是我们让一个代理同时承担了太多职责。复杂系统天然需要多人分工,Agent也不例外。

1.2 多代理的本质:角色分离与状态隔离

多代理架构的核心可以归纳成两个词:角色分离、状态隔离。角色分离是指每个子代理只负责一个高度内聚的子任务,拥有独立的system prompt和工具集。状态隔离是指子代理之间不共享全部上下文,只传递结构化的工作成果,而不是把一整段对话历史甩给下一个代理。

打个生活化的比方。单代理就像一个全栈全能员工,什么事情都自己扛,流程一旦复杂就手忙脚乱。多代理则是一个项目组:有人负责需求分析,有人负责开发,有人负责测试,有人负责汇总报告。每个人只掌握自己需要的信息,项目进度通过会议纪要和交接文档推进。这个比喻基本就是多代理架构的真实形态。

LangGraph的价值在于,它用“图”的方式把这些角色和状态串起来。你可以显式定义节点(每个节点是一个函数或一个Agent)、边(执行顺序)、状态(跨节点的数据传递)和条件路由(根据中间结果决定下一步)。它没有让Agent变得无所不能,但给了你一个可控的执行框架。出了问题你能一眼看出是哪个环节挂了,这是Multi-Agent架构相比单Agent最大的生产力优势。

2. LangGraph多代理不是什么魔法:核心概念速览

2.1 StateGraph、Node、Edge:一张图看懂LangGraph

LangGraph的核心概念不多,四个概念就能讲明白:StateGraph是整张图的“总控容器”,管理所有状态;Node是图中的一个执行单元,可以是一个普通函数、一个LangChain Agent或任何可调用对象;Edge是节点之间的连线,控制执行顺序;State是跨节点传递的全局状态,通常在定义图时用TypedDict声明。

一个最小流程示例是这样的:

from typing import TypedDict from langgraph.graph import StateGraph, START, END class MyState(TypedDict): input_text: str output_text: str def process_node(state: MyState) -> dict: # 模拟一个处理步骤 return {"output_text": state["input_text"].upper()} graph = StateGraph(MyState) graph.add_node("process", process_node) graph.add_edge(START, "process") graph.add_edge("process", END) app = graph.compile() result = app.invoke({"input_text": "hello multi-agent"}) print(result["output_text"])

每次invoke,LangGraph会从START节点进入,按有向图的顺序执行节点,每个节点的返回值会更新全局State,最终走到END。你可以把State想象成一个在各岗位之间流转的工单,每个节点只修改工单上的某个字段,然后交给下一个节点。就是这样一个简单的模型,组合上条件分支后,能力会超出你的想象。

2.2 三种主流多代理协作模式的取舍

多代理的协作方式五花八门,但我用下来最常见的是三种模式。

Supervisor模式,类似“领导分配任务”。一个Supervisor Agent负责接收用户需求,把它拆解成多个明确的子任务,分派给底下的Worker Agent,最后汇总结果。这个模式适合任务边界清晰、分工固定的场景。优势是职责清晰、易扩展;劣势是Supervisor拆任务的提示词需要反复调,拆不好容易给Worker派了模糊指令。

Pipeline模式,类似“流水线”。任务按固定顺序流经多个处理节点,每个节点只做一件事,把结果传给下一个。这个模式适合流程稳定、不需要中途回环的场景,比如“文本清洗 -> 摘要生成 -> 格式转换”。优势是简单可靠,劣势是不够灵活,中间节点想根据结果跳步就麻烦了。

层级协作模式,类似“多级管理”。顶层Agent把任务下发给中层Agent,中层继续拆给底层Agent,结果逐级汇总。这种模式适合任务规模大、子任务还能继续拆的场景。优势是能处理超复杂任务,劣势是层级多了延迟高、token消耗大,调试也费劲。

三种模式不是互斥的,实际项目中经常混合使用。下面要讲的代码评审案例,本质就是“Pipeline + Supervisor”的混合:专家Agent按流水线依次处理,最后由一个Supervisor节点汇总所有报告。

3. 案例实战:搭建“代码评审多代理系统”

3.1 需求拆解:四个角色怎么分工

就用我最近在模拟项目X里落地的一个系统来讲。需求很简单:用户提交一段代码变更(diff),系统自动产出一份代码评审报告,包括逻辑Bug、安全隐患、文档更新建议三个维度。如果单代理做,它又要懂业务代码,又要懂安全,又要会写文档,效果很难保证。多代理拆法是将系统拆成四个角色。

第一个是协调者Coordinator,负责接收初始diff,调度其他Agent,并把所有子结果汇总成最终报告。第二个是代码审查Agent,focus在代码逻辑,检查潜在Bug、异常处理、可维护性,它只关心代码质量和架构合理性。第三个是安全审查Agent,专门看安全视角,包括注入风险、敏感信息泄露、依赖漏洞。第四个是文档Agent,根据代码变更内容和前两个Agent的发现,生成或更新对应的技术文档说明。

这四个角色背后是四个不同的系统提示词、四套不同的工具集、四个相互隔离的上下文。代码审查Agent不需要知道安全漏洞库的查询语法,安全审查Agent也不需要在代码风格规则上浪费时间。这就是角色分离带来的第一层红利:每个模型实例都在一个窄任务上运行,专注度明显提高。

3.2 状态Schema与多代理图结构设计

先定义全局状态。我在这个案例里没有让子代理共享messages字段,而是给每个角色单独建了输出字段:

from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages from langgraph.graph import StateGraph, START, END class CodeReviewState(TypedDict): diff_content: str # 输入:代码变更内容 code_issues: List[str] # 代码审查Agent的输出 security_issues: List[str] # 安全审查Agent的输出 doc_updates: List[str] # 文档Agent的输出 analysis_history: Annotated[List[str], add_messages] # 记录子Agent分析动作 final_report: str # 协调者生成的最终报告

这里的重点在于:每个子代理负责不同的字段,输出不会互相覆盖。analysis_history我用了add_messages这个Reducer,它在每次节点返回时会自动把新消息追加到历史列表里,而不是覆盖旧消息。这一步是后面不踩状态污染坑的关键。

图的结构和方向是:

  • START -> 代码审查Agent
  • 代码审查Agent -> 安全审查Agent
  • 安全审查Agent评估风险等级,通过条件路由决定:风险高则回到代码审查Agent再深入检查,风险低则继续
  • 安全审查Agent -> 文档Agent
  • 文档Agent -> 协调者Coordinator
  • Coordinator -> END

你可以先画一个手写流程图,把“哪些节点之间有边、哪些地方要条件判断”写清楚,再动手写代码。图设计阶段多花十分钟,后面调试能省两个小时。

3.3 子代理节点与Coordinator节点实现

先看代码审查Agent节点的核心实现。它要做两件事:基于diff内容做代码质量分析,然后调用一个历史问题检索工具看看有没有类似issue。

from langchain_core.tools import tool @tool def search_known_issues(keyword: str) -> str: """在历史问题库中搜索与该关键词相关的已知Bug。""" # 实际实现中可以替换为向量检索接口 return f"找到与"{keyword}"相关的历史问题2条" def code_review_node(state: CodeReviewState) -> dict: prompt = ( "你是资深代码评审工程师。审查以下代码变更,只关注逻辑正确性、" "边界条件、异常处理和可维护性。输出JSON数组,每项包含severity" "和description字段。不要讨论安全问题,那是另一个专家的职责。\n\n" f"代码变更:\n{state['diff_content']}" ) llm_with_tools = llm.bind_tools([search_known_issues]) response = llm_with_tools.invoke(prompt) # 实际生成中需要解析工具调用,这里为了演示做了简化 issues = parse_issues(response.content) return {"code_issues": issues}

代码里面我只塞了针对代码质量的prompt,并把工具集限定为一个。这个节点没有权限去查漏洞库,也不可能去更新文档。这种“权限最小化”是工具编排的基本原则,能直接减少模型在工具之间乱跳的概率。

安全审查节点的结构类似,但prompt主题完全不同,工具绑定也不同。它会输出security_issues字段。文档节点则输入前面两个节点的结构化发现,以及原始diff,输出doc_updates字段。最后看协调者Coordinator节点:

def coordinator_node(state: CodeReviewState) -> dict: report_prompt = f""" 你是技术评审负责人。下面是三位专家对同一份代码变更的结论: - 代码质量结论:{state['code_issues']} - 安全结论:{state['security_issues']} - 文档更新建议:{state['doc_updates']} 请整合成一份带执行优先级的技术评审报告,用Markdown格式输出。 不要补充专家没有提到的内容。 """ final = llm.invoke(report_prompt) return {"final_report": final.content}

节点与节点之间的调用链已经完整了。有一个重要细节:每个子节点返回的字段名必须在StateSchema里声明过,否则LangGraph更新状态时会直接报错。这既是约束,也是保护,逼着你提前把数据契约设计好。

3.4 条件路由:让系统自己决定深入审查还是直接收尾

光有固定顺序还不算真正的图编排,条件路由才是LangGraph的灵魂。我在这个案例里设计了一个回环逻辑:安全审查Agent发现高风险项时,需要回到代码审查Agent再深入检查一次,看看这道安全隐患是不是因为某个代码设计缺陷导致的。

def route_after_security(state: CodeReviewState) -> str: # 判断标准:高风险问题数量大于等于3,并且没有超过最大重试次数 high_risk_count = len( [s for s in state.get("security_issues", []) if s.get("severity") == "high"] ) if high_risk_count >= 3 and state["analysis_history"].get("retry_count", 0) < 2: return "more_review" return "to_docs" graph.add_node("code_review", code_review_node) graph.add_node("security", security_node) graph.add_node("docs", docs_node) graph.add_node("coordinator", coordinator_node) graph.add_edge(START, "code_review") graph.add_edge("code_review", "security") graph.add_conditional_edges( "security", route_after_security, { "more_review": "code_review", # 回到代码审查,二次深入 "to_docs": "docs", # 正常流程,进入文档节点 }, ) graph.add_edge("docs", "coordinator") graph.add_edge("coordinator", END)

这样实现后,系统特别接近真实团队的运作方式:先常规审查,发现严重安全风险就回炉重查,查完再写文档,最后汇总。条件路由让图具备了“基于中间结果自适应”的能力,还因为设置了重试次数上限,不会出现无限循环。如果你有多个条件分支,记得在route函数里穷举所有返回路径,否则LangGraph会报“分支目标不匹配”的运行时错误。

4. 状态管理与工具编排的进阶细节

4.1 Reducers机制:为什么不能用普通dict覆盖状态

LangGraph状态更新的默认行为是“覆盖”:当某个节点返回的dict里有某个key时,这个key的整个旧值会被新值替换。这在很多场景下是对的,但在多代理协作里,多个节点写同一个字段的情况太常见了。比如两个子Agent都想往findings字段里追加自己的结论,如果没有Reducer,后面那个Agent会把前面那个Agent的结论清零。

LangGraph提供的解决方案是Reducer函数。在TypedDict字段声明中,用Annotated[字段类型, reducer函数]来告诉框架“两个值相遇时怎么合并”。

def merge_findings(existing: list, new: list) -> list: return existing + new class BetterState(TypedDict): findings: Annotated[List[str], merge_findings]

再看看LangChain官方常见的add_messages,它本质上就是一个Reducer:把新消息追加到消息列表后面,而不是覆盖旧消息。凡是多个节点要共享“记录性数据”的,都应该用Reducer;凡是节点之间传递“结果性数据”的,比如final_report这种只被写一次、只被读一次的字段,用普通覆盖没问题。这个判断很重要,盲目使用Reducer也可能造成状态无限膨胀。

我在实际做模拟项目X时踩过的坑是这样的:两个子Agent都返回了messages字段,结果因为没加Reducer,后跑的Agent把前面Agent的记忆全覆盖了,导致最终汇总时缺失了一半数据。检查了半小时才发现是字段声明的问题。所以多代理的状态设计原则很简单:每个角色一个专属输出字段,需要累计的历史记录加Reducer,不要在状态里塞大块全文。

4.2 工具集隔离与上下文传递的实务建议

多代理不是单纯把Prompt拆碎,工具也必须跟着隔离。代码审查Agent只绑定代码检索和issue搜索工具,安全Agent只绑定漏洞库查询工具,文档Agent只绑定文档读写工具。这样每个Agent待调用的工具数量从二十个降到两三个,模型在工具选择上的认知负担大幅下降。

上下文传递方面,我建议遵循“只传结论,不传过程”。子Agent经过多轮推理后,把最终结论整理成结构化JSON再写入State;下一个Agent拿到的是一份干净的结构化数据,而不是几十轮对话日志。这能显著控制token消耗。有人会问,那消息历史怎么办?放到analysis_history里但没有必要传给后续每个Agent。Coordinator汇总时只读code_issues、security_issues和doc_updates这些结构化字段,完全不需要看原始对话过程。

还有一个容易忽略的细节:State里放引用,不放大文件。如果diff内容超长,不要直接把全文塞进State,可以存一个仓库路径和分支号,子Agent需要时自己去读。State越大,图的执行越慢,LangGraph的checkpoint开销也越大。这个优化在长期运行的生产系统里收益非常明显。

5. 多代理开发避坑实录:问题定位与调试技巧

5.1 四个高频事故现场复盘

第一个事故是状态覆盖。前面提过,多个节点写同一个key导致数据互相覆盖。修复方式就是给累计型字段加Reducer,或者拆成独立字段。排查时看每个节点返回的key和在StateSchema里声明的对应关系,基本能一眼锁定。

第二个事故是递归跑飞了。条件路由设计不完整或重试次数没设上限,图在几个节点之间反复转圈,直到触发LangGraph的RecursionLimit。我遇到过最夸张的一次是图循环了49次才被强制终止。修复方式很简单:条件路由函数里加计数器,到达上限后强制走正常出口。写条件边时一定要问自己:每个可能的返回值都映射到了节点吗?循环路径有终止条件吗?

第三个事故是子代理输出格式不稳定。有的Agent返回的JSON里多了一个逗号,有的把severity字段拼成了severty,解析端拿到后直接崩溃。这个问题在单Agent时代也存在,但多代理时代影响更大,因为下游Agent拿到的是一堆无法解析的字符串。修复方式是用模型的结构化输出能力或JsonOutputParser,在子Agent节点内部就强制格式校验,而不是等到汇总节点才处理。

第四个事故是上下文越来越长。我见过有人把每个Agent的全部对话历史都塞进State,导致每次invoke的token消耗成倍增长。修复方式是控制State字段粒度、只传结构化结果、必要时用摘要型Agent压缩长上下文。

这四个事故基本覆盖了多代理开发中80%的问题。下面给一张速查表方便快速定位:

现象可能原因快速处理方式
结果数据缺失节点写入了未声明字段,或字段被覆盖检查StateSchema字段声明,给累计字段加Reducer
运行不结束或报RecursionLimit条件路由缺少终止条件设重试上限,穷举路由映射表
下游节点解析失败上游输出不是稳定结构化数据用结构化解析器或JSON模式约束输出
Token消耗异常上涨State里塞入了大量历史消息只传结果摘要,不传原始对话记录
节点行为不符合预期Prompt粒度过粗,职责不聚焦拆分子Agent,缩小每个节点的职责范围

5.2 问题排查三板斧与调试环境配置

第一板斧是“脱离图去测节点”。把每个节点当成普通Python函数,手动构造一个假state,传入节点函数,断言返回的字段对不对。不要每次都跑整张图,那样出了问题很难定位是哪一环。我经常写一个小型测试脚本,对四个节点分别跑一遍,再测条件路由函数,最后才集成测试整图。

第二板斧是“检查状态快照”。LangGraph提供了获取中间状态的能力,可以在运行过程中打印或查看每一步更新后的State,确认哪个节点的输出没有被正确写入。

第三板斧是“开启可观测性”。在LangGraph应用里接入LangSmith或OpenTelemetry,可以看到节点执行时序、各节点耗时、Token消耗和工具调用明细。有一次我怀疑某个子Agent根本没正确接收上游数据,打开追踪后发现它的输入context里完全缺少对应字段,问题一目了然。

调试环境配置上,建议把recursion_limit调小,比如先设成6,跑通流程之后再恢复默认值。这样即使路由出问题,也不会白白等上几分钟。还有一个实用小技巧:在条件路由函数里临时加一个print,输出返回的路径名。这种“打点式调试”虽然土,但定位条件分支问题特别快。

多代理开发日志最好统一格式,每个节点输出的JSON里都带上source节点名和时间戳。我习惯在子Agent的返回内容里加一个agent_name字段,Coordinator汇总时可以根据这个字段做来源标注,排查问题时也能知道每段结论是哪个专家给的。

结尾

这套代码评审多代理系统在我这边的模拟项目X里已经跑过不少真实仓库的diff,我的直观感受是:多代理并没有让大模型变聪明,它只是把“聪明”用在了刀刃上。每个Agent只操心自己那点事,上下文干净,工具少而精准,出问题也知道该去找谁负责。这种可控感,是单Agent模式下怎么调prompt都换不来的。

如果你也想把自己的单Agent应用拆成多代理,我的建议是别一上来就搭一个五个节点十个工具的大系统。先从最痛的一个场景开始,拆成两个节点跑通图结构,再加上条件路由,最后慢慢补角色和工具。状态设计上宁可多开几个字段,也别图省事让所有Agent共用messages,否则你很快会回来读这篇避坑清单。

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/10/21 8:09:29

基于Python的文言文详解文档结构化与自动出题实践

简介&#xff1a;《小学文言文阅读50篇详解》是一份面向小学生的文言文学习文档&#xff0c;旨在通过经典短篇的逐篇解析&#xff0c;帮助孩子掌握文言基础知识、提升阅读理解能力&#xff0c;同时兼顾品格教育与传统文化熏陶。资源包内含1个docx文件&#xff0c;大小约86KB&am…

作者头像 李华
网站建设 2026/10/21 8:09:10

LoadRunner与JMeter实战对比:从脚本效率到压测调优的全面解析

做性能测试这些年&#xff0c;被问得最多的一个问题就是&#xff1a;LoadRunner和JMeter到底该选哪个&#xff1f;网上的对比文章一搜一大堆&#xff0c;但大部分都是功能清单罗列&#xff0c;看完了还是不知道怎么落地。这篇文章我不想重复整理对照表&#xff0c;而是想站在实…

作者头像 李华
网站建设 2026/10/21 8:08:19

从免费额度到统一接入:GLM 5.3 Flash 在聚合平台的实战指南

几年前&#xff0c;开发者想给项目配一个“能写代码的模型”&#xff0c;通常意味着走完复杂的商务流程&#xff1a;申请预算、签合同、等审批&#xff0c;然后拿到的 Key 还有一堆调用限制。现在情况变了&#xff1a;不少模型厂商把轻量档位放进免费或极低价格区间&#xff0c…

作者头像 李华
网站建设 2026/10/21 8:08:14

循环工程(Loop Engineering):大模型生成重复的检测与打破实践

项目做到一半&#xff0c;模型开始原地复读。我当时在生成一篇长文&#xff0c;先是同一个词在连续三句话里反复出现&#xff0c;接着整段复读&#xff0c;最后连着章节标题都开始自我循环。那是我第一次认真面对一个事实&#xff1a;在语言模型应用里&#xff0c;最隐蔽的故障…

作者头像 李华
网站建设 2026/10/21 8:07:45

UWB+IMU融合定位:TDOA与UKF的MATLAB实现及锚点自适应

我们直接进入正题。UWB配合IMU做定位融合&#xff0c;在二维环境里跑通精度和稳定性兼顾的方案&#xff0c;这是目前室内定位、机器人导航、AGV调度里非常常见的需求。如果你看过不少论文&#xff0c;大概率会发现它们都在提“TDOA”和“UKF”&#xff0c;但真正能把这两者结合…

作者头像 李华
网站建设 2026/10/21 8:07:27

基于Matlab的水箱液位级联控制动态系统模型与整定

水箱液位级联控制的动态系统模型&#xff0c;这是我用Matlab搭过最典型的一套过程控制仿真。项目本身不算复杂&#xff1a;一个水箱、一个调节阀、一台流量计&#xff0c;外环测液位&#xff0c;内环测流量&#xff0c;两个PID串起来&#xff0c;目标就是让液位顶住进水扰动。真…

作者头像 李华