什么是LangGraph?

发表于 |

做「能自己查资料、自己决定下一步」的 AI 助手，几乎一定会碰到 LangGraph。

四个名字经常被说成一个。其实各管一层：LangChain 是零件盒，LangGraph 是现场调度——下一步去哪、做一半能不能停、关机之后病历还在不在。

一句话：LangGraph 是给 AI 应用用的流程图运行时。它允许转圈、允许等人、每一步都存档。它不让模型更聪明，它让「聪明」按你画的规矩干活，并且中途不把病历弄丢。

模型、工具、提示词这些零件怎么拼，上一篇《什么是 LangChain?》写过。这篇只讲调度。文中的说法以 LangGraph 1.x 为准。

先看一个柜台

客人说：订单迟到了，我要退款。

只会「问一句、答一句」的模型，多半直接编一段安抚。能办事的助手得分几步走：

  1. 听清楚要退哪一单
  2. 查物流，别靠猜
  3. 金额不大就按规矩办；超过 200 元，等主管点头
  4. 点头之后才去碰支付
  5. 主管去吃饭了，这单不能丢

左边是流水线。点完就往下，地上没有回头的箭头。客人突然说不要香菜，要么整单重来，要么假装没听见。

右边允许转圈。厨房发现不对，箭头指回点单，「不要香菜」还写在病历上，然后才上菜。

AI 助手更像右边。它经常是：想一想，发现还缺资料，去查，看完再想。中间还可能要人盖章。这张「能回头、能暂停、关了机还能找回来」的图，就是 LangGraph 在管的事。

四个名字，各管一层

LangGraph 是开源的编排运行时，Python 和 JavaScript 都有。官方把它放在很底层：用来构建可以跑很久、中途状态还在的 Agent。Klarna、Uber、J.P. Morgan 出现在官方介绍里，因为这类系统最在意三件事：中途不能丢，花钱之前要有人点头，能同时干的步骤别排成一队干等。

可以不装 LangChain，单独用 LangGraph。反过来，LangChain 里现在推荐的 create_agent，底下就是一张已经画好的 LangGraph：模型工位和工具工位来回走。

名字 管什么 记成
LangChain 模型、工具、提示词、标准助手循环 零件盒
LangGraph 状态、分支、转圈、暂停、存档 调度台 + 病历柜
LangSmith 追踪、评测、部署 监控录像
Deep Agents 规划、分身、文件、上下文，建在 LangGraph 之上 作业规范

它不训练模型，也不保证回答正确。模型照样会胡说。图保证的是：胡说发生在哪一步，看得见，拦得住，重得来。

三块积木

整张图只由三样东西拼成。

状态：一本共享病历

所有工位看同一本。上面通常有对话、查到的订单、批了没有。技术上它是一份带类型的数据，常见写法是 TypedDict。每个工位只交回「我改了哪几栏」，整本病历由调度台来合并。

默认合并很粗：后写的覆盖先写的。订单状态适合这样，「已揽收」就该替换「运输中」。对话不行。对话要追加，否则助手会失忆。这个合并规则叫 reducer。消息列表最常用的是 add_messages。

同一拍里如果两个工位改同一栏，也得先说好怎么合并。没说好，后写完的覆盖先写完的，而且谁先写完并不稳定。

节点：一个工位一件事

节点就是一个函数。输入是当前病历，输出是要改的那几栏。里面可以调用模型，也可以是普通代码：查数据库、做加法、校验格式。

分工可以记死：拿不准的判断交给模型，必须算对的动作交给代码。退款金额用代码算，要不要升级给主管可以让模型先看语气，真正扣款仍然走代码。

边：地上的箭头

普通边：做完永远去下一站。条件边：看病历再决定，比如「还要查」或者「可以结束」。

图上还有两个特殊位置：START 和 END。客人从 START 进场，走到 END 这单就办完了。

画完要 compile()。这一步检查有没有没人去的孤岛工位，也是把病历柜装上去的地方。没编译的图不能跑。

会转圈，才像一个助手

动图里那颗点，走的是助手真正的节奏：

  1. 想一想。读病历，决定直接答，还是用工具
  2. 用工具。查订单、算数、搜文档，把结果追加进病历
  3. 再想一想。看结果。还缺，就再查；够了，就去回答

以前这段藏在 Agent 的黑盒里，出了问题只能看最后一句。LangGraph 把它画在明处：每一步改了病历的哪一栏，下一步为什么走这边，都能指出来。

转圈得有刹车。两种就够：

  • 条件边指向 END：这一轮可以回答了
  • recursion_limit：转太多次就停下，避免死循环把钱烧光

想把某一拍按住看，播放下面这段。路径和动图相同。

按拍子走

图不是「哪个函数先返回就先往下冲」。它按拍子走。这个办法受 Google 的 Pregel 启发，名字可以不记，记节拍器就行。

  • 同一拍里，几个工位可以同时干。查物流和查库存不用排队
  • 这一拍全部写完，才进入下一拍
  • 每一拍结束，给病历拍一张快照，叫 checkpoint
  • 快照按 thread_id 归档。同一张图，不同客人是不同档案

停电、重启、换一台机器，都从最近一拍继续。同一拍里已经成功的工位不用重做。被打断的那一站会从头再跑一遍。

所以扣款、发短信这种动作，要么做成做两次结果仍然一样，要么放到「人点头之后」的下一站。存档发生在拍与拍之间，不发生在某个工位做到一半的时候。

两本记忆

检查点 Checkpointer 仓库 Store
记住 这一次任务的病历快照 跨很多次都想留的事实
范围 一个 thread_id 跨 thread
例子 这单聊到哪、工具返回了什么、停在盖章 这位客人要中文回复、不要打电话
生产环境 放数据库。内存版一关进程就没了 同样要持久化

内存存档（InMemorySaver，老教程里的 MemorySaver 是同类东西）只适合在自己电脑上试。用它做「等人审批」，进程一关，章就没处盖了。

这两本别塞进同一个抽屉。长期习惯写进某一单的对话里，这一单归档，习惯就跟着封存了。新的一单开场，助手又像第一次见面。

人来盖章

interrupt() 有点像 Python 的 input()：跑到这里，把一个问题抛出去，等答复。差别是它能活在生产里。进程可以关掉。三天后，另一个进程用同一个 thread_id，带着 Command(resume=...) 把答复送回去。图从档案里醒来，接着办。

三种常见用法：

  1. 批准或拒绝。退款、发邮件、删数据之前
  2. 人来改。模型起草，人改完再继续
  3. 人来补。缺订单号，停下来问

继续的时候，有个必须记住的细节：

停住所在的那一站会从头再做一遍。再次走到 interrupt()，不再暂停，而是直接拿到人的答复。

interrupt() 前面如果已经扣了款，人一点头，扣款代码会再跑一次。扣款要放在下一站：人说可以，才碰真实世界。

两种写法，同一台引擎

Graph API Functional API
你在写 节点、边、共享病历 普通函数，加上 @entrypoint / @task
适合 分支多、要并行、要给人看图 手头已有一段流程，想少改代码，先加上存档和暂停
状态 大家读写同一本 默认收在函数里面

能力是同一套：存档、流式输出、人机协同、记忆。流程还简单时，用函数写法更快。分支多到自己都指不清时，再画成图。

把柜台画成图

下面这段不调用真模型，只把调度画出来。看箭头就行。真实系统里，「接待」会去调模型；「查物流」是普通代码，不靠模型猜运单号。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
from typing import Annotated, Literal
from typing_extensions import TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.types import Command, interrupt


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


def last_text(state: State) -> str:
    msg = state["messages"][-1]
    return msg["content"] if isinstance(msg, dict) else msg.content


def receptionist(state: State) -> dict:
    # 还没查过物流，就去查；查过了，才送去盖章
    if not state.get("order"):
        return {"messages": [{"role": "assistant", "content": "先查物流"}]}
    return {"messages": [{"role": "assistant", "content": "进入审批"}]}


def route(state: State) -> Literal["lookup", "ask_human"]:
    if "先查" in last_text(state):
        return "lookup"
    return "ask_human"


def lookup(state: State) -> dict:
    return {
        "order": "已揽收",
        "messages": [{"role": "assistant", "content": "物流：已揽收"}],
    }


def ask_human(state: State) -> dict:
    # 第一次走到这里会暂停；人答复后这一站重跑，interrupt 直接返回答复
    decision = interrupt({"question": "退款 200 元，是否放行？", "order": state.get("order")})
    return {"approved": decision}


def after_human(state: State) -> Literal["refund", "end"]:
    return "refund" if state.get("approved") == "yes" else "end"


def refund(state: State) -> dict:
    return {"messages": [{"role": "assistant", "content": "已退款"}]}


builder = StateGraph(State)
builder.add_node("receptionist", receptionist)
builder.add_node("lookup", lookup)
builder.add_node("ask_human", ask_human)
builder.add_node("refund", refund)
builder.add_edge(START, "receptionist")
builder.add_conditional_edges(
    "receptionist", route, {"lookup": "lookup", "ask_human": "ask_human"}
)
builder.add_edge("lookup", "receptionist")
builder.add_conditional_edges("ask_human", after_human, {"refund": "refund", "end": END})
builder.add_edge("refund", END)

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "order-10086"}}

graph.invoke(
    {
        "messages": [{"role": "user", "content": "订单迟到，我要退款"}],
        "order": "",
        "approved": "",
    },
    config,
)
# 此时停在盖章。人同意之后，用同一份 config 再调用：
# graph.invoke(Command(resume="yes"), config)

按这张图走一遍：

  1. 客人说订单迟到、要退款
  2. 病历里还没有物流，接待把单子送到「查物流」
  3. 查完，「已揽收」写进病历，箭头回到接待
  4. 接待看到已经查过，送到「等人盖章」
  5. 图暂停。界面上可以出现：退 200 元，放行吗
  6. 人同意，用同一个 thread_id 送回 yes
  7. 「退款」这一站才调用支付，然后结束

人下班了也没关系。明天打开同一个 thread，还停在盖章那一拍。上面用内存存档，是为了把结构放在一个屏幕里。真正上线，要换成数据库里的 checkpointer，否则进程一重启，这单就找不回来。

什么时候值得自己画

你的情况 用什么
问一句，答一句 直接调模型
固定两三步，不暂停、不恢复 普通函数
标准的「想 → 用工具 → 再想」 LangChain 的 create_agent，底下已经是 LangGraph
要审批、要分支、要并行汇总、要崩溃后续跑、要多个角色交接 自己画 LangGraph

自己画，通常是因为标准循环包不住规矩：哪些步必须是代码，哪一步必须等人，哪些查询应该同时发出去。

同一套调度上还有几件今天知道名字就够的事：

  • 流式输出。界面不用干等整段话。工具在查的时候，先显示「正在查物流」
  • Send。一个问题拆给多个分身并行查，再汇总。比如同时核对三家物流
  • 时间旅行。回到某一拍的快照，改一笔状态，从那里重跑。调试时很有用

上了也容易踩的坑

现象 原因
一点头就扣了两次款 扣款写在 interrupt() 前面，恢复时这一站重跑
助手突然失忆 对话那一栏用了默认覆盖，没有追加
试的时候能暂停，一部署就不能 用了内存存档，进程重启档案就没了
两个查询结果互相覆盖 同一拍写同一栏，没写 reducer
转圈把账单打满 没有结束条件，也没设 recursion_limit
标准工具循环却从零画了一张图 create_agent 已经是这张图。规矩不够再拆开

图负责纪律，不负责正确

物流接口返回了过期数据，图会很忠实地写进病历，再让模型基于它往下说。拦这种错，靠的是工具结果要不要校验、金额门槛要不要人盖章、回答要不要带出处。

LangGraph 把这些纪律放进看得见的步骤里。判断对不对，仍然是你的提示词、你的工具、还有那个盖章的人一起扛。

官方文档从 Graph API 和 概览 看起就够。interrupt 的官方演示在这段视频。