飞污熊小站
飞污熊小站

使用LangChain+LangGraph开发AI应用（8） - LangGraph 基础

XiongNeng 收录于 AI
 约 4100 字 预计阅读 9 分钟 

https://static.xiongneng.me/langchain-langgraph-cover-20261017184837.png

学习进度

LangGraph 是 agent 脚下的引擎。你在 create_agent 里用到的循环、限轮数、钩子时序，全是这张图上长出来的东西。这一篇把引擎拆开看，状态 Schema 定字段，节点干活，边定走向，Reducers 管并发写入怎么合并，这是图语言的全部词汇，学会了，中断恢复、人机协同、多智能体编排才有地基。

示例在 source/hello_graph_deep.py，九个 demo，全部纯函数不依赖模型。完整案例是一个文本体检流水线 source/textcheck.py，清洗、分诊、统计、汇总，一条能看见每一步的图。

前言用途

写流程有三个层次。最朴素的是顺序脚本，从头跑到尾，遇不到分叉。往上一点是工作流引擎，有分支有循环，但每一步写死在代码里。LangGraph 走的是第三条路，把流程声明成一张图，节点是函数，边是走向，状态是所有节点共享的一块数据，执行交给引擎。你描述图，引擎管调度，并行、重试、断点恢复、状态持久化这些脏活都不用你写。

这带来三个直接的好处。画出来的流程可审查，评审的人看图说话，不用读代码脑补控制流。节点之间只通过状态通信，任何一个节点换成新实现，别的代码不动。引擎知道每个节点的边界，哪些节点能并行、哪一步该存档，都由图结构推出来，后续篇章的中断恢复、人机协同全建立在这上面。

什么时候需要直接写图。流程里有分支或循环，或者你想让某几步并行跑，或者要做中途等人确认的交互，这些场景下手写图比套 agent 模板更可控。单轮问答这种直线流程，用 create_agent 就够，不必为它画图。

原理概念

LangGraph 把工作流建模成三样东西。State 是共享数据结构，全图只有一份，记录流程此刻的快照。Node 是函数，读状态、干活、返回更新。Edge 决定下一个跑哪个节点。官方文档的说法很省字，nodes do the work, edges tell what to do next。

https://static.xiongneng.me/graph-exec-20261017184649.png

执行引擎的算法来自 Google 的 Pregel，按离散的 super-step 推进。每个 super-step 里，所有被激活的节点并行执行，各自把更新交回状态，然后引擎按边激活下一批节点，直到没有节点被激活。demo 4 里一个节点两条出边、两个目标并行跑，就是这个机制在工作。这个模型还解释了一个设计约束，同一个 super-step 里多个节点写同一个字段，引擎必须知道怎么合并，这就轮到 Reducer 出场。

节点函数的参数是可选的，写几个框架注几个。最常见是只收 state。加一个 config 参数能拿到 RunnableConfig，thread_id、tags 这些运行时配置都在里面。再加一个 runtime 参数拿到 Runtime 对象，context、store、stream_writer 都从这来。探针实测过 (state, config) 双参写法，config["configurable"] 里传什么节点就收到什么。参数是按声明注入的，不声明就不给，函数保持好测的纯函数长相。

State 的 schema 支持 TypedDict、dataclass 和 Pydantic 三种，官方主要推荐 TypedDict，需要默认值用 dataclass，需要递归校验才上 Pydantic，注意 langchain 的 create_agent 不支持 Pydantic 状态。每个字段可以挂一个 Reducer，它是个二元函数，签名固定，左边是字段当前值，右边是节点返回的新值，返回合并结果。Annotated[list[str], operator.add] 的意思就是这个字段用列表相加来合并。没挂 Reducer 的字段走覆盖语义，节点返回什么就是什么。节点只返回更新的那几个字段，不用带全量状态。

图还支持多套 schema 分工。定义一个只含输入字段的 InputState 和只含输出字段的 OutputState，构造时传 StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)，对外看到的输入输出就收窄了，内部节点之间随便传中间字段。官方管这种中间字段叫私有 channel，专门给节点内部通信用。有个实测过的细节要留意，私有字段在流式输出时不会被隐藏，stream_mode="values" 会把所有 channel 都吐出来，要藏就用 output_keys 限制，或者改看 updates 模式只看每步的更新。

图编译后才能跑，compile() 会做结构检查。入口和出口是两个常量，START 和 END，从 langgraph.graph 导入，add_edge(START, "a") 定入口，add_edge("b", END) 定出口。

核心用法

九个用法，示例在 source/hello_graph_deep.py，全部纯函数图，不依赖任何模型，python 直接跑。

1. 静态边

最基础的图，两个节点一条直线。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

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

class State(TypedDict):
    text: str

def upper(state: State):
    return {"text": state["text"].upper()}

def wrap(state: State):
    return {"text": f"[{state['text']}]"}

g = StateGraph(State)
g.add_node("upper", upper)
g.add_node("wrap", wrap)
g.add_edge(START, "upper")
g.add_edge("upper", "wrap")
g.add_edge("wrap", END)
result = g.compile().invoke({"text": "hello"})
1
2

== 静态边 ==
  [HELLO]

节点返回的是部分更新，{"text": ...} 一个字段就够，引擎负责合并进状态。静态边是无条件的，走到这就一定走那条。

2. Reducer 两种语义

同一个图里两种字段，看合并和覆盖怎么各干各的。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

from operator import add
from typing import Annotated

class State(TypedDict):
    log: Annotated[list[str], add]
    last: str

def node_a(state: State):
    return {"log": ["a"], "last": "a"}

def node_b(state: State):
    return {"log": ["b"], "last": "b"}
1
2

== Reducer ==
  log 合并成 ['a', 'b']，last 被覆盖成 b

log 挂了 add，两个节点的写入相加成 ['a', 'b']。last 没挂，后写的 b 把 a 盖掉。什么时候用哪个，判断标准就一条，这个字段会不会被多个节点写，会就挂 reducer，只被一个节点写就覆盖。

operator.add 之外，reducer 就是个普通函数，自己写也行。官方的示例叫 append_strings，签名两个位置参数 left 和 right，返回 left + right，效果跟 operator.add 一样。想做去重、排序、截断，都在这个函数里做。

1
2
3
4
5

def merge_unique(left: list[str], right: list[str]) -> list[str]:
    return list(dict.fromkeys(left + right))

class State(TypedDict):
    tags: Annotated[list[str], merge_unique]

3. 条件路由

add_conditional_edges 按状态决定下一跳。

1
2
3
4
5
6

from typing import Literal

def check(state: State) -> Literal["even", "odd"]:
    return "even" if state["n"] % 2 == 0 else "odd"

g.add_conditional_edges("check", check)
1
2
3

== 条件路由 ==
  n=4 走 even
  n=7 走 odd

路由函数读状态返回下一个节点名。返回值可以直接是节点名，也可以配映射字典 {True: "node_b", False: "node_c"} 让可读性更好。路由函数的类型注解写 Literal[...]，图可视化时靠它画分支。

4. 多出边并行

一个节点画多条出边，目标节点并行执行。

1
2

g.add_edge("fan", "x1")
g.add_edge("fan", "x2")
1
2

== 多出边并行 ==
  执行了 ['fan', 'x1', 'x2']

fan 的两条出边让 x1、x2 落进同一个 super-step，引擎并行跑完再汇合。两个节点都往 log 写，所以 log 挂了 reducer，没挂的话这里直接抛 InvalidUpdateError。

5. Overwrite 绕过 reducer

合并型字段想清空，返回空列表是没用的。

1
2
3
4

from langgraph.types import Overwrite

def node_clear(state: State):
    return {"errors": Overwrite([])}
1
2

== Overwrite ==
  清空后 []

挂了 add 的字段，节点返回 [] 会被合并进旧值，等于什么都没做。Overwrite([]) 绕过 reducer，直接把字段替换成空列表。这个类型专治「我想覆盖一个带 reducer 的字段」。

6. add_messages 按 id 更新

消息列表的专属 reducer，按消息 id 对账。

1
2
3
4
5
6
7

from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]

def bot(state: State):
    return {"messages": [AIMessage(id="m1", content="revised answer")]}
1
2

== add_messages ==
  同 id 更新后: revised answer

新 id 追加，同 id 覆盖，输入的第 m1 条被原地改写而不是追加一条。这个语义是第 9 篇人机协同的基础，人工修订一条消息就是用同 id 重发。第 1 篇用过的 MessagesState 就是预置了这套 reducer 的现成 schema，自己加字段继承它就行。

1
2
3
4

from langgraph.graph import MessagesState

class State(MessagesState):
    documents: list[str]

messages 字段不用再声明，子类里只写新增的 documents。

7. Command 节点内路由

条件边的路由逻辑挂在边上，Command 把它收回节点内部。

1
2
3
4
5
6

from langgraph.types import Command

def judge(state: State) -> Command[Literal["pass", "fail"]]:
    if state["score"] >= 60:
        return Command(update={"where": "pass"}, goto="pass")
    return Command(update={"where": "fail"}, goto="fail")
1
2
3

== Command ==
  score=90 -> pass
  score=50 -> fail

update 和 goto 打包在一个返回值里，状态更新和跳转天然同步，不会出现改了 A 字段却跳去不认 A 的节点。官方的选型建议很清楚，只路由用条件边，要同时更新状态和路由用 Command。类型注解 Command[Literal[...]] 必须写，图渲染靠它。

8. Send map-reduce

任务条数运行时才知道，固定边画不了，Send 把每个子任务连各自的 state 发出去。

1
2
3
4
5
6

from langgraph.types import Send

def dispatch(state: State):
    return [Send("make", {"subject": s}) for s in state["subjects"]]

g.add_conditional_edges(START, dispatch, ["make"])
1
2

== Send map-reduce ==
  ['关于狗的笑话', '关于猫的笑话', '关于鸟的笑话']

dispatch 返回 Send 列表，几个条目就发几份，每份带独立的输入 state，三个统计节点并行执行，结果经 reducer 汇总。处理文章列表、批量文档这类 map-reduce 场景，这就是标准姿势。

9. 递归上限

循环图必须有出口，引擎的兜底是 recursion_limit。

1

app.invoke({"n": 0}, config={"recursion_limit": 5})
1
2
3

== 递归上限 ==
  循环到 3 正常结束: n=3
  死循环被 recursion_limit=5 叫停

每个 super-step 计一步，默认上限一千，超了抛 GraphRecursionError。demo 里循环三次正常结束，死循环图被 limit=5 叫停。调试循环图时把这个值调小，炸得快好定位。

完整案例

九个词都认识了。收个尾，用它们搭一条真的流水线，textcheck.py，文本体检。输入一段文字，清洗空白，判断中英文，按语言走不同的统计分支，最后汇总一份体检报告。

1

uv run python textcheck.py "LangGraph 让工作流变成一张能看见的图，节点做活，边定走向，条件边让流程学会分叉。" --trace
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

执行轨迹:
  [clean] cleaned=LangGraph 让工作流变成一张能看见的图，节点做活，边定走向...
  [detect] lang=zh
  [zh_count]
      + stats: 中文字数 31
      + stats: 字频前五 让x2、流x2、边x2
      + stats: 句子数 1
  [summary] report=文本体检报告

文本体检报告
  · 语言判定 zh
  · 中文字数 31
  · 字频前五 让x2、流x2、边x2
  · 句子数 1

–trace 用 updates 流把每一步打印出来，哪个节点跑了、改了哪个字段一目了然，这就是「可审查」的直观长相。图的结构是 clean 到 detect，detect 条件路由到 zh_count 或 en_count，两个统计节点都写 stats 这个 Annotated[list[str], add] 字段，最后汇进 summary。

1

uv run python textcheck.py "StateGraph turns a workflow into a visible graph. Nodes do the work."
1
2
3
4
5

文本体检报告
  · 语言判定 en
  · english words 12
  · top words ax2, stategraphx1, turnsx1, workflowx1, intox1
  · sentences 2

英文输入自动换分支，词频、词数、句子数按英文规则统计。空输入还有一条特殊路径，summary 节点里判断 cleaned 为空就 Command(goto=END) 提前收工，报告写「空文本，没什么可体检的」，一个节点同时干完更新和跳转。

1

echo "" | uv run python textcheck.py
1

空文本，没什么可体检的。

整个案例没有一个模型调用，每一步都是确定性函数，跑一百遍结果一致。测试在 tests/test_hello_graph_deep.py 九条加 tests/test_textcheck.py 八条，链式顺序、两种 reducer 语义、条件路由、并行分支、Overwrite、同 id 更新、Command 路由、Send 发散、递归兜底、空文本短路都钉住了。加上前七篇的，全仓一百三十三条，全绿。

源码

https://github.com/yidao620c/xneng-langchain-langgraph

参考

更新于 2026-10-17 
LangChainLangGraph
返回 | 主页

目录