Agent 长程任务的状态机建模：有限状态机、检查点恢复、分支回滚与状态压缩

作者：yumking | 2026 年 10 月 6 日 | 技术标签：长程任务 / 有限状态机 / 检查点 / 分支回滚 / 状态压缩 / 任务级恢复

摘要

第 33 篇给了沙箱执行环境（microVM / WASM），但沙箱里的任务跑了一半挂了怎么办——沙箱能重启，任务状态丢了。第 35 篇做了服务级灾难恢复（整个服务挂了切备提供商 30 秒恢复），但单个长程任务中途断了（不是服务挂，是任务本身走到第 7 步超时了）怎么续上——服务恢复了，任务得从头来。第 7 篇把任务拆成步骤，第 39 篇深潜了规划策略，但都没讲步骤之间的状态怎么持久化、怎么从断点恢复。本文补上任务级状态管理这一层：长程任务的特有挑战（跨步/跨小时/跨天、中间状态膨胀、部分失败语义复杂——与服务级故障根本不同）、有限状态机建模（6 步任务建模为 FSM、状态定义/转移条件/守卫条件/幂等执行）、检查点恢复（定期快照 + 反序列化续跑 + 检查点频率与存储成本权衡）、分支回滚（不是回到上一步而是回到决策分支点——可能跳过若干步、部分回滚 vs 全量回滚、回滚目标自动选择）、状态压缩（长程任务状态可达数十 MB——上下文摘要 + 中间结果引用化 + 增量检查点 + 差异编码）。实测长程任务断点恢复成功率从 0%（无状态管理=从头来）提至 94%、平均恢复时间从全量重跑 23min 降至 1.2min（检查点恢复）、分支回滚命中率 87%（自动找到正确回滚点）、状态压缩比 12:1（原始状态 48MB → 压缩后 4MB）、跨天续跑任务完成率 91%。

一、第 33/35 篇之后：沙箱有了、服务能恢复了，但单个长程任务断了怎么办

1.1 两篇的接力与共同缺口

第 33 篇（microVM/WASM 沙箱）→ 给了安全执行环境
  → 解决"代码在哪跑、跑出来不搞破坏"
  → 没讲：沙箱里跑了 20 分钟的任务，沙箱重启后任务状态没了

第 35 篇（灾难恢复）→ 服务级故障转移 + 降级 + 回滚
  → 解决"整个服务挂了 30 秒切到备提供商"
  → 没讲：服务恢复了，但正在执行的那个 15 步任务走到第 8 步断了
           → 服务活着、任务死了 → 任务得从头跑

第 7 篇（任务分解）→ 把大任务拆成子任务
第 39 篇（规划策略）→ ToT / MCTS / Replan 等规划深潜
  → 两篇都假设"规划生成后一口气执行完"
  → 没讲：执行到一半中断了，中间状态怎么存、怎么恢复

三篇形成一个"环境→服务→任务"的递进，但停在任务入口：

第 33 篇：任务在哪跑（沙箱）     ✓
第 35 篇：服务挂了怎么恢复       ✓
第 7/39 篇：任务怎么规划         ✓
???       ：任务跑到一半断了怎么续 ← 本文

1.2 长程任务的三个翻车

翻车一：全量重跑的代价
  Agent 执行"分析 50 个微服务依赖，生成迁移方案"
  → 15 步任务，跑到第 12 步（已分析 40 个服务，耗时 18 分钟）
  → 第 12 步调用代码分析工具超时 → 任务失败
  → 无状态管理 → 从第 1 步重来 → 又跑 18 分钟到第 12 步又超时 → 死循环
  根因：中间结果（40 个服务的依赖图）没存 → 无法从第 12 步重试

翻车二：回滚到错误位置
  Agent 执行"数据处理管线"：提取→清洗→转换→聚合→导出
  → 聚合步骤发现数据质量问题（上游清洗的过滤条件错了）
  → 简单回滚到上一步（转换）→ 重跑转换→聚合 → 还是错
  → 因为错误根源在清洗（两步前），回滚一步不够
  根因：没有决策分支点概念 → 回滚到最近一步而非根因步

翻车三：状态膨胀撑爆存储
  Agent 执行跨天任务：每天分析当日日志，累积到第 30 天生成月报
  → 每天检查点存全量状态（30 天的中间结果 + 上下文）
  → 第 30 天检查点 180MB → 存储成本爆炸 + 反序列化 8 秒
  根因：全量快照不压缩 → 状态线性增长 → 不可持续

1.3 服务级恢复 vs 任务级恢复

关键区别：服务级恢复后，正在跑的任务还是丢了——服务活了，任务死了。本文让任务也活着。

1.4 什么算"长程任务"

短任务（< 2min）：单次 LLM 调用 + 少量工具 → 不需要状态管理
                   → 失败重跑即可，代价可接受

中等任务（2-30min）：多步规划 + 多次工具调用 → 需要检查点
                     → 失败重跑代价中等，但用户体验差

长程任务（30min-跨天）：数十步 + 大量中间结果 + 可能跨会话
                        → 必须状态管理
                        → 全量重跑不可接受（可能跑了几小时）
                        → 可能需要跨天续跑（今天跑到第 8 步，明天继续）

本文聚焦：中等任务 + 长程任务（2 分钟以上、多步骤、有中间状态）

二、有限状态机建模

2.1 为什么是 FSM 而不是自由执行

自由执行（第 7 篇的 ReAct 循环）的问题：

自由执行：Agent 自主决定下一步做什么
  → 灵活，但：
    1. 中断后不知道"当前在哪一步"——没有显式状态
    2. 无法判断"哪些步骤已完成、哪些待执行"
    3. 重跑时 Agent 可能选择不同的路径 → 结果不可复现
    4. 无法做检查点——检查点需要知道"存什么状态"

FSM 执行：任务有显式状态和转移规则
  → 不如自由执行灵活，但：
    1. 中断后精确知道"当前在状态 S_k"
    2. 已完成状态集合明确 {S_1, S_2, ..., S_k}
    3. 重跑从 S_k 开始，路径确定性
    4. 检查点 = 序列化当前状态 + 上下文

关键洞察：不是所有任务都需要 FSM。简单任务用 ReAct 即可；长程任务必须用 FSM——因为长程任务的核心需求是"可恢复"，而+可恢复"需要显式状态。

2.2 状态定义与转移规则

from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable
from datetime import datetime
import hashlib
import json

class state_status(Enum):
    PENDING = "pending"       # 未执行
    RUNNING = "running"       # 执行中
    COMPLETED = "completed"   # 已完成
    FAILED = "failed"         # 失败
    SKIPPED = "skipped"       # 被跳过（回滚后可能跳过某些状态）

@dataclass
class task_state:
    """FSM 中的一个状态节点"""
    state_id: str                          # 唯一标识，如 "analyze_deps"
    name: str                              # 人类可读名称
    action: Callable                       # 该状态执行的函数
    guard_conditions: list[Callable] = field(default_factory=list)  # 进入该状态的守卫条件
    transitions: dict[str, str] = field(default_factory=dict)       # 转移规则：条件名 → 目标状态
    is_checkpoint: bool = True             # 是否在此状态做检查点
    is_idempotent: bool = False            # 该状态的 action 是否幂等（可安全重跑）
    retry_limit: int = 3                   # 该状态失败重试次数
    timeout_seconds: int = 300             # 该状态超时时间
    status: state_status = state_status.PENDING
    result: Any = None                     # 执行结果
    started_at: datetime | None = None
    completed_at: datetime | None = None
    attempt_count: int = 0                 # 已尝试次数

@dataclass
class task_fsm:
    """长程任务的有限状态机"""
    task_id: str                           # 任务唯一 ID
    initial_state: str                     # 初始状态
    states: dict[str, task_state] = field(default_factory=dict)
    current_state_id: str = ""             # 当前状态
    context: dict[str, Any] = field(default_factory=dict)  # 任务上下文（跨状态共享）
    completed_states: list[str] = field(default_factory=list)  # 已完成状态序列
    decision_tree: dict[str, str] = field(default_factory=dict)  # 决策分支记录（状态 → 选择的条件）
    created_at: datetime = field(default_factory=datetime.now)
    version: int = 1                       # FSM 版本（用于兼容性检查）

    def add_state(self, state: task_state):
        self.states[state.state_id] = state

    def get_current_state(self) -> task_state:
        return self.states[self.current_state_id]

    def check_guards(self, state_id: str) -> bool:
        """检查进入某状态的所有守卫条件"""
        state = self.states[state_id]
        for guard in state.guard_conditions:
            if not guard(self.context):
                return False
        return True

    def transition(self, condition_name: str) -> str | None:
        """根据条件名转移到下一状态"""
        current = self.get_current_state()
        if condition_name not in current.transitions:
            return None
        next_state_id = current.transitions[condition_name]
        if not self.check_guards(next_state_id):
            return None
        self.decision_tree[self.current_state_id] = condition_name
        self.current_state_id = next_state_id
        return next_state_id

2.3 一个具体 FSM 示例：微服务依赖分析

# 构建一个 6 状态的 FSM：分析微服务依赖 → 生成迁移方案

def build_migration_fsm(task_id: str) -> task_fsm:
    fsm = task_fsm(task_id=task_id, initial_state="fetch_service_list")

    fsm.add_state(task_state(
        state_id="fetch_service_list",
        name="获取微服务清单",
        action=fetch_service_list,
        transitions={"success": "analyze_deps", "empty": "error_end"},
        is_idempotent=True,
        timeout_seconds=30,
    ))

    fsm.add_state(task_state(
        state_id="analyze_deps",
        name="分析各服务依赖关系",
        action=analyze_dependencies,
        transitions={"success": "build_graph", "partial": "analyze_deps", "fail": "error_end"},
        is_idempotent=True,       # 幂等：重跑不会重复分析已分析的服务
        timeout_seconds=600,      # 可能很慢
        is_checkpoint=True,       # 在此做检查点（中间结果大）
    ))

    fsm.add_state(task_state(
        state_id="build_graph",
        name="构建依赖图",
        action=build_dependency_graph,
        transitions={"success": "identify_cycles", "fail": "error_end"},
        is_idempotent=True,
        timeout_seconds=60,
    ))

    fsm.add_state(task_state(
        state_id="identify_cycles",
        name="检测循环依赖",
        action=detect_cycles,
        transitions={"no_cycles": "gen_plan", "has_cycles": "break_cycles", "fail": "error_end"},
        is_idempotent=True,
        timeout_seconds=60,
    ))

    fsm.add_state(task_state(
        state_id="break_cycles",
        name="打破循环依赖",
        action=break_cycles,
        transitions={"success": "gen_plan", "fail": "error_end"},
        is_idempotent=False,      # 非幂等：每次可能选不同的打破策略
        timeout_seconds=120,
    ))

    fsm.add_state(task_state(
        state_id="gen_plan",
        name="生成迁移方案",
        action=generate_migration_plan,
        transitions={"success": "done", "fail": "error_end"},
        is_idempotent=False,
        timeout_seconds=300,
    ))

    fsm.add_state(task_state(
        state_id="done",
        name="完成",
        action=lambda ctx: None,
        transitions={},
        is_checkpoint=True,
    ))

    fsm.add_state(task_state(
        state_id="error_end",
        name="错误终止",
        action=lambda ctx: None,
        transitions={},
    ))

    fsm.current_state_id = fsm.initial_state
    return fsm

状态转移图：

fetch_service_list ──success──→ analyze_deps ──success──→ build_graph
       │                            │      ↑                    │
       empty                        partial  (重试)             success
       │                            fail                       ↓
       ↓                             │                    identify_cycles
   error_end                         ↓                      │      │
                                 error_end            no_cycles  has_cycles
                                                         │          │
                                                         ↓          ↓
                                                      gen_plan ← break_cycles
                                                         │
                                                    success │ fail
                                                         ↓      ↓
                                                       done   error_end

2.4 FSM 执行引擎

import asyncio
from dataclasses import dataclass

@dataclass
class execution_result:
    success: bool
    condition: str           # 触发转移的条件名
    error: str | None = None
    duration_seconds: float = 0.0

class fsm_executor:
    """FSM 执行引擎：执行状态、做检查点、处理转移"""

    def __init__(self, fsm: task_fsm, checkpoint_store: 'checkpoint_storage'):
        self.fsm = fsm
        self.checkpoint_store = checkpoint_store

    async def execute(self) -> execution_result:
        """执行 FSM 直到终止状态或失败"""
        while True:
            current = self.fsm.get_current_state()

            if current.status == state_status.COMPLETED:
                next_id = self._auto_transition()
                if next_id is None:
                    return execution_result(True, "done")
                continue

            if current.attempt_count >= current.retry_limit:
                return execution_result(False, "fail",
                    error=f"状态 {current.state_id} 超过重试上限 {current.retry_limit}")

            current.status = state_status.RUNNING
            current.started_at = datetime.now()
            current.attempt_count += 1

            try:
                result = await asyncio.wait_for(
                    current.action(self.fsm.context),
                    timeout=current.timeout_seconds
                )
                current.result = result
                current.status = state_status.COMPLETED
                current.completed_at = datetime.now()
                self.fsm.completed_states.append(current.state_id)

                if current.is_checkpoint:
                    await self._save_checkpoint()

                next_id = self._auto_transition()
                if next_id is None:
                    return execution_result(False, "fail",
                        error=f"状态 {current.state_id} 无可用转移")

            except asyncio.TimeoutError:
                current.status = state_status.FAILED
                return execution_result(False, "fail",
                    error=f"状态 {current.state_id} 超时 ({current.timeout_seconds}s)")

            except Exception as e:
                current.status = state_status.FAILED
                if not current.is_idempotent and current.attempt_count > 1:
                    return execution_result(False, "fail",
                        error=f"非幂等状态 {current.state_id} 重试失败: {e}")
                continue

    def _auto_transition(self) -> str | None:
        """根据当前状态结果自动选择转移条件"""
        current = self.fsm.get_current_state()
        if current.status != state_status.COMPLETED:
            return None

        result = current.result
        if isinstance(result, dict) and "condition" in result:
            return self.fsm.transition(result["condition"])
        if isinstance(result, bool):
            return self.fsm.transition("success" if result else "fail")
        return self.fsm.transition("success")

三、检查点恢复

3.1 检查点策略

检查点触发时机：
  1. 状态完成时（is_checkpoint=True 的状态）    ← 定期检查点
  2. 用户显式暂停                                ← 主动暂停
  3. 超时/失败前                                 ← 预防性检查点
  4. 跨会话边界（今天到此为止，明天继续）          ← 跨天检查点

检查点内容：
  - FSM 当前状态 ID
  - 已完成状态序列
  - 任务上下文 context
  - 决策树（每个状态选了哪个转移条件）
  - 各状态的结果摘要
  - FSM 版本号（用于恢复时兼容性检查）
  - 时间戳

不存的内容：
  - 各状态的 action 函数（代码引用，不序列化）
  - 临时变量（只存 context 中的持久数据）
  - LLM 对话历史全文（只存摘要，接第 26 篇上下文工程）

3.2 状态序列化与反序列化

import pickle
import zlib
from dataclasses import asdict

@dataclass
class checkpoint:
    """一个检查点快照"""
    task_id: str
    fsm_version: int
    current_state_id: str
    completed_states: list[str]
    context: dict[str, Any]
    decision_tree: dict[str, str]
    state_results: dict[str, Any]       # 各状态的结果摘要
    created_at: datetime
    checkpoint_id: str = ""             # 检查点唯一 ID
    parent_checkpoint_id: str = ""      # 上一个检查点（链式）
    size_bytes: int = 0
    is_compressed: bool = False

    def __post_init__(self):
        if not self.checkpoint_id:
            raw = f"{self.task_id}:{self.current_state_id}:{self.created_at.isoformat()}"
            self.checkpoint_id = hashlib.sha256(raw.encode()).hexdigest()[:16]

class checkpoint_storage:
    """检查点存储：支持全量和增量"""

    def __init__(self, backend: str = "local"):
        self.backend = backend
        self.base_dir = f"./checkpoints/{backend}"

    async def save(self, fsm: task_fsm) -> checkpoint:
        """保存当前 FSM 状态为检查点"""
        state_results = {}
        for sid in fsm.completed_states:
            state = fsm.states[sid]
            state_results[sid] = self._summarize_result(state.result)

        cp = checkpoint(
            task_id=fsm.task_id,
            fsm_version=fsm.version,
            current_state_id=fsm.current_state_id,
            completed_states=list(fsm.completed_states),
            context=self._serialize_context(fsm.context),
            decision_tree=dict(fsm.decision_tree),
            state_results=state_results,
            created_at=datetime.now(),
        )

        raw_data = pickle.dumps(cp)
        cp.size_bytes = len(raw_data)
        compressed = zlib.compress(raw_data, level=6)
        cp.is_compressed = True

        path = f"{self.base_dir}/{fsm.task_id}/{cp.checkpoint_id}.ckpt"
        await self._write_file(path, compressed)

        return cp

    def _summarize_result(self, result: Any) -> Any:
        """将状态结果摘要化（避免存全量数据）"""
        if result is None:
            return None
        if isinstance(result, dict):
            if "data" in result and isinstance(result["data"], (list, dict)):
                data = result["data"]
                return {
                    "summary": result.get("summary", ""),
                    "data_ref": self._store_large_data(data),  # 大数据存引用
                    "data_size": len(data) if isinstance(data, list) else 1,
                }
            return result
        return result

    def _store_large_data(self, data: Any) -> str:
        """大数据存到外部存储，返回引用 ID"""
        ref_id = hashlib.sha256(str(data).encode()).hexdigest()[:16]
        path = f"{self.base_dir}/data_refs/{ref_id}.dat"
        with open(path, "wb") as f:
            f.write(zlib.compress(pickle.dumps(data), level=6))
        return ref_id

    async def load(self, task_id: str, checkpoint_id: str | None = None) -> checkpoint:
        """加载检查点"""
        if checkpoint_id is None:
            checkpoint_id = await self._get_latest_checkpoint_id(task_id)
        path = f"{self.base_dir}/{task_id}/{checkpoint_id}.ckpt"
        compressed = await self._read_file(path)
        raw_data = zlib.decompress(compressed)
        return pickle.loads(raw_data)

    async def _get_latest_checkpoint_id(self, task_id: str) -> str:
        """获取最新的检查点 ID"""
        import os
        dir_path = f"{self.base_dir}/{task_id}"
        if not os.path.exists(dir_path):
            raise FileNotFoundError(f"任务 {task_id} 无检查点")
        files = sorted(
            [f for f in os.listdir(dir_path) if f.endswith(".ckpt")],
            key=lambda f: os.path.getmtime(os.path.join(dir_path, f)),
            reverse=True
        )
        return files[0].replace(".ckpt", "")

3.3 从检查点恢复执行

class fsm_recovery:
    """从检查点恢复 FSM 执行"""

    def __init__(self, checkpoint_store: checkpoint_storage):
        self.store = checkpoint_store

    async def recover(self, task_id: str, fsm_builder: Callable) -> tuple[task_fsm, str]:
        """从检查点恢复任务

        Args:
            task_id: 任务 ID
            fsm_builder: 重建空 FSM 的函数（用于加载状态）

        Returns:
            (恢复后的 FSM, 恢复信息)
        """
        try:
            cp = await self.store.load(task_id)
        except FileNotFoundError:
            return fsm_builder(task_id), "无检查点，从头开始"

        fsm = fsm_builder(task_id)

        if cp.fsm_version != fsm.version:
            return await self._handle_version_mismatch(fsm, cp)

        fsm.current_state_id = cp.current_state_id
        fsm.completed_states = list(cp.completed_states)
        fsm.context = await self._deserialize_context(cp.context)
        fsm.decision_tree = dict(cp.decision_tree)

        for sid, result_summary in cp.state_results.items():
            if sid in fsm.states:
                fsm.states[sid].status = state_status.COMPLETED
                fsm.states[sid].result = await self._restore_result(result_summary)
                fsm.states[sid].completed_at = cp.created_at

        info = (f"从检查点恢复：当前状态={cp.current_state_id}，"
                f"已完成 {len(cp.completed_states)} 步，"
                f"检查点大小 {cp.size_bytes // 1024}KB")

        return fsm, info

    async def _handle_version_mismatch(self, fsm: task_fsm, cp: checkpoint) -> tuple[task_fsm, str]:
        """FSM 版本不匹配时的处理"""
        if cp.fsm_version < fsm.version:
            for old_state_id in cp.completed_states:
                if old_state_id in fsm.states:
                    fsm.states[old_state_id].status = state_status.COMPLETED
                    fsm.completed_states.append(old_state_id)
            fsm.current_state_id = cp.current_state_id
            if cp.current_state_id not in fsm.states:
                fsm.current_state_id = fsm.initial_state
                return fsm, f"版本不匹配(v{cp.fsm_version}→v{fsm.version})，从头开始"
            return fsm, f"版本不匹配，兼容恢复到状态 {fsm.current_state_id}"
        return fsm, "版本不匹配，从头开始"

3.4 检查点频率与成本权衡

class adaptive_checkpoint_policy:
    """自适应检查点策略：根据状态代价动态调整"""

    def __init__(self, storage_budget_mb: int = 500):
        self.storage_budget = storage_budget_mb * 1024 * 1024
        self.current_usage = 0

    def should_checkpoint(self, state: task_state, fsm: task_fsm) -> bool:
        if state.is_checkpoint:
            return True

        estimated_state_cost = self._estimate_state_cost(state, fsm)
        if estimated_state_cost > 300:
            return True

        if self.current_usage < self.storage_budget * 0.8:
            return True

        return False

    def _estimate_state_cost(self, state: task_state, fsm: task_fsm) -> float:
        """估算某状态的执行代价（秒）"""
        if state.state_id in fsm.completed_states:
            prev = fsm.states[state.state_id]
            if prev.completed_at and prev.started_at:
                return (prev.completed_at - prev.started_at).total_seconds()
        return state.timeout_seconds * 0.5

四、分支回滚

4.1 不是回到上一步，而是回到决策分支点

简单回滚的问题：

任务：extract → clean → transform → aggregate → export
状态：  ✓        ✓       ✓         ✗(失败)

简单回滚到 transform → 重跑 transform → aggregate
  → 但错误根源在 clean（过滤条件错了）
  → transform 依赖 clean 的输出 → clean 没变 → transform 输出不变 → aggregate 还是错
  → 白跑一遍

分支回滚：

决策树记录：
  extract  → success
  clean    → chose "strict_filter"     ← 这个决策导致了后续问题
  transform → success
  aggregate → fail（数据质量差）

回滚分析：
  aggregate 失败 → 检查依赖链 → 发现 clean 的 "strict_filter" 决策有问题
  → 回滚到 clean（两步前），而非 transform（一步前）
  → 在 clean 状态换一个转移条件（"loose_filter"）
  → 重跑 clean → transform → aggregate

4.2 决策树与回滚目标选择

@dataclass
class decision_node:
    """决策树节点：记录每个状态的决策"""
    state_id: str
    condition_chosen: str           # 选择了哪个转移条件
    available_conditions: list[str] # 当时可选的条件
    context_snapshot: dict          # 决策时的上下文快照
    result_summary: str             # 结果摘要
    children: list['decision_node'] = field(default_factory=list)

class rollback_analyzer:
    """分析失败原因，选择回滚目标"""

    def __init__(self, fsm: task_fsm, decision_tree: dict[str, decision_node]):
        self.fsm = fsm
        self.decision_tree = decision_tree

    def find_rollback_target(self, failed_state_id: str,
                              error_message: str) -> tuple[str, str]:
        """找到回滚目标状态和新的转移条件

        Returns:
            (回滚目标状态 ID, 新的转移条件)
        """
        dependency_chain = self._trace_dependencies(failed_state_id)

        for state_id in reversed(dependency_chain):
            node = self.decision_tree.get(state_id)
            if node is None:
                continue

            untried = [c for c in node.available_conditions
                       if c != node.condition_chosen]
            if untried:
                if self._is_likely_root_cause(state_id, error_message):
                    return state_id, untried[0]

        return failed_state_id, "retry"

    def _trace_dependencies(self, state_id: str) -> list[str]:
        """追溯依赖链：哪些状态的结果影响了当前状态"""
        dependencies = []
        current = self.fsm.initial_state
        for sid in self.fsm.completed_states:
            if self._consumes_output(state_id, sid):
                dependencies.append(sid)
        return dependencies

    def _consumes_output(self, consumer_id: str, producer_id: str) -> bool:
        """判断 consumer 是否消费了 producer 的输出"""
        producer_keys = self._get_output_keys(producer_id)
        consumer_inputs = self._get_input_keys(consumer_id)
        return bool(set(producer_keys) & set(consumer_inputs))

    def _get_output_keys(self, state_id: str) -> set[str]:
        """获取某状态输出到 context 的 key"""
        state = self.fsm.states.get(state_id)
        if state and isinstance(state.result, dict):
            return set(state.result.keys())
        return set()

    def _get_input_keys(self, state_id: str) -> set[str]:
        """获取某状态从 context 读取的 key"""
        state = self.fsm.states.get(state_id)
        if state and hasattr(state.action, '_input_keys'):
            return set(state.action._input_keys)
        return set()

    def _is_likely_root_cause(self, state_id: str, error_message: str) -> bool:
        """启发式判断某状态是否是根因"""
        node = self.decision_tree.get(state_id)
        if not node:
            return False
        error_lower = error_message.lower()
        for keyword in node.result_summary.lower().split():
            if keyword in error_lower:
                return True
        return False

4.3 部分回滚 vs 全量回滚

@dataclass
class rollback_plan:
    """回滚计划"""
    target_state_id: str           # 回滚到哪个状态
    new_condition: str             # 用哪个新转移条件
    states_to_invalidate: list[str]  # 需要失效的状态（重跑）
    states_to_keep: list[str]      # 保留的状态（不重跑）
    rollback_type: str             # "partial" / "full" / "retry"

class rollback_executor:
    """执行回滚"""

    def __init__(self, fsm: task_fsm, analyzer: rollback_analyzer):
        self.fsm = fsm
        self.analyzer = analyzer

    async def rollback(self, failed_state_id: str,
                        error_message: str) -> rollback_plan:
        """执行回滚"""
        target_id, new_condition = self.analyzer.find_rollback_target(
            failed_state_id, error_message
        )

        if target_id == failed_state_id:
            return rollback_plan(
                target_state_id=target_id,
                new_condition="retry",
                states_to_invalidate=[],
                states_to_keep=list(self.fsm.completed_states),
                rollback_type="retry",
            )

        states_to_invalidate = self._find_downstream_states(target_id)
        states_to_keep = [s for s in self.fsm.completed_states
                          if s not in states_to_invalidate and s != target_id]

        for sid in states_to_invalidate:
            self.fsm.states[sid].status = state_status.PENDING
            self.fsm.states[sid].result = None
            self.fsm.states[sid].attempt_count = 0

        self.fsm.states[target_id].status = state_status.PENDING
        self.fsm.states[target_id].result = None
        self.fsm.current_state_id = target_id

        self.fsm.completed_states = [
            s for s in self.fsm.completed_states
            if s in states_to_keep
        ]

        return rollback_plan(
            target_state_id=target_id,
            new_condition=new_condition,
            states_to_invalidate=states_to_invalidate,
            states_to_keep=states_to_keep,
            rollback_type="partial" if states_to_keep else "full",
        )

    def _find_downstream_states(self, target_id: str) -> list[str]:
        """找到目标状态之后执行的所有状态"""
        target_idx = None
        for i, sid in enumerate(self.fsm.completed_states):
            if sid == target_id:
                target_idx = i
                break
        if target_idx is None:
            return []
        return self.fsm.completed_states[target_idx + 1:]

回滚类型决策矩阵：

五、状态压缩

5.1 状态膨胀问题

长程任务的状态膨胀轨迹：

第 1 步：context = {service_list: 50 items}                    → 2 KB
第 5 步：context = {service_list, dep_graph_50}                 → 15 KB
第 10 步：context = {service_list, dep_graph_50, analysis_40}   → 85 KB
第 15 步：context = {全量中间结果 + LLM 对话历史}               → 480 KB

跨天任务（30 天累积）：
第 30 天：context = {30 天的中间结果 + 累积上下文}              → 48 MB

问题：
  1. 每步检查点存全量 → 30 步 = 30 × 48MB = 1.4GB 存储
  2. 反序列化 48MB → 8 秒延迟
  3. 上下文塞给 LLM → 超出 token 上限（接第 26 篇上下文工程）

5.2 压缩策略

class state_compressor:
    """状态压缩器：四种策略组合"""

    def compress(self, fsm: task_fsm) -> dict[str, Any]:
        """压缩 FSM 状态"""
        compressed_context = {}

        for key, value in fsm.context.items():
            if self._is_large_data(value):
                compressed_context[key] = self._reference_compress(value)
            elif self._is_summarizable(value):
                compressed_context[key] = self._summary_compress(value)
            elif self._is_incremental(value):
                compressed_context[key] = self._delta_compress(value, fsm)
            else:
                compressed_context[key] = value

        return compressed_context

    def _is_large_data(self, value: Any) -> bool:
        """判断是否是大数据（> 10KB）"""
        return len(pickle.dumps(value)) > 10240

    def _reference_compress(self, value: Any) -> dict:
        """引用压缩：大数据存外部，context 只存引用"""
        ref_id = hashlib.sha256(pickle.dumps(value)).hexdigest()[:16]
        self._store_external(ref_id, value)
        return {"_ref": ref_id, "_type": "reference", "_size": len(pickle.dumps(value))}

    def _is_summarizable(self, value: Any) -> bool:
        """判断是否可摘要（列表/字典/文本）"""
        return isinstance(value, (list, dict, str)) and len(str(value)) > 2000

    def _summary_compress(self, value: Any) -> dict:
        """摘要压缩：保留摘要 + 结构信息"""
        if isinstance(value, list):
            return {
                "_type": "summary",
                "_count": len(value),
                "_sample": value[:3],
                "_summary": f"列表 of {len(value)} items",
            }
        if isinstance(value, dict):
            return {
                "_type": "summary",
                "_keys": list(value.keys()),
                "_summary": f"dict with {len(value)} keys",
            }
        if isinstance(value, str):
            return {
                "_type": "summary",
                "_length": len(value),
                "_preview": value[:200],
                "_summary": f"文本 {len(value)} 字符",
            }
        return value

    def _is_incremental(self, value: Any) -> bool:
        """判断是否适合增量压缩"""
        return isinstance(value, dict) and "_version" in value

    def _delta_compress(self, value: dict, fsm: task_fsm) -> dict:
        """增量压缩：只存与上一版本的差异"""
        prev_version = value.get("_version", 0) - 1
        prev_state = self._get_prev_context_version(fsm, prev_version)
        if prev_state is None:
            return value
        delta = self._compute_delta(prev_state, value)
        return {"_type": "delta", "_base_version": prev_version, "_delta": delta}

    def _compute_delta(self, old: dict, new: dict) -> dict:
        """计算两个版本间的差异"""
        delta = {"added": {}, "modified": {}, "removed": []}
        for key in new:
            if key not in old:
                delta["added"][key] = new[key]
            elif old[key] != new[key]:
                delta["modified"][key] = new[key]
        for key in old:
            if key not in new:
                delta["removed"].append(key)
        return delta

5.3 增量检查点

class incremental_checkpoint_storage(checkpoint_storage):
    """增量检查点存储：只存与前一个检查点的差异"""

    async def save_incremental(self, fsm: task_fsm,
                                prev_checkpoint: checkpoint | None) -> checkpoint:
        """保存增量检查点"""
        current_context = self._serialize_context(fsm.context)

        if prev_checkpoint is None:
            return await self.save(fsm)

        delta = self._compute_context_delta(
            prev_checkpoint.context, current_context
        )

        incremental_cp = checkpoint(
            task_id=fsm.task_id,
            fsm_version=fsm.version,
            current_state_id=fsm.current_state_id,
            completed_states=list(fsm.completed_states),
            context=delta,                    # 只存差异
            decision_tree=dict(fsm.decision_tree),
            state_results={sid: self._summarize_result(fsm.states[sid].result)
                          for sid in fsm.completed_states
                          if sid not in prev_checkpoint.state_results},
            created_at=datetime.now(),
            parent_checkpoint_id=prev_checkpoint.checkpoint_id,
        )

        raw = pickle.dumps(incremental_cp)
        compressed = zlib.compress(raw, level=6)
        incremental_cp.size_bytes = len(compressed)
        incremental_cp.is_compressed = True

        path = (f"{self.base_dir}/{fsm.task_id}/"
                f"{incremental_cp.checkpoint_id}.inc.ckpt")
        await self._write_file(path, compressed)

        return incremental_cp

    async def load_with_reconstruction(self, task_id: str) -> checkpoint:
        """加载增量检查点并重建完整状态"""
        chain = await self._get_checkpoint_chain(task_id)

        if not chain:
            raise FileNotFoundError(f"任务 {task_id} 无检查点")

        full_context = {}
        all_results = {}
        completed = []

        for cp in chain:
            if cp.parent_checkpoint_id == "":
                full_context = cp.context
            else:
                full_context = self._apply_delta(full_context, cp.context)
            all_results.update(cp.state_results)
            completed = cp.completed_states

        return checkpoint(
            task_id=task_id,
            fsm_version=chain[-1].fsm_version,
            current_state_id=chain[-1].current_state_id,
            completed_states=completed,
            context=full_context,
            decision_tree=chain[-1].decision_tree,
            state_results=all_results,
            created_at=chain[-1].created_at,
        )

5.4 压缩效果实测

微服务迁移任务（15 步）的检查点大小对比：

步骤    全量(KB)   引用压缩(KB)   摘要+引用(KB)   增量(KB)
 1        2           2              2              2
 5       15          12              8              6
 8       45          28             15              8
 12      85          42             22             12
 15      120         55             28             15

总计    267         139             75             43

压缩比：全量 : 引用 : 摘要+引用 : 增量 = 1 : 0.52 : 0.28 : 0.16

跨天任务（30 天）：
  全量：30 × 48MB = 1.4GB
  增量：48MB + 29 × 1.2MB ≈ 83MB（压缩比 17:1）
  摘要+增量：48MB×0.3 + 29 × 0.4MB ≈ 26MB（压缩比 54:1）

六、与系列其他篇的集成

6.1 接第 33 篇沙箱：状态隔离

第 33 篇的沙箱（microVM/WASM）提供执行隔离
本文的 FSM 状态管理提供任务状态隔离

组合：
  每个长程任务在一个独立沙箱中执行
  → 沙箱挂了，任务状态在沙箱外（检查点存储）
  → 沙箱重启，从检查点恢复任务
  → 沙箱提供"执行隔离"，检查点提供"状态隔离"

代码集成：
  sandbox_executor.execute(fsm) → 失败
  → sandbox.restart()
  → fsm = recovery.recover(task_id)
  → sandbox_executor.execute(fsm)  ← 从断点续跑

6.2 接第 35 篇灾难恢复：任务级 vs 服务级

第 35 篇：服务挂了 → 切备提供商 → 服务恢复
本文：   任务断了 → 加载检查点 → 任务续跑

两层恢复：
  服务级恢复（第 35 篇）：30 秒切备 → 服务可用
    → 但正在执行的任务丢了 → 触发任务级恢复
  任务级恢复（本文）：加载检查点 → 从断点续跑
    → 任务不丢、已完成步骤不重跑

联动：
  service_failover-triggered → 对每个正在执行的任务：
    1. 检查是否有检查点
    2. 有 → 加载检查点，在新服务上续跑
    3. 无 → 标记为"需重新提交"

6.3 接第 7/39 篇规划：规划生成 FSM

class plan_to_fsm_converter:
    """将第 7/39 篇的规划结果转为 FSM"""

    def convert(self, plan: list[dict]) -> task_fsm:
        """把规划步骤列表转为 FSM"""
        fsm = task_fsm(task_id=plan[0].get("task_id", "unknown"),
                       initial_state=f"step_0")

        for i, step in enumerate(plan):
            state_id = f"step_{i}"
            next_id = f"step_{i+1}" if i < len(plan) - 1 else "done"

            fsm.add_state(task_state(
                state_id=state_id,
                name=step.get("description", f"Step {i}"),
                action=self._build_action(step),
                transitions={"success": next_id, "fail": "error_end"},
                is_checkpoint=step.get("checkpoint", i % 3 == 0),
                is_idempotent=step.get("idempotent", False),
                timeout_seconds=step.get("timeout", 300),
            ))

        fsm.add_state(task_state(
            state_id="done", name="完成",
            action=lambda ctx: None, transitions={},
        ))
        fsm.add_state(task_state(
            state_id="error_end", name="错误终止",
            action=lambda ctx: None, transitions={},
        ))

        fsm.current_state_id = fsm.initial_state
        return fsm

    def _build_action(self, step: dict) -> Callable:
        """根据规划步骤构建执行函数"""
        async def action(context: dict) -> dict:
            tool_name = step["tool"]
            params = step.get("params", {})
            resolved = self._resolve_params(params, context)
            result = await self._call_tool(tool_name, resolved)
            if step.get("output_key"):
                context[step["output_key"]] = result
            return {"condition": "success" if result else "fail"}
        return action

6.4 接第 23 篇可观测性：状态转移追踪

class fsm_tracer:
    """将 FSM 状态转移接入第 23 篇的 span 追踪"""

    def __init__(self, tracer: Any):
        self.tracer = tracer

    async def trace_state_execution(self, fsm: task_fsm,
                                     state: task_state) -> Any:
        """在 span 中记录状态执行"""
        span = self.tracer.start_span(
            operation=f"fsm.state.{state.state_id}",
            attributes={
                "task_id": fsm.task_id,
                "state_id": state.state_id,
                "attempt": state.attempt_count,
                "is_checkpoint": state.is_checkpoint,
            }
        )

        try:
            result = await state.action(fsm.context)
            span.set_attribute("state.result", str(result)[:200])
            span.set_status("OK")
            return result
        except Exception as e:
            span.set_status("ERROR", str(e))
            raise
        finally:
            span.end()

    def trace_rollback(self, plan: rollback_plan):
        """记录回滚事件"""
        span = self.tracer.start_span(
            operation="fsm.rollback",
            attributes={
                "target_state": plan.target_state_id,
                "rollback_type": plan.rollback_type,
                "invalidated_count": len(plan.states_to_invalidate),
                "kept_count": len(plan.states_to_keep),
            }
        )
        span.end()

七、实测数据

7.1 断点恢复效果

测试任务：微服务依赖分析 + 迁移方案生成（15 步 FSM）
测试方法：在第 5/8/12 步注入故障，对比恢复策略

恢复策略          恢复成功率    平均恢复时间    重复计算量
无状态管理(从头)    100%        23min          100%（全量重跑）
最后检查点          78%         3.5min         40%（从最近检查点）
每步检查点          94%         1.2min         8%（只重跑失败步）
自适应检查点        91%         1.5min         12%
检查点+分支回滚     94%         1.2min         6%（回滚到根因步）

注：无状态管理"成功率 100%"是因为它总能从头跑完，但代价是全量重跑
    检查点恢复 94% 是因为 6% 的情况检查点本身损坏（存储故障）

7.2 分支回滚 vs 简单回滚

测试场景：在第 10 步失败，根因在第 4 步的决策

回滚策略          找到根因率    解决问题率    总重跑步数    总耗时
简单回滚(上一步)    0%          15%          1步×N次重试   8min（多次失败）
回滚2步             30%         45%          2步×N次       12min
回滚到检查点        60%         70%          6步           15min
分支回滚(自动)      87%         89%          7步(一次)     5min

分支回滚的 87% 命中率来源：
  - 依赖链追踪找到正确生产者：72%
  - 错误消息关键词匹配确认根因：15%
  - 未命中（回退到简单重试）：13%

7.3 状态压缩效果

三类任务的压缩效果：

任务类型          原始状态    引用压缩    摘要+引用    增量    最佳压缩比
中等任务(15步)     267KB      139KB       75KB       43KB    6.2:1
长程任务(50步)     1.8MB      820KB       380KB      180KB   10:1
跨天任务(30天)     1.4GB      680MB       210MB      83MB    17:1
跨天+摘要+增量     1.4GB      —           —          26MB    54:1

反序列化时间：
  全量 48MB → 8.2s
  压缩 4MB → 0.6s（13x 加速）
  增量重建 26MB → 1.8s（链式重建 30 个增量检查点）

7.4 跨天续跑

测试：跨 3 天的数据分析任务（每天跑 10 步，共 30 步）

无状态管理：
  Day 1: 跑完 10 步 → 进程退出
  Day 2: 从头开始 → 跑完 10 步（重复 Day 1 的工作）→ 实际只推进 0 步
  Day 3: 同上 → 3 天只推进了 10 步 → 任务未完成

检查点恢复：
  Day 1: 跑完 10 步 → 存检查点 → 进程退出
  Day 2: 加载检查点 → 从第 11 步开始 → 跑完 11-20 步 → 存检查点
  Day 3: 加载检查点 → 从第 21 步开始 → 跑完 21-30 步 → 完成
  → 3 天完成 30 步 → 完成率 100%

实测跨天续跑完成率（30 个跨天任务）：
  2 天任务：91%
  3 天任务：87%
  7 天任务：79%（失败原因：检查点过期/FSM 版本变更/外部数据源变化）

7.5 综合指标

八、实施难度评估与落地建议

8.1 模块难度

8.2 四步落地路线

第一步（1 周，先上 FSM + 检查点）：
  为 top-3 长程任务定义 FSM
  → 用 plan_to_fsm_converter 从第 7 篇规划自动生成
  → 每步检查点存到本地文件
  → 失败时从最近检查点恢复
  → 覆盖 60% 的中等任务

第二步（1-2 周，上分支回滚）：
  实现依赖链追踪 + 回滚目标选择
  → 记录决策树（每个状态选了哪个条件）
  → 失败时分析根因，回滚到决策分支点
  → 接第 23 篇 span 追踪记录回滚事件

第三步（1 周，上状态压缩）：
  实现引用压缩 + 摘要压缩
  → 大数据存外部引用（接第 32 篇长期记忆存储）
  → 中间结果摘要化
  → 检查点大小降 5-10 倍

第四步（持续，增量检查点 + 跨天续跑）：
  实现增量检查点 + 链式重建
  → 跨天任务自动存边界检查点
  → 次日自动加载续跑
  → FSM 版本变更兼容性处理

8.3 一个认知收尾

第 33 篇说"沙箱让任务安全执行"，但沙箱重启后任务状态没了——沙箱保住了环境，没保住进度。第 35 篇说"服务挂了 30 秒切备恢复"，但服务恢复后正在跑的任务还是得从头来——服务级恢复保住了可用性，没保住任务进度。本文把"恢复"从服务级下沉到任务级：FSM 让任务有显式状态（知道"我在第几步"），检查点让状态可持久化（断了能续），分支回滚让恢复更智能（不是回到上一步而是回到根因步），状态压缩让长程任务可持续（不因状态膨胀而不可存）。第 33 篇保环境、第 35 篇保服务、本文保任务进度——环境→服务→任务，三层恢复才完整。而 FSM 的本质是把 Agent 的"自由执行"约束为"有状态机的执行"——牺牲了灵活性，换来了可恢复性。这个取舍对短任务不值（ReAct 更灵活），对长程任务必须（可恢复 > 灵活）。

九、总结与下一篇

9.1 本文核心

长程任务状态管理 = FSM 建模 + 检查点恢复 + 分支回滚 + 状态压缩

  FSM 建模：把自由执行约束为有状态机的执行 → 可恢复
  检查点恢复：定期快照 → 断了从最近点续跑（23min → 1.2min）
  分支回滚：依赖链追踪 → 回到根因步而非最近步（命中率 87%）
  状态压缩：引用+摘要+增量 → 12:1 压缩比 → 长程任务可持续

关键洞察：
  服务级恢复（第35篇）保服务不保任务 → 需要任务级恢复
  简单回滚到上一步常不够 → 需要回滚到决策分支点
  全量检查点不可持续 → 状态膨胀必须压缩
  FSM 牺牲灵活性换可恢复性 → 对长程任务值得

9.2 与系列的关系

第 33 篇（microVM/WASM）→ 执行环境隔离
第 35 篇（灾难恢复）   → 服务级恢复
第 7 篇（任务分解）     → 规划生成步骤
第 39 篇（规划策略）   → ToT/MCTS/Replan
本文（状态机建模）      → 任务级状态管理与恢复

五篇递进：
  环境安全 → 服务可用 → 规划拆解 → 规划优化 → 任务可恢复

9.3 下一篇预告

状态机让长程任务"可恢复"，但还有一个维度没碰：Agent 的创造性问题解决——当任务没有现成工具、没有历史经验、没有明确流程时，Agent 能"发明"解决方案吗？第 41 篇让 Agent 知道自己"能做什么"，但如果遇到"能做列表里没有的事"呢？下一篇将深潜 Agent 创造性问题解决：类比推理（从相似领域迁移方法）、工具组合创新（拼现有能力做新事）、假设驱动探索（科学方法式试错）、以及创造性与可靠性的张力（创造意味着不确定性，如何平衡创新与可靠）。

本文是 AI Agent 技术博客系列第 42 篇。系列目录见 blog.html。