LangGraph 检查点三次生产重写避坑经验

我的第一个 LangGraph Agent 悄悄丢弃了所有的任务，持续了数周。AgentState 中的 memory 字段定义为 list[str]，但我的 Agent 写入的是 list[dict]。LangGraph 的 SqliteSaver 没有抛出错误，只是截断了状态，留下一个空列表。直到客户抱怨他们的多步骤请求从未超过第一轮，我才发现了这个问题。修复方法是一个单行 schema 变更，但代价是数小时的调试和一个丢失的客户。这不是唯一的 checkpoint 陷阱。

SqliteSaver 的「宽容」：隐式的 Schema 不匹配

我最初的 LangGraph 设置使用 SqliteSaver 做 checkpointing。它简单、嵌入，似乎在早期生产环境足够稳健。Agent 的目的是处理来自 Telegram 的传入请求，将其分解，并管理多步骤对话。AgentState 大致如下：

class AgentState(TypedDict):
    chat_id: str
    thread_id: str
    user_input: str
    memory: list[str] # 罪魁祸首
    current_step: int
    # ... other fields

Agent 的 memory 字段本意是存储对话片段的历史。然而，我的 Agent 代码设计用于存储更结构化的数据，比如 {"role": "user", "content": "..."}。所以，它写入的是 list[dict] 而不是 list[str]。

SqliteSaver 使用 json.dumps 将状态序列化为 BLOB。当 json.dumps 遇到 list[dict] 而它根据初始推断或给出的 schema 期望的是 list[str] 时，它没有失败。相反，它将 list[dict] 静默地序列化为一个字符串。问题出在反序列化步骤。当 SqliteSaver 加载状态时，它尝试将那个字符串反序列化为 list[str]。由于 list[dict] 的字符串表示不是有效的 list[str]，它通常会导致空列表或格式错误的对象，从而有效地抹去了该特定字段的对话历史。

修复方法是在 AgentState 中将 memory 字段显式定义为 list[dict]。这突出了一条关键教训：LangGraph 的 SqliteSaver 宽容得过头了。它在序列化过程中优先保存内容而不是严格的 schema 验证，导致反序列化时出现静默数据损坏。对于生产环境，你需要显式验证或更健壮的 ORM。

SqliteSaver 的竞态条件：Checkpoint 损坏

修复 schema 后，我开始看到 sqlite3.OperationalError: database is locked 错误。我的 Agent 运行在 Oracle Cloud Infrastructure（OCI）的无服务器函数（OCI Functions）环境中。每条传入消息都会触发一个新的函数调用。虽然 OCI Functions 是无状态的，但我的 LangGraph Agent 需要状态。SqliteSaver 写入文件。在无服务器环境中，这个文件需要外部化。我使用 OCI Object Storage 存储 SQLite 数据库文件，通过 FUSE 挂载。

问题：多个并发函数调用可能尝试同时写入同一个 SQLite 文件。即使使用 FUSE，底层的 sqlite3 库也不是为来自独立进程的并发写入而设计的——它没有适当的锁定机制，而 FUSE 挂载的对象存储在数据库层面并不原生提供这种机制。这导致 checkpoint 损坏。数据库锁定错误会使 SQLite 文件处于不一致状态，使其对后续调用不可读。

我的解决方案是硬切换：RedisSaver。Redis 为并发访问而设计，提供原子操作。我部署了一个 OCI Cache（Redis）并切换了我的 CheckpointSaver 实现。

from langgraph.checkpoint.redis import RedisSaver
import redis

# ...
# In my agent initialization
redis_client = redis.Redis(host=os.environ.get("REDIS_HOST"), port=6379, db=0)
memory = RedisSaver(redis_client=redis_client)
# ...

这立即解决了 database is locked 错误和 checkpoint 损坏。Redis 的原子 SET 操作确保即使多个调用尝试更新同一个 checkpoint，也只有一个会成功，其他调用会在下次读取时获取最新状态。代价是一个额外的托管服务（OCI Cache），基础实例每月 $25，但稳定性是值得的。

使多步骤管道稳定的唯一模式：显式状态转换

即使使用 RedisSaver，我的多步骤 Agent 仍然偶尔会卡住。用户发送消息后，Agent 会处理它，但下一步不会触发，或者 Agent 会重复上一步。这本身不是 checkpointing 问题，而是 LangGraph 图执行中的状态管理问题。

我最初的图设计严重依赖条件边，这些边检查 user_input 的内容或状态中某些字段的存在。例如：

def route_next_step(state: AgentState):
    if state.get("user_input") == "confirm":
        return "confirm_action"
    if state.get("task_completed"):
        return "notify_user"
    return "process_input"

这种方法很脆弱。如果 user_input 不是精确的 "confirm"，或者 task_completed 被设置但另一个条件也匹配了，Agent 可能会循环或跳转到错误的节点。问题是状态本身没有显式地引导转换。

突破来自我在 AgentState 中引入一个显式的 next_action 字段，并让每个节点负责设置它。

class AgentState(TypedDict):
    chat_id: str
    thread_id: str
    user_input: str
    memory: list[dict]
    current_step: Literal["start", "process_input", "confirm_action", "notify_user", "end"] # 显式状态
    next_action: Literal["process_input", "confirm_action", "notify_user", "end", "wait_for_user"] # 引导图
    # ... other fields

现在，我的条件边变得简单得多，也更健壮：

def route_next_action(state: AgentState):
    return state["next_action"]

# In my graph definition:
graph.add_conditional_edges(
    "start",
    route_next_action,
    {
        "process_input": "process_input_node",
        "confirm_action": "confirm_action_node",
        "notify_user": "notify_user_node",
        "end": END,
        "wait_for_user": "wait_for_user_node", # 一个只是等待新输入的节点
    }
)

每个节点的职责现在不仅包括处理数据，还包括根据其结果显式设置 next_action。例如，一个 process_input_node 可能会决定：

def process_input_node(state: AgentState):
    # ... process input with LLM (Groq/Claude routing based on complexity)
    if requires_confirmation:
        state["next_action"] = "confirm_action"
    elif task_is_done:
        state["next_action"] = "notify_user"
    else:
        state["next_action"] = "wait_for_user" # 等待更多用户输入
    return state

这个模式将我的 Agent 从脆弱的隐式状态机转变为健壮的显式状态机。图的流程变得确定性，由 next_action 字段驱动。这也使调试变得显著更容易：我可以在 checkpoint 中检查 next_action，立即理解 Agent 为什么转换（或不转换）到特定节点。这对于生产环境中的 LangGraph 有状态 Agent checkpointing 至关重要。

这种方法还自然支持 Agent 需要等待用户输入的多轮对话。wait_for_user 动作只是路由到一个什么都不做、只是返回状态的节点，有效地暂停图直到新的 user_input 到达。

健壮性的代价

用零 VC 资金构建这些 Agent 意味着每一分钱都很重要。

OCI Functions：每 GB 秒 $0.000015，每次调用 $0.0000002。我的 Agent 运行时间以毫秒计，因此计算成本可以忽略不计（数千用户每月不到 $5）。

OCI Cache（Redis）：1GB 实例每月 $25。这是状态的主要基础设施成本。

LLM API：Groq（Llama 3 8B）用于快速简单任务，Claude 3 Opus 用于复杂推理。Groq 成本是每百万 token 几分钱。Claude 3 Opus 是输入 $15/M，输出 $75/M。我的路由逻辑将 Opus 的使用降到最低。LLM 总成本通常每月 $50-$150。

OCI Object Storage：每 GB 每月 $0.0255。用于日志和偶尔的大型数据 blob，不是主要状态。可以忽略不计。

运行多个为真实用户服务的产品级 LangGraph Agent 的总基础设施成本每月低于 $200。最大的成本是我调试最初脆弱实现的时间。转向 RedisSaver 和显式状态转换显著减少了调试时间。

常见问题

Q：为什么不使用更健壮的数据库（如 PostgreSQL）做 checkpointing 而用 Redis？

A：PostgreSQL 提供更强的 ACID 保证和复杂查询能力，但 Redis 为键值查找和原子更新提供更低的延迟，这对于 LangGraph checkpoint 中频繁的小状态变更来说是理想的。对于我的用例，Redis 的简单性和速度胜过了对完整关系数据库的需求。

Q：如何使用 RedisSaver 处理生产环境中 AgentState 的 schema 迁移？

A：RedisSaver 将状态存储为 JSON 字符串。对于 schema 变更，我在 AgentState 中实现一个版本字段和一个在加载时运行的迁移函数。如果加载的状态版本比当前 Agent 版本旧，迁移函数会在 Agent 处理状态之前将状态转换为新 schema。

Q：基于任务复杂度在不同 LLM（Groq、Claude）之间路由的策略是什么？

A：我使用一个小而快的 LLM（如 Groq 上的 Llama 3 8B）作为路由器。路由器分析用户输入和当前 AgentState，以确定任务是否需要复杂推理（路由到 Claude 3 Opus）或可以由更便宜更快的模型处理（路由到 Groq）。这个决策是 process_input_node 逻辑的一部分。

Q：如何管理单个 Agent 实例的并发用户交互？

A：每个用户交互（例如一个 Telegram chat ID）映射到 LangGraph 中一个唯一的 thread_id。RedisSaver 按 thread_id 存储 checkpoint。当新消息传入一个已存在的 thread_id 时，LangGraph 加载该 thread 的特定状态，确保隔离的对话。

— Elena Revicheva · AIdeazz · Portfolio