专栏第 4 篇：把“能跑”的 Agent，升级为“出错也能稳住”的工程系统。

---

一、先给标题与结构（可直接用于 CSDN）

标题建议：
《状态机设计实战：从 INIT 到 FAILED 的可恢复执行流（含幂等与重试策略）》

文章结构：

1. 问题描述：没有状态机为什么会“偶发失控”
2. 状态模型：最小可用状态集合与流转规则
3. 核心实现：状态机伪代码 + 事件驱动
4. 关键机制：幂等、超时、重试、降级
5. 验证方法：如何证明“真的可恢复”
6. 常见报错与排查
7. 总结与下一篇预告

---

二、问题描述：为什么第 4 篇必须讲状态机

前 3 篇我们已经完成：

• 确立单 Agent 路线；
• 明确最小可用架构；
• 搭好可运行脚手架。

但一到真实请求，常见问题会集中爆发：

• 工具超时后，流程卡死；
• 重试触发重复执行，导致脏数据；
• 同一请求多次进入同一步骤，结果不一致；
• 故障时只能“看日志猜”，无法确定系统当前处于什么阶段。

本质原因：执行流缺少明确状态边界。

---

三、最小可用状态模型

3.1 状态定义

建议先用 5 个状态（足够覆盖 MVP）：

• INIT：请求进入，完成参数校验与上下文装载
• PLAN_READY：任务规划完成，可执行
• TOOL_RUNNING：工具执行阶段（可多步）
• ANSWER_READY：答案完成，可返回
• FAILED：执行失败，等待重试或降级

3.2 状态流转主路径

INIT -> PLAN_READY -> TOOL_RUNNING -> ANSWER_READY

异常路径（任一步可能）：

任意状态 -> FAILED

可恢复路径：

FAILED -> TOOL_RUNNING（可重试场景）
FAILED -> ANSWER_READY（降级兜底场景）

---

四、状态流转规则

每条状态迁移都要满足 3 个约束：

1. 前置条件（Guard）
   例如：INIT -> PLAN_READY 前，必须完成 query 校验。

2. 动作（Action）
   迁移时触发的执行动作，如生成计划、调用工具、写状态。

3. 后置记录（Audit）
   每次迁移必须记录日志：request_id / from / to / reason / latency

---

五、核心伪代码：可恢复执行流

from enum import Enum

class State(str, Enum):
    INIT = "INIT"
    PLAN_READY = "PLAN_READY"
    TOOL_RUNNING = "TOOL_RUNNING"
    ANSWER_READY = "ANSWER_READY"
    FAILED = "FAILED"

class TransitionError(Exception):
    pass

ALLOWED = {
    State.INIT: {State.PLAN_READY, State.FAILED},
    State.PLAN_READY: {State.TOOL_RUNNING, State.FAILED},
    State.TOOL_RUNNING: {State.ANSWER_READY, State.FAILED},
    State.FAILED: {State.TOOL_RUNNING, State.ANSWER_READY},  # 重试/降级
    State.ANSWER_READY: set()
}

def transit(ctx, to_state: State, reason: str):
    frm = ctx.state
    if to_state not in ALLOWED[frm]:
        raise TransitionError(f"invalid transition: {frm} -> {to_state}")
    ctx.state = to_state
    log_transition(
        request_id=ctx.request_id,
        from_state=frm,
        to_state=to_state,
        reason=reason
    )

def run(ctx):
    try:
        transit(ctx, State.PLAN_READY, "request validated")

        plan = build_plan(ctx.query)
        ctx.plan = plan

        transit(ctx, State.TOOL_RUNNING, "plan created")

        tool_outputs = []
        for step in plan.steps:
            out = run_step_with_guard(ctx, step)   # 幂等+超时+重试
            tool_outputs.append(out)

        ctx.answer = summarize(tool_outputs)
        transit(ctx, State.ANSWER_READY, "answer generated")
        return success(ctx.answer)

    except Exception as e:
        transit(ctx, State.FAILED, f"error: {type(e).__name__}")
        fallback = degrade(ctx, e)                # 兜底回答
        transit(ctx, State.ANSWER_READY, "fallback response")
        return fail_with_fallback(fallback)

---

六、4 个关键机制：状态机要“稳”，必须配齐

6.1 幂等键（防重复执行）

建议：idem_key = request_id + step_id

• 若该 key 已有执行结果，直接返回缓存；
• 防止重试或网络抖动导致重复调用外部工具。

6.2 超时控制（防卡死）

• 每个工具调用必须配置 timeout；
• 超时不等于系统失败，应进入 FAILED 分支处理。

6.3 重试策略（防瞬时故障）

• 建议最多 2 次；
• 使用指数退避（如 200ms -> 500ms）；
• 达到上限后进入降级逻辑。

6.4 降级兜底（防不可用）

• 工具不可达时返回“部分结果 + 明确说明”；
• 保证用户收到可用响应，而不是 500 崩溃。

---

七、验证清单：如何证明状态机“真可恢复”

上线前至少跑这 3 类测试：

1. 正常链路
   期望：状态按主路径流转，最终到 ANSWER_READY

2. 异常链路（工具超时）
   期望：进入 FAILED，触发重试或降级，最终仍可返回

3. 重复请求链路（同 request_id）
   期望：命中幂等缓存，不重复执行外部步骤

建议监控指标：

• 状态迁移成功率
• FAILED 占比
• 平均恢复耗时
• 重试命中率
• 幂等命中率

---

八、常见报错与排查

1. 报错：invalid transition

   • 原因：状态图定义不完整或调用顺序错误
   • 处理：统一通过 transit() 修改状态，禁止“手改 state”

2. 现象：同一步骤被执行多次

   • 原因：幂等键缺失
   • 处理：落地 request_id + step_id 级别缓存

3. 现象：FAILED 后直接中断无返回

   • 原因：缺少降级出口
   • 处理：补 FAILED -> ANSWER_READY 的 fallback 路径

4. 现象：线上问题难复现

   • 原因：没记录迁移日志
   • 处理：补齐 from/to/reason/latency/request_id 字段

---

九、本篇总结

状态机不是“多写几行枚举”，而是 Agent 工程可控性的核心。

一句话：没有状态机，你是在“碰运气跑流程”；有状态机，你是在“可验证地执行系统”。