☰
LangGraph 核心概念与实战:状态、节点、边与检查点
2026/11/4 14:26:46 网站建设 项目流程

1. 从 LangChain 到 LangGraph:为什么需要一张“图”

如果你已经用 LangChain 写过几个 Demo,大概率经历过这样的时刻:一个简单的问答链跑得挺顺,可一旦想加入“先判断意图、再决定调哪个工具、工具失败要重试、重试超过三次就转人工”这类逻辑,代码就开始变成一堆if/else嵌套,调试时根本说不清上一步到底走到哪了。LangGraph 出现的直接原因,就是给这种“有状态、有分支、有循环”的流程提供一个显式的结构——用图(Graph)来描述节点和边,而不是靠代码里的条件判断硬堆。

我最初接触 LangGraph 时的第一反应是:这不就是把工作流画成流程图吗?但用下来发现,它真正解决的不是“画图好看”,而是状态管理和控制流这两件在复杂 Agent 里最要命的事。LangChain 的 Chain 本质是一条单向流水线,数据从 A 流到 B 再到 C,中间很难回头,也很难根据中间结果动态改变路线。而 LangGraph 允许你定义节点(Node)、边(Edge)和条件边(Conditional Edge),并且所有节点共享一份可持久化的状态(State)。这意味着循环、重试、人工介入这些在真实业务里绕不开的需求,终于有了一个相对干净的实现方式。

这篇文章面向的是已经了解 LangChain 基础、想进一步搞懂 LangGraph 到底怎么用、什么时候该用的人。我会从核心概念讲起,把状态、节点、边、检查点这些机制拆开揉碎,再给出一套可以直接跑起来的实操流程,最后聊聊我在实际项目里踩过的坑。全文不堆概念,尽量用“为什么这么设计”的角度来讲,让你看完能自己判断某个场景到底该不该上 LangGraph。

2. LangGraph 的核心抽象:状态、节点与边到底怎么协作

2.1 State 不是普通变量,而是贯穿全图的共享内存

LangGraph 里最核心的概念是 State。你可以把它理解成一张全局的“黑板”,每个节点都能读到上面的内容,也能往上写东西。和普通函数传参最大的区别在于:State 是跨节点持久存在的,而且它的更新方式决定了整个图的运行逻辑。

State 通常用一个 TypedDict 或者 Pydantic 模型来定义。比如一个最简单的对话 Agent,State 里可能包含messages(对话历史)、user_intent(识别出的意图)、retry_count(重试次数)这几个字段。每个节点执行完,返回一个字典,LangGraph 会把这个字典合并回全局 State。

这里有个关键点很多人一开始会忽略:State 的字段可以定义“归约函数”(reducer)。默认情况下,节点返回的新值会直接覆盖旧值。但像messages这种列表,你往往希望是“追加”而不是“覆盖”。这时候就要用Annotated配合operator.add来声明:

from typing import Annotated, TypedDict import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] retry_count: int

这样每次节点返回新的 message,都会被追加到列表末尾,而不是把整个历史冲掉。我见过不少人第一次写 LangGraph 时消息莫名其妙丢失,八成就是忘了配 reducer。这个设计其实很符合直觉:状态更新语义应该由字段自己决定,而不是让每个节点都去手动处理合并逻辑。

2.2 节点是计算单元,但它的返回值有讲究

节点本质上就是一个 Python 函数,接收当前 State,返回一个字典表示要更新的字段。听起来简单,但有几个细节决定了你的图能不能正常工作。

第一,节点返回的字典只需要包含要更新的字段,不需要把整个 State 都返回。LangGraph 会负责合并。第二,节点可以是同步函数也可以是异步函数,这在调用外部 API 时很实用。第三,节点内部不要直接修改传入的 State 对象,而是通过返回值来更新——这是函数式编程的思路,保证了状态变更的可追踪性。

举个实际例子,一个“意图识别”节点大概长这样:

def classify_intent(state: AgentState): last_message = state["messages"][-1] intent = llm.invoke(f"判断这句话的意图:{last_message}") return {"user_intent": intent}

它只更新了user_intent一个字段,其他字段原封不动。这种“只声明增量”的方式,让每个节点的职责非常清晰,也方便后续做单元测试——你给一个固定的输入 State,断言输出字典是否符合预期就行。

2.3 边决定了“下一步去哪”,条件边才是真正的灵魂

普通边(Edge)是确定性的,从节点 A 直接连到节点 B。条件边(Conditional Edge)则根据当前 State 动态决定下一个节点是谁。这是 LangGraph 区别于普通工作流引擎的关键。

条件边的写法是给一个路由函数,它读取 State,返回一个字符串,这个字符串对应你预先注册的节点名。比如:

def route_after_classify(state: AgentState): if state["user_intent"] == "查询订单": return "query_order" elif state["user_intent"] == "退款": return "refund" else: return "fallback"

然后在构图时:

graph.add_conditional_edges( "classify", route_after_classify, { "query_order": "query_order", "refund": "refund", "fallback": "fallback" } )

这个路由函数就是整个 Agent 的“大脑”。我个人的经验是,把路由逻辑集中在一个或少数几个函数里,比散落在各个节点内部要好得多。因为当流程出问题时,你只需要看路由函数就能知道分支条件,不用满图找if。

2.4 检查点机制让“断点续跑”和“人工介入”成为可能

LangGraph 另一个让我觉得设计得很到位的地方是 Checkpointer。每次图执行到一个节点结束,它可以把当前 State 存下来。这个存储可以是内存、SQLite、Postgres 等。存下来之后能干什么?两件事:一是失败后从上一个检查点恢复,不用从头跑;二是实现 Human-in-the-loop,也就是在某个节点前暂停,等人工确认后再继续。

比如一个自动发邮件的流程,你可以在“发送”节点前设置中断,把草稿展示给用户,用户点确认后再恢复执行。这在 LangChain 的 Chain 里几乎没法优雅实现,而在 LangGraph 里只需要配置interrupt_before=["send_email"],然后用graph.update_state()修改状态再graph.invoke(None, config)继续跑。

提示:Checkpointer 在生产环境建议用持久化后端,内存版只适合本地调试。因为一旦进程重启,内存里的检查点就没了,断点续跑也就失效了。

3. 手把手搭一个带分支和重试的 LangGraph 流程

3.1 环境准备与依赖安装

先把环境搭起来。LangGraph 是独立包,不需要装整个 LangChain 全家桶,但通常会配合 langchain-core 和某个模型 SDK 一起用。

pip install langgraph langchain-core langchain-openai

如果你要用持久化检查点,再加一个:

pip install langgraph-checkpoint-sqlite

版本方面,LangGraph 迭代挺快,建议锁定一个较新的稳定版。我写这篇时用的是 0.2.x 系列,API 相对稳定。装完后先跑一个最小示例验证环境没问题,别一上来就写复杂图,不然报错都分不清是环境问题还是逻辑问题。

3.2 定义 State 与三个基础节点

我们来做一个“客服工单处理”的小流程:用户发一句话,先分类意图,如果是“查订单”就去查订单,如果是“退款”就走退款流程,其他情况走兜底回复。查订单失败要重试,最多两次。

先定义 State:

from typing import Annotated, TypedDict import operator class TicketState(TypedDict): messages: Annotated[list, operator.add] intent: str order_result: str retry_count: int

然后写三个节点:分类、查订单、兜底。

def classify_node(state: TicketState): user_msg = state["messages"][-1] intent = llm.invoke(f"分类以下客服消息的意图,只返回'查订单'、'退款'或'其他':{user_msg}") return {"intent": intent.strip()} def query_order_node(state: TicketState): # 模拟查询,可能失败 if state["retry_count"] < 2 and random.random() < 0.5: return {"order_result": "查询失败", "retry_count": state["retry_count"] + 1} return {"order_result": "订单号 12345,已发货"} def fallback_node(state: TicketState): return {"order_result": "抱歉,我暂时无法处理,已转人工"}

注意query_order_node里我用了retry_count来控制重试次数,并且每次失败都自增。这个字段就是靠 State 在节点间传递的。

3.3 用条件边实现“失败重试”的循环

重试的本质是一个循环:查订单失败就回到查订单节点,直到成功或超过次数。在 LangGraph 里,循环就是一条指回自己的条件边。

def route_after_query(state: TicketState): if state["order_result"] == "查询失败" and state["retry_count"] < 2: return "retry" return "done"

构图:

from langgraph.graph import StateGraph, END builder = StateGraph(TicketState) builder.add_node("classify", classify_node) builder.add_node("query_order", query_order_node) builder.add_node("fallback", fallback_node) builder.set_entry_point("classify") builder.add_conditional_edges("classify", lambda s: s["intent"], { "查订单": "query_order", "退款": "fallback", "其他": "fallback" }) builder.add_conditional_edges("query_order", route_after_query, { "retry": "query_order", "done": END }) builder.add_edge("fallback", END) graph = builder.compile()

这里route_after_query返回"retry"时,边指向query_order自己,形成循环。返回"done"时指向END,流程结束。整个图跑起来后,如果查询一直失败,最多循环两次就会退出,不会死循环。

3.4 加上检查点,让流程可以暂停和恢复

现在给图加上 SQLite 检查点:

from langgraph.checkpoint.sqlite import SqliteSaver memory = SqliteSaver.from_conn_string("checkpoints.db") graph = builder.compile(checkpointer=memory)

调用时要传一个config,里面带thread_id,用来区分不同的会话:

config = {"configurable": {"thread_id": "ticket-001"}} result = graph.invoke({"messages": ["我的订单到哪了"], "retry_count": 0}, config)

有了thread_id,同一个会话的状态会被持久化。你可以在任意时刻用graph.get_state(config)查看当前状态,也可以用graph.update_state(config, {...})手动改状态。这个能力在做“人工审核后继续”时特别有用:先跑到审核节点前中断,人工看完改一下 State,再继续跑。

注意:retry_count这种计数字段一定要在 State 里初始化,否则第一次读取会 KeyError。我习惯在入口调用时把所有非列表字段都显式传一遍。

4. 那些文档里不会明说的坑与调优经验

4.1 循环没有出口是最常见的翻车方式

LangGraph 给了你写循环的能力,但也意味着你得自己保证循环能结束。我见过最典型的事故是:条件边在某个边界情况下一直返回“重试”,而重试计数因为 reducer 配置问题没有正确累加,结果图跑了几百轮把额度耗光。

避免这个问题的办法有两个。第一,所有循环都必须有一个基于 State 的硬性退出条件,比如retry_count >= max_retry,而且这个判断要放在路由函数的最前面。第二,给图设置递归限制。LangGraph 编译时可以传recursion_limit,默认是 25,超过会抛异常。这个默认值其实是个保护,别轻易调太高,调高之前先确认你的退出条件真的可靠。

graph = builder.compile(checkpointer=memory) result = graph.invoke(inputs, config={"recursion_limit": 50, **config})

4.2 State 字段设计要“够用但不臃肿”

State 是全局共享的,字段越多,每个节点能看到的上下文就越多,但同时也越容易产生意料之外的耦合。我的经验是:只把需要跨节点传递的数据放进 State,节点内部的临时变量不要塞进去。比如某个节点里算了一个中间结果,只有它自己用,那就用局部变量,别加到 State。

另外,列表类字段一定要想清楚是覆盖还是追加。messages用追加,但像tool_results这种每次只关心最新一次的,可能覆盖更合适。reducer 配错了,轻则数据冗余,重则逻辑错乱。

4.3 调试时善用 stream 和 get_state

LangGraph 提供了stream方法,可以逐节点输出执行过程。调试复杂图时,这比invoke好用得多:

for event in graph.stream(inputs, config, stream_mode="updates"): print(event)

stream_mode="updates"会打印每个节点返回的增量字典,你能清楚看到每一步 State 变了什么。如果某个分支没走到,或者某个字段没更新,一眼就能定位。配合graph.get_state(config)查看完整状态快照,基本能覆盖大部分调试场景。

4.4 和 LangChain 的关系:不是替代,是互补

经常有人问 LangGraph 和 LangChain 到底什么关系。我的理解是:LangChain 提供的是“组件”,比如模型封装、工具、检索器、提示模板;LangGraph 提供的是“编排”,把这些组件按图的方式组织起来。你完全可以在 LangGraph 的节点里调用 LangChain 的 Chain 或 Agent,两者不冲突。

那什么时候该从 LangChain 迁到 LangGraph?我的判断标准是:当你的流程出现“需要根据中间结果决定下一步”“需要循环”“需要人工介入”“需要持久化状态”这四个特征中的任意一个时,就该考虑 LangGraph 了。如果只是简单的“输入→模型→输出”,用 LangChain 的 LCEL 反而更简洁。

5. 从 Planning 模式看 LangGraph 的进阶玩法

5.1 Planning 模式为什么天然适合用图来表达

Planning 模式的核心是:先让模型生成一个计划(一串步骤),然后逐步执行,每执行完一步根据结果决定是否调整后续计划。这个“生成计划→执行→评估→可能重新规划”的过程,本质上就是一个带循环和条件分支的图。

用 LangGraph 实现 Planning,通常会有一个planner节点负责生成或更新计划,一个executor节点负责执行当前步骤,一个evaluator节点判断是否完成或需要重新规划。条件边根据 evaluator 的结果决定回到 planner 还是继续 executor 还是结束。这种结构用 Chain 写会非常别扭,因为 Chain 不支持“回到前面某一步”这种控制流。

5.2 多 Agent 协作其实是同一套机制的延伸

LangGraph 常被用来做多 Agent 系统,比如一个“主管 Agent”分配任务给“研究员 Agent”和“写作 Agent”。听起来很高级,但拆开看还是 State + 节点 + 条件边。主管节点读取 State 决定派谁,条件边路由到对应 Agent 节点,Agent 执行完把结果写回 State,再回到主管节点判断是否继续。

关键设计点在于:每个 Agent 的职责边界要清晰,State 里要有明确的“当前任务”和“完成标志”字段。否则多个 Agent 互相覆盖状态,很容易乱套。我一般会给每个 Agent 的输出单独开一个字段,而不是都往messages里塞,这样回溯问题时能清楚知道哪条信息是谁产生的。

5.3 性能与成本上的取舍

图越复杂,节点越多,调用模型的次数就越多,成本和延迟都会上升。我做过一个对比:同样一个客服场景,用单次 Prompt 直接回答,平均 1 次模型调用;用 LangGraph 做意图分类+工具调用+结果润色,平均 3 到 4 次调用。延迟大概翻倍,成本也翻倍。

所以我的建议是:不要为了用 LangGraph 而用。先用最简单的方案跑通,当简单方案确实撑不住业务复杂度时,再引入图编排。引入之后,也要定期回看哪些节点是必要的,能不能合并。比如意图分类如果准确率够高,可以和小模型结合,不一定每次都调大模型。

6. 我实际落地时总结的几条硬经验

第一条,先把图画在纸上再写代码。LangGraph 的代码结构和图结构是一一对应的,如果图本身没想清楚,代码写出来一定是乱的。我现在习惯先用纸笔画出节点和边,标清楚每个节点的输入输出字段,确认没有死循环和孤立节点,再动手写。

第二条,State 字段命名要能自解释。data、result、info这种名字在单节点里没问题,但在跨节点的 State 里就是灾难。用order_query_result、intent_label、retry_count这种一看就知道是什么的名字,能省下大量调试时间。

第三条,检查点不是万能的,状态迁移要幂等。断点续跑的前提是节点可以重复执行而不产生副作用。如果你的节点里有“发邮件”“扣款”这类操作,一定要做幂等处理,比如用唯一 ID 去重,否则恢复执行时可能重复触发。

第四条,别忽略异常处理。LangGraph 的节点抛异常会中断整个图。对于可能失败的外部调用,我习惯在节点内部 try/except,把错误信息写进 State,然后通过条件边决定是重试还是走兜底。这样图不会直接崩,用户也能得到反馈。

第五条,版本升级要谨慎。LangGraph 还在快速演进,小版本之间偶尔会有 API 调整。生产项目里建议锁定版本,升级前先在测试环境跑一遍完整流程,重点看条件边和检查点相关的行为有没有变化。

这套东西我用了大半年,从最初的小心翼翼到现在基本能比较顺手地设计中等复杂度的 Agent 流程。它确实不是银弹,但对于那些“流程本身就很复杂”的场景,LangGraph 提供的结构化表达能力,比在代码里堆条件判断要清晰太多。如果你正在被复杂的 Agent 逻辑折磨,不妨从一个小图开始试试,先跑通一个带条件分支的流程,感受一下状态在节点间流动的方式,后面再逐步加循环、加检查点、加人工介入。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询