LangGraph Checkpointing 导致生产数据丢失的三个阶段与解法

Originally published on AIdeazz — cross-posted here with canonical link.

我的第一个 LangGraph Agent（智能体），一个简单的文档摘要工具，在三周内悄无声息地丢弃了 80% 的任务。日志显示 Agent finished successfully，但输出队列始终是空的。问题不在 LLM、不在 prompt、也不在向量数据库——而是一种状态 schema 不匹配，是 LangGraph 检查点机制中的一个静默杀手，在被我发现之前，它让我在 Oracle Cloud 上浪费了 1200 个 CPU 小时。

这并不是我唯一的 LangGraph 生产环境头疼问题。在为 AIdeazz 交付我的第一个多 Agent 系统之前，我经历了三次对核心 LangGraph 状态管理和检查点逻辑的完整重写。每次重写都针对不同的失败模式：静默数据丢失、损坏的检查点，以及最终为多步骤、有状态 Agent 管道带来稳定性的模式。

The Silent State Schema Mismatch（静默的状态 Schema 不匹配）

我的初始 LangGraph Agent 处理传入的文档，对其进行摘要，然后将摘要路由到特定的输出渠道（Telegram、WhatsApp、email）。状态是一个简单的 TypedDict：

class AgentState(TypedDict):
    document_id: str
    raw_text: str
    summary: Optional[str]
    output_channel: str
    status: Literal["processing", "summarized", "failed"]

Agent 在本地运行正常。部署到 Oracle Cloud 后，每小时处理 100 个文档，表面上看也是正常的。status 字段会在数据库中更新为 summarized，Agent 日志也确认了完成。但数据库中的 summary 字段始终是 NULL。

我花了好几天调试摘要步骤本身，确信是 LLM 在产生幻觉或者 prompt 格式有问题。我添加了更多日志、打印中间步骤，甚至在 LangGraph 之外单独运行摘要逻辑——它总是能生成摘要。

问题出在 summary: Optional[str] 字段上。我最初的状态定义是 summary: str。后来我将其更新为 Optional[str]，以处理摘要可能失败或不需要立即生成的情况。LangGraph 的 SqliteSaver（以及由此推及的其他 BaseCheckpointSaver 实现）会将存储的状态反序列化为当前状态 schema。如果存储状态中存在某个字段，但在新 schema 中被移除或类型改变了，该值会在反序列化过程中被静默丢弃。反之亦然，如果添加了新字段，它的值会是 None。

我的数据库仍然保存着旧 schema 的状态。当 Agent 加载检查点时，存储状态中的 summary 字段（当时是 str 类型）被反序列化到新的 Optional[str] schema 中。LangGraph 内部的 _load_state 方法在遇到类型不匹配或新 schema 中缺少的字段时，会直接丢弃旧检查点中的值。没有错误，没有警告。摘要明明在数据库里，但它从未进入 Agent 的运行时状态。

修复方案：显式的 schema 版本控制和迁移。我现在在每个 AgentState 中嵌入一个 schema_version: int，并实现一个 migrate_state(state: AgentState, target_version: int) -> AgentState 函数。在加载检查点之前，我会检查其版本并应用必要的迁移。这增加了样板代码，但防止了静默数据丢失。

class AgentStateV1(TypedDict):
    document_id: str
    raw_text: str
    summary: str # Old schema
    schema_version: Literal[1]

class AgentStateV2(TypedDict):
    document_id: str
    raw_text: str
    summary: Optional[str] # New schema
    output_channel: str
    status: Literal["processing", "summarized", "failed"]
    schema_version: Literal[2]

def migrate_state(state: dict, target_version: int) -> dict:
    current_version = state.get("schema_version", 1) # Assume V1 if not present
    if current_version == target_version:
        return state

    if current_version == 1 and target_version == 2:
        # Example migration: add new fields with defaults
        state["output_channel"] = "default"
        state["status"] = "processing"
        state["schema_version"] = 2
        return state

    raise ValueError(f"Unsupported migration from V{current_version} to V{target_version}")

# Before loading:
# loaded_state = checkpoint_saver.get_tuple(thread_id).checkpoint["v"]
# current_state = migrate_state(loaded_state, TARGET_SCHEMA_VERSION)
# graph.invoke(current_state, config={"configurable": {"thread_id": thread_id}})

Corrupted Checkpoints and Race Conditions（损坏的检查点与竞态条件）

我的第二次重写是在经历了一周的 sqlite3.DatabaseError: database disk image is malformed 错误之后。这些错误发生在运行同一个 LangGraph Agent 的多个实例时——每个实例都有自己的 SqliteSaver，指向共享 NFS 卷上的同一个 SQLite 文件。

问题是一个经典的竞态条件。SQLite 在单写多读场景下是健壮的。LangGraph 的 SqliteSaver 执行多个操作：get_tuple、反序列化、修改状态、序列化、put_tuple。如果两个 Agent 试图同时更新同一个 thread ID 的检查点，其中一个会覆盖另一个的更改，或者更糟——写入一个部分更新或损坏的 blob。SqliteSaver 没有为来自独立进程的并发写操作实现文件级锁或事务管理。

我的 Agent 以 Docker 容器形式部署在 Oracle Container Engine for Kubernetes (OKE) 上。每个 Pod 有自己的 SqliteSaver 实例，而我当时把 SQLite 数据库挂载到了一个共享的 NFS 卷上。对于需要并发写的 SqliteSaver 来说，这个架构从根本上就是有缺陷的。

修复方案：集中化、原子化的检查点存储。我从 SqliteSaver 切换到了一个自定义的 BaseCheckpointSaver 实现，后端是 Oracle Autonomous Database (ADB) 和 Redis。

Redis 用于临时状态和锁：在 Agent 开始处理一个 thread 之前，它在 Redis 中为该 thread_id 获取一个带过期时间的锁。如果无法获取锁，它就重试或将任务加入队列。

ADB 用于持久化检查点：实际的检查点数据（序列化的 LangGraph 状态）存储在 ADB 表的一个 JSON 列中。更新操作在数据库事务内执行，确保原子性。put_tuple 方法现在执行 UPSERT 操作来更新 JSON 列。

这个模式确保了任意时刻只有一个 Agent 能修改给定 thread 的状态，且更新是原子的和持久的。ADB 的成本比 SQLite 高，但对于生产环境来说稳定性是不可妥协的。

# Simplified custom saver logic
class ADBCheckpointSaver(BaseCheckpointSaver):
    def __init__(self, db_connection_pool, redis_client):
        self.db_pool = db_connection_pool
        self.redis = redis_client

    def get_tuple(self, thread_id: str) -> Optional[CheckpointTuple]:
        # Acquire Redis lock
        lock_key = f"langgraph_lock:{thread_id}"
        if not self.redis.set(lock_key, "locked", ex=60, nx=True): # 60s expiry, only if not exists
            raise LockAcquisitionError(f"Could not acquire lock for thread {thread_id}")

        try:
            with self.db_pool.acquire() as conn:
                with conn.cursor() as cursor:
                    cursor.execute("SELECT checkpoint_data FROM checkpoints WHERE thread_id = :1", [thread_id])
                    row = cursor.fetchone()
                    if row:
                        checkpoint_data = json.loads(row[0])
                        # Deserialize into CheckpointTuple
                        return CheckpointTuple(
                            config={"configurable": {"thread_id": thread_id}},
                            checkpoint=checkpoint_data,
                            parent_config=None, # Or retrieve if stored
                        )
                    return None
        finally:
            self.redis.delete(lock_key) # Release lock

    def put_tuple(self, checkpoint_tuple: CheckpointTuple) -> None:
        thread_id = checkpoint_tuple.config["configurable"]["thread_id"]
        lock_key = f"langgraph_lock:{thread_id}"
        if not self.redis.get(lock_key):
            raise LockAcquisitionError(f"Lock for thread {thread_id} not held during put_tuple")

        with self.db_pool.acquire() as conn:
            with conn.cursor() as cursor:
                checkpoint_json = json.dumps(checkpoint_tuple.checkpoint)
                cursor.execute(
                    """
                    MERGE INTO checkpoints c
                    USING (SELECT :1 AS thread_id, :2 AS checkpoint_data FROM DUAL) d
                    ON (c.thread_id = d.thread_id)
                    WHEN MATCHED THEN UPDATE SET c.checkpoint_data = d.checkpoint_data
                    WHEN NOT MATCHED THEN INSERT (thread_id, checkpoint_data) VALUES (d.thread_id, d.checkpoint_data)
                    """,
                    [thread_id, checkpoint_json]
                )
                conn.commit()

The "Always Restart" Pattern for Multi-Step Pipelines（多步骤管道中的"始终重启"模式）

即使有了 schema 版本控制和原子化检查点，我的多 Agent 系统——尤其是那些涉及外部 API 调用或人工介入步骤的系统——仍然很脆弱。一个 Agent 可能发起 API 调用、收到 200 OK，但随后由于网络故障或意外载荷而无法解析响应。状态会被保存，但 Agent 就此卡住了。用相同步骤重试通常会导致重复操作（例如：同一封邮件发出两次）。

我的 Agent 通常涉及：

• 接入与预处理（Groq 用于快速分类）
• 复杂推理（Claude 3.5 Sonnet 用于多步骤规划）
• 外部 API 调用（CRM、支付网关）
• 人工审核（通过 Telegram/WhatsApp）
• 最终输出生成

第三步的失败意味着 Agent 会卡住。LangGraph 的默认行为是从上一个保存的状态恢复。如果那个状态是"恰好在失败的 API 调用之前"，它就会重试 API 调用。如果 API 调用是幂等的，还好。如果不是，就是个大问题。

修复方案："始终重启"模式。我没有让 LangGraph 从精确的失败点恢复，而是将我的 Agent 节点设计为幂等的，并始终从头重新评估当前逻辑步骤的当前状态。

每个逻辑步骤（例如"摘要文档"、"发起 API 调用"、"等待人工批准"）都是一个 LangGraph 节点。在每个节点内部，在执行任何操作之前，Agent 首先根据当前状态检查该操作是否已经完成。

例如，在一个"执行 API 调用"节点中：

def execute_api_call_node(state: AgentState) -> AgentState:
    if state.api_call_status == "completed":
        print("API call already completed, skipping.")
        return state

    try:
        # Perform API call
        response = make_external_api_call(state.api_payload)
        state["api_response"] = response.json()
        state["api_call_status"] = "completed"
        return state
    except Exception as e:
        state["api_call_status"] = "failed"
        state["error_message"] = str(e)
        return state

这个模式意味着：如果一个 Agent 在节点中途失败，然后被重启，它会重新进入该节点，看到 api_call_status 不是 "completed"，尝试 API 调用，然后更新状态。如果再次失败，状态保持 "failed"。如果成功，状态变为 "completed"。下次图运行时，它会看到 "completed" 并跳过 API 调用。

从这个角度看，这个模式使得每个节点在图的角度上是幂等的。图可以从任意点重启，会优雅地在上次离开的地方继续，而不会重复工作或陷入非幂等操作的重试循环。这也简化了错误处理：我不再依赖 LangGraph 内部的复杂重试逻辑，而是依靠一个外部编排器（一个简单的 Python 脚本，运行在 cron job 上）定期重新调用处于 "failed" 或 "pending" 状态的 Agent。

这个"始终重启"模式，结合健壮的检查点和显式的 schema 管理，终于为 Oracle Cloud 上的生产多 Agent 系统带来了所需的稳定性。我目前的 Agent 每天处理数千条消息，在 Groq（用于快速初始处理）、Claude 3.5 Sonnet（用于复杂推理）和自定义工具（用于外部交互）之间路由，同时在整个可能运行很长时间的进程中维护状态。

Frequently Asked Questions（常见问题）

Q: How do you handle schema changes for in-flight agents with the versioning approach? A: When a new schema version is deployed, agents processing older checkpoints will first load the old state, then migrate_state will transform it to the new schema. This transformed state is then saved back to the checkpoint store, effectively upgrading the checkpoint. Agents starting new threads will use the latest schema.

Q: What's the overhead of using Redis for locking and ADB for checkpoints compared to a simpler solution? A: Redis adds ~2-5ms latency for lock acquisition/release. ADB adds ~10-50ms for checkpoint UPSERT operations, depending on network latency and payload size. This is acceptable for most multi-agent systems where LLM calls dominate latency (hundreds of ms to seconds). The stability gain far outweighs this overhead for production.

Q: How do you manage the LockAcquisitionError in your ADBCheckpointSaver? A: When LockAcquisitionError is raised, the agent's current invocation is aborted. The external orchestrator (e.g., a message queue consumer or a cron job) responsible for invoking agents will catch this error and typically re-queue the message or mark the thread for a later retry. This ensures that only one agent attempts to process a specific thread at a time.

Q: Does the "Always Restart" pattern mean you re-run LLM calls if a node fails after the LLM call but before saving state? A: Yes, if an LLM call completes but the subsequent state update or external action fails before the LangGraph node returns and its state is checkpointed, the LLM call might be re-run on restart. To prevent this for expensive LLM calls, I often add a llm_response_cached: bool flag to the state and save the raw LLM response. The node then checks this flag and uses the cached response if available.

Q: Why Oracle Autonomous Database (ADB) specifically? A: ADB offers fully managed, auto-scaling, and highly available PostgreSQL-compatible or Oracle Database instances. For AIdeazz, it integrates seamlessly with other Oracle Cloud Infrastructure (OCI) services I use (OKE, OCI Functions, OCI AI Services) and provides strong performance guarantees without requiring dedicated DBA resources, which is critical for a lean operation.

— Elena Revicheva · AIdeazz · Portfolio