上一章的 ReAct 是个心智模型，代码不长。本章把 ReAct 真正做成"长跑循环"——加入预算、反思、终止、错误恢复，让它能扛 30 分钟级别任务而不会"撒手就跑飞"。

10.1 最小 Loop 的三处不完美

把上一章那 30 行 loop 真正跑长跑，它会失败在哪？

现象 根因 需要加什么
token 烧到天亮 没有预算 max_tool_calls / max_tokens
模型反复试同一个错 没有"反思" 失败后回看 trace，让模型给假设
任务没完就被模型自己说终了 缺乏 “DONE validation” 终止前问"DONE"三问
一遇工具 exception 就直接 crash 没有异常通道 tool errors 当 message 喂回
Agent 跟上下文对话头尾都说不止 缺乏沉淀 在循环里维护 state.json

10.2 一个稍微长一点的最小 Agent Loop

from dataclasses import dataclass, field

@dataclass
class AgentState:
    history: list = field(default_factory=list)
    tool_calls: int = 0
    max_tool_calls: int = 20
    failures: list = field(default_factory=list)

def step(state: AgentState, user_msg: str) -> tuple[str, AgentState]:
    if state.tool_calls >= state.max_tool_calls:
        return "budget reached: stop", state

    msg = client.messages.create(
        tools=TOOLS,
        system=SYSTEM,
        messages=state.history + [{"role": "user", "content": user_msg}],
    )
    state.history.append({"role": "assistant", "content": msg.content})

    if msg.stop_reason == "end_turn":
        # STOP 之前先做"完成验证" 3 问
        answer = msg.content[-1].text
        valid, reason = check_done(answer, state)
        if valid:
            return answer, state
        # 答不上来 → 给它一次反思机会
        return step(state, f"你说 done 了但还校验不过: {reason}")

    # 否则继续 ReAct 循环
    for block in msg.content:
        if block.type != "tool_use":
            continue
        state.tool_calls += 1
        try:
            result = run_tool(block.name, **block.input)
        except Exception as e:
            result = f"[TOOL ERROR] {type(e).__name__}: {e}"
            state.failures.append((block.name, str(e)))
        state.history.append({
            "role": "user",
            "content": [{"type": "tool_result",
                         "tool_use_id": block.id,
                         "content": result}],
        })
    return step(state, "")                   # 自循环

def check_done(answer: str, state: AgentState) -> tuple[bool, str]:
    if "DONE" not in answer.upper():
        return False, "未给出 DONE 标记"
    if state.failures and not any("[resolved]" in f[1] for f in state.failures):
        return False, f"{len(state.failures)} 次 tool 失败未修复"
    return True, ""

注意 4 点：

  1. 预算：max_tool_calls=20 是上限，到点就停
  2. DONE validation：不是模型说停止就停，要符合"成功"的判据
  3. 失败被自动捕获 + 喂回模型：模型能看到自己的 exception
  4. 递归循环：把 state 在调用之间传递

10.3 五种"健康循环"

健康循环 形态 作用
ReAct Loop thought → action → obs → next 推进任务的主循环
Reflection Loop 失败后回包 trace → 让模型给假设 纠错
Verification Loop 终止前问 “DONE 3 问” 防自欺
Compression Loop 上下文爆掉前压缩早期 messages 续航
Trajectory Loop 每次 step 写 trace 到文件 可观测（→ 第 14 章）

成熟的 harness（OpenCode、Claude Code）都内置了这五种。当你要自己写 harness，要逐个补齐。

10.4 终止条件如何写：DONE 三问

AI Agent 在长任务的最大坑是过早自认完成。一个简单或殊化判定方法：

def check_done(answer: str, state: AgentState) -> tuple[bool, str]:
    # 1. 显式标记
    if "DONE" not in answer.upper():
        return False, "answer 缺少 DONE 标记"
    # 2. 工具失败收敛
    if state.failures and not all_resolved(state.failures):
        return False, f"还有 {count_unresolved(state.failures)} 个 failure 未解"
    # 3. 必要产出存在
    for required in state.required_artifacts:    # 预先约定
        if not exists(required):
            return False, f"required artifact 缺失: {required}"
    return True, ""

更严格、更工程化的做法是让模型自己回答 3 个 yes/no：

在你输出 DONE 之前，先是没有提及 output 中：
1) Are all assertions satisfied? (yes/no)
2) Are all tools'exited_code == 0? (yes/no)
3) Is the goal stated in first user_msg achieved? (yes/no)
答中所有 yes 才输出 DONE。

OpenCode 的 verification-before-completion Skill 就是把它做成框架化。第 22 章我们会展开。

10.5 压缩 Loop：让 agent 不爆 context

长任务在 20 步以后，history 几十 KB 很常见。两步压缩：

async def maybe_compress(state: AgentState, threshold: int = 80_000) -> AgentState:
    if tokens(state.history) < threshold:
        return state
    # 1. 分别总结早期 tool 调用，把 "tool_use + tool_result" 对压成一行摘要
    compressed_history, summary = await summarize_early(state.history)
    # 2. 把 summary 放在最前面（采用 [SUMMARY] 标签，模型看得见）
    new_history = [{"role": "user",
                    "content": [{"type": "text",
                                 "text": f"[EARLY SUMMARY] {summary}"}]}]
    return replace(state, history=new_history + compressed_history)

关键点是把最相关的内容（末尾步骤、最关键的产出）保住，弱的工具调用记录压成一行。Claude Code 默认在 80k 左右触发，OpenCode 类似。

10.6 一个长跑 Loop 的成本与决策

选项 单步成本 起飞停？ 适合
不压 + 终止预算 50 步 高 中期会掉准确度 安全健身
压缩 + 终止预算 50 步 中 准确度稳 推荐主线
不压 + 终止预算 100 步 极高 中后期崩 accuracy 极重 schedule 任务
压缩 + 终止预算 100 步 中 110k+ 才压 大型重构任务

预算的本源是任务复杂度 → 工具数量 → token 量 → 价格。一个中型 Refactor，30 步 × 8K tokens × Claude Opus 4.5 ≈ 0.05 USD/step，30 步 ≈ 1.5 USD。两个 step 没干好、第 31 步到底这正是为什么"Token 燃烧"是 Agent stress。

10.7 Loop 调试：把"史"打印下来

import json
from pathlib import Path

async def log_step(state: AgentState, output_dir: Path = Path(".agent-trace")):
    output_dir.mkdir(exist_ok=True)
    step_n = state.tool_calls
    (output_dir / f"step-{step_n:03d}.json").write_text(
        json.dumps({
            "tool_calls": state.tool_calls,
            "history_tail": state.history[-4:],
            "failures": state.failures,
        }, ensure_ascii=False, indent=2, default=str)
    )

完成后 trace 文件夹长这样：

.agent-trace/
├── step-000.json     # 第一次 tool_use
├── step-001.json
├── step-002.json
└── ...

出错可回放、可诊断、可改 prompt。这就是第 14 章讲可观测性的原料。

10.8 小结

  • 最小 loop 三件事：预算 / 失败 fallback / DONE validation
  • 五种健康循环：ReAct / Reflection / Verification / Compression / Trajectory
  • DONE 三问：显式 + 失败收敛 + 必要产出齐
  • 压缩在 80k 上下文窗口接近触发
  • trace 文件夹是 Agent 调试的金矿

下一篇：《11 Graph 工作流》——单 loop 不够时，把任务拆成图：哪条不再装在 ReAct 里、而从有向图开始编排。

Summary: Loop = 预算 + 反思 + DONE 校验 + 压缩 + Trace，从 30 行到能扛 30 分钟的执行引擎。