LangGraph 状态管理指南：流程编排、检查点与持久化｜云选文章

复杂 AI 应用的问题通常不在“能否调用模型”，而在流程能否被解释、恢复和维护。一次请求里只做提示词填充与模型调用时，普通函数或 LangChain 的模型、工具抽象已经够用；当任务开始出现条件分支、工具重试、人工审批和跨轮恢复，才需要把执行过程建模为状态图。

先分清两套工具的职责

LangChain 提供模型、提示词、工具和 Agent 等上层抽象，适合快速组装常见的大模型能力。LangGraph 则是偏底层的编排框架与运行时，重点处理长时间运行、有状态的工作流。两者可以组合，但并不存在“使用 LangGraph 就必须先构造 LangChain 链”的限制：LangGraph 节点可以调用 LangChain 组件，也可以直接执行普通 Python 函数或其他 SDK。

判断是否需要状态图，可以看流程是否具备以下特征：

下一步取决于模型输出、工具结果或业务规则；
某些节点失败后需要重试、降级或转人工；
执行可能暂停，稍后从原位置继续；
多个节点需要共享结构化上下文；
团队需要追踪每一步输入、输出及状态变化。

如果流程始终是“检索—生成—返回”的固定顺序，先用简单函数或链式组合通常更易维护。不要为了画图而引入图。

LangGraph 状态管理的三个核心概念

状态是工作流的数据契约

StateGraph 接收一个状态 Schema。每个节点读取当前状态，并返回本节点需要更新的字段，而不是随意修改一组全局变量。这样，业务输入、中间结果、错误信息和最终输出都有明确边界。

下面是一个精简的审核流程：

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

class ReviewState(TypedDict):
    question: str
    draft: str
    approved: bool
    error: str

def generate(state: ReviewState):
    # 实际项目中可在这里调用模型
    return {"draft": f"待审核答案：{state['question']}"}

def review(state: ReviewState):
    passed = len(state["draft"]) >= 10
    return {"approved": passed}

def revise(state: ReviewState):
    return {"draft": state["draft"] + "（已补充必要说明）"}

def route(state: ReviewState) -> Literal["finish", "revise"]:
    return "finish" if state["approved"] else "revise"

builder = StateGraph(ReviewState)
builder.add_node("generate", generate)
builder.add_node("review", review)
builder.add_node("revise", revise)
builder.add_edge(START, "generate")
builder.add_edge("generate", "review")
builder.add_conditional_edges(
    "review", route, {"finish": END, "revise": "revise"}
)
builder.add_edge("revise", "review")
graph = builder.compile()

节点应尽量保持单一职责：模型推理、检索、校验和数据写入分别处理。路由函数只负责根据状态选择路径，不要把耗时副作用藏在条件判断中。这样更便于测试，也能避免循环逻辑失控。

并发更新需要合并规则

当多个分支可能同时写入同一字段时，必须先定义更新语义：是覆盖旧值、追加列表，还是按业务键去重。LangGraph 可通过 reducer 指定字段的合并方式。若没有明确规则，并发分支可能造成状态覆盖或难以复现的结果。

状态也不宜塞入全部上下文。大型文档、二进制文件和可重复获取的数据更适合放在对象存储或数据库中，状态里保存标识、摘要和必要元数据即可。这样能控制检查点体积，并减少序列化开销。

检查点与长期记忆不是一回事

官方文档将两类持久化能力区分得很清楚：checkpointer 保存单个线程的图状态快照，适合会话连续性、故障恢复、人工介入和回看历史；store 保存跨线程的应用数据，适合用户偏好、事实或共享知识。

编译图时可以接入检查点，并在调用时提供稳定的 thread_id：

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "review-42"}}
result = graph.invoke(
    {"question": "如何设计可恢复的 Agent？", "draft": "",
     "approved": False, "error": ""},
    config,
)

InMemorySaver 只适合开发和测试，进程重启后数据会丢失。生产环境应选择持久化后端，并设计备份、保留周期、访问控制和清理策略。检查点长期不清理会增加存储占用，也可能拖慢状态读取。

从原型迁移到状态图的步骤

第一步是画出真实业务流程，而不是照着框架 API 拆节点。列出输入、确定性规则、模型决策、外部工具、人工操作和失败出口，再判断哪些步骤必须保留状态。

第二步是设计最小状态。通常可以分成四类：

请求数据：用户输入、租户与权限上下文；
过程数据：检索结果、工具返回和当前阶段；
控制数据：重试次数、审批状态、错误码；
输出数据：最终回答或结构化业务结果。

第三步是把不稳定边界隔离成节点。模型、网络 API 和数据库写入都可能失败，应分别设置超时、有限重试和可观察的错误状态。循环必须有退出条件，例如最大重试次数或人工接管路径，不能把“让模型再想一次”设计成无限回路。

第四步再接入持久化与人工审批。先验证状态转换正确，再增加恢复能力，调试成本通常更低。上线前至少测试正常路径、工具超时、模型输出不合规、重复调用与中途恢复五类场景。

选型时不要忽略运行成本

状态图提升了复杂流程的可控性，也会引入额外成本：节点越多，模型调用和网络往返越多；检查点越频繁，存储与序列化开销越大；保留完整提示词和工具结果，还会带来敏感数据治理问题。

适合采用 LangGraph 的，通常是需要长任务、动态路由、人工介入或故障恢复的 Agent。简单问答、单次内容生成和固定 ETL 流程未必需要它。更稳妥的选择顺序是：先用普通代码实现最小闭环，再在分支、循环和恢复需求明确后引入状态图，最后根据运行量选择持久化与观测方案。

状态管理的价值不是让架构显得复杂，而是让每次决策都有数据依据、每次中断都有恢复位置、每条失败路径都有明确出口。做到这三点，LangGraph 才真正从演示框架变成可维护的 AI 应用基础设施。

LangGraph 状态管理：从流程编排到持久化选型