一文搞懂 AI Agent 的 Replan：何时重试、何时重规划、何时放弃（附 Python 决策引擎）

📖 摘要：ReAct 循环只会"重试"，不会"换路"——这是长程 Agent 最容易卡死的地方。本文用信号驱动的决策模型讲清"重试 / 重规划 / 放弃"的边界，并手写一个可运行的 ReplanDecider（纯标准库），把你 08 月已经在写的 Harness 主循环补上控制智能层。读完你能直接把这套决策引擎接进自己的 Agent，告别"同一个坑反复跳"。

🏷️ 关键词：AI Agent，Replan，ReAct，错误恢复，Agent 工程

目录

• 一、背景与痛点：ReAct 循环只会重试，不会换路
• 二、核心原理：重试、重规划与放弃的边界
  • 2.1 三种失败的形态
  • 2.2 信号驱动的决策模型
  • 2.3 何时该 Replan 的五个判断
• 三、实战落地：手写一个 ReplanDecider
  • 3.1 环境准备
  • 3.2 执行状态与信号采集
  • 3.3 决策引擎实现
  • 3.4 接入 Harness 主循环
• 四、四个生产避坑
• 五、总结

一、背景与痛点：ReAct 循环只会重试，不会换路

一个最朴素的 Agent 主循环长这样：

观察 → 思考 → 行动 → 观察 → 思考 → 行动 → ...

只要行动失败，最常见的写法就是 retry：再试一次、再试一次。问题来了——如果失败的根因不是"网络抖了一下"，而是"这条路从一开始就是错的"，那么重试一万次也只是把同样的错误刷一万遍。

这在 CSDN Agent 社区本周的热文《Agent 何时该 Replan：五个关键判断》里被点名了：Replan 的触发条件不应是单纯的"重试次数到了"。真正的工程化 Agent（Claude Code、Cursor、各类编码 Agent）之所以稳，是因为它们在主循环里多了一层"决策"：这次失败，是再试一次就好，还是得推翻当前计划重画路线，还是干脆认输放弃？

💡 一句话：重试解决"偶发"，重规划解决"路线错了"，放弃解决"外部不可控"。把三者混为一谈，Agent 要么在死循环里烧 token，要么在错误路线上越走越远。

本文的目标，就是给你一个可运行、可接入的决策引擎，把"该重试 / 该重规划 / 该放弃"的判断从拍脑袋变成可观测、可配置的信号逻辑。

二、核心原理：重试、重规划与放弃的边界

2.1 三种失败的形态

要先做对决策，得先分清失败长什么样。我们把它归成三类：

形态	特征	正确对策	错误对策
瞬时失败	网络超时、限流 429、连接断开	重试（预算内）	立刻重规划（浪费）
结构失败	同样的错误反复出现、产出永远不达标	重规划（换路线）	一直重试（死循环）
外部失败	依赖的第三方服务长期不可用、权限被收回	放弃（外部不可控）	反复重试（无意义）

注意"结构失败"和"瞬时失败"的区分关键：错误是否重复。偶发一次超时可重试；同一个 API 401 连着出现三次，说明不是网络问题，是凭证或路线本身错了——这时候重试只是重复踩坑。

2.2 信号驱动的决策模型

决策不靠"重试了几遍"一个维度，而是靠一组**信号（Signal）**驱动。每个行动结束后，环境给决策器回一个信号：

TRANSIENT_ERROR  瞬时错误（可重试）
REPEATED_ERROR   重复错误（路线失效）
PROGRESS_STALL   步数在走但无实质进展
BLOCKED_DEPENDENCY 外部依赖不可用
VALIDATION_FAILED 产出不符合校验约束
SUCCESS          产出通过校验

决策器拿到信号 + 当前运行态（已尝试次数、重复错误连击、停滞步数、计划版本），输出一个动作：

RETRY   同计划再试
REPLAN  重画计划（计划版本 +1）
ABORT   放弃任务
DONE    任务成功

这套模型的好处是：决策可解释。每一次"为什么重规划"都能回溯到具体信号和阈值，而不是一个黑盒里的随机行为。

2.3 何时该 Replan 的五个判断

把社区里讨论的"五个关键判断"泛化沉淀成可执行的规则：

1. 同一错误重复出现 ≥ 阈值（如 2 次）→ 路线结构失效，Replan。
2. 连续 N 步无实质进展（步数在走但目标没靠近）→ 当前策略卡住，Replan。
3. 产出校验反复不通过 → 不是执行问题，是计划本身定义错了目标，Replan。
4. 外部依赖长时间不可用 → 不是 Agent 能解决的，直接 Abort，别空转。
5. 重规划预算耗尽后仍失败 → 及时认输，避免无限重画计划烧光上下文与成本。

⚠️ 注意第 5 条：Replan 必须有预算。没有预算的"永远重规划"比"永远重试"更可怕——它会不停重写计划，把上下文窗口吃得干干净净。

三、实战落地：手写一个 ReplanDecider

3.1 环境准备

纯标准库实现，无需任何第三方依赖，Python 3.10+ 直接可跑：

# 准备一个虚拟环境（可选）
python -m venv .venv && source .venv/bin/activate

# 直接运行示例脚本
python replan_decider.py

示例数据全程为虚构演示，域名、IP、服务名均用 example.com / 127.0.0.1 / order_service 等泛化占位。

3.2 执行状态与信号采集

先用 enum 定义信号与决策，用 dataclass 记录运行态：

from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum


class Signal(str, Enum):
    TRANSIENT_ERROR = "transient_error"        # 瞬时错误：超时、限流
    REPEATED_ERROR = "repeated_error"          # 同一错误反复出现
    PROGRESS_STALL = "progress_stall"          # 步数推进但无实质进展
    BLOCKED_DEPENDENCY = "blocked_dependency"  # 外部依赖不可用
    VALIDATION_FAILED = "validation_failed"    # 产出不符合约束
    SUCCESS = "success"                        # 产出通过校验


class Decision(str, Enum):
    RETRY = "retry"
    REPLAN = "replan"
    ABORT = "abort"
    DONE = "done"


@dataclass
class RunState:
    plan_version: int = 0            # 当前计划版本（每次重规划 +1）
    attempts: int = 0                # 累计行动次数
    last_error: str = ""
    repeated_error_streak: int = 0   # 同一错误连续出现次数
    stall_steps: int = 0            # 连续无进展步数
    signals: list = field(default_factory=list)

    def record(self, signal: Signal, error: str = "") -> None:
        self.attempts += 1
        self.signals.append(signal)
        # 同一错误连击计数；错误变化则重置
        if error and error == self.last_error:
            self.repeated_error_streak += 1
        elif error:
            self.last_error = error
            self.repeated_error_streak = 1
        # 非进展类信号不计入停滞（这里仅作演示，真实环境按 advance 标记）
        if signal == Signal.PROGRESS_STALL:
            self.stall_steps += 1

3.3 决策引擎实现

核心就是 decide()：一组 if 按优先级收口，返回决策与可解释的原因。

# 可调阈值（生产环境建议放进配置中心）
RETRY_BUDGET = 3        # 瞬时错误最多重试次数
REPLAN_BUDGET = 2       # 最多重规划次数
STALL_THRESHOLD = 5     # 连续无进展多少步触发重规划
REPEAT_THRESHOLD = 2    # 同一错误重复几次判定为结构失效


def decide(state: RunState, signal: Signal, error: str = "") -> tuple[Decision, str]:
    state.record(signal, error)

    # 1) 进展停滞：优先于重试预算判断，避免被 attempts 误杀
    if signal == Signal.PROGRESS_STALL:
        if state.stall_steps >= STALL_THRESHOLD:
            if state.plan_version < REPLAN_BUDGET:
                return Decision.REPLAN, f"连续 {STALL_THRESHOLD} 步无进展，触发重规划"
            return Decision.ABORT, "重规划后仍停滞，放弃任务"
        return Decision.RETRY, "步数推进中，继续当前计划"

    # 2) 结构失败：重复错误 / 产出不达标 → 换路线
    if signal in (Signal.REPEATED_ERROR, Signal.VALIDATION_FAILED):
        if state.plan_version < REPLAN_BUDGET:
            return Decision.REPLAN, "路线或产出失效，触发重规划"
        return Decision.ABORT, "重规划预算耗尽，放弃任务"

    # 3) 外部失败：依赖不可用，放弃
    if signal == Signal.BLOCKED_DEPENDENCY:
        return Decision.ABORT, "外部依赖持续不可用，放弃任务"

    # 4) 瞬时错误：预算内重试
    if signal == Signal.TRANSIENT_ERROR:
        if state.attempts <= RETRY_BUDGET:
            return Decision.RETRY, "瞬时错误，预算内重试"
        return Decision.ABORT, "瞬时错误重试预算耗尽，放弃任务"

    # 5) 成功
    if signal == Signal.SUCCESS:
        return Decision.DONE, "产出通过校验，任务成功"

    return Decision.RETRY, "默认重试"

3.4 接入 Harness 主循环

把决策器嵌进主循环：拿到信号 → 决策 → 执行对应动作。重规划时把 plan_version + 1 并重置计数，画一条新路线。

def run_task(events):
    """events: List[(Signal, error)]，按顺序回放，演示决策路径。"""
    state = RunState()
    print(f"计划 v{state.plan_version} 启动，事件序列共 {len(events)} 步")
    for idx, (signal, error) in enumerate(events, 1):
        decision, reason = decide(state, signal, error)
        print(f"  步{idx:>2} 信号={signal.value:<18} 决策={decision.value:<6} | {reason}")
        if decision == Decision.REPLAN:
            state.plan_version += 1
            state.stall_steps = 0
            state.repeated_error_streak = 0
            print(f"       ↳ 重画计划 -> 计划 v{state.plan_version}")
            continue
        if decision == Decision.ABORT:
            print("  ✅ 任务终止（放弃）")
            return
        if decision == Decision.DONE:
            print("  ✅ 任务完成")
            return
    print("  ✅ 事件序列结束")


if __name__ == "__main__":
    print("=== 场景 A：同一错误反复出现 -> 重规划 -> 放弃 ===")
    run_task([(Signal.REPEATED_ERROR, "API 401")] * 4)

    print("\n=== 场景 B：长期无进展 -> 重规划 -> 仍停滞 -> 放弃 ===")
    run_task([(Signal.PROGRESS_STALL, "")] * 15)

    print("\n=== 场景 C：瞬时错误 -> 重试 -> 成功 ===")
    run_task([(Signal.TRANSIENT_ERROR, "timeout")] * 2 + [(Signal.SUCCESS, "")])

运行输出（节选，示例数据，仅作演示）：

=== 场景 A：同一错误反复出现 -> 重规划 -> 放弃 ===
计划 v0 启动，事件序列共 4 步
  步 1 信号=repeated_error     决策=replan  | 路线或产出失效，触发重规划
       ↳ 重画计划 -> 计划 v1
  步 2 信号=repeated_error     决策=replan  | 路线或产出失效，触发重规划
       ↳ 重画计划 -> 计划 v2
  步 3 信号=repeated_error     决策=abort   | 重规划预算耗尽，放弃任务
  ✅ 任务终止（放弃）

=== 场景 C：瞬时错误 -> 重试 -> 成功 ===
计划 v0 启动，事件序列共 3 步
  步 1 信号=transient_error    决策=retry   | 瞬时错误，预算内重试
  步 2 信号=transient_error    决策=retry   | 瞬时错误，预算内重试
  步 3 信号=success           决策=done    | 产出通过校验，任务成功
  ✅ 任务完成

把 run_task 换成你真实的 Agent 调度：每执行完一个工具，把环境返回的信号喂给 decide()，按返回动作 retry / replan / abort 推进即可。计划重画逻辑（怎么生成 v1 计划）交给你的 Planner 模块，本文的 ReplanDecider 只负责"何时该换"。

四、四个生产避坑

1. Replan 必须设预算。没有 REPLAN_BUDGET 的 Agent 会陷入"重写计划 → 又失败 → 再重写"的旋涡，上下文瞬间爆炸。预算耗尽直接 ABORT，把"认输"当作一等公民。

2. 停滞判定要用"进展"而非"步数"。本文为演示用 PROGRESS_STALL 信号，真实环境应让每个工具返回 advanced: bool——只把"真正让目标靠近"的步骤算作进展，避免 Agent 用"打日志、读文件"这类伪动作骗过停滞检测。

3. 重复错误要归一化。比较"是不是同一个错误"前，先把错误做指纹归一（去掉时间戳、随机 ID、行号），否则 timeout at 12:00:01 和 timeout at 12:00:02 会被当成两个不同错误，漏掉结构失效信号。

4. Abort 也要可恢复。外部依赖不可用导致的放弃，最好带上"断点状态"落盘（参考长程 Agent 的执行状态运行时层思路），等依赖恢复后能从中断点续跑，而不是从头再来。

五、总结

ReAct 循环只解决了"怎么动"，没解决"动错了怎么办"。本文补上的 ReplanDecider 用信号驱动 + 阈值收口把"重试 / 重规划 / 放弃"变成可解释、可配置的工程决策，核心三点：

• 区分失败形态：瞬时可重试、结构要换路、外部该认输。
• 决策靠信号而非拍脑袋：重复错误、进展停滞、校验失败各自触发对应动作。
• 一切皆有预算：重试有预算、重规划有预算、放弃要可恢复。

把它接进你 8 月写好的 Harness 主循环，Agent 就从"会重试的循环"升级成"会思考要不要换路的系统"。下一篇可以继续聊：重规划时怎么生成更好的 v1 计划（结合长程执行状态运行时层 Ledger 的思路），把"换路"本身也做成数据驱动的。

觉得有用的话点个赞 / 收藏，评论区聊聊你的 Agent 是怎么处理"卡住"的。