LangGraph 保姆级入门：State、Node、Edge 三大件，加上检查点、流式输出与人工审核

说明：本文系统梳理 LangGraph 的核心概念与工程用法，参考 LangGraph 官方文档与社区实践经验整理而成，面向"会一点 Python、但完全没接触过 LangGraph"的同学。全文按"是什么 → 为什么 → 怎么跑 → 每个概念拆开讲 → 踩坑清单"的顺序展开，建议按顺序阅读。

目录

• 一、先讲人话：为什么需要 LangGraph
• 二、LangGraph 到底是什么
• 三、LangGraph 和 LangChain 到底差在哪
• 四、把环境跑起来
• 五、快速入门：20 行代码跑通第一个图
• 六、State：整个应用的数据中枢
• 七、状态存储：Checkpointer 与 thread_id
• 八、揭开底牌：Pregel 是怎么把图跑起来的
• 九、检查点快照：怎么把历史状态翻出来看
• 十、Node：节点怎么写才不出错
• 十一、图内外数据传递：流式输出六种模式
• 十二、Human-in-the-Loop：让用户参与关键决策
• 十三、Edge：边和两种流转方式
• 十四、踩坑清单：这些坑我帮你先踩了
• 十五、学习路线建议
• 十六、总结

一、先讲人话：为什么需要 LangGraph

如果你用过 LangChain，大概率写过这样的代码：

chain = prompt | llm | output_parser
result = chain.invoke({"question": "什么是 LangGraph"})

这条"链"就像工厂里的流水线：原料进去，按固定顺序经过几道工序，成品出来。做翻译、摘要、单轮问答这类任务，非常顺手。

但真实的 Agent 需求，往往不是一条笔直的流水线，而是一张会拐弯的流程图：

• 模型说"我要查一下天气"，程序就得先去调工具，拿到结果再回到模型 —— 这是循环；
• 用户问"我订单到哪了"就走订单分支，问"这个商品有货吗"就走商品分支 —— 这是条件分支；
• 要转出 8000 元，得先停下来，等人点一下"同意" —— 这是人工介入；
• 跑到第 3 步服务器崩了，重启后希望能从第 3 步继续，而不是从头再算一遍 —— 这是持久化 / 断点续跑。

这四件事，用一条 A | B | C 的链去硬写，会非常别扭。LangGraph 就是为了解决这类问题而生的。一句话概括两者的分工：

当 LLM 应用需要「有状态、可循环、可分支的多步骤控制流（像一个可回溯的流程图 / 状态机）」时，LangChain 很难优雅完成，而必须用 LangGraph。

换个比方：LangChain 是零件库 + 流水线，LangGraph 是车间的调度中心。 官方也是这么定位的——LangChain 提供模型、工具、向量库这些"零件"，LangGraph 负责把这些零件编排成一台能跑、能停、能恢复的机器。

二、LangGraph 到底是什么

2.1 官方定义翻译成人话

LangGraph 是一个低级编排框架和运行时环境，用于构建、管理和部署长期运行的有状态智能体（Agent）。它的核心理念是把 Agent 工作流建模成一张图（Graph）。

既然是一张图，那就只有三样东西，记住这三个词，后面的内容都能挂上去：

所以写一个 LangGraph 应用，本质就是在画一张带方向的流程图，再给这张图配一个共享记事本。

2.2 LangGraph 提供的五大核心能力

官方强调它有五样东西，正好对应生产环境的五个痛点：

• 持久化执行：能从故障中恢复，支持长时间运行；
• 人机协作：可以在任意时刻检查、甚至修改智能体的状态；
• 记忆管理：同时支持"短期工作记忆"和"跨会话的长期记忆"；
• 流式处理：天生为流式工作流设计（能边跑边往外吐数据）；
• 生产级部署：为有状态、长期运行的工作流提供可扩展的基础设施。

官方地址，建议收藏：

• GitHub：<https://github.com/langchain-ai/langgraph>
• 官方文档：<https://docs.langchain.com/oss/python/langgraph/overview>

三、LangGraph 和 LangChain 到底差在哪

3.1 一张表看清区别

3.2 因为这种区别，LangGraph 在这些场景更吃香

• 复杂的多智能体系统（一个主 Agent 带一堆子 Agent）；
• 需要长期记忆的应用（跨天、跨会话记得你是谁）；
• 需要人工审核的工作流（审批、转账、发邮件、改数据库）；
• 后台处理任务和实时交互并存的系统；
• 需要精细控制的定制化智能体编排。

3.3 三个常见误区，先帮你排掉

四、把环境跑起来

先准备一个干净的 Python 环境，避免污染你电脑上已有的包。本文对应 langgraph 1.0.5、Python 3.12。

# 1. 创建并激活虚拟环境（名字随便起）
conda create –name=langgraph_venv python=3.12
conda activate langgraph_venv

# 2. 安装 langgraph
pip install langgraph==1.0.5

# 3. 确认装上了
pip show langgraph

如果你还想把图结构打印成 ASCII 图看，再装一个小依赖：

pip install grandalf

五、快速入门：20 行代码跑通第一个图

5.1 记住这五步，所有 LangGraph 代码都是这个套路

下面这个例子做一个"RAG + 联网搜索"的问答系统：收到问题后，同时去知识库检索、去网上搜索，两路结果都回来之后，再交给大模型总结成一句话回答。

注意这里的"同时"——这就是图的力量，两条边从 START 一起出发，天然并行。

5.2 完整代码

from typing import TypedDict

from pydantic import BaseModel
from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.types import Send, Command

# 第一步：定义状态（共享记事本上要记哪些东西）
class MyState(TypedDict):
rag_result: str # 知识库检索结果
web_search_result: str # 联网搜索结果
query: str # 用户问题
final_answer: str # 最终回答

# 第三步的准备：用状态创建 builder
graph = StateGraph(state_schema=MyState)

# 第二步：定义节点函数
def rag_search_node(state: MyState):
query = state["query"]
rag_result = f"关于{query}的rag_result"
return {"rag_result": rag_result} # 只返回自己更新的字段

def web_search_node(state: MyState):
query = state["query"]
web_search_result = f"关于{query}的web_search_result"
return {"web_search_result": web_search_result}

def final_answer_node(state: MyState):
rag_result = state["rag_result"]
web_search_result = state["web_search_result"]
final_answer = f"LLM基于{rag_result}和{web_search_result}的最终回复"
return {"final_answer": final_answer}

# 注册节点
graph.add_node(rag_search_node)
graph.add_node(web_search_node)
graph.add_node(final_answer_node)

# 连边：START 同时指向两个检索节点 => 并行
graph.add_edge(START, "rag_search_node")
graph.add_edge(START, "web_search_node")
# 两路都汇合到总结节点 => 等两路都完成才执行
graph.add_edge("rag_search_node", "final_answer_node")
graph.add_edge("web_search_node", "final_answer_node")
graph.add_edge("final_answer_node", END)

# 第四步：编译
compiled_graph = graph.compile()

# 第五步：调用
res = compiled_graph.invoke({"query": "如何使用LangGraph"})
print(res)

5.3 逐行拆解几个关键点

• class MyState(TypedDict)：这不是普通类，是用类型注解描述字典结构的"协议"。它告诉 LangGraph：这本记事本上有 4 个字段，各自是什么类型。
• graph = StateGraph(state_schema=MyState)：把协议交给图，图从此知道自己的完整状态空间。
• 节点函数返回的是字典，不是整个 state。这一点极其重要，后面第十节会专门讲错误示范。
• START 和 END 是两个特殊字符串常量，分别代表"图开始"和"图结束"，用它们来标记入口和出口。
• .compile() 是必须的一步，编译之后才能 invoke；编译时还可以顺便挂上检查点、缓存等能力。

5.4 把图"画"出来看一眼

光看代码没感觉？编译后的对象可以导出图结构，直接打印成 ASCII 图：

# 需要先 pip install grandalf
graph_structure = compiled_graph.get_graph()
print(graph_structure.draw_ascii())

你会看到一张清楚的"两路并行再汇合"的图。这种可视化在调试复杂工作流时非常有用——你画的图和你脑子里想的图是否一致，一眼就能看出来。

六、State：整个应用的数据中枢

用 LangGraph 建应用，第一步永远是定义 State。因为 State 决定了"这张图的产出目标是什么"，也决定了"每个节点可以改哪些数据"。

6.1 定义状态的三种写法

State 的 schema 有三种定义方式，复杂度递增，按需选择，别过度设计。

写法一：TypedDict（推荐，最简单）

TypedDict 是 Python 自带的类型提示工具，用来给字典的键和值指定精确类型。零依赖、写起来最短。

from typing import TypedDict

class MyStateFull(TypedDict):
rag_result: str
web_search_result: str
final_answer: str
query: str
a_new_key: str

写法二：Pydantic BaseModel（带运行时校验）

Pydantic 会在运行时检查数据类型对不对，字段还能带默认值和说明文档，IDE 提示也更友好。项目里最常用。

from pydantic import BaseModel

class MyStateFull(BaseModel):
rag_result: str
web_search_result: str
final_answer: str
query: str
a_new_key: str

写法三：dataclass（标准库风格）

dataclass 是 Python 标准库的装饰器，会自动帮你生成 __init__、__repr__、__eq__ 等方法，适合当纯数据容器用。

from dataclasses import dataclass

@dataclass
class MyStateFullTwo:
rag_result: str
web_search_result: str
final_answer: str
query: str
a_new_key: str

6.2 输入输出数据隔离：三个 schema

这是很多新手会忽略、但做企业项目时必须会的技巧：你可以精确控制"外面能传进来什么"和"最后能拿到什么"。

做法是在初始化 StateGraph 时传三个参数：

为什么要这么设计？打个比方：厨房内部可能有三十种调料（state_schema），但顾客点单时只需要说"要一份番茄炒蛋"（input_schema），端上桌时也只需要那盘菜（output_schema）。中间过程对调用方不可见，接口就干净了。

from typing import TypedDict
from langgraph.constants import START, END
from langgraph.graph import StateGraph

class MyStateFull(TypedDict):
rag_result: str
web_search_result: str
final_answer: str
query: str
a_new_key: str

class InputSchema(TypedDict):
query: str # 只允许外部传 query

class OutputSchema(TypedDict):
final_answer: str # 只把 final_answer 交出去

graph = StateGraph(
state_schema=MyStateFull,
input_schema=InputSchema,
output_schema=OutputSchema,
)

def rag_search_node(state: MyStateFull):
print(state)
query = state["query"]
rag_result = f"关于{query}的rag_result"
return {"rag_result": rag_result, "a_new_key": "a_new_key_value"}

def web_search_node(state: MyStateFull):
print(state)
query = state["query"]
web_search_result = f"关于{query}的web_search_result"
return {"web_search_result": web_search_result}

def final_answer_node(state: MyStateFull):
print("在final_answer_node当中的state", state)
rag_result = state["rag_result"]
web_search_result = state["web_search_result"]
final_answer = f"LLM基于{rag_result}和{web_search_result}的最终回复"
return {"final_answer": final_answer}

graph.add_node(rag_search_node)
graph.add_node(web_search_node)
graph.add_node(final_answer_node)

graph.add_edge(START, "rag_search_node")
graph.add_edge(START, "web_search_node")
graph.add_edge("rag_search_node", "final_answer_node")
graph.add_edge("web_search_node", "final_answer_node")
graph.add_edge("final_answer_node", END)

compiled_graph = graph.compile()
res = compiled_graph.invoke({"query": "如何使用LangGraph", "a_new_key": "a_new_key_value"})
print("最终结果为", res)

实战小提示：如果你在 input_schema 里没有声明 a_new_key，那么调用时传它进去是不生效的（会被过滤）。这正好可以用来防止调用方乱塞数据。

6.3 节点间数据隔离：每个节点也能有自己的"私有状态"

除了图级别的状态，LangGraph 还允许某个节点只看到自己关心的一部分字段。做法是给这个节点函数换个 state 类型注解：

from typing import TypedDict
from langgraph.constants import START, END
from langgraph.graph import StateGraph

class MyState(TypedDict):
query: str
final_answer: str

class SearchState(TypedDict):
rag_result: str
web_search_result: str

graph = StateGraph(state_schema=MyState)

def rag_search_node(state: MyState):
query = state["query"]
return {"rag_result": f"关于{query}的rag_result"}

def web_search_node(state: MyState):
query = state["query"]
return {"web_search_result": f"关于{query}的web_search_result"}

def final_answer_node(state: SearchState): # 注意：这个节点只关心检索结果
print("在final_answer_node当中的state", state)
rag_result = state["rag_result"]
web_search_result = state["web_search_result"]
final_answer = f"LLM基于{rag_result}，和{web_search_result}的最终回复"
return {"final_answer": final_answer}

graph.add_node(rag_search_node)
graph.add_node(web_search_node)
graph.add_node(final_answer_node)

graph.add_edge(START, "rag_search_node")
graph.add_edge(START, "web_search_node")
graph.add_edge("rag_search_node", "final_answer_node")
graph.add_edge("web_search_node", "final_answer_node")
graph.add_edge("final_answer_node", END)

compiled_graph = graph.compile()
res = compiled_graph.invoke({"query": "如何使用LangGraph"})
print("最终结果为", res)

为什么有用？ 两个好处：一是这个节点看不到也不需要看不到无关数据，逻辑更纯粹；二是当状态字段很多（几十个）时，节点只会收到它声明的那几个，能减少意外改错字段的风险。这有点像函数只接收自己需要的参数，而不是把整个全局变量都塞进去。

6.4 Reducer：状态到底怎么"合并"

这是 LangGraph 里最容易踩坑、也最精髓的机制，请务必看懂。

前面反复说"节点返回的只是增量"。那么问题来了：增量拿到之后，怎么和全局状态合并？ 答案就是每个字段都有自己的 Reducer（归约函数）。

Reducer 就是一个函数：接收"左边的旧值"和"右边的增量"，返回合并后的新值。

情况一：不写 Reducer = 默认覆盖

这是默认行为。新值直接把旧值盖掉，就像给变量重新赋值。

from typing import List
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class DefaultReducerState(TypedDict):
foo: int
bar: List[str]

def node_default_1(state: DefaultReducerState) -> dict:
return {"foo": 2} # foo: 1 -> 2

def node_default_2(state: DefaultReducerState) -> dict:
return {"bar": ["bye"]} # bar: ["hi"] -> ["bye"]，注意 "hi" 没了！

builder = StateGraph(DefaultReducerState)
builder.add_node("node1", node_default_1)
builder.add_node("node2", node_default_2)
builder.add_edge(START, "node1")
builder.add_edge("node1", "node2")
builder.add_edge("node2", END)

graph = builder.compile()
result = graph.invoke({"foo": 1, "bar": ["hi"]})
print(f"初始状态: {{'foo': 1, 'bar': ['hi']}}")
print(f"执行结果: {result}")

跑一下你会发现，bar 最终是 ["bye"]，原来的 "hi" 被覆盖掉了。如果你想要的是"追加"，就必须自己指定 Reducer。

情况二：自定义 Reducer（最常用：消息追加）

用 Annotated[类型, 函数] 的语法把 Reducer 挂到字段上。下面这个 add_message 就是把旧消息列表和新消息列表拼起来：

from typing import Annotated, TypedDict, List
from langgraph.graph import StateGraph
from langgraph.constants import START
from langchain_core.messages import ToolMessage, HumanMessage, AIMessage, BaseMessage

# 自定义 Reducer：左边 + 右边
def add_message(message_list_left: list, message_list_right: list):
print("=" * 20)
print("正在执行add_message")
print("左边的", message_list_left)
print("右边的", message_list_right)
print("=" * 20)
return message_list_left + message_list_right

class MyAgent(TypedDict):
# 关键：用 Annotated 挂上自定义 reducer
messages: Annotated[List[BaseMessage], add_message]

def tool_node(state: MyAgent):
return {"messages": [ToolMessage(content="来自tool_node的内容", tool_call_id="xxx")]}

def llm_node(state: MyAgent):
return {"messages": [AIMessage(content="来自llm_node的内容")]}

builder = StateGraph(state_schema=MyAgent)
builder.add_node(tool_node)
builder.add_node(llm_node)
builder.add_edge(START, "tool_node")
builder.add_edge("tool_node", "llm_node")

graph = builder.compile()
graph.invoke({"messages": [HumanMessage(content="你好")]})

跑完你会看到 messages 里依次是：用户说的"你好" → 工具的返回 → 模型的回复。三条都在，一条没丢。这也是为什么对话类应用几乎都用追加型 Reducer——聊天记录必须是累加的，不能互相覆盖。

情况三：使用内置 Reducer

LangGraph 已经内置了一些常用 Reducer，比如 langgraph.graph.messages 里的 add_messages，以及做列表相加的 operator.add。后者在并行场景特别好用：

import operator
from typing import Annotated
from typing_extensions import TypedDict

class State(TypedDict):
aggregate: Annotated[list, operator.add] # 多个节点同时写，结果会拼接起来

注意：当一个字段有多个并行节点同时写，又没有配置 Reducer 时，LangGraph 会因为"不知道该以谁为准"而报错。这是新手最常见的报错来源之一，看到 InvalidUpdateError 之类的提示，先检查是不是并行的节点写了同一个键却没配 Reducer。

七、状态存储：Checkpointer 与 thread_id

7.1 先说两个真实场景，你就知道为什么需要它

在前面所有例子里，每次 invoke，LangGraph 都会初始化一个空状态，然后把你的初始状态合进去，跑完就丢。简单任务没问题，但遇到下面两种情况就废了。

场景一：同一个会话里要记住上一轮

用户第一句说"帮我查一下北京明天天气"，第二句问"适合出去玩吗"。第二次调用时，状态被重置，messages 是空的，模型根本不知道"它"指什么——上下文断了。

场景二：跑到一半报错了，不想从头再来

假设 Node2 因为某个 bug 抛异常。你修好代码后想"从 Node2 继续"，但内存里的状态已经随进程退出消失了，Node1 的结果得重新算一遍。如果 Node1 刚刚花了两毛钱调用大模型，你会很心疼。

7.2 解决方案：Checkpointer + thread_id

思路很朴素：把每一步执行完的状态存起来，下次用的时候按 id 取回来，而不是让 LangGraph 每次都从空白开始。

LangGraph 提供的存储机制叫 Checkpointer（检查点），相关的类都在 langgraph.checkpointer 包下，可以存到内存、SQLite、Postgres 等地方。

同时，因为一个应用可能同时服务很多用户、每个用户又有很多会话，所以需要给状态打个"编号"来隔离。这个编号就是 thread_id。

划重点：thread_id 和编程里的"线程"没有半点关系！ 它只是一个用来隔离状态空间的字符串，你可以把它理解成"会话编号"。叫张三的编号 user_1_session_1，叫李四的编号 user_2_session_1，两人的记忆互不干扰。

编码上就三步：

7.3 实战一：用 InMemorySaver 做"短期记忆"

下面这段代码用 create_agent 演示（它底层也是 LangGraph 图），重点是看注释里标了序号的四步：

def scenario_1_demo():
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI

# 1、引入 checkpointer
from langgraph.checkpoint.memory import InMemorySaver

llm = ChatOpenAI(model="gpt-4o-mini")

# 2、定义 checkpointer 实例
checkpointer = InMemorySaver()

@tool
def weather_tool(city: str, date: str) -> str:
"""查天气的工具"""
return f"{city}在{date}的天气是晴朗的"

# 3、构建 Agent 时引入 checkpointer
agent = create_agent(
model=llm,
tools=[weather_tool],
checkpointer=checkpointer,
)

# 4、用户第一次调用
user1_res = agent.invoke(
{"messages": "北京2026-01-14天气怎么样"},
config={"configurable": {"thread_id": "user_1_session_1"}},
)
print("第一次调用", user1_res["messages"][-1])

# 5、同一个会话里，第二次调用（thread_id 相同！）
user1_res_2 = agent.invoke(
{"messages": "适合出去玩吗"},
config={"configurable": {"thread_id": "user_1_session_1"}},
)
print("第二次调用", user1_res_2["messages"][-1])

第二次提问"适合出去玩吗"，模型能答上来，因为它从 checkpointer 里读到了上一轮的"北京、晴天"。这就是所谓的 Agent 短期记忆。

7.4 实战二：用 SqliteSaver 做"断点续跑"

InMemorySaver 只存在内存里，进程一挂就全没了，所以它解决不了场景二。生产环境要落到数据库，本地开发最方便的是 SQLite。

第一次运行：Node2 故意报错

def scenario_2_demo():
"""场景二：如何从 checkpointer 中恢复状态"""
from langgraph.graph import StateGraph, START
from typing import TypedDict

# 1、引入 sqlite3 和 SqliteSaver
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

# 2、构建 Connection 对象
conn = sqlite3.connect(database="./sqlite_data/langgraph_sqlite", check_same_thread=False)
# 3、通过 connection 对象构建 checkpointer 实例
memory = SqliteSaver(conn)

class MyState(TypedDict):
key_1: str
key_2: str
key_3: str

def node_1(state: MyState) -> MyState:
print("node_1初始状态为", state)
return {"key_1": "value_1"}

def node_2(state: MyState) -> MyState:
# 5、模拟生产环境下的 bug 或不可预见情况
raise Exception("node_2节点执行报错")
return {"key_2": "value_2"}

def node_3(state: MyState) -> MyState:
return {"key_3": "value_3"}

builder = StateGraph(MyState)
builder.add_node(node_1)
builder.add_node(node_2)
builder.add_node(node_3)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
builder.add_edge("node_1", "node_3")

config = {"configurable": {"thread_id": "123"}}

# 6、compile 时传入 checkpointer
graph = builder.compile(checkpointer=memory)

# 7、调用时传入 config
graph.invoke({}, config=config)

这一跑，程序会在 Node2 抛异常中断。但因为挂了 checkpointer，node_1 的结果已经落盘了。

第二次运行：修好 bug，从断点继续

把 Node2 里那行 raise 注释掉，然后只改两个地方：

def node_2(state: MyState) -> MyState:
# 1、修复错误，注释掉报错代码
# raise Exception("node_2节点执行报错")
return {"key_2": "value_2"}

# …（其他部分与上面完全相同）

# 2、注意！从故障中恢复，第一个参数传 None
graph.invoke(None, config=config)

此时你会发现，node_1 里的打印语句没有再执行一次——它是从 Node2 接着跑的。这就是"断点续跑"。

一个必须记住的坑：恢复执行时，被中断的那个节点函数会从函数开头重新跑一遍，而不是从 interrupt 那一行接着跑。所以如果节点里有"发邮件""扣款""写数据库"这类操作，一定保证幂等，或者把它拆成单独节点。这个坑在第十二节讲 interrupt 时还会再强调一次。

八、揭开底牌：Pregel 是怎么把图跑起来的

前面说"图会并行""会循环"，那它内部到底怎么调度的？理解这一层，你调试复杂图时会轻松很多。

8.1 两个主角：Actor 和 Channel

LangGraph 底层有个叫 Pregel 的类，负责管理整个图从开始到结束的迭代执行。它里面只有两大组件：

• Actors（节点）：对应你写的 Node，在底层叫 PregelNode。它订阅某些通道（从通道读数据），也会写入某些通道。
• Channels（通道）：Actor 之间通信的管道。每个通道都有值类型、更新类型、以及一个更新函数（接收一批更新，修改存储的值）。

因为这种设计，LangGraph 的图不是传统意义上的有向无环图（DAG）。在传统 DAG 里，一个节点必须等它所有上游都跑完才能跑，靠"边"来决定顺序。

而在 LangGraph 里，节点什么时候执行，是由一个个 超步（SuperStep） 决定的。

8.2 一个超步分三个阶段

这三个阶段组成的一个"超步"会不断重复，直到没有 Actor 被选中，或者达到最大步数。

换句话说：不是"边通向谁"决定执行顺序，而是"谁写了这个通道、谁订阅了这个通道"决定下一批跑谁。 边只是帮你把"订阅关系"自动生成出来的语法糖。

8.3 跟着一个例子数一遍 Step

假设图是这样的：START → a → (b, c)；b → b_2 → d；c → d；d → END。

编译时，builder.compile() 会把这张逻辑图转换成 Pregel 实例：节点变成 Actors，边变成 Channels。

执行过程（Step 从 0 开始数）：

• Step -1：__START__ 节点执行（图中看不见），把状态写进 branch:to:a；
• Step 0：节点 a 执行，把状态写进 branch:to:b 和 branch:to:c；
• Step 1：节点 b 和 c 并行执行，分别写进 branch:to:b_2 和 branch:to:d；
• Step 2：节点 b_2 和节点 d 执行。注意 d 这时候已经被触发过一次（因为 c 写了它），b_2 又往 branch:to:d 写了一次；
• Step 3：节点 d 再执行一次，这次没有新的状态更新，图的执行结束。

这个例子说明了一件重要的事：同一个节点可以在多个 Step 里被多次执行（d 跑了两次）。这就是为什么"循环"对 LangGraph 来说是天然的，而不是靠特殊语法实现。

import operator
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
aggregate: Annotated[list, operator.add]

def a(state: State, config):
print(f'Adding "A" to {state["aggregate"]}')
return {"aggregate": ["A"]}

def b(state: State, config):
print(f'Adding "B" to {state["aggregate"]}')
return {"aggregate": ["B"]}

def b_2(state: State, config):
print(f'Adding "B_2" to {state["aggregate"]}')
return {"aggregate": ["B_2"]}

def c(state: State, config):
print(f'Adding "C" to {state["aggregate"]}')
return {"aggregate": ["C"]}

def d(state: State, config):
print(f'Adding "D" to {state["aggregate"]}')
return {"aggregate": ["D"]}

builder = StateGraph(State)
builder.add_node("a", a)
builder.add_node("b", b)
builder.add_node("b_2", b_2)
builder.add_node("c", c)
builder.add_node("d", d)

builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("a", "c")
builder.add_edge("b", "b_2")
builder.add_edge("b_2", "d")
builder.add_edge("c", "d")
builder.add_edge("d", END)

graph = builder.compile()

# 1、查看当前图的 channels
print("当前图的channels为", graph.channels)
# 2、查看当前图的节点
print("当前图的节点为", graph.nodes)
# 3、查看 a 节点的触发条件（订阅）和写入目标
print("节点a的triggers为", graph.nodes["a"].triggers)
print("节点a的writers为", graph.nodes["a"].writers)
# 4、查看 d 节点的触发条件和写入目标
print("节点d的triggers为", graph.nodes["d"].triggers)
print("节点d的writers为", graph.nodes["d"].writers)

# 5、执行图，查看结果
output_state = graph.invoke({"aggregate": []})
print("执行图后的状态为", output_state)

打印出来的 triggers 就是"这个节点订阅了哪些通道"，writers 就是"这个节点会写哪些通道"。看懂这两行输出，你就能预测任何一张图的执行顺序了。

九、检查点快照：怎么把历史状态翻出来看

有了 checkpointer，LangGraph 会在每个超步结束时存一份快照。那怎么读出来？两个方法：

• graph.get_state(config)：拿最近一次的状态；
• graph.get_state_history(config)：拿所有历史状态，返回一个迭代器，按时间步倒序排列（最新的在最前）。

每个快照是一个 StateSnapshot 对象，字段如下：

这里最值得关注的是 next：它不仅告诉你"现在数据是什么"，还告诉你"接下来该跑谁"。断点续跑能实现，靠的正是这个信息。

import operator
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

connection = sqlite3.connect("checkpointer.db", check_same_thread=False)
checkpointer = SqliteSaver(connection)

class State(TypedDict):
aggregate: Annotated[list, operator.add]

def a(state: State, config):
print(f'Adding "A" to {state["aggregate"]}')
return {"aggregate": ["A"]}

def b(state: State, config):
print(f'Adding "B" to {state["aggregate"]}')
return {"aggregate": ["B"]}

def b_2(state: State, config):
print(f'Adding "B_2" to {state["aggregate"]}')
return {"aggregate": ["B_2"]}

def c(state: State, config):
print(f'Adding "C" to {state["aggregate"]}')
return {"aggregate": ["C"]}

def d(state: State, config):
print(f'Adding "D" to {state["aggregate"]}')
return {"aggregate": ["D"]}

builder = StateGraph(State)
builder.add_node("a", a)
builder.add_node("b", b)
builder.add_node("b_2", b_2)
builder.add_node("c", c)
builder.add_node("d", d)

builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("a", "c")
builder.add_edge("b", "b_2")
builder.add_edge("b_2", "d")
builder.add_edge("c", "d")
builder.add_edge("d", END)

graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}
res = graph.invoke({}, config=config)

# 1、最终状态
print(res)

# 2、查看历史所有状态
all_states = graph.get_state_history(config=config)
all_states_list = list(all_states)
print("历史所有状态如下：")
for state in all_states_list:
print(state)
print("=" * 30)

# 3、最近一次状态
print("\\n最近一次状态如下：")
print(graph.get_state(config=config))

跑完这段，你会看到一串快照，每个里都有 values 和 next。顺着看下去，就相当于给图装了一个"行车记录仪"——出了问题能精确知道在哪一步、当时的数据长什么样。

十、Node：节点怎么写才不出错

10.1 节点能接收哪些参数

节点一般是 Python 函数（同步异步都行），LangGraph 会在运行时自动注入三个可选参数：

这三个参数是按关键字传的，所以顺序可以随便调。下面的例子演示一个很实用的技巧：依赖注入——把 LLM 和数据库对象通过 runtime.context 传进节点，而不是写死在全局变量里。这样测试时可以轻松替换成 mock。

import time
from typing import TypedDict, List
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime

class MockLLM:
def invoke(self, prompt: str):
return f"AI Generated: Answer for '{prompt}'"

class MockDatabase:
def get_user_info(self, user_id: str):
return {"id": user_id, "role": "vip" if "vip" in user_id else "standard"}

class CustomerSupportState(TypedDict):
query: str # 用户问题
response: str # 客服回复
log: List[str] # 处理日志

def node_customer_service(
state: CustomerSupportState,
config: RunnableConfig,
runtime: Runtime,
) -> dict:
"""客服节点：从 config 取配置，从 runtime.context 取注入的依赖，并往外推进度。"""
user_query = state["query"]

# 1、从 runtime 里取出依赖
llm = runtime.context["llm"]
db = runtime.context["db"]

configurable = config.get("configurable", {})
user_id = configurable.get("user_id", "guest")
print(f"\\n[Node] 开始处理，User ID: {user_id}")

# 2、校验依赖是否存在
if not llm or not db:
return {"response": "System Error: Dependencies not injected!", "log": ["Error"]}

# 3、用 db 查用户角色
user_info = db.get_user_info(user_id)
user_tier = user_info.get("role")
print(f"[Node] 从 DB 获取用户角色: {user_tier}")

# 4、用 runtime.stream_writer 向图外推自定义进度（对应 custom 流式模式）
writer = runtime.stream_writer
writer({"status": "thinking", "message": f"正在调用 LLM 为 {user_tier} 用户生成回复…"})
time.sleep(0.5)

# 5、按角色构造不同 Prompt，模拟 LLM 调用
prompt = f"User({user_tier}) asks: {user_query}"
llm_response = llm.invoke(prompt)

return {
"response": llm_response,
"log": [f"Processed by {llm.__class__.__name__}"],
}

def run_demo():
builder = StateGraph(CustomerSupportState)
builder.add_node("customer_service", node_customer_service)
builder.add_edge(START, "customer_service")
builder.add_edge("customer_service", END)
graph = builder.compile()

my_llm = MockLLM()
my_db = MockDatabase()

initial_state = {"query": "如何升级会员？"}
config = {"configurable": {"user_id": "vip_user_999"}}
context = {"llm": my_llm, "db": my_db}

print("[System] 开始运行图，并注入依赖对象…")
result = graph.invoke(initial_state, config=config, context=context)
print(result)

注意最后一行 graph.invoke(initial_state, config=config, context=context)——context 就是"外挂依赖包"，节点里通过 runtime.context 取用。这个模式在真实项目里非常推荐：数据库连接、HTTP 客户端、配置对象都走这条路，节点函数就成了纯逻辑，好测、好换。

10.2 节点输出：只能返回"增量"，千万别返回整个 state

这是新手最容易犯的错，而且报错信息不一定直观，值得单独用一节讲。

规则：节点的返回值表示"本节点对状态的增量修改"，不能把收到的整个 state 原样返回。

为什么？因为 LangGraph 会把每个节点的返回值当作增量，去和全局状态做一次合并。如果你返回了整个 state：

• 那些没有配 Reducer 的键，LangGraph 不知道该以谁为准，会直接抛异常；
• 那些配了 Reducer 的键，因为你把没改动的旧值也一起返回了，合并后可能产生重复数据（比如消息被复制一遍）。

下面就是错误示范：

from typing import TypedDict
from langgraph.constants import START, END
from langgraph.graph import StateGraph

class MyState(TypedDict):
query: str
file_result: str
web_result: str
final_answer: str

def query_web(state: MyState) -> dict:
# 错误演示：直接 return 整个 state，而不是当前节点增量修改的部分
query = state["query"]
state["web_result"] = f"{query}的网络搜索结果"
return state # <– 这里错了

def query_file(state: MyState) -> dict:
query = state["query"]
state["file_result"] = f"{query}的文件搜索结果"
return state # <– 这里错了

def answer(state: MyState) -> dict:
web_result = state["web_result"]
file_result = state["file_result"]
state["final_answer"] = f"LLM基于{web_result}，{file_result} 的最终结果"
return state # <– 这里错了

graph = StateGraph(state_schema=MyState)
graph.add_node(answer)
graph.add_node(query_web)
graph.add_node(query_file)
graph.add_edge(START, "query_web")
graph.add_edge(START, "query_file")
graph.add_edge("query_web", "answer")
graph.add_edge("query_file", "answer")

compiled_graph = graph.compile()
res = compiled_graph.invoke({"query": "什么是Langgraph"})
print(res["final_answer"])

正确写法只有一行之差：return {"web_result": …}，只写你改动的那个键。

10.3 START 和 END：两个虚拟节点

它们本质上就是两个字符串，源码里写得很清楚：

END = sys.intern("__end__")
"""The last (maybe virtual) node in graph-style Pregel."""

START = sys.intern("__start__")
"""The first (maybe virtual) node in graph-style Pregel."""

• START 代表"把用户输入送进图里的那个点"。引用它的目的是声明哪些节点是入口。如果你写了多条 add_edge(START, …)，这些节点就会并行开始。
• END 代表终止。哪条边指向它，就表示那条路走到头了，后面没有动作。

一个小细节：START 这个"特殊节点"执行完之后，也会产生一个 checkpoint。所以你在历史快照里往往会看到第一步的 values 是空的——那就是 START 留下的。

10.4 节点缓存：同样的输入，别算第二遍

有些节点很贵：调一次大模型几毛钱，跑一次复杂计算要 5 秒。如果输入没变，结果应该直接复用。LangGraph 支持按节点输入做缓存。

两步搞定：

CachePolicy 有两个参数：

• key_func：怎么根据输入生成缓存键。默认对输入做 pickle 再 hash；
• ttl：缓存活多久（秒）。不写就永不过期。

import time
from typing_extensions import TypedDict
from langgraph.graph import StateGraph
from langgraph.cache.memory import InMemoryCache
from langgraph.types import CachePolicy
from langgraph.constants import START
from langgraph.checkpoint.memory import InMemorySaver

class State(TypedDict):
x: int
result: int

builder = StateGraph(State)
checkpointer = InMemorySaver()

# 1、定义节点：模拟有耗时的计算
def expensive_node(state: State) -> dict:
print("expensive_node 被调用")
time.sleep(5)
print("expensive_node 计算完成")
return {"result": state["x"] * 2}

# 2、添加节点时配置缓存策略：缓存 10 秒
builder.add_node("expensive_node", expensive_node, cache_policy=CachePolicy(ttl=10))
builder.add_edge(START, "expensive_node")

# 3、编译时传入缓存器
graph = builder.compile(cache=InMemoryCache(), checkpointer=checkpointer)

# 4、第一次调用，x=5，会真的算 5 秒
print(graph.invoke({"x": 5}, config={"configurable": {"thread_id": "1"}}))

# 5、第二次调用，x 还是 5，10 秒内 => 直接命中缓存，秒回
print(graph.invoke({"x": 5}, config={"configurable": {"thread_id": "2"}}))

跑一下你能明显感觉到：第二次几乎是瞬间返回的。注意这里用了不同的 thread_id，说明缓存是跨会话共享的（因为它只依赖输入，不依赖会话）。

10.5 节点重试：调 API、调模型必备

调外部接口总会偶发失败：网络抖一下、限流、超时。给节点配一个重试策略，比自己在函数里写 for i in range(3) 优雅得多。

在 add_node 时传 retry_policy，它接受一个 RetryPolicy 对象，两个关键属性：

• max_attempts：最多尝试几次；
• retry_on：遇到哪些异常类型才重试。

from typing import Dict, Any
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import RetryPolicy

class State(TypedDict):
result: str

attempt_counter = 0

def unstable_api_call(state: State) -> Dict[str, Any]:
"""模拟一个不稳定的 API 调用，前两次失败，第三次成功"""
global attempt_counter
attempt_counter += 1
print(f"尝试调用API，这是第 {attempt_counter} 次尝试")

if attempt_counter < 3:
raise Exception(f"模拟API调用失败 (尝试 {attempt_counter})")
else:
return {"result": f"API调用成功，经过 {attempt_counter} 次尝试"}

def value_error_call(state: State) -> Dict[str, Any]:
"""抛出 ValueError，默认策略不会重试它"""
print("调用会抛出 ValueError 的节点")
raise ValueError("模拟 ValueError 异常")

def run_demo():
global attempt_counter

# 演示1：默认重试策略
attempt_counter = 0
builder1 = StateGraph(State)
builder1.add_node(
"unstable_call",
unstable_api_call,
retry_policy=RetryPolicy(max_attempts=5),
)
builder1.add_edge(START, "unstable_call")
builder1.add_edge("unstable_call", END)
graph1 = builder1.compile()

try:
print("最终结果:", graph1.invoke({"result": ""}))
except Exception as e:
print(f"最终失败: {type(e).__name__}: {e}")

# 演示2：不会被重试的异常
builder3 = StateGraph(State)
builder3.add_node(
"value_error_call",
value_error_call,
retry_policy=RetryPolicy(max_attempts=3),
)
builder3.add_edge(START, "value_error_call")
builder3.add_edge("value_error_call", END)
graph3 = builder3.compile()

try:
print("最终结果:", graph3.invoke({"result": ""}))
except Exception as e:
print(f"最终失败: {type(e).__name__}: {e}")

重点：默认策略并不会重试所有异常。 下面这些"明显是代码写错了"的异常，重试一万次也没用，所以默认跳过：

ValueError、TypeError、ArithmeticError、ImportError、LookupError、NameError、SyntaxError、RuntimeError、ReferenceError、StopIteration、StopAsyncIteration、OSError。

所以演示 2 里，ValueError 只会执行一次就失败退出，不会尝试 3 次。反过来，如果你想让某个异常重试（比如自定义的 RateLimitError），就在 retry_on 里显式指定。

十一、图内外数据传递：流式输出六种模式

11.1 为什么需要流式

默认情况下，graph.invoke 要等整张图跑完才返回。对 Agent 来说这体验很糟：

• 模型正在逐字生成答案，用户却盯着转圈圈等 10 秒；
• 流程很长，用户不知道现在跑到哪一步了。

好在 LangGraph 实现了 langchain_core 的 Runnable 接口，所以自带 stream / astream 方法，可以边跑边把内部数据吐出来。

11.2 六种模式对照表

11.3 完整演示

import time
from typing import TypedDict, Annotated, List
import operator
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.runtime import Runtime

llm = ChatOpenAI(model="gpt-4o-mini")

# — 1. 定义状态 —
class State(TypedDict):
messages: Annotated[List[BaseMessage], operator.add]
current_step: str

# — 2. 定义节点 —
def node_input(state: State):
"""模拟接收用户输入"""
return {
"messages": [HumanMessage(content="请帮我写一段Python代码")],
"current_step": "input_received",
}

def node_processing(state: State, runtime: Runtime):
"""中间处理过程，用 writer 输出自定义流式数据（对应 custom 模式）"""
steps = ["正在分析意图…", "正在检索知识库…", "正在构建Prompt…"]
writer = runtime.stream_writer
for i, step in enumerate(steps):
time.sleep(0.5)
writer({
"step_index": i + 1,
"description": step,
"timestamp": time.time(),
})
return {"current_step": "processing_complete"}

def node_generation(state: State):
"""LLM 生成；messages 模式会自动捕获它的流式输出"""
response = llm.invoke(state["messages"])
return {"messages": [response], "current_step": "generation_complete"}

# — 3. 构建图 —
builder = StateGraph(State)
builder.add_node("input", node_input)
builder.add_node("process", node_processing)
builder.add_node("generate", node_generation)
builder.add_edge(START, "input")
builder.add_edge("input", "process")
builder.add_edge("process", "generate")
builder.add_edge("generate", END)
graph = builder.compile()

# — 4. 演示不同模式 —
def run_demo():
initial_state = {"messages": [], "current_step": "start"}

print("\\n===== 1. Mode: values（完整状态）=====")
for event in graph.stream(initial_state, stream_mode="values"):
print(f"State: keys={list(event.keys())}, step={event.get('current_step')}")

print("\\n===== 2. Mode: updates（增量更新）=====")
for event in graph.stream(initial_state, stream_mode="updates"):
print(f"Update: {event}")

print("\\n===== 3. Mode: custom（节点内自定义数据）=====")
for event in graph.stream(initial_state, stream_mode="custom"):
print(f"Custom Data: {event}")

print("\\n===== 4. Mode: messages（LLM Token）=====")
for chunk, metadata in graph.stream(initial_state, stream_mode="messages"):
node_name = metadata.get("langgraph_node", "unknown")
print(f"[{node_name}] Token: {chunk.content!r}")

print("\\n===== 5. Mode: debug（调试）=====")
count = 0
for event in graph.stream(initial_state, stream_mode="debug"):
if count < 3:
print(f"Debug Event: {event['type']} – {event.get('payload', {}).get('name')}")
count += 1
print("… (省略后续 debug 信息)")

print("\\n===== 6. Mixed Mode（混合模式）=====")
for mode, data in graph.stream(initial_state, stream_mode=["updates", "custom"]):
if mode == "updates":
print(f"[Updates] 来自节点 {list(data.keys())[0]}")
elif mode == "custom":
print(f"[Custom] {data['description']}")

run_demo()

怎么选？ 一个实用的组合是：前端要"打字机效果 + 当前步骤提示"，就用 stream_mode=["messages", "custom"]；要调试"为什么没走那条分支"，临时换成 debug，一眼就能看到路由函数的返回值。

十二、Human-in-the-Loop：让用户参与关键决策

12.1 为什么需要人工审核

Agent 再聪明，有些决定也不该它自己拍板：转账、下单、发公告、改生产数据库。这时候需要"跑到一半暂停，等人确认，再继续"。

LangGraph 为此提供了一个原语：interrupt。它可以在节点内部任意位置插入，执行到那里就停下来，同时把"需要人看的数据"带出图外；用户处理完之后，再让图继续往下跑。

12.2 完整示例：转账前人工审核

"""
LangGraph interrupt 演示：让用户参与关键决策
场景：准备执行一笔"转账"，在真正执行前让用户审核/修改转账信息。
"""
from __future__ import annotations
from typing import Any
from typing_extensions import TypedDict
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.types import Command, interrupt

class TransferState(TypedDict):
recipient: str
amount: int
memo: str
approved: bool
final_status: str

def review_transfer(state: TransferState) -> dict[str, Any]:
print("\\n[Node] review_transfer：生成待执行的转账请求")
pending_transfer = {
"recipient": state["recipient"],
"amount": state["amount"],
"memo": state["memo"],
}

# 关键：执行到这里会暂停，payload 被送到图外
user_review = interrupt({
"title": "转账审核",
"pending_transfer": pending_transfer,
"instruction": "请返回 bool(是否批准) 或 dict(可改 recipient/amount/memo，并带 approved 字段)。",
})

approved = False
updated_transfer = dict(pending_transfer)

if isinstance(user_review, bool):
approved = user_review
elif isinstance(user_review, dict):
approved = bool(user_review.get("approved", True))
for k in ("recipient", "amount", "memo"):
if k in user_review:
updated_transfer[k] = user_review[k]

print(f"[Node] review_transfer：用户审核结果 approved={approved}，transfer={updated_transfer}")
return {
"approved": approved,
"recipient": updated_transfer["recipient"],
"amount": updated_transfer["amount"],
"memo": updated_transfer["memo"],
}

def execute_transfer(state: TransferState) -> dict[str, str]:
if not state["approved"]:
print("\\n[Node] execute_transfer：用户未批准，取消转账")
return {"final_status": "已取消：用户未批准转账"}

print("\\n[Node] execute_transfer：模拟执行转账")
return {
"final_status": (
f"已转账：收款人={state['recipient']}，"
f"金额={state['amount']}，备注={state['memo']}"
)
}

def build_graph():
builder = StateGraph(TransferState)
builder.add_node("review_transfer", review_transfer)
builder.add_node("execute_transfer", execute_transfer)
builder.add_edge(START, "review_transfer")
builder.add_edge("review_transfer", "execute_transfer")
builder.add_edge("execute_transfer", END)
return builder.compile(checkpointer=InMemorySaver())

def run_demo():
graph = build_graph()
config = {"configurable": {"thread_id": "interrupt-demo-1"}}

initial_state = {
"recipient": "Alice",
"amount": 100,
"memo": "午饭AA",
"approved": False,
"final_status": "",
}

# 第一次 invoke：会停在 interrupt
first = graph.invoke(initial_state, config=config)
interrupt_payload = first["__interrupt__"][0].value
print("\\n[System] 图已暂停，等待用户审核。可给用户展示的数据如下：")
print(interrupt_payload)

# 用户审核完了，带着决定恢复
user_resume_value = {"approved": True, "amount": 80, "memo": "改为80元（实际应付）"}
final = graph.invoke(Command(resume=user_resume_value), config=config)
print("\\n[System] 已恢复执行，最终结果：")
print(final["final_status"])

run_demo()

12.3 三个必须知道的细节

十三、Edge：边和两种流转方式

13.1 边的本质

前面讲 Pregel 时已经埋了伏笔：定义边，本质上是在定义"节点的订阅通道"和"节点执行后要更新的通道"。

边的类型主要有两种：

• 普通边（Normal Edges）：从一个节点直接连到下一个节点，无条件。
• 条件边（Conditional Edges）：调用一个函数来判断接下来去哪（哪些）节点。

13.2 条件边：本质是一个路由函数

条件边就是一个普通 Python 函数：接收 state，返回"下一个（或下一批）节点名"。它决定了流程的分叉。

from typing import Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class GraphState(TypedDict):
value: int
step: str

def node_a(state: GraphState) -> dict:
print("执行节点A")
return {"value": state["value"], "step": "A执行完毕"}

def node_b(state: GraphState) -> dict:
print("执行节点B")
return {"value": state["value"] * 2, "step": "B执行完毕"}

def node_c(state: GraphState) -> dict:
print("执行节点C")
return {"value": state["value"] – 1, "step": "C执行完毕"}

# 路由函数：根据 value 的奇偶性决定去哪
def route_condition(state: GraphState) -> Literal["node_b_alias", "node_c_alias"]:
if state["value"] % 2 == 0:
return "node_b_alias" # 偶数去 B
else:
return "node_c_alias" # 奇数去 C

def main():
builder = StateGraph(GraphState)

builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_node("node_c", node_c)

builder.add_edge(START, "node_a")

# 条件边：源节点 + 路由函数 + 路由映射表
builder.add_conditional_edges(
"node_a",
route_condition,
{
"node_b_alias": "node_b",
"node_c_alias": "node_c",
},
)

builder.add_edge("node_b", END)
builder.add_edge("node_c", END)

graph = builder.compile()

print("输入值为偶数:")
print(f"执行结果: {graph.invoke({'value': 2})}")

print("\\n输入值为奇数:")
print(f"执行结果: {graph.invoke({'value': 1})}")

main()

为什么要有"路由映射表"这第三个参数？ 因为路由函数返回的只是一个"标记名"（alias），映射表负责把它翻译成真实节点名。这样做的好处是：路由函数不需要知道真实节点叫什么，解耦得更彻底，也方便一个函数复用到多张图里。

如果你嫌麻烦，也可以让路由函数直接返回真实节点名，省略映射表。小项目里这么写更省事。

13.3 可控循环：用条件边转圈，用 recursion_limit 刹车

把"条件边"和"回边"组合起来，就有了循环。最典型的就是 ReAct 模式：模型思考 → 调工具 → 结果回模型 → 再思考……直到模型说"不用调工具了"才结束。

但这里埋着一颗雷：如果路由条件永远不满足，图会一直转圈，把你的 token 和钱一起烧掉。

LangGraph 给的刹车是 recursion_limit：它限制图在抛异常之前允许执行的超步数量，默认 25，在 graph.invoke 的 config 里指定。超了就会抛 GraphRecursionError。

from typing import Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.errors import GraphRecursionError

class LoopState(TypedDict):
count: int
result: str
max_count: int

def node_a(state: LoopState) -> dict:
print(f"执行节点a，当前计数: {state['count']}")
return {
"count": state["count"] + 1,
"result": f"已处理{state['count']}次",
}

def node_b(state: LoopState) -> dict:
print(f"执行节点b，当前计数: {state['count']}")
return {"result": f"已处理{state['count']}次 – 辅助处理"}

def route(state: LoopState) -> Literal["b", END]:
"""条件路由：决定继续循环还是终止"""
if state["count"] >= state["max_count"]:
print(f"满足终止条件，计数 {state['count']} >= {state['max_count']}，返回END")
return END
else:
print(f"未满足终止条件，计数 {state['count']} < {state['max_count']}，返回b")
return "b"

builder = StateGraph(LoopState)
builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.add_edge(START, "a")
builder.add_conditional_edges("a", route)
builder.add_edge("b", "a") # 回边，形成循环

graph = builder.compile()

print("=== 开始执行工作流 ===")
try:
result = graph.invoke(
input={"count": 0, "result": "", "max_count": 300},
config={"recursion_limit": 6}, # 刹车：最多 6 个超步
)
print("=== 执行结果 ===")
print(result)
except GraphRecursionError as e:
print(f"递归错误: {e}")

这个例子里 max_count 设成 300，但 recursion_limit 只有 6，所以图会在第 6 个超步被强制中断，抛出 GraphRecursionError。这正是你想要的保护——宁可报错，也别无限循环。

一个易混点：recursion_limit 的单位是"超步"，不是"节点执行次数"。 因为一个超步里可能有多个节点并行执行，所以 limit=6 不代表只跑 6 个节点。调试时如果发现"明明只设了 6 却跑了十几个节点"，就是这个原因。

十四、踩坑清单：这些坑我帮你先踩了

下面这些是社区里反复被提到的真实问题，按"最容易踩"排序。

14.1 State 不是垃圾桶

• 只放节点之间需要共享传递的数据，节点内部的临时变量就留在函数里；
• 字段名要有语义：用 messages、user_input、tool_result，别用 data、info；
• 嵌套别超过两层，复杂结构尽量扁平化，否则读写心智负担会爆炸。

记住一句话：State 的设计质量决定了系统的可维护性。多花十分钟设计，少花十小时调试。

14.2 thread_id 的命名方式，决定了你的记忆会不会串

如果每次用户刷新页面就生成一个新 thread_id，短期记忆立刻断掉。推荐用"用户标识-会话标识"的组合，例如 user_1021-session_0003。这样既能区分会话，又能靠前缀做用户级聚合。

14.3 历史消息不能无限增长

把几十轮对话全塞给模型，token 会被吃光，而且模型还会"记混"。建议设一个阈值（比如超过 30 轮或接近 token 上限），触发一个摘要节点，把早期对话压缩成一段摘要存进 State。

14.4 人工审批要做成状态机，不是一条边

审批其实有四个状态：等待、通过、拒绝、超时。建议在 State 里显式放一个 approval_state 字段，进审批节点先检查状态，只处理 waiting 的请求。否则可能出现"两个人同时收到审批、各点一次，业务被执行两遍"的事故。

14.5 生产环境别用 InMemorySaver

InMemorySaver 只适合本地调试。生产请换 SqliteSaver / PostgresSaver / RedisSaver，否则服务一重启，所有会话记忆和断点全部蒸发。

14.6 日志和指标别塞进 State

有人喜欢把"每一步的执行日志"累积到 State 里。问题是每次状态变更都要存一份快照，State 越胖，checkpoint 越慢，成本越高。日志该走日志系统，别走图状态。

14.7 不要在 interrupt 之前做有副作用的操作

恢复时节点会从头重跑，前面已经说过，但值得再强调一次。转账、发邮件、扣积分这类操作，务必放到 interrupt 之后，或者做成幂等。

十五、学习路线建议

如果看到这里你已经有点晕，别慌，这是正常的。建议按下面这个顺序动手，每一阶段都写代码，别只读：

走完这五步，你已经能覆盖 80% 的实际需求了。再往后就是进阶话题：子图（把一张图当成另一个节点的内部）、Send 做动态扇出（Map-Reduce 风格）、多智能体协作（Supervisor 架构）——这些等你把基础打牢再看，会顺很多。

十六、总结

最后用一段话把全文串起来：

• LangGraph 是干什么的：把 Agent 工作流画成图，从而支持分支、循环、并行、暂停、恢复。它比 LangChain 更底层、更可控。
• 三大件：State（共享记事本）、Node（工位）、Edge（传送带）。写应用就是"定义记事本 → 写工位 → 连传送带 → 编译 → 调用"。
• Reducer：决定增量怎么合并。不配就覆盖，配了才能追加。并行写同一个键又没配 Reducer，必报错。
• Checkpointer + thread_id：这是"记忆"和"断点续跑"的地基。invoke(None) 加同一个 thread_id 就能从断点继续。
• Pregel：底层是"超步"驱动的，Plan → Execute → Update 不断循环，直到没有节点被选中。所以同一个节点可以被执行多次，循环是天然的。
• 节点输出必须是增量，返回整个 state 是最常见的错误。
• 流式输出有六种模式，messages 给打字机效果，custom 给进度提示，debug 给排错。
• interrupt 让人参与决策，但恢复时整个节点会重跑，务必保证幂等或拆分节点。
• 循环一定要有刹车：recursion_limit 数的是超步，不是节点。

LangGraph 的学习曲线主要不在 API，而在思维方式——从"写一串函数调用"切换到"设计一张状态机"。一旦这个思维建立起来，你会发现很多以前写得很别扭的 Agent 逻辑，突然变得清晰可控。

希望这篇能帮你少走点弯路。