从崩溃重跑到 3 秒续跑：智能体断点续跑实战

长任务智能体为什么必须支持断点续跑

在代码迁移、批量数据分析、多轮网页操作、自动化测试修复等长任务中，智能体通常要连续执行几十甚至上百个步骤。问题不在于它会不会失败，而在于失败后是否必须从零开始：一次网络抖动、工具限流、上下文超限或进程重启，如果导致前面 30 分钟的规划、工具调用和中间结果全部作废，用户体验和成本都不可接受。

断点续跑的目标不是简单保存聊天记录，而是让智能体在恢复后能够回答三个问题：

1. 我已经完成了哪些业务步骤；
2. 每个工具调用是否真正生效，能否安全重试；
3. 当前还缺哪些状态，才能继续生成下一步动作。

因此，agent-checkpoint 应覆盖任务状态、消息历史、工具结果、外部资源指针、重试元数据与版本信息，而不是只持久化最后一次 LLM 输出。

三种状态恢复方案对比

常见实现可以分为三类：

方案	持久化内容	优点	局限	适用场景
对话日志重放	messages、prompt、响应	实现简单	工具副作用难撤销，恢复成本高	纯问答、短流程
步骤事件溯源	每一步计划、动作、结果	可审计、可回放到任意点	事件结构设计复杂	高可靠工作流
状态快照 + 事件日志	周期性快照加增量事件	恢复快，存储可控	要处理快照一致性	长任务、生产级 Agent

推荐采用第三种方式：每隔 N 个步骤或在关键边界写快照，步骤之间追加事件。恢复时先加载最近快照，再重放快照之后的确定性事件。

设计 agent-checkpoint 的核心数据结构

一个可恢复检查点至少要包含六类字段：

• run_id：一次长任务的稳定标识；
• step_id：单调递增的步骤编号；
• phase：规划、工具调用、校验、完成等阶段；
• messages：压缩后的对话与工具消息；
• business_state：业务变量，例如已处理文件、迁移结果；
• pending_action：崩溃时可能已发出但未确认的动作；
• idempotency_key：工具调用幂等键；
• schema_version：检查点结构版本，用于兼容升级。

不要把临时文件句柄、数据库连接、WebSocket 对象直接放入快照。状态中应保存可重新打开的资源描述，例如文件路径、对象存储 key、数据库主键。

可运行的 Python 断点续跑示例

下面示例用内存存储模拟持久化。生产环境可将 CheckpointStore 替换为 SQLite、PostgreSQL、Redis 或对象存储，但写入顺序和状态机语义保持不变。

from __future__ import annotations

import json
import time
import uuid
from dataclasses import dataclass, asdict, field
from typing import Any, Callable


@dataclass
class Checkpoint:
    run_id: str
    step_id: int
    phase: str
    messages: list[dict[str, Any]]
    business_state: dict[str, Any]
    pending_action: dict[str, Any] | None = None
    schema_version: int = 1
    updated_at: float = field(default_factory=time.time)


class CheckpointStore:
    """生产环境可替换为 SQLite/PostgreSQL；接口保持一致。"""

    def __init__(self) -> None:
        self._data: dict[str, dict[str, Any]] = {}

    def save(self, cp: Checkpoint) -> None:
        # 实际系统中建议使用事务：先写事件，再更新当前快照指针
        self._data[cp.run_id] = asdict(cp)

    def load(self, run_id: str) -> Checkpoint | None:
        raw = self._data.get(run_id)
        return Checkpoint(**raw) if raw else None


class ResumableAgent:
    def __init__(
        self,
        store: CheckpointStore,
        planner: Callable[[Checkpoint], dict[str, Any]],
        tools: dict[str, Callable[[dict[str, Any]], Any]],
    ) -> None:
        self.store = store
        self.planner = planner
        self.tools = tools

    def start(self, task: str) -> str:
        run_id = str(uuid.uuid4())
        cp = Checkpoint(
            run_id=run_id,
            step_id=0,
            phase="planning",
            messages=[{"role": "user", "content": task}],
            business_state={"completed": [], "results": {}},
        )
        self.store.save(cp)
        return run_id

    def run_once(self, run_id: str) -> str:
        cp = self.store.load(run_id)
        if cp is None:
            raise ValueError("检查点不存在，无法恢复")

        if cp.phase == "completed":
            return "任务已完成"

        # 如果存在 pending_action，优先处理上次可能已执行但未确认的动作
        action = cp.pending_action or self.planner(cp)
        cp.step_id += 1
        cp.pending_action = action
        cp.phase = "tool_running"
        self.store.save(cp)

        tool_name = action["tool"]
        arguments = action["arguments"]
        result = self.tools[tool_name](**arguments)

        cp.business_state["completed"].append({
            "step_id": cp.step_id,
            "tool": tool_name,
            "arguments": arguments,
        })
        cp.business_state["results"][str(cp.step_id)] = result
        cp.messages.append({
            "role": "tool",
            "name": tool_name,
            "content": json.dumps(result, ensure_ascii=False),
        })

        # 工具结果确认后，才能清除 pending_action
        cp.pending_action = None
        cp.phase = "planning"
        self.store.save(cp)
        return f"步骤 {cp.step_id} 执行完成"

    def finish(self, run_id: str) -> None:
        cp = self.store.load(run_id)
        assert cp is not None
        cp.phase = "completed"
        self.store.save(cp)


def fake_planner(cp: Checkpoint) -> dict[str, Any]:
    # 示例：根据已完成步骤决定下一个动作；真实系统由 LLM + 结构化输出生成
    next_index = len(cp.business_state["completed"]) + 1
    return {
        "idempotency_key": f"{cp.run_id}:{next_index}",
        "tool": "echo",
        "arguments": {"text": f"第 {next_index} 个长任务步骤"},
    }


def echo(text: str) -> dict[str, str]:
    return {"echo": text}


if __name__ == "__main__":
    store = CheckpointStore()
    agent = ResumableAgent(store, fake_planner, {"echo": echo})

    run_id = agent.start("请批量处理三个长任务步骤")
    for _ in range(3):
        print(agent.run_once(run_id))
    agent.finish(run_id)

    # 即使进程重启，只要 run_id 和持久化存储还在，就能加载状态
    restored = store.load(run_id)
    print(json.dumps(restored.business_state, ensure_ascii=False, indent=2))

这段代码的关键是两阶段提交思想：执行工具前先把动作写入 pending_action，工具结果被业务状态确认后再清空。如果进程恰好在工具调用期间退出，恢复流程不会盲目跳过，而是进入补偿或查询逻辑。

工具幂等是状态恢复能否成立的前提

LLM 可以重发请求，但外部工具未必能安全重复执行。发送邮件、创建工单、支付、写数据库都可能因为重试产生重复副作用。实践中应为每个动作生成稳定的 idempotency_key，格式可采用 run_id:step_id:action_type。

工具侧需要支持三种处理：

1. 首次请求：正常执行并记录幂等键；
2. 完全相同的重试：返回首次结果，不重复执行；
3. 相同键但参数不同：拒绝执行并提示状态冲突。

对于无法原生幂等的 HTTP 工具，可以增加动作状态表：pending、succeeded、failed、unknown。恢复时看到 unknown，应先查询远端状态，而不是直接重试。

上下文压缩与快照粒度

长任务的消息历史可能超过模型上下文窗口。恢复时不能机械加载全部 messages，建议采用三层结构：

• 原始事件日志：完整审计，不直接进入提示词；
• 周期快照：保存关键业务变量和摘要；
• 短期窗口：最近若干轮完整消息，供模型继续推理。

快照粒度过细会增加写入成本，过粗则恢复后要重放大量事件。经验值是每 5 到 10 个工具步骤生成一次快照；遇到外部写操作、阶段切换、人工审批等边界，则强制生成快照。

避坑总结：生产环境最常见的 5 个问题

1. 只保存对话，不保存业务状态

常见现象是恢复后模型能看到历史聊天，却不知道哪些文件已处理、哪些工单已创建，于是重复执行。修正方法是把工具结果转成显式 business_state，并让规划器基于状态而不是仅基于自然语言记忆决策。

2. 工具调用后崩溃，恢复时重复扣款或重复发消息

典型错误是“先调用工具，成功后再写检查点”。一旦中间崩溃，系统不知道工具是否已执行。应先持久化 pending_action 与幂等键，调用后再记录结果；恢复时对未知结果执行查询或补偿。

3. 快照中存入连接对象和本地临时路径

反序列化后出现 object is not JSON serializable，或者任务迁移到另一台机器后路径不存在。应保存资源描述符，例如对象存储 key、数据库 ID、文件哈希，并通过资源注册表重新打开。

4. 多执行器同时恢复同一个 run

两个 Worker 同时加载同一检查点，会造成工具重复调用。需要给 run 加租约锁或乐观锁，例如 version 字段更新时执行条件更新；拿不到锁的执行器只能等待或转为观测模式。

5. 检查点结构升级后旧任务无法恢复

直接增加必填字段会让历史快照反序列化失败。必须保留 schema_version，加载后执行迁移函数；删除字段采用宽进严出策略，先双写、再灰度读取、最后清理旧字段。

结语

断点续跑本质上是在为智能体建立可靠的状态机，而不是给聊天记录加一个保存按钮。生产级长任务应围绕 agent-checkpoint、幂等工具、事件日志和周期快照设计。只有当“崩溃在任意一步发生”都成为可预期、可恢复的常规情况，智能体才能真正承担持续数十分钟甚至数小时的业务流程。

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。