资讯详情 LangGraph 实战:用状态图构建带分支与循环的多步 Agent
📅 2026/11/4 20:59:10
Agent 开发这两年火得一塌糊涂但真正动手写过多步协作逻辑的人都知道最难受的不是调模型而是“流程怎么管”。一个能查资料、能算数、能根据中间结果决定下一步干什么的 Agent本质上是一张会自己走的状态图可大多数人上手时还在用 if-else 硬堆堆到第三层分支就开始怀疑人生。LangGraph 就是冲着这个痛点来的——它把 Agent 的执行过程抽象成“节点 边 共享状态”让你用画流程图的方式写代码。这篇内容适合两类人一类是刚接触 Agent、想找个能跑通的最小闭环的新手另一类是已经用链式写法踩过坑、想换成状态图思路的中级开发者。我会从概念讲到可运行代码把状态设计、节点拆分、条件路由、循环控制这些关键环节拆开揉碎最后附上我实际调试时踩过的坑和排查表。看完你至少能独立搭出一个带分支和循环的多步 Agent而不是停留在“调个 API 就完事”的阶段。1. 先搞清楚 LangGraph 到底在解决什么问题1.1 从“链”到“图”为什么线性流程不够用大部分人入门 Agent 开发时接触的第一个范式是“链”——把几个步骤串成一条直线输入 → 调模型 → 解析结果 → 输出。这种写法在单轮任务里很顺手但一旦任务需要“根据结果决定下一步”链式结构就开始露怯。举个具体场景你让 Agent 处理一个用户提问它需要先判断这个问题要不要查资料要查就调检索工具查完再判断资料够不够不够就换个关键词再查一次够了才生成答案。这条路径里有判断、有回退、有循环用链式写法你得嵌套一堆条件分支代码很快就变成一团乱麻。LangGraph 的核心思路是把这种流程建模成有向图每个处理步骤是一个节点Node节点之间的跳转是一条边Edge整张图共享一份状态State。节点读取状态、修改状态边决定下一步去哪个节点。这么一抽象原本藏在 if-else 里的控制流就被显式画了出来你能一眼看出“哪里会分叉、哪里会绕回来”。我个人的体会是图结构最大的价值不是让代码变短而是让流程变得可读、可调试、可局部替换——你想换掉某个环节只动那一个节点就行不用碰整条链。这里要强调一个容易混淆的点LangGraph 不是替代大模型调用框架的它管的是“编排”。模型调用、工具调用这些具体动作还是你自己写LangGraph 负责的是“什么时候调谁、调完去哪”。把它理解成一个流程调度器更准确别指望它帮你写 prompt。1.2 状态图式 Agent 的三个核心概念刚上手时只要抓住三个词后面所有代码都是它们的组合。State状态整张图共享的一份数据通常是个字典或者带类型定义的类。所有节点都能读它也能返回增量去更新它。状态设计得好不好直接决定后面写起来顺不顺。我的经验是状态里至少要放三类东西原始输入、中间产物、控制标记比如“还需要再查一次吗”这种布尔位。Node节点一个普通的函数入参是当前状态出参是要合并进状态的更新。节点本身不关心自己从哪来、到哪去它只负责“拿到状态、干活、交回结果”。这种职责单一的设计让每个节点都能单独测试这点在实际项目里非常救命。Edge边决定节点之间的走向。普通边是固定的“A 完去 B”条件边Conditional Edge则是“根据状态里的某个值决定去 B 还是去 C”。条件边是整张图的大脑Agent 的“决策能力”基本都体现在这里。把这三个概念串起来看状态是黑板节点是往黑板上写字的人边是决定下一个谁来写字的规则。理解了这层LangGraph 就没有神秘感了。1.3 它和常见编排方式的取舍对比很多人会问我用普通函数调用、用现成的链式框架不行吗能行但适用场景不同。下面这张表是我根据实际项目整理的对比帮你判断什么时候该上图结构。编排方式适合场景主要短板什么时候该换纯函数顺序调用步骤固定、无分支一有判断就写死出现第二个条件分支时链式框架线性流水线、单轮任务循环和回退表达别扭需要“重试直到满足条件”时状态图LangGraph多步决策、带循环与分支初期设计成本略高任务本身就有决策逻辑时我的判断标准很简单只要你的流程里出现“根据中间结果决定下一步”或者“可能要重复做某件事直到达标”就该考虑状态图。反过来如果任务就是一条直线硬上 LangGraph 反而是过度设计多出来的状态定义和边配置只会增加维护负担。工具是拿来解决问题的不是拿来炫技的。2. 动手前的环境与状态设计2.1 依赖安装与最小可运行环境环境这块没什么玄学Python 3.9 以上就行建议用虚拟环境隔离。核心依赖就一个 LangGraph 包模型调用部分看你用哪家我示例里用通用的对话模型接口你替换成自己手头的即可。python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install langgraph装完之后先别急着写业务跑一个“两节点直线图”验证环境通不通这是我一直推荐的习惯。很多人一上来就写复杂逻辑结果报错时分不清是环境问题还是逻辑问题白白浪费时间。最小验证图长这样from langgraph.graph import StateGraph, END from typing import TypedDict class S(TypedDict): text: str def node_a(state: S): return {text: state[text] - A} def node_b(state: S): return {text: state[text] - B} g StateGraph(S) g.add_node(a, node_a) g.add_node(b, node_b) g.set_entry_point(a) g.add_edge(a, b) g.add_edge(b, END) app g.compile() print(app.invoke({text: start}))跑通会输出start - A - B。这一步的意义在于确认三件事状态能正常传递、节点能正常执行、END能正常终止。三件事都对了再往上叠业务逻辑才踏实。提示END是一个特殊常量代表图的终点。忘了给某条路径接上END编译时可能不报错但运行时会卡住或者抛异常这是新手最常见的低级错误之一。2.2 状态字段怎么设计才不返工状态设计是整张图的地基改起来最伤筋动骨所以值得多花十分钟想清楚。我一般按“输入 / 中间 / 控制”三段式来规划字段下面用一个“带检索的问答 Agent”举例说明。from typing import TypedDict, List class AgentState(TypedDict): question: str # 输入用户原始问题 docs: List[str] # 中间检索到的资料 answer: str # 中间生成的答案 retry_count: int # 控制已重试次数 need_retry: bool # 控制是否需要再查一次几个设计要点值得展开说。第一控制字段一定要显式存在状态里不要藏在函数内部的局部变量里否则条件边读不到路由就无从谈起。第二中间产物尽量结构化比如资料用列表而不是拼成一个大字符串后面节点处理起来方便得多。第三给循环加计数器这是防止死循环的保险丝后面讲循环控制时会详细说。还有一个坑我踩过状态字段的类型标注别偷懒用Any。LangGraph 在合并状态更新时依赖类型信息标注清楚能帮你提前发现“节点返回了错误类型”这类问题。用TypedDict定义比裸字典强很多编辑器还能给你补全提示。2.3 节点拆分粒度粗一点还是细一点节点拆多细没有标准答案但有个实用原则一个节点只做一件能被单独描述清楚的事。比如“检索资料”是一个节点“判断资料够不够”是另一个节点别把两件事塞进一个函数。拆细的好处是每个节点都能单独 invoke 测试出问题时定位快坏处是节点太多图会变得琐碎状态字段也跟着膨胀。我的经验值是一个中等复杂度的 Agent节点数量控制在 4 到 8 个之间比较舒服。低于 4 个往往意味着某个节点承担了过多职责高于 8 个就要考虑是不是该拆成子图了。拆分的另一个隐性收益是可复用——比如“格式化输出”这种节点换个项目直接搬过去就行。3. 核心机制条件边、循环与状态合并3.1 条件边让 Agent 学会“看情况办事”条件边是整张图最关键的零件写法上分两步先写一个路由函数它读状态、返回一个字符串标签再在add_conditional_edges里把这个标签映射到目标节点。下面这段是“资料够不够决定是否重查”的典型实现。def route_after_check(state: AgentState) - str: if state[need_retry] and state[retry_count] 3: return retry return generate g.add_conditional_edges( check, route_after_check, { retry: retrieve, # 回去再查 generate: generate, # 直接生成 }, )路由函数必须是纯判断不要在里面做业务操作否则调试时你会分不清“是判断错了还是操作错了”。返回的标签要和映射表里的键严格对应拼错一个字母就会在运行时抛KeyError而且报错信息不一定直观这点要留意。条件边还有一个容易忽略的细节它不改变状态。路由函数只读不写真正改状态的是它指向的节点。理解这一点你就不会在路由函数里写state[xxx] ...这种无效操作了。3.2 循环控制重试逻辑怎么写才不死循环Agent 里最常见的循环就是“重试直到满足条件”。LangGraph 允许你把边指回前面的节点形成环但必须自己加终止条件框架不会帮你兜底。我见过不止一个新手写出“检索 → 判断不够 → 再检索 → 判断不够”的无限循环跑起来直接把额度烧光。防死循环的标准做法是“计数器 上限判断”双保险。计数器在每次进入重试分支时自增路由函数里同时检查“是否还需要重试”和“是否已达上限”。上面route_after_check里的state[retry_count] 3就是这道保险。计数器自增要放在节点里做比如检索节点每次执行时retry_count 1这样逻辑集中、不易漏。def retrieve(state: AgentState): # 模拟检索实际替换成你的工具调用 new_docs [fdoc_{state[retry_count]}] return { docs: state[docs] new_docs, retry_count: state[retry_count] 1, }注意上限值别拍脑袋定。检索类任务一般 2 到 3 次足够再多往往说明问题本身有歧义继续重试也是浪费。上限到了还没满足条件应该走“降级生成”而不是报错用户体验会好很多。3.3 状态合并规则为什么你的字段被覆盖了这是新手最容易懵的地方明明节点 A 往列表里加了一条到了节点 C 发现只剩最后一条。原因在于 LangGraph 默认对状态字段做覆盖式合并——节点返回什么值就直接替换掉原字段而不是追加。列表字段尤其容易中招。解决办法是给字段指定合并策略。用Annotated配合operator.add可以让列表字段变成“追加”语义from typing import Annotated import operator class AgentState(TypedDict): question: str docs: Annotated[List[str], operator.add] # 追加而非覆盖 retry_count: int加上Annotated[List[str], operator.add]之后节点返回{docs: [新条目]}就会自动拼到原列表后面不用手动state[docs] [...]。这个机制理解透了状态管理会顺很多。反过来像retry_count这种标量字段用默认覆盖语义正合适别给它也加operator.add否则计数会越加越离谱。4. 完整实操搭一个带分支和循环的问答 Agent4.1 整体流程图与节点职责划分这一节把前面的零件组装成一个能跑的完整例子。任务设定用户提问 → 判断是否需要检索 → 需要就检索 → 检查资料是否充分 → 不充分且未超限就重查 → 充分则生成答案。节点职责如下表。节点职责读状态写状态decide判断问题是否需要检索questionneed_retry 初值retrieve调用检索、累加计数question, retry_countdocs, retry_countcheck评估资料是否充分docsneed_retrygenerate基于资料生成答案question, docsanswer这张表建议你在动手前先画出来节点之间的数据依赖一目了然写代码时就不会出现“这个字段哪来的”这种困惑。职责划分清楚之后每个节点都是十几行的小函数测试起来毫无压力。4.2 逐节点实现与关键代码注释先定义状态再逐个写节点。为了让例子能独立运行检索和生成部分用模拟逻辑代替你替换成真实的工具调用和模型调用即可。from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, END class AgentState(TypedDict): question: str docs: Annotated[List[str], operator.add] answer: str retry_count: int need_retry: bool def decide(state: AgentState): # 简单规则问题里带查字就认为需要检索 need 查 in state[question] return {need_retry: need} def retrieve(state: AgentState): # 模拟检索真实场景替换为向量库或搜索工具 fake_doc f关于「{state[question]}」的第{state[retry_count] 1}份资料 return { docs: [fake_doc], retry_count: state[retry_count] 1, } def check(state: AgentState): # 模拟评估资料少于2份就认为不充分 enough len(state[docs]) 2 return {need_retry: not enough} def generate(state: AgentState): context .join(state[docs]) if state[docs] else 无参考资料 return {answer: f基于[{context}]对「{state[question]}」的回答是……}几个实现细节值得点出来。retrieve里返回的docs是单元素列表靠operator.add自动追加所以不用手动拼接。check只改need_retry不碰其他字段职责很干净。generate里对空资料做了兜底避免join报错——这种边界处理在实际项目里能省掉不少线上问题。4.3 组装图、编译与运行验证节点齐了接下来连边、设入口、编译。注意条件边的映射要覆盖路由函数所有可能的返回值漏一个就会在特定分支上崩。g StateGraph(AgentState) g.add_node(decide, decide) g.add_node(retrieve, retrieve) g.add_node(check, check) g.add_node(generate, generate) g.set_entry_point(decide) # decide 之后需要检索就去 retrieve否则直接 generate g.add_conditional_edges( decide, lambda s: retrieve if s[need_retry] else generate, {retrieve: retrieve, generate: generate}, ) g.add_edge(retrieve, check) # check 之后不够且未超限就重查否则生成 def route_after_check(state: AgentState) - str: if state[need_retry] and state[retry_count] 3: return retry return generate g.add_conditional_edges( check, route_after_check, {retry: retrieve, generate: generate}, ) g.add_edge(generate, END) app g.compile() result app.invoke({ question: 帮我查一下状态图怎么用, docs: [], answer: , retry_count: 0, need_retry: False, }) print(result[answer]) print(检索次数:, result[retry_count])跑起来你会看到检索执行了两次因为check要求至少 2 份资料然后进入生成。把问题改成不带“查”字的比如“你好”decide会直接跳到generate检索次数为 0。这两个用例一跑分支和循环两条路径就都验证到了。提示invoke的入参要把状态里所有字段都给全缺字段可能触发KeyError。如果嫌麻烦可以在状态定义时给字段设默认值或者写个initial_state()工厂函数统一构造。4.4 用流式输出观察每一步的执行轨迹图跑通只是第一步调试时你更想知道“它到底走了哪条路、每步状态变成什么样”。LangGraph 提供了流式接口能把每个节点执行后的状态增量吐出来这对排查路由错误特别有用。for chunk in app.stream({ question: 帮我查一下状态图怎么用, docs: [], answer: , retry_count: 0, need_retry: False, }): for node_name, update in chunk.items(): print(f[{node_name}] - {update})输出会按节点顺序打印你能清楚看到retrieve被调了几次、每次retry_count怎么变、check给出的need_retry是什么。我调试循环逻辑时基本全靠这个比在节点里到处打日志干净得多。流式输出还有个好处是能配合前端做“执行过程可视化”用户能看到 Agent 正在干什么体验会好不少。5. 常见问题与排查技巧实录5.1 报错速查表从症状反推原因下面这张表是我和身边同行实际踩过的坑汇总按“症状 → 可能原因 → 处理方式”组织遇到问题先对号入座能省掉大量瞎猜时间。症状可能原因处理方式运行直接卡住不结束某条路径没接END检查所有分支是否都有出口抛KeyError找不到节点条件边映射的键拼错核对路由返回值与映射表列表字段只剩最后一条默认覆盖合并给字段加Annotated[..., operator.add]无限循环烧额度缺计数器或上限判断路由函数加retry_count N状态字段类型报错节点返回类型与定义不符检查TypedDict标注与返回值入口节点不执行忘了set_entry_point编译前补上入口设置这张表建议收藏新手阶段八成的问题都能在里面找到答案。尤其是前两条几乎每个刚上手的人都至少撞过一次。5.2 三个我踩过的真实坑与规避方法第一个坑是在路由函数里改状态。早期我以为路由函数能顺便把retry_count加一结果发现根本不生效——路由函数只读不写改了也白改。后来把计数逻辑挪进retrieve节点问题消失。这个教训让我记住路由函数只做判断副作用一律放节点里。第二个坑是状态字段命名太随意。有次我用flag表示“是否需要重试”过两天又用flag表示“是否已生成”同一个字段两种语义调试时差点把自己绕晕。后来养成习惯控制字段一律用need_xxx、is_xxx这种自解释命名再没出过类似问题。第三个坑是节点里做了太多事。有个节点既检索又评估又生成看着省事结果某次检索失败导致整个节点抛异常连评估逻辑都没跑到排查时完全不知道是哪一步出的问题。拆成三个节点后异常定位一目了然。这件事让我彻底信了“一个节点一件事”的原则。5.3 性能与成本上的几个实用建议图结构本身开销很小真正花钱的是节点里的模型调用和工具调用。几个能立竿见影的优化点把不需要模型参与的判断做成纯规则节点比如上面decide用关键词判断比调一次模型便宜太多检索结果做去重重试时很容易拿到重复资料白白增加上下文长度给循环上限设小一点2 到 3 次足够再多往往是问题本身需要澄清而不是检索不够。还有一个容易被忽略的点状态里别塞大对象。有人把整个文档原文塞进状态每次节点更新都要复制一遍图一复杂内存就上去了。正确做法是状态里只存引用或 ID需要时再去取。这个习惯在原型阶段看不出差别上了规模就是生死线。6. 从能跑到好用下一步可以怎么扩展把上面这个例子跑通你已经掌握了 LangGraph 的核心用法。接下来往哪个方向走取决于你的实际需求。想做多轮对话就在状态里加messages列表并配上追加合并策略想让多个 Agent 协作可以把每个 Agent 做成子图再拼成父图想加人工审核环节LangGraph 支持在节点间中断等待外部输入这个能力在做高风险操作时特别有用。我个人建议的进阶顺序是先把单图的循环和分支玩熟再尝试子图嵌套最后再碰中断与恢复这类高级特性。跳着学容易在状态传递上翻车因为子图的状态和父图状态怎么对接是有讲究的没把单图吃透很难理解。另外别急着上复杂框架LangGraph 本身够轻很多能力用原生 Python 配合它就够了堆太多依赖反而增加调试难度。最后分享一个我自己的习惯每加一个新节点先单独invoke测一遍它的输入输出确认没问题再连进图里。这个笨办法看着慢实际能省掉大量“连进图后不知道哪坏了”的排查时间。状态图式开发最大的好处就是每个零件都能独立验证把这个优势用足你的 Agent 会越搭越顺。