重启后 Agent 状态对不上：检测漂移并重新同步

workflow 执行到一半崩溃。你从最近的 checkpoint 重启，Agent 内部状态显示「文件 A、B、C 都已重构完成」——但崩溃发生在它记下 step 3 complete 之后、真正把文件 C 写到磁盘之前。Agent 以为 C 已经写好了就跳过它，于是代码库恰恰在 Agent 自以为修好的地方坏掉了。另一种情况：LangGraph 流水线在 MemorySaver/InMemorySaver 进程重启后恢复，内存里的 checkpoint 全没了，Agent 把每一步重跑一遍——在不支持幂等写入的数据库上把所有变更又做了一次。

TL;DR——最快的修法： 90% 的状态漂移都是这两种之一：(1) 状态在副作用完成之前就写了（把顺序改成副作用完成之后再写状态）；(2) 你把运行时状态存在了进程内存里（Python 的 set/dict，或 LangGraph 的 InMemorySaver），kill -9 一下就没了（换成持久化 saver——LangGraph 用 PostgresSaver/RedisSaver，或交给 Temporal 的事件历史来保存）。然后加一个「恢复前校验」步骤，让 Agent 永远不要盲目信任 checkpoint。下面是细节。

你属于哪一类？

重启后的症状	最可能的根因	跳转
跳过了一个其实没真正完成的步骤	状态在副作用之前就写了	根因 1
重跑了已经做过的工作（重复发邮件、重复写库）	checkpoint 粒度太粗 / 步骤不幂等	根因 2
重启后从零开始	状态只存在进程内存里	根因 3
用了过期数据（旧 schema、旧文件内容）	停机期间外部世界变了	根因 4
部署后某个状态字段出现 KeyError / None	checkpoint schema 版本不匹配	根因 5
checkpoint 能加载，但部分字段是默认值	checkpoint 写入不完整 / 损坏	根因 6
nondeterministic /「command does not match event history」（Temporal）	没做版本控制就改了 workflow 代码	Temporal 说明

常见原因

1. 状态在副作用完成之前就写了

Agent 先把「step 3 complete」写进状态库，然后在真正的文件写入或 API 调用完成之前崩溃。重启后状态说「已完成」，但副作用从未发生。这是经典的「先提交后执行」顺序 bug，也是状态漂移最常见的单一根因。

怎么判断：找所有在真正的文件写入、API 调用、数据库变更之前（而不是之后）就调用 state.mark_done(step)（或写 checkpoint）的代码。状态更新必须放在最后，或与副作用做成原子操作。

2. Checkpoint 粒度太粗——一次覆盖多步

Checkpoint 每 10 步写一次，但崩溃发生在第 7 步。重启后 Agent 重放第 1-10 步，把已经跑过的第 1-7 步又执行一遍。如果这些步骤不幂等（追加写文件、自增计数器、扣款、发邮件），重放就会产生重复副作用。

怎么判断：找到 workflow 代码里的 checkpoint 粒度，算 latest_checkpoint_step 减 crash_step。差值大于 0 就是被丢弃并重放的工作量。只要一个 checkpoint 覆盖多个有副作用的操作，崩溃在 checkpoint 中间就必然导致漂移。

3. 运行时状态只存在进程内存里

很多流水线把状态攒在普通的 Python dict/set 或类属性里。进程重启会把它全部清空，于是 Agent 要么从头来过，要么做一半再造出重复数据。在 LangGraph 里，InMemorySaver（旧名 MemorySaver 的新叫法）正是这样——它把 checkpoint 存在 RAM 的 defaultdict 里，仅供测试。截至 2026 年 6 月，LangChain 官方文档说得很明确：任何需要扛住重启的场景都要用 PostgresSaver/AsyncPostgresSaver（或 RedisSaver）。

怎么判断：列出每一个状态变量，标记它的存储位置（内存 / 文件 / Redis / Postgres）。任何在恢复之后还要用、却只存在内存里的变量，kill -9 之后都会丢。

4. Checkpoint 与恢复之间外部状态变了

Agent 在 checkpoint 里记下 db_schema = v4。它停机期间，有人手动把库迁移到了 v5。Agent 从 v4 的 checkpoint 恢复，生成的迁移 SQL 现在是错的。同一类 bug 还包括：Agent 依赖的临时文件被系统清理了，或者它「已经写过」的文件被别人改了。

怎么判断：把恢复时刻的外部世界与 checkpoint 记录的对比——文件 hash、schema 版本、行数、API 资源版本。任何差异都是漂移，这也是为什么你需要一个明确的「恢复前校验」步骤（见下面 Step 1）。

5. 部署后 checkpoint schema 版本不匹配

你上线了新代码，它的状态结构和磁盘上已有的 checkpoint 不一样。旧 checkpoint 里没有 state["new_field"]，于是 Agent 拿到 None 或 KeyError，然后用错误的默认值继续跑。在 Temporal 里这个问题表现形式不同（见确定性说明），但对自研状态和 LangGraph 状态来说，这就是单纯的序列化版本断层。

怎么判断：把当前代码的 state schema 与最新 checkpoint 的字段列表做 diff。缺失或被重命名的字段就是触发点。给每个 checkpoint 打上 schema_version，并在加载时做迁移。

6. Checkpoint 写入不完整或损坏

Checkpoint 写到一半被打断（崩溃、OOM kill、磁盘满），读回来时部分字段停在了零值/默认值。Agent 用一份从未真实共存过的「真状态 + 默认状态」混合体恢复。多个写入者不加锁并发写也会造成同样的后果——一半字段来自一个写入者，一半来自另一个。

怎么判断：给每个 checkpoint 加 checksum 和 is_complete: true 标记。加载时先校验这两项再信任任何字段。检查 updated_at 时间戳是否有同一毫秒内的两次写入，那是无锁并发写的信号。

最短修复路径

Step 1：加一个恢复前状态校验步骤

从任何 checkpoint 恢复之前，先确认现实世界与 checkpoint 的假设一致。一旦不一致，不要自动恢复——先告警让人来对账。

def verify_checkpoint(checkpoint: dict) -> list[str]:
    discrepancies = []
    for file_path, expected_hash in checkpoint.get("file_hashes", {}).items():
        actual = hash_file(file_path) if os.path.exists(file_path) else None
        if actual != expected_hash:
            discrepancies.append(
                f"{file_path}: expected {expected_hash}, got {actual}"
            )
    for key, expected_val in checkpoint.get("db_state", {}).items():
        actual_val = db.get_value(key)
        if actual_val != expected_val:
            discrepancies.append(
                f"DB {key}: expected {expected_val}, got {actual_val}"
            )
    return discrepancies

Step 2：状态写在副作用之后，而不是之前

# 错误 —— 状态在副作用之前写
def execute_step(step):
    state.mark_done(step.id)   # 在这里崩溃 => 状态说已完成，副作用没做
    write_file(step.output)

# 正确 —— 状态在副作用之后写
def execute_step(step):
    write_file(step.output)    # 先做副作用
    state.mark_done(step.id)   # 在这里崩溃 => 状态说未完成，副作用已做（只要幂等，重跑是安全的）

这样保证崩溃后状态总是比现实落后一步，绝不会超前一步。只要满足下面 Step 3，重跑最后一步就是安全的。

Step 3：让副作用幂等，重跑才安全

def write_file_idempotent(path: str, content: str, expected_hash: str):
    if os.path.exists(path) and hash_file(path) == expected_hash:
        return  # 已经正确写过 —— 跳过
    with open(path, "w") as f:
        f.write(content)

# 数据库操作：
def upsert_record(table: str, key: str, value: dict):
    # INSERT ON CONFLICT DO UPDATE（幂等）
    db.execute(
        f"INSERT INTO {table} (key, data) VALUES (?, ?) "
        "ON CONFLICT(key) DO UPDATE SET data = excluded.data",
        (key, json.dumps(value)),
    )

对于不幂等的外部调用（扣款、发邮件），用 workflow_id + step_id 派生一个幂等键传给服务方，让重放在服务端被去重。

Step 4：把 checkpoint 边界细化到「一步一个 checkpoint」

for step in steps:
    execute_step(step)                                    # 一个副作用
    checkpoint.save(step_id=step.id, state=current_state) # 紧接着立即写

这样重启最多只重跑一步（最后一步），而不是一批。如果你用 LangGraph，持久化 saver 给你的正是这个能力——用 thread_id 给每次运行划范围，让 saver 在每个节点之后持久化：

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # 首次使用必须调用 —— 创建 checkpoint 相关表
    graph = builder.compile(checkpointer=checkpointer)
    # 恢复：不传 checkpoint_id 就取该 thread 的最新 checkpoint
    graph.invoke(state, {"configurable": {"thread_id": run_id}})

截至 2026 年 6 月，PostgresSaver（和 AsyncPostgresSaver）首次使用前必须调用一次 .setup()。要定位到历史中的某个具体点，用 {"configurable": {"thread_id": run_id, "checkpoint_id": "<uuid>"}}。

Step 5：给每个 checkpoint 加 checksum 和完成标记

import hashlib, json

def save_checkpoint(state: dict, path: str):
    payload = json.dumps(state, sort_keys=True, default=str)
    checksum = hashlib.sha256(payload.encode()).hexdigest()
    tmp = path + ".tmp"
    with open(tmp, "w") as f:
        json.dump({"state": state, "checksum": checksum,
                   "is_complete": True, "schema_version": 2}, f)
    os.replace(tmp, path)  # 原子重命名 —— `path` 处永远不会是写到一半的文件

def load_checkpoint(path: str) -> dict:
    with open(path) as f:
        record = json.load(f)
    if not record.get("is_complete"):
        raise CorruptedCheckpointError(f"Incomplete checkpoint at {path}")
    payload = json.dumps(record["state"], sort_keys=True, default=str)
    if hashlib.sha256(payload.encode()).hexdigest() != record["checksum"]:
        raise CorruptedCheckpointError(f"Checksum mismatch at {path}")
    return record["state"]

tmp + os.replace() 让写入变成原子操作，写到一半崩溃也绝不会在真实路径上留下损坏文件——它留下的是上一份完好的 checkpoint。

关于 Temporal 的确定性

Temporal 不保存内存快照。重启时它会拿你的 workflow 代码对着记录下来的事件历史重放，跳过所有已经成功的 activity 并复用其结果——所以上面那些漂移类别在 workflow 函数内部基本会消失。但仍有两点会咬你：

1. activity 仍可能重跑：worker 崩溃或重试时 activity 会被重新执行，所以每个碰外部世界的 activity 都必须幂等（用 Step 3 里的 workflow_id + activity 幂等键模式）。
2. 运行中改 workflow 代码会破坏重放：如果一个正在运行的执行是用旧代码起的，却在一个加了/删了/调整了步骤的 worker 上恢复，你会拿到 nondeterminism 错误（「command does not match event history」）。改代码要用 Workflow Versioning API 保护，并在 CI 里跑 replay 测试。注意：2025 年之前的实验性 Worker Versioning 已于 2026 年 3 月从 Temporal Server 移除——请用当前的 versioning API，别用旧的那套。

值得记住的一点：checkpointer 不等于 durable execution（持久化执行）。checkpoint 只给你一个存档点；你仍要自己负责判断何时需要它、触发恢复、并协调以避免重复工作。而 durable-execution 引擎（Temporal，以及越来越成熟的 LangGraph 持久化执行模式）会替你做这些协调。

如何确认已修复

1. 在 CI 里做崩溃测试。 写一个测试：跑 workflow，在随机某一步硬杀进程（步骤里 os._exit(1)，或对子进程 kill -9），从 checkpoint 重启，断言最终状态和副作用恰好正确一次。
2. 对账检查。 一次运行结束后，把 checkpoint 里的「已处理数量」与真正的事实来源（数据库行数、磁盘文件数）对比。两者必须一致。一旦对不上，说明还有漂移。
3. Temporal replay 测试。 拿你当前的 workflow 代码对着生产抓下来的事件历史跑一遍；replay 测试通过，说明最新部署不会让运行中的执行抛 nondeterminism 错误。

预防建议

• 状态永远写在副作用完成之后——绝不在之前。
• 用存储允许的最小粒度做 checkpoint——一步一个 checkpoint。
• 加一个恢复前校验步骤，继续执行前先确认现实世界与 checkpoint 假设一致。
• 让每个副作用幂等；对不幂等的外部调用，传一个稳定的幂等键。
• 给 checkpoint 加 checksum、完成标记和 schema_version；对不完整或损坏的 checkpoint 拒绝并告警，加载时做迁移。
• checkpoint 里要包含关键外部产物（文件、schema 版本、API 资源版本）的 hash，而不只是内部状态。
• 任何长于一次 LLM 调用的 workflow，都不要把进程内存（裸 dict/set 或 LangGraph InMemorySaver）当作唯一的状态库。
• 把外部数据库当作事实来源（source of truth）；恢复时从它重建 processed_ids，而不是信任 checkpoint 里的那份拷贝。
• 在 CI 里显式测试恢复路径，每次部署前跑 Temporal replay 测试。

常见问答 (FAQ)

Q: Temporal 能保证不出现状态漂移吗？ A: 在 workflow 函数内部基本可以——Temporal 用基于事件溯源的 durable execution，事件历史是权威，已经成功的 activity 在重放时会被跳过并复用结果。但产生外部副作用的 activity（写文件、调 API）仍必须幂等，因为 worker 重启时 Temporal 可能重新执行它们；而且不做版本控制就改 workflow 代码会导致 nondeterminism 错误。

Q: LangGraph 的 MemorySaver/InMemorySaver、SqliteSaver 和 PostgresSaver 有什么区别？ A: InMemorySaver（现用名；MemorySaver 是旧别名）把 checkpoint 存在 RAM 里，进程一重启全丢——仅供测试。SqliteSaver 写入本地 SQLite 文件，能在同一台机器上扛住重启。生产和多 worker 场景请用 PostgresSaver/AsyncPostgresSaver（或 Redis saver），它们能在多个 worker 间共享 checkpoint。Postgres saver 首次使用需调用一次 .setup() 来建表。

Q: 如果 checkpoint 和外部数据库对不上，以哪个为准？ A: 以外部数据库为准。它是事实来源，checkpoint 只是执行进度的记录。恢复时，先从数据库查哪些 item 真正被处理过（通过唯一 ID + 状态字段），据此重建 processed_ids，不要盲目信任 checkpoint 里的 processed_ids。

Q: 如果一个 checkpoint 只有部分有效——一些字段对、一些字段错，怎么办？ A: 不要自动把部分有效的 checkpoint 和现实世界合并。把损坏的那份归档，跑恢复前校验，从上一份完全有效的 checkpoint 重放。自动部分合并会造出极难调试的混合状态。

Q: 旧 checkpoint 要保留多久？ A: 每个 workflow run 至少保留最近 3 份。只留一份的话，下次写 checkpoint 失败你就没有退路；三份能在最近一份损坏时给你回退。至于 schema 迁移代码，要保留到覆盖你恢复时仍可能加载的所有版本——保留两个大版本是常见经验法则。

相关阅读

• Checkpoint 恢复出来的状态是损坏的
• Agent 交接时上下文丢失了
• Agent 预算在任务中途被吃光
• 共享 memory 被多 Agent 写覆盖
• Agent 回滚不完整
• Claude Code 中途停下：撞了哪种额度上限怎么办
• 多个 AI agent 同时改文件冲突了

标签: #AI 编程 #Agents #排查

相关文章

常见问题解决库

Agent 预算在任务中途被吃光

Agent 执行长任务时 token 或费用预算耗尽，任务半途中止、半成品代码留在磁盘上。本文教你定位预算去向、抢救中间结果，并从断点续跑。

2026/05/25 #AI 编程 #Agents

常见问题解决库

Checkpoint 恢复出来的状态是损坏的

Agent 从 checkpoint 恢复后状态字段缺失、类型错乱或值异常。本文教你定位损坏、回滚到完好的历史版本，并写出真正抗崩溃的 checkpoint。

2026/05/25 #AI 编程 #Agents

常见问题解决库

成本统计漏算了子 Agent 用量

成本仪表盘显示 2 美元，账单却是 18 美元。子 Agent 的 token 消耗没有归因到父级运行。本文给出补齐缺口的方法，2026 年 6 月核实。

2026/05/25 #AI 编程 #Agents

常见问题解决库

Agent 调用图出现循环但没人发现

多 Agent 之间互相交接，形成永不终止的环，因为没有任何东西在检测循环。本文教你抓出这个环、配好框架自带的上限、并把循环彻底限定住。

2026/05/25 #AI 编程 #Agents

常见问题解决库

Agent 交接时上下文丢失了

下游 Agent 重复询问已答过的问题，或输出与前几步的决策相矛盾。本文帮你定位丢失上下文的交接边界，并在一小时内接上结构化、可持久的状态传递。

2026/05/25 #AI 编程 #Agents

常见问题解决库

Agent 编排器互相等待出现死锁

两个 Agent 互相等待对方的输出，整条流水线永远挂起。本文教你找到环、加超时、几分钟内打破死锁，并给出 LangGraph、Temporal、AutoGen 的具体做法。

2026/05/25 #AI 编程 #Agents