Agent 多步任务可靠性实战：死循环、跑偏、幻觉、人工兜底与可观测

单轮 Agent 还能用，一旦让它"连续跑 20 步完成一个复杂任务"，问题全暴露：反复调同一个工具（死循环）、目标越跑越偏、中间结果开始编、第 30 步还在兜圈、失败后无法重来、出问题完全查不到。可靠性不是"换个更强的模型"能解决的，而是流程控制 + 守卫 + 兜底 + 可观测的系统工程。本文全部围绕真实问题：现象 → 根因 → 可运行方案（ReAct 循环检测、预算守卫、终态判定、Plan-Execute-Review、Checkpoint 断点续跑、HITL 人工介入、Trace 可观测、任务评测集），带完整 Python 代码。

---

一、五个真实故障现场

1.1 故障一：死循环

现象：
- Agent 反复调用 query_order(order_id=10086)
- 连续 12 次相同参数，每次返回一样
- token 烧掉十几万，任务没进展

根因：
- 没有"重复检测"
- 没有"最大步数"硬约束
- 模型看到结果不满意 → 换个说法再调一次（其实同一动作）

1.2 故障二：目标漂移（跑偏）

现象：
- 任务：给登录接口加限流
- 中途它开始改日志格式、重构 UserService、升级依赖……
- 最后交付了一大堆无关改动

根因：
- 目标没有被"钉住"
- 每步决策只看"下一步"，不看"离目标多远"
- 没有验收标准作为停止条件

1.3 故障三：中间结果幻觉

现象：
- Agent 说"订单已创建，订单号 10088"
- 实际工具返回失败（库存不足）
- 它把"计划要做的"当成"已经做的"

根因：
- 结果没做结构化校验
- 模型自己"编"完成状态

1.4 故障四：失败后无法重来

现象：
- 任务跑到第 15 步崩了（工具超时）
- 只能从头再跑一遍，前面 15 步白干
- 重复执行有副作用（重复下单）

根因：
- 没有 checkpoint（状态快照）
- 没有断点续跑

1.5 故障五：出事查不到

现象：
- 用户说"它给我改错了"
- 你不知道它每一步想了什么、调了什么、返回什么
- 只能猜

根因：
- 没有 trace（每步的输入/输出/决策）
- 日志只记了最终结果

---

二、可靠性框架

2.1 四层守卫

1. 循环守卫：防死循环（重复检测 + 步数上限）
2. 预算守卫：防成本失控（token/时间/工具调用次数）
3. 目标守卫：防跑偏（目标锚定 + 偏离检测）
4. 结果守卫：防幻觉（工具结果校验 + 完成判定）

外加：
5. 兜底：HITL 人工介入 + checkpoint 续跑
6. 可观测：trace + 指标 + 评测

2.2 任务生命周期

规划（Plan）→ 执行（Act）→ 观察（Observe）→ 反思（Reflect）
   ↑                                              ↓
   └──────────── 每轮检查守卫 ────────────────────┘

任一守卫触发 → 停止/降级/转人工
完成条件满足 → 输出结果 + 归档 trace

---

三、方案一：死循环检测

3.1 重复动作检测（状态哈希）

import hashlib, json

class LoopDetector:
    def __init__(self, max_repeat: int = 2, max_steps: int = 25):
        self.max_repeat = max_repeat
        self.max_steps = max_steps
        self.history = []          # 每步动作签名
        self.steps = 0

    def signature(self, action: dict) -> str:
        """动作签名：工具名 + 规范化参数"""
        raw = json.dumps(action, sort_keys=True, ensure_ascii=False)
        return hashlib.md5(raw.encode()).hexdigest()[:16]

    def check(self, action: dict) -> tuple[bool, str]:
        self.steps += 1

        # 1. 步数上限
        if self.steps > self.max_steps:
            return False, f"超过最大步数 {self.max_steps}，停止"

        sig = self.signature(action)
        # 2. 同一动作重复次数
        repeat = self.history.count(sig)
        if repeat >= self.max_repeat:
            return False, f"检测到重复动作（第 {repeat + 1} 次相同调用），停止"

        self.history.append(sig)
        return True, ""

要点：
- 签名要"规范化"（参数排序、忽略无关字段）
- 只比最近 N 步（避免历史误判）
- 触发后不是硬停，可以让模型"换个思路"（见下）

3.2 语义重复检测（进阶）

# 动作不同但实质相同（query_order vs get_order_detail）
# 用 embedding 相似度判断
def semantic_duplicate(new_action: str, recent_actions: list,
                        threshold: float = 0.92) -> bool:
    """需要 embedding 模型；相似度超阈值视为重复"""
    emb = embed(new_action)
    for old in recent_actions[-5:]:
        if cosine(emb, embed(old)) > threshold:
            return True
    return False

3.3 触发后的正确处置

不要直接"报错退出"，而是：
1. 把"重复检测"结果告诉模型：
   "你已经用相同参数查了 3 次订单 10086，结果相同。
    请换一种方式：检查参数是否正确，或告知用户当前信息不足。"
2. 让模型反思（Reflect）→ 换策略
3. 仍重复 → 强制停止 + 转人工

---

四、方案二：预算守卫

4.1 多维预算

class BudgetGuard:
    def __init__(self, max_tokens: int = 200_000,
                 max_seconds: float = 300,
                 max_tool_calls: int = 40):
        self.max_tokens = max_tokens
        self.max_seconds = max_seconds
        self.max_tool_calls = max_tool_calls

        self.tokens = 0
        self.start = time.time()
        self.tool_calls = 0

    def consume_tokens(self, n: int):
        self.tokens += n
        self._check()

    def consume_tool(self):
        self.tool_calls += 1
        self._check()

    def _check(self):
        if self.tokens > self.max_tokens:
            raise BudgetExceeded(f"token 超限: {self.tokens}")
        if self.tool_calls > self.max_tool_calls:
            raise BudgetExceeded(f"工具调用超限: {self.tool_calls}")
        if time.time() - self.start > self.max_seconds:
            raise BudgetExceeded("执行超时")

# 或软守卫：接近上限时提醒模型收敛
def budget_hint(guard) -> str:
    if guard.tokens > guard.max_tokens * 0.8:
        return "预算已用 80%，请尽快收敛到结论，不要继续探索。"
    return ""

原则：
- 硬上限必须有（防无限烧钱）
- 软提醒更优雅（让模型自己收尾）
- 超限时输出"已完成的部分 + 未完成原因"（别白跑）

---

五、方案三：目标锚定与偏离检测

5.1 目标声明（写死）

GOAL = {
    "task": "给登录接口增加失败锁定（5 次错误锁定 10 分钟）",
    "constraints": [
        "只改 AuthService/AuthController",
        "不改数据库表结构",
        "用现有 Redis 客户端",
    ],
    "acceptance": [
        "mvn test 全绿",
        "第 6 次错误登录返回 423",
        "锁定 10 分钟后自动解锁",
    ],
}

每轮把 GOAL 注入 system（不参与裁剪）
每步执行后检查：这一步是否在服务 acceptance？

5.2 偏离检测

def check_drift(step_action: str, goal: dict, llm) -> tuple[bool, str]:
    """让模型判断当前动作是否偏离目标（成本低，可每 3 步做一次）"""
    prompt = f"""目标：{goal['task']}
约束：{goal['constraints']}
验收：{goal['acceptance']}

当前动作：{step_action}

判断：这个动作是否直接服务于目标？
只回答 DRIFT 或 OK，并给一句理由。"""
    verdict = llm.invoke(prompt).content.strip()
    return verdict.startswith("OK"), verdict

触发偏离时：
1. 停止当前动作
2. 重新规划（回到 Plan）
3. 连续偏离 2 次 → 转人工确认方向

5.3 完成判定（验收驱动停止）

def is_complete(goal: dict, evidence: dict) -> bool:
    """必须基于"证据"判定，不能凭模型自述"""
    checks = {
        "tests_green": evidence.get("test_result") == "PASS",
        "locked_after_5": evidence.get("http_status_6th") == 423,
        "unlock_after_10m": evidence.get("unlock_verified") is True,
    }
    return all(checks.values())

关键：
- 完成 = 所有 acceptance 有客观证据
- 不接受"我觉得做完了"
- 缺证据 → 继续执行或报告"部分完成"

---

六、方案四：结果校验（防幻觉）

6.1 工具结果结构化

def normalize_tool_result(raw):
    """统一成 {ok, data, error}，禁止模型自由解读"""
    if isinstance(raw, dict) and "ok" in raw:
        return raw
    return {"ok": True, "data": raw, "error": None}

6.2 关键断言校验

def verify_step(step: dict) -> tuple[bool, str]:
    """每步执行后做客观校验"""
    result = normalize_tool_result(step["result"])

    # 1. 工具明确失败 → 不能宣称成功
    if not result["ok"]:
        return False, f"工具返回失败: {result['error']}"

    # 2. 关键字段非空
    required = step.get("expect_fields", [])
    data = result.get("data") or {}
    for f in required:
        if data.get(f) in (None, "", []):
            return False, f"结果缺少字段: {f}"

    # 3. 业务规则（示例：订单状态）
    if step["tool"] == "query_order":
        if data.get("status") not in {"paid", "pending", "shipped"}:
            return False, f"订单状态异常: {data.get('status')}"

    return True, ""

6.3 禁止"计划当事实"

Prompt 里明确：
"只有工具返回 ok=true 且字段齐全，才能声称该步骤完成。
 计划中的动作不算完成，必须等执行结果。"

并在每步后强制写"事实清单"：
done: ["query_order 返回 status=paid"]
not_done: ["cancel_order 尚未执行"]

---

七、方案五：Plan-Execute-Review

7.1 三阶段

1. Plan：把任务拆成有序步骤（含每步验收）
2. Execute：逐步执行，每步后校验
3. Review：整体验收（对照 acceptance）

好处：
- 有明确步骤，减少"边想边跑"的漂移
- 每步可校验，早发现错误
- 失败可定位到具体步骤

7.2 代码骨架

class PlanExecuteAgent:
    def __init__(self, llm, tools, guard, loop_detector):
        self.llm = llm
        self.tools = tools
        self.guard = guard
        self.loop = loop_detector

    def plan(self, goal: dict) -> list:
        prompt = f"""把任务拆成步骤，每步含：动作、工具、验收标准。
任务：{goal['task']}
约束：{goal['constraints']}
输出 JSON 数组：[{{"step":1,"action":"...","tool":"...","acceptance":"..."}}]
"""
        return json.loads(self.llm.invoke(prompt).content)

    def execute(self, steps: list) -> dict:
        evidence = {}
        for i, s in enumerate(steps, 1):
            ok, why = self.loop.check({"tool": s["tool"], "args": s.get("args")})
            if not ok:
                return {"status": "aborted", "reason": why, "evidence": evidence}

            self.guard.consume_tool()
            res = self.tools.call(s["tool"], s.get("args", {}))
            ok2, why2 = verify_step({"tool": s["tool"], "result": res,
                                     "expect_fields": s.get("expect_fields", [])})
            if not ok2:
                return {"status": "failed", "step": i, "reason": why2,
                        "evidence": evidence}

            evidence[s["tool"]] = res
        return {"status": "done", "evidence": evidence}

7.3 Review

def review(goal, evidence, llm) -> dict:
    prompt = f"""对照验收标准逐条核对证据，输出 JSON：
{{"passed":true/false,"missing":["..."],"notes":"..."}}
验收：{goal['acceptance']}
证据：{json.dumps(evidence, ensure_ascii=False)[:2000]}
"""
    return json.loads(llm.invoke(prompt).content)

---

八、方案六：Checkpoint 断点续跑

8.1 存什么

class Checkpoint:
    def __init__(self, path: str):
        self.path = path

    def save(self, state: dict):
        with open(self.path, "w", encoding="utf-8") as f:
            json.dump(state, f, ensure_ascii=False)

    def load(self) -> dict:
        with open(self.path, encoding="utf-8") as f:
            return json.load(f)

# state 内容
{
    "goal": GOAL,
    "plan": [...],
    "step_index": 7,            # 已完成到第 7 步
    "evidence": {...},
    "done_signatures": [...],   # 已执行动作（防重复副作用）
    "budget": {"tokens": 120000, "tool_calls": 18},
}

8.2 续跑逻辑

def resume_or_start(ckpt: Checkpoint, agent, goal):
    try:
        state = ckpt.load()
        if state["goal"]["task"] == goal["task"] and not state.get("finished"):
            print(f"从第 {state['step_index']} 步续跑")
            return agent.continue_from(state)
    except FileNotFoundError:
        pass
    return agent.start(goal)

关键：
- 有副作用的步骤要用幂等键（防续跑重复下单）
- 每步后保存 checkpoint（崩溃只丢最后一步）
- checkpoint 含"已完成签名"，续跑时跳过已做过的事

---

九、方案七：HITL 人工介入

9.1 什么时候必须转人工

1. 高危动作：删除数据、批量更新、付款、发外部消息
2. 置信度低：偏离检测失败、重复检测触发
3. 预算临界：token/时间快超
4. 验收不通过且无法自修

9.2 审批门

HIGH_RISK = {"delete_data", "send_email", "payment", "batch_update"}

def need_approval(tool_name: str, args: dict) -> bool:
    if tool_name in HIGH_RISK:
        return True
    # 批量操作超过阈值
    if tool_name == "batch_update" and args.get("count", 0) > 100:
        return True
    return False

def execute_with_approval(tool_name, args, approve_fn):
    if need_approval(tool_name, args):
        ok = approve_fn(tool_name, args)     # 人工确认
        if not ok:
            return {"ok": False, "error": "人工拒绝执行"}
    return tool_call(tool_name, args)

9.3 人工介入的体验设计

好的人工介入：
- 明确展示"要做什么、影响什么、为什么需要你确认"
- 提供"批准 / 拒绝 / 修改参数"三选项
- 超时未响应 → 默认拒绝（安全）

差的人工介入：
- 每一步都弹窗（烦）
- 只给"是/否"没有上下文

---

十、方案八：可观测与评测

10.1 Trace（每步可复盘）

class Tracer:
    def __init__(self, task_id: str):
        self.task_id = task_id
        self.spans = []

    def record(self, step: int, thought: str, action: dict,
               observation, ok: bool, duration_ms: int):
        self.spans.append({
            "task_id": self.task_id,
            "step": step,
            "thought": thought,
            "action": action,
            "observation": str(observation)[:500],
            "ok": ok,
            "duration_ms": duration_ms,
        })

    def dump(self, path: str):
        with open(path, "w", encoding="utf-8") as f:
            json.dump(self.spans, f, ensure_ascii=False, indent=2)

Trace 能回答：
- 第几步开始跑偏？（目标 vs 动作）
- 哪次工具调用失败导致了后续连锁？
- 重复检测在第几步触发？

10.2 关键指标

- 任务完成率（按任务类型）
- 平均步数 / P95 步数（是否异常膨胀）
- 循环触发率（死循环多不多）
- 幻觉率（人工抽检"声称完成但实际没成"）
- 人工介入率（兜底频率）
- 平均成本（token/次）

10.3 评测集（回归）

TASKS = [
    {"id": "T1", "goal": "给登录接口加限流",
     "checks": ["tests_green", "429_on_6th"]},
    {"id": "T2", "goal": "修复订单查询 NPE",
     "checks": ["tests_green", "no_npe_in_log"]},
]

def run_eval(agent, tasks) -> dict:
    results = {}
    for t in tasks:
        out = agent.run(t["goal"])
        results[t["id"]] = all(c in out["evidence"] for c in t["checks"])
    return results

用途：
- 改 prompt/工具/流程后跑一遍，防回归
- 新模型上线前对比通过率
- 定位"哪类任务总失败"（针对性优化）

---

十一、常见坑速查

现象	根因	解决
反复调同一工具	无重复检测	状态哈希 + 重复上限
任务跑偏	目标没锚定	GOAL 每轮注入 + 偏离检测
声称完成实际没做	计划当事实	工具结果校验 + 证据判定
费用失控	无预算	token/时间/调用数硬上限
崩了要从头跑	无 checkpoint	每步存档 + 断点续跑
高危操作误执行	无审批	HITL 审批门
出错查不到	无 trace	每步记录 thought/action/observation
改完更差	无评测	任务集回归测试

---

本章小结

Agent 多步任务可靠性的主线：循环守卫（签名去重 + 步数上限）→ 预算守卫（token/时间/调用数）→ 目标守卫（GOAL 锚定 + 偏离检测 + 验收驱动停止）→ 结果守卫（工具结果结构化 + 证据校验，禁止计划当事实）→ Plan-Execute-Review → Checkpoint 断点续跑（配幂等）→ HITL 高危审批 → Trace 可观测 + 评测集回归。核心心法：可靠性 = 每一步都有客观校验 + 每类风险都有硬上限 + 每次失败都能定位和续跑；模型负责"想"，工程负责"守"。把守卫和兜底做扎实，Agent 才能从"演示能跑"变成"生产敢用"。

三篇 Agent 实际问题与解决方案主题文档全部完成：上下文工程与记忆、工具调用稳定性、多步任务可靠性。

（全文完）