用状态机编排任务

学习目标

用状态和转移定义一个多步任务。
保存执行快照，并从快照恢复。
区分安全重试与可能重复产生外部影响的操作。

前置要求：工具调用、Agent 循环与 JSON。本节用 Python 标准库写一个固定工作流，不需要编排框架。

把自由度放在真正需要的位置

不是所有任务都需要模型选择下一步。研究简报可以先固定为“收集问题 → 检索 → 整理证据 → 审核 → 交付”。模型只在整理证据等需要语言理解的节点工作，路线由程序控制。

状态机明确回答三个问题：现在在哪一步，进入下一步需要什么，失败后去哪里。相比一串互相调用但没有记录的函数，它更容易恢复、观察和解释。

先设计状态

{
  "schema_version": 1,
  "run_id": "demo-001",
  "question": "报销有哪些要求？",
  "stage": "retrieve",
  "evidence": [],
  "draft": null,
  "errors": []
}

阶段名应稳定、明确，不使用含义模糊的数字。每个节点只修改自己负责的字段；输入不满足条件时明确失败。状态结构变化后用版本号处理旧快照，避免旧数据被新代码误读。

运行一个能恢复的工作流

保存为 workflow.py 并运行。示例会在当前目录创建 workflow-checkpoint.json，第二次运行会读取已完成状态。

import json
from pathlib import Path

CHECKPOINT = Path("workflow-checkpoint.json")

def save(state):
    temporary = CHECKPOINT.with_suffix(".tmp")
    temporary.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8")
    temporary.replace(CHECKPOINT)

def advance(state):
    if state["stage"] == "retrieve":
        state["evidence"] = [{"id": "expense-v2", "text": "报销须附有效发票。"}]
        state["stage"] = "draft"
    elif state["stage"] == "draft":
        if not state["evidence"]:
            state.update(stage="failed", error="no_evidence")
        else:
            state["draft"] = "\n".join(
                f'[{item["id"]}] {item["text"]}' for item in state["evidence"]
            )
            state["stage"] = "review"
    elif state["stage"] == "review":
        # 本例是格式检查节点，不代表真实人工审核。
        state["review"] = {"citation_present": "[expense-v2]" in state["draft"]}
        state["stage"] = "completed" if state["review"]["citation_present"] else "failed"
    else:
        raise ValueError("无法推进当前阶段")
    return state

if CHECKPOINT.exists():
    state = json.loads(CHECKPOINT.read_text(encoding="utf-8"))
    if state.get("schema_version") != 1:
        raise ValueError("不支持的快照版本")
else:
    state = {
        "schema_version": 1, "run_id": "demo-001",
        "question": "报销有哪些要求？", "stage": "retrieve", "evidence": [],
    }

while state["stage"] not in ("completed", "failed"):
    state = advance(state)
    save(state)
    print("已保存阶段：", state["stage"])
print(json.dumps(state, ensure_ascii=False, indent=2))

临时文件替换减少了写到一半留下不完整 JSON 的风险。它不能代替生产数据库事务，也没有解决多个进程同时更新同一任务的问题。本例只适合单进程本地实验。

重试为什么可能危险

假设节点发送邮件成功，但保存“已发送”状态时崩溃。恢复后重复执行这个节点，就可能发送第二封邮件。因此，有副作用的操作需要幂等键、执行记录，或能查询既有结果的外部服务。

只读查询通常更容易重试，但也要考虑查询费用和限流。把每个节点的重试次数、超时和可重试错误写清楚。不要用“失败后重新跑全部步骤”覆盖所有情况。

什么时候引入 LangGraph

当你需要复杂分支、持久化、暂停恢复或多节点协作时，可以考虑编排框架。先理解状态契约，再把节点映射到框架，可以减少对隐含行为的依赖。框架不会替你决定哪些操作可重试、哪些字段属于可信事实。

练习与验收

在保存 draft 阶段后主动结束程序，再运行一次，确认从该阶段继续。增加一个 no_evidence 分支，确认不会生成无依据草稿。

恢复后保留原始问题和证据。
已完成任务不会重复执行节点。
不认识的快照版本会明确报错。
能解释邮件节点为何需要额外幂等设计。

延伸阅读

下一课认识 MCP，用一致的方式发现和调用外部工具。