LangGraph：构建有状态智能体的图形框架

LangGraph 解决的问题和 LangChain 不同。LangChain 回答「如何调用 LLM」，LangGraph 回答的是另一个问题：把多步 Agent 执行改造成可观测、可恢复、可干预的状态机。从 demo 走到生产，这一步往往比换一个更强的模型更关键——GitHub 上 41K+ Stars 的关注（截至 2026 年 9 月）也主要来自这里。

一个典型的 Agent 失败场景：用户问「帮我订下周去上海的机票并通知同事」，Agent 调了 5 个工具，第 6 步调用邮件 API 时网络抖动。在传统链式实现里，进程一重启，前 5 步的中间结果全部丢失，用户只能从头再来。生产环境里这种体验直接等于流失。LangGraph 把链式执行拆成节点和边，每个节点执行完都把状态写进 Checkpoint（状态快照），下一次恢复时从最近的成功节点继续——Agent 从「一次性脚本」变成了「可断点续传的状态机」。

学习目标

读完后你应该能够：

• 说清 LangGraph 与 LangChain AgentExecutor 在控制流、状态、Human-in-the-Loop（HITL）三个维度上的本质差异
• 独立画出 StateGraph、Node、Edge、State、Reducer 五个抽象的依赖关系，并解释 Reducer 在多写者场景下的作用
• 用 PostgresSaver + thread_id 实现一条断点续传流程，并说明节点幂等性为何是隐含契约
• 用 interrupt() 和 Command(resume) 实现「批准继续 / 修改后继续 / 拒绝终止」三条人工决策路径，并说清恢复时节点从头重跑的后果
• 给出一个不该用 LangGraph 的场景，并说明替代方案

目录

• 全景地图：五个抽象与三项能力
• 为什么 Agent 需要图模型而不是线性链
• 五个抽象的工程用法
• Durable Execution：从一次性脚本到可断点续传
• Human-in-the-Loop：生产门槛而非可选功能
• Memory：Working Memory 与 Persistent Memory 的边界
• 和 LangChain Agent / CrewAI / AutoGen 的工程取舍
• 适用边界：什么时候不该用 LangGraph
• 三个真实场景的架构形态
• 调试与部署
• 常见问题与排查
• 自测题
• 从哪里开始落地
• 相关资源

全景地图：五个抽象与三项能力

LangGraph 的 API 表面不大，但抽象层次需要先理清。五个核心抽象构成图的骨架，三项关键能力构建在骨架之上。

三项关键能力都建立在 Checkpoint 之上。这里先分清两个名字：Checkpoint 是每次写入的那份状态快照，Checkpointer 是负责读写快照的存储组件（InMemorySaver、PostgresSaver 都是 Checkpointer 的实现）。

• Durable Execution：每个节点执行后自动持久化状态，故障后从检查点恢复
• Human-in-the-Loop：通过 interrupt() 在节点内暂停执行，等人工输入后恢复
• Comprehensive Memory：Working Memory 由 Checkpoint 管理，Persistent Memory 接 Store 存储

为什么 Agent 需要图模型而不是线性链

LangChain 的 AgentExecutor 是一条链：模型决定下一步 → 执行工具 → 把结果塞回 prompt → 再问模型。这条链在 demo 阶段够用，但生产环境会撞上三堵墙。

第一堵墙在控制流。链式执行里，「先查数据库还是先调外部 API」「失败后是重试还是降级」这些决策全部塞在 prompt 里，由模型决定。出了问题，你拿到的只有一段对话历史，看不到执行路径。LangGraph 把这些决策外化成 Edge，每条边的选择都被记录在 Checkpoint 里，调试时能精确看到「第 3 步为什么走了 search 节点而不是 recall 节点」。

第二堵墙在状态层。链式执行的状态留在内存里，进程崩了就没了。LangGraph 的每个节点返回一个 State delta，框架合并进全量 State 后写入 Checkpointer。PostgresSaver 把状态写进数据库，进程重启后用同一个 thread_id 调用 invoke(None, config) 就能从断点继续。

第三堵墙在人工介入。链式执行一旦启动就跑到底，中间想暂停等人工确认，得自己造一套暂停-恢复机制。LangGraph 的 interrupt() 是框架级原语：节点在需要人工决策的位置调用它，框架把当前状态写进 Checkpoint，审核完成后调 Command(resume) 注入人工决定，interrupt() 带着这份决定恢复返回，节点继续往下走——图的其它节点感知不到暂停发生过。

三堵墙的根子相同：链式模型把执行、状态、控制流耦合在一起，图模型把它们拆开。Node 管计算，Edge 管路由，State 管数据，Checkpoint 管持久化——每层出问题都能独立定位，不用在对话历史里猜执行路径。

五个抽象的工程用法

StateGraph：图的容器，也是编译器入口

StateGraph 看起来像配置对象，但它的真正角色是编译器入口。compile() 之后返回的 app 是一个不可变的可执行图，所有节点、边、Reducer 在编译时已经固定，运行时不能再修改图结构。这个限制是为了保证同一次编译产出的 app 在多线程环境下行为一致——运行时改图会引入竞态条件，框架直接禁掉。

from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import InMemorySaver
from typing import TypedDict, Annotated

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # Reducer 指定追加而非覆盖
    current_step: str
    context: dict

def chat_node(state: AgentState) -> dict:
    # 对话节点，这里示意为直接记录状态，实际会调用模型
    return {"current_step": "chat_done"}

def search_node(state: AgentState) -> dict:
    # 检索节点，示意为写入检索结果，实际会调用工具
    return {"current_step": "search_done", "context": {"hits": []}}

def router_fn(state: AgentState) -> str:
    # 条件边路由：最后一条消息若带工具调用则去 search，否则结束
    last = state["messages"][-1]
    return "search" if getattr(last, "tool_calls", None) else "end"

graph = StateGraph(AgentState)
graph.add_node("chat", chat_node)
graph.add_node("search", search_node)
graph.add_edge(START, "chat")
graph.add_conditional_edges("chat", router_fn, {"search": "search", "end": END})
# 开发环境先挂内存版 Checkpointer，生产环境换 Postgres（见下文 Checkpoint 小节）
app = graph.compile(checkpointer=InMemorySaver())

Annotated[list, add_messages] 这一行容易被忽略，但它是 State 设计的关键。默认情况下，节点返回的字段会覆盖原 State；指定 Reducer 后，多个节点写同一字段时按策略合并。add_messages 会按 message id 去重追加，避免每次节点返回都把整个消息列表重写一遍。

Reducer 在多 Agent 场景下尤其重要。假设 Supervisor 同时调度 researcher 和 coder 两个子 Agent，两者都往 messages 字段写结果，没有 Reducer 时后写的会覆盖先写的；指定 add_messages 后，两条结果都会保留，调用方能看到完整的协作轨迹。

Node：纯函数，不是方法

Node 的契约是 f(state) -> state_delta。它接收完整 State，返回需要更新的字段子集。不返回的字段保持不变。这个限制带来一个直接好处：每个节点的输出自包含，重放时不依赖隐式状态，Checkpoint 恢复才走得通。

def search_node(state: AgentState) -> dict:
    query = state["messages"][-1].content
    results = search_tool.invoke(query)
    # 只返回需要更新的字段，messages 由 add_messages Reducer 追加
    return {"messages": [ToolMessage(content=str(results))], "current_step": "search_done"}

Node 内部不应该有跨调用的可变状态。如果需要计数器、缓存这类东西，写进 State 让 Checkpoint 管理。把状态藏在闭包或全局变量里，故障恢复时这些状态会丢失，行为不可复现。

Edge：路由的两种形态

固定边 add_edge("chat", "search") 表示 chat 完成后必然到 search，用于线性流程。条件边 add_conditional_edges("router", fn, mapping) 用于分支：fn 接收 State 返回字符串，mapping 把字符串映射到目标节点。

def router(state: AgentState) -> str:
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        return "tools"
    return "end"

graph.add_conditional_edges(
    "agent",
    router,
    {"tools": "tools_node", "end": END}
)

条件边的 mapping 参数可以省略，省略时 fn 返回的字符串直接当节点名。但显式写出 mapping 是更稳妥的做法——它把所有可能的路由路径在编译时暴露出来，配合 compile() 的图校验能提前发现「漏了一个分支」这类错误。线上出过这样的 bug：模型偶尔返回一个未在 mapping 里的字符串，框架抛 KeyError，整个会话中断。显式 mapping 让这种错误在编译期就被拦住。

Checkpoint：持久化的最小单位

每次节点执行后，框架把当前完整 State 写进 Checkpointer，附带执行到哪个节点、走了哪条边。恢复时按 thread_id 找到最近的 Checkpoint，从下一个节点继续。

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.postgres import PostgresSaver
from langchain_core.messages import HumanMessage

# 开发环境用内存，重启即丢
checkpointer = InMemorySaver()

# 生产环境用 Postgres，跨进程跨重启
# from_conn_string 返回 context manager，退出时释放连接
with PostgresSaver.from_conn_string("postgresql://user:pass@host/db") as saver:
    saver.setup()  # 首次运行时建表，之后可跳过
    app = graph.compile(checkpointer=saver)
    config = {"configurable": {"thread_id": "user_123_session_456"}}
    result = app.invoke({"messages": [HumanMessage(content="你好")]}, config=config)

两个容易踩的细节：一是 from_conn_string 必须按 context manager 使用，直接赋值拿到的对象没建立连接；二是 setup() 负责建表，首次部署时必须执行一次。常驻服务里不适合用 with 块——更常见的做法是用连接池构造 Checkpointer，让它和应用同生命周期（PostgresSaver(pool)），setup() 在应用启动时调用。异步服务对应 AsyncPostgresSaver，用法见官方文档。

thread_id 是会话维度的标识。同一个用户的多次请求用同一个 thread_id，Agent 自动延续上下文；不同用户用不同 thread_id，状态互相隔离。这种设计让多租户场景天然支持，不需要自己在业务层做状态分桶。

Checkpointer 的选型：官方维护 InMemory、SQLite、Postgres、Redis 四种实现，社区还有 MongoDB、DynamoDB、Cassandra。单机开发用 InMemory，单机持久化用 SQLite，多实例生产用 Postgres——后两者支持跨进程恢复，InMemory 只在进程内有效。

另一个容易忽略的细节是 Checkpoint 的写入时机。调用图时可以用 durability 参数控制落盘强度：sync（默认）在下一步开始前把状态写满，持久性最好但每个节点都要等 IO 完成；async 边执行边异步落盘，性能更好，但进程在写盘前崩溃可能丢掉最近一次 Checkpoint；exit 只在整次执行退出时落盘，性能最好，中途崩溃则无法断点续传。对延迟敏感的场景，先评估把 durability 调成 async 或 exit，比换更快的后端来得直接；如果还不行，再把多个轻量节点合并成一个，或改用吞吐更高的后端（Redis 比 Postgres 快，但持久性保证弱一些）。

Durable Execution：从一次性脚本到可断点续传

Durable Execution 是 Checkpoint 机制的必然结果，不是 LangGraph 的某个开关。每个节点执行完都持久化，故障后从最近的成功节点继续，这就是「durable」的全部含义。

一条完整的 Checkpoint 恢复流程

假设有一个客服 Agent，流程是：分类用户问题 → 查询订单 → 查询物流 → 生成回复。用户问「我的订单 #12345 物流到哪了」，Agent 执行到「查询物流」时数据库连接超时。

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool
from typing import TypedDict

class ServiceState(TypedDict):
    messages: list
    order_info: dict
    logistics_info: dict
    reply: str

def classify_node(state): ...
def query_order_node(state): ...
def query_logistics_node(state):
    # 这里抛出数据库超时异常
    raise ConnectionError("DB timeout")
def generate_reply_node(state): ...

graph = StateGraph(ServiceState)
graph.add_node("classify", classify_node)
graph.add_node("query_order", query_order_node)
graph.add_node("query_logistics", query_logistics_node)
graph.add_node("reply", generate_reply_node)
graph.add_edge(START, "classify")
graph.add_edge("classify", "query_order")
graph.add_edge("query_order", "query_logistics")
graph.add_edge("query_logistics", "reply")
graph.add_edge("reply", END)

# 常驻服务用连接池构造，Checkpointer 与应用同生命周期
pool = ConnectionPool("postgresql://...")
checkpointer = PostgresSaver(pool)
checkpointer.setup()  # 应用启动时调用一次，之后跳过
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "user_abc_ticket_789"}}

执行过程：

1. classify 执行成功，State 写入 Checkpoint：{messages: [...], order_info: {}, logistics_info: {}, reply: ""}，当前节点 classify，下一节点 query_order
2. query_order 执行成功，State 更新为 {..., order_info: {id: "12345", status: "shipped"}, ...}，写入 Checkpoint
3. query_logistics 抛出 ConnectionError，异常向上抛，但 Checkpoint 仍停留在第 2 步的状态
4. 业务层捕获异常，记录日志，向用户返回「正在处理中」的临时响应
5. 几分钟后数据库恢复，业务层用同一个 thread_id 再次调用：

# input=None 表示从 Checkpoint 继续，不重新输入
result = app.invoke(None, config=config)

6. 框架从 Checkpoint 读取状态，发现执行到 query_logistics 的入口，重新执行这个节点
7. 这次数据库正常，query_logistics 成功，reply 节点继续执行，最终返回完整回复

整个恢复过程对业务代码透明。业务层要做的只有两件事：捕获异常时记录 thread_id，恢复时用同一个 thread_id 调 invoke(None)。中间状态由 Checkpoint 管，从哪一步继续由框架判断。

幂等性是 Durable Execution 的隐含契约

Checkpoint 恢复会重新执行失败的那个节点。如果这个节点有副作用（发邮件、扣款、写数据库），重试时可能产生重复操作。Node 设计必须满足幂等性：

• 写数据库：用唯一键 upsert，不用 insert
• 调外部 API：传幂等键（idempotency key），让对端去重
• 发消息：先查是否已发，再决定是否重发

这类问题往往在第一次故障恢复时才暴露。开发环境一切正常，因为没触发过重试；生产环境第一次数据库抖动，同一个邮件发了两次，同一个订单扣了两次款。这是 LangGraph 新手付出过最多学费的坑。

Human-in-the-Loop：生产门槛而非可选功能

很多团队把 Human-in-the-Loop（HITL）当成「高级功能」，觉得 demo 阶段不需要。到了生产环境，这个判断会反过来：HITL 在金融、医疗、法律、HR 这些领域属于合规和安全的硬性要求，没它过不了审查，谈不上上线。

考虑一个财务 Agent，自动审批员工报销。如果完全自动化，Agent 误判一笔 5 万元的报销直接打款，损失由谁承担？监管和内部风控都要求关键决策必须有人工签字。

interrupt()：在任意位置挂起

LangGraph 的 HITL 原语是 interrupt() 函数。注意，早期版本的 NodeInterrupt 异常已被官方弃用，现行方案一律用 interrupt()。节点在需要人工决策的位置调用它，执行在调用点挂起，当前状态连同挂起位置一起写入 Checkpoint，审核负载交回调用方，然后无限期等待。恢复时，传入的 resume 值会成为 interrupt() 的返回值。

from langgraph.types import interrupt
from typing import TypedDict

class FinanceState(TypedDict):
    messages: list
    amount: float
    recipient: str
    email_content: str   # 上游 draft 节点生成的邮件草稿
    status: str

def send_email_node(state: FinanceState) -> dict:
    draft = state["email_content"]
    if state["amount"] > 10000:
        # 执行在这里挂起，审核负载会出现在返回值的 __interrupt__ 里
        decision = interrupt({
            "question": "金额超过 1 万元，需要人工确认",
            "recipient": state["recipient"],
            "amount": state["amount"],
            "draft": draft,
        })
        # 恢复后 decision 就是审核员传入的 resume 值（注意节点是从头重跑的，见下方红线）
        if not decision["approved"]:
            return {"status": "rejected_by_reviewer"}
        draft = decision["email_content"]  # 审核员可能改过内容
    send_email(state["recipient"], draft)
    return {"status": "sent"}

send_email 是你自己的业务函数，示例里只关心它在哪个时机被调用。调用方如何知道发生了中断？invoke() 的返回值里带 __interrupt__ 字段：

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

result = app.invoke(input, config=config)
interrupts = result.get("__interrupt__")
if interrupts:
    # 把 thread_id 和审核负载推给人工审核队列
    enqueue_for_review(
        thread_id="user_abc_ticket_789",
        payload=interrupts[0].value,
    )

上面用的是 invoke() 同步接口。如果调用方要做逐 token 流式展示，改用事件流式接口 stream_events(..., version="v3")：中断负载在 stream.interrupts 里，stream.interrupted 表示这次执行是否因人工介入暂停，跑完的最终状态在 stream.output。对不需要流式投影的场景，invoke() 的 __interrupt__ 字段够用，两种方式恢复动作完全一样。

审核员在后台系统看到这条待办，决定批准、修改还是拒绝。恢复一律通过 Command(resume=...)：

from langgraph.types import Command

# 路径一：批准继续。resume 值原样回传草稿，成为 interrupt() 的返回值
app.invoke(
    Command(resume={"approved": True, "email_content": payload["draft"]}),
    config=config,
)

# 路径二：修改后继续。审核员的修改放进 resume 载荷
app.invoke(
    Command(resume={"approved": True, "email_content": "审核员修改后的内容"}),
    config=config,
)

# 路径三：拒绝终止。resume 值带拒绝标记，节点内写入状态后收尾
app.invoke(
    Command(resume={"approved": False}),
    config=config,
)

三条路径复用同一份 Checkpoint：业务代码要做的只是把审核负载放进 interrupt()，把人工决定放进 Command(resume)，暂停期间的状态保存由框架完成，恢复时节点从头重跑，具体语义见下文红线。审核拖几个小时甚至几天都没关系——Checkpoint 会一直停在原地。如果只是想人工修正历史状态（比如回填一笔数据）而不是恢复中断，update_state 仍然可用，它的定位是修正和调试，不是恢复中断的标准入口。

三条使用红线

interrupt() 的恢复语义里藏着三个坑，官方文档明确列出，这里提前讲：

恢复时节点从头重跑。恢复不是从挂起行继续，而是把整个节点重新执行一遍，interrupt() 之前的代码会再次运行。上面的例子挂起前只读了 State，重复执行无害；如果挂起前调过风控 API，恢复后这笔调用会发生两次。把副作用放在 interrupt() 之后，或者做成幂等。

不要把 interrupt() 包在裸 try/except 里。它靠抛出特殊异常实现挂起，被 except 吞掉等于把暂停机制整个废掉。

不要在节点里条件性跳过 interrupt()。恢复值按调用顺序严格匹配，跳过一次调用会让恢复值错位到下一次 interrupt() 上，产生难以排查的错乱。

Memory：Working Memory 与 Persistent Memory 的边界

LangGraph 的 Memory 模型容易混淆，因为「记忆」这个词在 LLM 语境下被用得太泛。LangGraph 把 Memory 明确分成两层，两层有不同的生命周期和存储后端。

Working Memory 是单次会话内的状态，由 Checkpoint 自动管理。一个 thread_id 对应一份 Working Memory，会话结束（用户离开）后是否保留取决于 Checkpointer 配置——InMemorySaver 进程退出就丢，PostgresSaver 永久保留。Working Memory 里放的是当前对话的消息历史、中间工具调用结果、当前执行到哪一步。

Persistent Memory 是跨会话的长期记忆，由 LangGraph 的 Store 抽象承载。Store 是一个按 namespace 组织的键值存储：namespace 用元组分层（比如 ("users", user_id, "preferences")），同一 namespace 下按 key 读写。编译时把 store 传给 compile()，节点声明 store 参数就能访问。需要语义检索时，给 Store 配置 embedding 索引，search() 就按相似度返回——向量检索依赖外部的 embedding 模型和向量后端（生产环境 PostgresStore 底下通常是 pgvector），LangGraph 提供的是统一的读写接口。

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
# 写入：namespace 按用户分层隔离，跨会话可查
store.put(("users", "u_123", "preferences"), "contact_style", {"style": "正式"})

def recall_node(state: AgentState, *, store) -> dict:
    """会话开始时检索该用户的长期记忆，填进上下文"""
    items = store.search(("users", "u_123", "preferences"))
    context = "\n".join(f"{item.key}: {item.value}" for item in items)
    return {"context": {"preferences": context}}

def persist_node(state: AgentState, *, store) -> dict:
    """会话结束时把关键决策写回长期记忆"""
    store.put(
        ("users", "u_123", "preferences"),
        "last_decision",
        {"decision": state["messages"][-1].content},
    )
    return {}

# 编译时同时传入两层存储
app = graph.compile(checkpointer=checkpointer, store=store)

要语义检索时，把 InMemoryStore 换成配置了索引的 Store：

from langchain_openai import OpenAIEmbeddings
from langgraph.store.memory import InMemoryStore

store = InMemoryStore(
    index={"embed": OpenAIEmbeddings(model="text-embedding-3-small"), "fields": ["text"]}
)
# search 按语义相似度排序，而不是只按 key 精确匹配
hits = store.search(("users", "u_123"), query="这位用户偏好的沟通风格", limit=3)

开发环境用 InMemoryStore，生产环境换 PostgresStore（同样需要 setup() 建表）。另外提醒一句：langchain.memory 里的 VectorStoreRetrieverMemory 这类旧方案已废弃，网上还能搜到大量旧教程，新项目一律用 Store。

两层的边界要划清楚：Working Memory 放「这次对话需要的东西」，Persistent Memory 放「下次对话可能需要的东西」。把所有历史都塞进 Working Memory 会导致 Context 爆炸；把当前会话的中间结果写进 Persistent Memory 会导致检索噪声。框架不会替你做这个判断，需要根据业务场景设计 State 结构和持久化策略。

和 LangChain Agent / CrewAI / AutoGen 的工程取舍

Agent 框架不止 LangGraph 一个，选型时需要清楚每个框架的定位差异。这里给出的是生产环境真实使用中的取舍，不是 feature 对比表。

LangChain Agent（AgentExecutor）是高层抽象，开箱即用。适合 demo、原型、简单场景：单步工具调用、不需要持久化、不需要 HITL。它的局限是控制流不透明、状态不可恢复、人工介入难插入。注意 AgentExecutor 已被官方标记废弃，LangChain 现在推荐的 Agent 构建 API（LangChain 1.0 的 create_agent，前身是 create_react_agent）就直接构建在 LangGraph 之上——撞上上面那几条局限时，官方指出的迁移方向就是 LangGraph。

CrewAI 强调多 Agent 角色协作，每个 Agent 有 role、goal、backstory，通过任务分配和角色对话完成复杂工作。适合内容生成、创意协作这类「多个角色一起讨论」的场景。它的局限是状态管理和故障恢复不如 LangGraph 细粒度——CrewAI 的抽象层次更高，调试时不容易看到具体哪一步出了问题。核心需求是「精细控制单 Agent 的执行流程」时，LangGraph 更合适；核心需求是「多个角色协作产出内容」时，CrewAI 更顺手。

AutoGen（Microsoft）主打多 Agent 对话，Agent 之间互相发消息完成工作。适合研究探索、对话式协作场景。它的对话模型灵活但缺乏图形化的控制流约束，复杂流程下容易出现「Agent 之间聊偏了」的情况。LangGraph 的图模型对控制流有强约束，不容易跑偏，但灵活性低于 AutoGen 的自由对话。

选型判断：先用 LangChain Agent 跑通 demo；当遇到状态丢失、控制流不可见、需要 HITL 这三个问题之一时，迁移到 LangGraph；如果场景天然是「多角色讨论」，考虑 CrewAI 或 AutoGen，但生产化时仍可能需要 LangGraph 做底座。

适用边界：什么时候不该用 LangGraph

下面这些场景用 LangGraph 属于过度设计。

单步工具调用。用户问一句话、调一个工具、返回结果，这种场景用 LangChain 的 chain 或直接调 LLM API 就够了。引入 StateGraph 反而增加心智负担。

纯流式对话。没有工具调用、没有多步推理、就是聊天，用 OpenAI SDK 的 streaming 接口更直接。LangGraph 的图模型对这种场景没有增值。

强确定性流程。如果流程是固定的「步骤 A → 步骤 B → 步骤 C」，没有条件分支、没有 LLM 决策，用普通的工作流引擎（Airflow、Temporal）更合适。LangGraph 的条件边是为「LLM 参与路由」设计的，纯确定性流程用不上。

极低延迟场景。Checkpoint 持久化有 IO 开销，每个节点都要写一次存储。如果要求毫秒级响应，Checkpoint 会成为瓶颈。可以禁用 Checkpointer（compile() 时不传），但这样就失去了 Durable Execution——回到链式执行的困境。

反过来，下面这些场景 LangGraph 是当前最成熟的选择：

• 多步工具调用，中间步骤可能失败需要重试
• 需要 HITL 的合规场景（金融、医疗、HR）
• 长对话需要跨会话恢复上下文
• 多 Agent 协作需要精细控制流转
• 需要可观测的执行路径用于调试和审计

三个真实场景的架构形态

每个场景对应一种常见的图结构：单图多节点 + HITL、ReAct 循环、Supervisor 多 Agent 协作。

客服 Agent：单图多节点 + HITL

电商客服 Agent 的典型流程：分类 → 查询订单 → 查询物流 → 生成回复。其中「退款」「投诉」类问题需要人工介入。

class ServiceState(TypedDict):
    messages: list
    intent: str          # classify 节点写入
    order_info: dict
    logistics_info: dict
    reply: str

graph = StateGraph(ServiceState)
graph.add_node("classify", classify_node)
graph.add_node("query_order", query_order_node)
graph.add_node("query_logistics", query_logistics_node)
graph.add_node("escalate", escalate_node)  # 升级到人工
graph.add_node("reply", generate_reply_node)

graph.add_edge(START, "classify")
graph.add_conditional_edges(
    "classify",
    lambda s: "escalate" if s["intent"] == "complaint" else "query_order",
    {"escalate": "escalate", "query_order": "query_order"}
)
graph.add_edge("query_order", "query_logistics")
graph.add_edge("query_logistics", "reply")
graph.add_edge("escalate", "reply")
graph.add_edge("reply", END)

escalate 节点内部调用 interrupt()，等待人工接管。人工处理后用 Command(resume) 把处理结果传回来，reply 节点基于这个结果生成最终回复。用户只看到「正在为您处理」然后「已解决」，感知不到中间的暂停和恢复。

代码生成 Agent：ReAct 循环 + 工具节点

代码生成 Agent 的核心是 ReAct 循环：模型决定调用什么工具 → 执行工具 → 把结果塞回上下文 → 再问模型。LangGraph 用条件边实现这个循环。

def agent_node(state):
    response = llm.bind_tools(tools).invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state):
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        return "tools"
    return "end"

graph = StateGraph(AgentState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_executor_node)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
graph.add_edge("tools", "agent")  # 工具执行完回到 agent，形成循环

tools 节点执行完回到 agent，形成循环。循环退出条件由 should_continue 判断：模型不再请求工具调用时走 end。这种模式是 LangGraph 官方推荐的 ReAct 实现方式，比 LangChain 的 AgentExecutor 更容易调试——每个循环迭代都有独立的 Checkpoint，可以精确看到第几次迭代出了问题。

多 Agent 协作：Supervisor 模式

多个 Agent 协作时，常见模式是 Supervisor：一个调度 Agent 决定把任务分给哪个子 Agent，子 Agent 完成后把结果交回 Supervisor。

from typing import Literal

class TeamState(TypedDict):
    messages: Annotated[list, add_messages]
    next: str

def supervisor(state):
    """决定下一步交给哪个子 Agent"""
    response = llm.invoke([
        SystemMessage(content="你是任务调度器。根据当前状态决定下一步交给谁：researcher / coder / end"),
        *state["messages"]
    ])
    return {"next": response.content.strip()}

graph = StateGraph(TeamState)
graph.add_node("supervisor", supervisor)
graph.add_node("researcher", researcher_agent)
graph.add_node("coder", coder_agent)
graph.add_edge(START, "supervisor")
graph.add_conditional_edges(
    "supervisor",
    lambda s: s["next"],
    {"researcher": "researcher", "coder": "coder", "end": END}
)
graph.add_edge("researcher", "supervisor")  # 子 Agent 完成后回到 Supervisor
graph.add_edge("coder", "supervisor")

Supervisor 模式的优势是控制流集中：所有路由决策都经过 Supervisor，调试时看 Supervisor 的输出就能理解流程走向。劣势也在这里——每个子任务都要经过它，延迟会累积。对延迟敏感的场景可以考虑 Swarm 模式（Agent 之间直接传递控制权），但 Swarm 的控制流更难追踪，调试成本会上升。

调试与部署

LangSmith：可观测性的标配

LangGraph 的执行路径天然适合可视化，但框架本身不提供 UI。LangSmith 是配套的可观测性平台，启用后每次 invoke 的完整轨迹都会被记录：每个节点的输入输出、Edge 的选择、Checkpoint 的写入、耗时分布。

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "lsv2_your-api-key"
os.environ["LANGSMITH_PROJECT"] = "my-agent-prod"

# 启用后所有 invoke 自动被追踪
app.invoke(input, config=config)

环境变量已从旧的 LANGCHAIN_TRACING_V2 系列更名为 LANGSMITH_* 系列，旧名在部分版本里还能识别，但新项目统一用新名。生产环境强烈建议开启：线上 Agent 出问题时，没有 LangSmith 你只能看日志猜；有 LangSmith 可以直接看到「第 3 步的 LLM 调用花了 8 秒，返回的 tool_calls 字段格式不对，导致第 4 步解析失败」这种级别的细节。可观测性贯穿优化全过程：执行路径看不到，性能调优和 bug 定位都缺依据。

LangGraph Platform：部署的三种形态

# 本地开发服务器，热重载
langgraph dev

# 本地用 Docker 起完整的 Agent Server，与生产环境同构
langgraph up

# 构建生产镜像，推到任意容器平台自行部署
langgraph build

上 Cloud 走 LangGraph Platform 的 GitHub 集成，绑定仓库后自动构建部署；Self-hosted 需要自己跑 LangGraph Server（依赖 Postgres 和 Redis），适合金融、医疗等不能把数据传出内网的场景。两种形态的 API 接口一致，迁移成本主要在数据层。

常见问题与排查

Q1：LangGraph 和 LangChain Agent 有什么区别？

LangChain Agent（AgentExecutor）是高层封装，开箱即用，适合 demo 和简单场景；LangGraph 是底层框架，把控制流、状态、人工介入交给你显式定义。AgentExecutor 已废弃，官方现在基于 LangGraph 提供 create_agent——当封装好的链式 Agent 撞上状态丢失、控制流不可见、人工介入难插入这些问题时，迁移方向就是 LangGraph。

Q2：什么时候该用 LangGraph？

满足以下任一条件就该考虑：需要持久化状态跨进程恢复、需要 HITL、需要可观测的执行路径、多 Agent 协作需要精细控制流转。如果只是单步工具调用或纯聊天，用 LangChain 或直接调 API 更合适——LangGraph 的图模型在这种场景下是过度设计。

Q3：Checkpoint 恢复时节点重试，副作用怎么处理？

Node 必须设计成幂等。写数据库用 upsert，调外部 API 传幂等键，发消息前先查是否已发。不满足幂等性的节点在重试时会出重复操作，而且问题往往在第一次故障恢复时才暴露。HITL 场景同理：interrupt() 之前的副作用在恢复时会重复执行。

Q4：支持哪些 Checkpointer？

官方维护 InMemory、SQLite、Postgres、Redis 四种，社区还有 MongoDB、DynamoDB、Cassandra。选型：单机开发用 InMemory，单机持久化用 SQLite，多实例生产用 Postgres。

Q5：能用于生产环境吗？

可以。据 LangGraph 官方公布的案例，Klarna（电商客服）、Uber（代码迁移与单测生成）、LinkedIn（AI 招聘）、Elastic（威胁检测）等公司在生产环境使用。LangGraph Platform 提供企业级部署支持。

Q6：有 JavaScript 版本吗？

有，见 LangGraph.js。API 与 Python 版本基本对齐，但生态和社区资源不如 Python 版本丰富。

Q7：invoke(None) 和 invoke(input) 有什么区别？

invoke(input) 是新会话或追加输入，框架从图的入口开始执行。invoke(None) 是从 Checkpoint 恢复，框架读取 thread_id 对应的最新 Checkpoint，从下一个节点继续。None 不是「没有输入」的意思，是「不提供新输入，从断点继续」的信号。

Q8：节点抛出异常后状态会丢吗？

不会。Checkpoint 在节点执行成功后才写入。节点抛异常时，Checkpoint 仍停留在上一个成功节点的状态。异常向上抛给调用方，调用方处理完异常后用 invoke(None) 恢复，会重新执行失败的那个节点。

Q9：Checkpoint 写入失败怎么办？

Checkpointer 抛出的异常会向上传播给调用方。生产环境需要监控 Checkpoint 写入的成功率，写入失败意味着这次节点的状态没落盘，后续 invoke(None) 恢复时会回退到上一个成功节点重跑。PostgresSaver 出现连接超时时，重试策略要放在业务层而不是框架层——框架不会自动重试 Checkpoint 写入。

Q10：同一个 thread_id 并发调用 invoke 会怎样？

LangGraph 默认对同一个 thread_id 加锁，保证状态写入的顺序一致性。并发调用会被串行化，后到的请求等待前一个完成。如果业务层需要并行处理同一用户的多个请求，要么用不同的 thread_id，要么在业务层做请求合并。

Q11：interrupt() 挂起后怎么恢复？和 invoke(None) 有什么不同？

interrupt() 挂起后用 Command(resume=...) 恢复，resume 值成为节点内 interrupt() 的返回值，节点从头重跑，把人工决策带回执行流。invoke(None) 用于失败重跑：不注入新输入，框架直接从 Checkpoint 里的下一个节点继续。一句话区分——人工决策用 Command(resume)，机器故障续跑用 invoke(None)。

自测题

下面这些问题用来检验你是否真的理解了上面的内容。建议先自己想答案，再回头看正文对照。

概念题

1. LangGraph 把链式执行拆成 Node、Edge、State、Checkpoint 四个独立维度。请说明「链式模型把哪三件事耦合在一起」，以及这种耦合在生产环境会撞上哪三堵墙。
2. Annotated[list, add_messages] 中的 add_messages 起什么作用？如果不指定 Reducer，Supervisor 同时调度 researcher 和 coder 写 messages 字段会发生什么？
3. invoke(None) 中的 None 表示什么？为什么不能用空字典 {} 代替？
4. interrupt() 挂起后，Command(resume) 传入的值去了哪里？恢复时节点从哪里开始重新执行？

场景题

5. 一个节点 send_email_node 内部调用了邮件 API，没有做幂等。生产环境数据库抖动一次后，用户收到两封相同邮件。请说明 Checkpoint 恢复流程中哪一步导致了重复发送，并给出修复方案。
6. 财务审批 Agent 中，审核员看了邮件草稿后决定修改收件人再发送。请写出对应的恢复代码（用 Command(resume)），并说明 resume 值在节点内如何被消费。
7. send_email_node 在 interrupt() 之前先调用了一次风控 API。审核员处理完恢复后，这次风控调用会发生什么？两种改法是什么？
8. 客服 Agent 的 escalate 节点调用 interrupt() 后，用户在前端一直看到「正在处理」。审核员处理完 30 分钟才恢复。这 30 分钟里 Checkpoint 状态有没有变化？reply 节点是否被执行？

选型题

9. 下面四个场景，哪些该用 LangGraph，哪些不该用？说明理由：
   • 用户输入一句话，调用一次天气 API 返回结果
   • 客服 Agent 调用 5 个工具，第 3 步可能失败需要重试
   • 固定流程「下载文件 → 解析 → 入库」，无 LLM 决策
   • 多个 Agent 角色讨论生成一份营销文案

进阶路径

如果你能答对上面 9 题中的 6 题以上，下一步可以深入：

• 阅读 LangGraph 官方文档 的 Persistence 和 Human-in-the-Loop 两节，对照本文的 Checkpoint 流程和 interrupt() 红线看官方实现细节
• 在 LangChain Academy 跑一遍 Intro to LangGraph 课程，重点做其中的 HITL 实验
• 用 PostgresSaver 在本地起一个 Postgres，把本文的客服 Agent 完整跑通，手动 kill 进程后验证 invoke(None) 能否恢复
• 把财务审批 Agent 跑通后，故意在 interrupt() 之前放一个写文件的副作用，用 Command(resume) 恢复，观察这个副作用执行了两次——亲手验证「节点从头重跑」的语义
• 研究 Swarm 模式与 Supervisor 模式的差异，思考什么场景下 Swarm 的延迟优势值得承担调试成本

从哪里开始落地

把一个 LLM 应用推上生产时，建议按以下顺序引入 LangGraph：

第一步：把现有链式 Agent 改造成 StateGraph。不改业务逻辑，只是把 chain.invoke() 拆成节点和边。改完之后配合 LangSmith 能看到完整执行路径，后续的性能调优和 bug 定位才有抓手。

第二步：接入 Checkpointer。先用 InMemorySaver 在开发环境验证，再切到 PostgresSaver。接入后进程重启不丢上下文，状态持久化这一关才算过——否则任何一次部署或重启都会让用户会话中断。

第三步：在关键节点加 HITL。识别出有合规风险或不可逆操作的节点，在决策点调用 interrupt()。这一步解决的是合规和安全性——在很多行业，没有 HITL 就没有上线资格。

第四步：优化 Reducer 和 State 设计。检查哪些字段需要 Reducer 合并、哪些字段应该排除在 Checkpoint 之外（比如大文件内容）。优化后性能和 Context 卫生都会改善，但属于精细化工作，可以等基础流程跑稳再做。

第五步：考虑多 Agent 协作。单 Agent 稳定运行后，再拆分出子 Agent 用 Supervisor 模式协作。不要一开始就上多 Agent——单 Agent 都没跑稳，多 Agent 的调试复杂度会指数级上升。

选框架阶段先把第一步和第二步走通。这两步的价值在生产环境第一次遇到故障时才会显出来——进程能从断点恢复，用户不用从头再来一次。

术语速查

相关资源

• GitHub：langchain-ai/langgraph
• 文档：docs.langchain.com/oss/python/langgraph
• API 参考：reference.langchain.com/python/langgraph
• 快速入门：docs.langchain.com/oss/python/langgraph/quickstart
• LangGraph.js：github.com/langchain-ai/langgraphjs
• LangChain Academy：academy.langchain.com