Workflow 系列（03）：状态管理中的持久化、幂等性与版本绑定

在AI工作流执行过程中，若运行至第五阶段时发生崩溃，重启后究竟该从头执行还是从第六阶段继续？这正是状态管理需要解决的核心矛盾。

为什么状态管理是核心问题

若缺少状态持久化机制，只能被迫重新开始，先前所有的LLM调用、工具执行及人工确认步骤都将彻底作废。状态管理的关键作用在于：当工作流在任意时刻意外中断时，它能从最近的检查点实现精确续接。这也是构建人工确认门机制的基础——当门被触发，工作流会暂停于某个确定状态，待人工响应后从该状态继续流转。

Durable Execution 模式

该模式的核心思想是将执行过程序列化为可恢复的检查点。无论何时中断，都能从最近的检查点重新执行，且最终结果与从未中断完全一致。Temporal.io 作为代码层面的典型实现，但使用简单的 JSON 文件同样能达成相同的语义效果。

状态文件结构

 复制代码{
  "workflow_id": "wf-bug-e2e-AE-33995-20260601",
  "workflow_version": "1.3.0",
  "jira_key": "AE-33995",
  "started_at": "2026-06-01T10:00:00+08:00",
  "phase": "phase_4",
  "phases": {
    "phase_1": {
      "status": "done",
      "completed_at": "2026-06-01T10:02:30+08:00",
      "output_file": "bug_info.json"
    },
    "phase_2": {
      "status": "done",
      "completed_at": "2026-06-01T10:05:00+08:00",
      "output_file": "log_analysis.json"
    },
    "phase_3": {
      "status": "done",
      "completed_at": "2026-06-01T10:18:00+08:00",
      "output_file": "analysis_final.json"
    },
    "phase_4": {
      "status": "in_progress",
      "step": "step_4_1",
      "steps": {
        "step_4_1": {"status": "done", "output_file": "candidate_a.json"},
        "step_4_2": {"status": "in_progress"},
        "step_4_3": {"status": "pending"}
      }
    }
  }
}

续接协议

 复制代码def resume_workflow(state_file: Path) -> None:
    state = json.loads(state_file.read_text())
    for phase_id, phase_data in state["phases"].items():
        if phase_data["status"] == "done":
            continue  # 跳过已完成的 Phase
        if phase_data["status"] == "in_progress":
            # in_progress 视同 pending 重新执行（幂等性保证安全）
            execute_phase(phase_id, state)
            return
        if phase_data["status"] == "pending":
            execute_phase(phase_id, state)
            return

核心设计原则：续接时仅信任状态文件，而不依赖任何内存记录。主 Agent 自身并不记忆已执行的操作，仅依据状态文件中的 status 字段做出决策。对于状态标记为 in_progress 的 Phase，系统会重新执行，这要求每个 Phase 的操作都必须具备幂等性。

两端写入

状态文件的写入必须在 Phase 开始前和完成后都进行，不能仅在其完成时写入：

 复制代码def execute_phase(phase_id: str, state: dict) -> None:
    # 开始前：标记 in_progress（即使崩溃，续接时也能找到这个 Phase）
    state["phases"][phase_id]["status"] = "in_progress"
    write_state(state)
    try:
        result = run_phase_logic(phase_id, state)
        # 完成后：标记 done，记录输出文件路径
        state["phases"][phase_id]["status"] = "done"
        state["phases"][phase_id]["output_file"] = result.output_file
        write_state(state)
    except Exception as e:
        state["phases"][phase_id]["status"] = "failed"
        state["phases"][phase_id]["error"] = str(e)
        write_state(state)
        raise

幂等性设计

续接协议中提及“in_progress 视同 pending 重新执行”，这意味着一个 Phase 可能会被执行两次。如果该 Phase 内的操作不具备幂等性，就会产生重复的副作用，例如两条 Jira 评论、两个 git commit 或两封通知邮件。

各类操作的幂等性分析

文件写入（天然幂等）

 复制代码# 覆盖写是幂等的——多次执行结果相同
output_file.write_text(json.dumps(result))  # 

Jira 评论（不幂等，需要检测）

 复制代码#  错误：直接写评论，重复执行会写两条
jira.add_comment(issue_key, comment_text)#  正确：写前先检查是否已有本次处理的评论
def add_comment_idempotent(issue_key: str, comment_text: str, run_id: str) -> None:
    existing = jira.get_comments(issue_key)
    marker = f"[run_id:{run_id}]"  # 每次工作流运行用唯一 ID 标记
    if any(marker in c.body for c in existing):
        return  # 已写过，跳过
    jira.add_comment(issue_key, f"{marker}n{comment_text}")

Git commit（不幂等，需要检测）

 复制代码#  错误：直接提交，重复执行会创建两个 commit
git.commit(message)#  正确：提交前检查 commit result 文件是否已存在且 passed=true
def commit_idempotent(message: str, output_file: Path) -> dict:
    if output_file.exists():
        result = json.loads(output_file.read_text())
        if result.get("passed"):
            return result  # 已提交成功，直接返回之前的结果
    # 执行提交
    commit_sha = git.commit(message)
    result = {"passed": True, "sha": commit_sha}
    output_file.write_text(json.dumps(result))
    return result

外部 API 触发（条件幂等）

 复制代码# Gerrit 添加 Reviewer：重复添加不报错，天然幂等 
gerrit.add_reviewer(change_id, reviewer)# 创建定时任务：重复创建会出现重复任务 
# → 修复：创建前先 list，检查是否已存在
def create_cron_idempotent(job_config: dict) -> None:
    existing_jobs = cron.list_jobs()
    if any(j["name"] == job_config["name"] for j in existing_jobs):
        return  # 已存在，跳过
    cron.create_job(job_config)

幂等性自检表

每个新增的 Step，在实现前都需要回答以下三个问题：

 复制代码□ 这个 Step 如果执行两次，会产生副作用吗？
□ 如果会，如何检测"已执行过"并跳过？
□ 检测逻辑本身是否也是幂等的？

最后一个问题最容易被忽略。如果检测逻辑依赖于内存状态或本身也产生副作用，那么在续接场景下同样会导致失效。

状态文件版本绑定

当工作流运行至中途时，如果工作流定义发生了变更（例如新增了一个 Step），续接操作便会出现不一致：旧的状态文件中没有新 Step 的记录，主 Agent 将无法确定如何正确处理。解决方法是，在状态文件中绑定工作流版本号，并在续接时校验版本一致性。

 复制代码def start_or_resume(state_file: Path, current_version: str) -> dict:
    if state_file.exists():
        state = json.loads(state_file.read_text())
        saved_version = state.get("workflow_version")
        if saved_version != current_version:
            # 版本不一致，告知用户选择
            raise WorkflowVersionMismatch(
                f"State file version: {saved_version}n"
                f"Current workflow version: {current_version}n"
                f"Options:n"
                f"  1. Continue with saved state using old workflow ({saved_version})n"
                f"  2. Start fresh with new workflow ({current_version})n"
                f"  3. Manually migrate state file"
            )
        return state  # 版本一致，正常续接
    # 新建状态文件
    state = {
        "workflow_version": current_version,
        "started_at": datetime.now(timezone.utc).isoformat(),
        "phases": {}
    }
    write_state(state, state_file)
    return state

版本号规则（MAJOR.MINOR.PATCH）

 复制代码MAJOR：Phase 结构变化（新增/删除 Phase，路由逻辑重大变更）
        → 对正在运行的工作流有断裂风险，必须处理版本不一致
        → 一般不可直接续接，需要人工决策MINOR：新增 Step、模板优化、新增确认门选项
        → 向后兼容，正在运行的工作流可以继续用旧版本完成
        → 新触发的工作流用新版本PATCH：措辞优化、配置调整、不影响行为
        → 安全更新，旧状态文件可以无感升级

设计 Checklist

状态持久化

1. 每个 Phase/Step 在开始前写入 in_progress，完成后写入 done
2. 续接协议只读状态文件，不依赖主 Agent 的对话历史
3. 状态文件包含 workflow_version 字段

幂等性

1. 所有外部写操作（Jira 评论、git commit、API 调用）都经过幂等性检查
2. 幂等性检测用唯一标识符（run_id 或 output_file 存在性）
3. 检测逻辑本身不产生副作用

版本绑定

1. 续接时校验状态文件版本与当前工作流版本是否一致
2. MAJOR 版本变更有明确的续接策略
3. 版本不一致时给用户提供可操作的选项，不是直接报错退出

总结

1. Durable Execution 的关键是两端写入：Phase 开始前写 in_progress，完成后写 done，崩溃在任何位置都能精确续接
2. 续接要求幂等性：in_progress 会被重新执行意味着每个外部写操作都要能安全执行两次，文件写入天然幂等，Jira 评论和 git commit 需要显式检测
3. 版本绑定防止静默错误：工作流修改后，旧状态文件和新工作流版本不匹配时应该告知用户，而不是用新逻辑处理旧状态，产生难以排查的错误