Agent 工程体系 · 第 20/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线；框架版本敏感能力会明确标注，不把实验行为写成通用保证。

Agent 终止与预算：最大步数、Deadline、Token、费用和循环检测

Agent 不是“一次调用模型并返回文本”，而是一个持续运行的控制循环：

1. 读取当前状态；
2. 调用模型进行决策；
3. 解析模型输出；
4. 执行工具、切换子 Agent 或等待人工审批；
5. 将结果写回状态；
6. 判断是否继续，或者终止运行。

OpenAI Agents SDK 将一次运行描述为应用层的一次 turn：运行器调用当前 Agent 的模型，检查输出，执行工具调用或 handoff，并在得到没有更多工具工作的最终答案后返回结果。Anthropic 也将 Agent 概括为“LLM 使用工具并根据环境反馈反复运行的循环”，并明确建议加入最大迭代次数等停止条件。(developers.openai.com)

因此，Agent 的终止不能只依赖模型说“任务完成”。生产系统必须同时控制：

• 最大步数：最多允许多少次决策循环；
• Deadline：最晚运行到什么时间；
• Token 预算：最多消耗多少输入、输出或推理 Token；
• 费用预算：最多允许产生多少模型和工具费用；
• 循环检测：如何识别没有产生有效进展的重复行为；
• 终止语义：终止后返回什么状态，以及是否允许恢复。

这些机制相互关联，但不能互相替代。最大步数防止无限迭代，Deadline 防止长时间阻塞，Token 预算控制模型上下文和生成量，费用预算控制金钱支出，循环检测则负责识别“虽然还没有超过上限，但实际上已经没有进展”的运行。

---

一、先定义 Agent 的“步”和“回合”

1.1 Agent 运行循环的状态模型

设 Agent 在第 $i$ 步开始时处于状态：

$$S_i$$

模型根据状态生成决策：

$$D_i = \pi_\theta(S_i)$$

其中：

• $S_i$ 是当前状态，包括用户请求、历史消息、工具结果、计划、预算和错误信息；
• $\pi_\theta$ 是由模型、系统提示词、工具定义和运行时上下文共同构成的决策策略；
• $D_i$ 可以是最终回答、工具调用、handoff、等待审批或无效输出。

如果 $D_i$ 是工具调用，工具执行后得到结果：

$$O_i = T(D_i)$$

然后状态更新为：

$$S_{i+1} = U(S_i, D_i, O_i)$$

整个运行可以表示为：

$$S_0 \rightarrow D_0 \rightarrow O_0 \rightarrow S_1 \rightarrow D_1 \rightarrow O_1 \rightarrow \cdots$$

当决策满足终止条件时，运行进入终止状态：

$$S_n \rightarrow \text{Terminal}$$

这里的关键是：步数应该统计控制循环，而不是简单统计 HTTP 请求数。

一次模型调用可能产生多个并行工具调用；一次工具调用也可能在内部访问数据库、HTTP 服务或执行脚本。若把每个 HTTP 请求都视为 Agent 步，会把工具内部实现细节错误地混入 Agent 的决策预算。

1.2 推荐的步数定义

对于大多数 Agent 运行时，可以采用如下定义：

一步是一次“模型决策完成并被运行时接受”的循环迭代。

具体来说：

• 模型输出最终答案：消耗一步；
• 模型输出一个或多个工具调用，运行时执行后继续：消耗一步；
• 模型 handoff 到另一个 Agent：通常消耗当前决策步，并继续计入同一个运行预算；
• 工具内部重试：默认不增加 Agent 步数，但必须单独计入工具调用次数和 Deadline；
• 人工审批暂停：不应自动增加步数；
• 从持久化状态恢复：继续原运行的预算，而不是重置预算。

OpenAI 文档也特别区分了“暂停的运行”和“新的用户回合”：如果运行因为审批暂停，应该从原状态恢复，而不是创建新 turn，否则会导致历史、turn 计数和续接 ID 不一致。(developers.openai.com)

1.3 “模型调用次数”和“Agent 步数”不一定相等

例如一次决策输出三个独立工具调用：

模型决策：
  1. search_orders(order_id=1001)
  2. get_customer(customer_id=42)
  3. query_policy(policy="refund")

运行时可以并行执行三个工具：

第 1 步：
  一次模型决策
  三个工具调用
  三个工具结果

此时可以记录：

agent_steps = 1
model_calls = 1
tool_calls = 3

如果工具失败并重试两次：

agent_steps = 1
model_calls = 1
tool_calls = 3
tool_attempts = 9

这四个计数器用途不同：

计数器	含义	主要用途
agent_steps	Agent 做了多少次决策	防止逻辑无限循环
model_calls	调用了多少次模型	统计 Token、模型费用和延迟
tool_calls	计划执行多少个工具动作	统计工作量和工具配额
tool_attempts	工具实际尝试了多少次	诊断重试、故障和外部费用

如果只保留一个 step_count，后续很难解释一次运行为什么“只走了 4 步，却产生了 20 次外部请求”。

---

二、终止条件不是一个条件，而是一组安全阀

一个生产级 Agent 通常同时具备以下终止条件：

$$\text{stop} = \text{success} \lor \text{failure} \lor \text{max\_steps} \lor \text{deadline} \lor \text{token\_budget} \lor \text{cost\_budget} \lor \text{loop\_detected} \lor \text{cancelled}$$

其中：

• success：完成目标；
• failure：不可恢复的错误；
• max_steps：超过最大决策步数；
• deadline：超过绝对截止时间；
• token_budget：Token 预算不足；
• cost_budget：费用预算不足；
• loop_detected：检测到重复或无进展循环；
• cancelled：用户、系统或上游服务主动取消。

这些条件应由运行时统一判断，而不是分散在模型提示词、工具函数和业务代码中。

stateDiagram-v2
    [*] --> Ready
    Ready --> Deciding: 调用模型
    Deciding --> ExecutingTools: 产生工具调用
    Deciding --> Handoff: 切换 Agent
    Deciding --> Completed: 产生最终答案
    ExecutingTools --> Ready: 工具结果写回状态
    Handoff --> Ready: 新 Agent 接管
    Ready --> Paused: 等待审批
    Paused --> Ready: 从原状态恢复

    Ready --> Terminated: 预算检查失败
    Deciding --> Terminated: Deadline/取消/Token 超限
    ExecutingTools --> Terminated: 工具超时或费用超限
    Ready --> Terminated: 循环检测命中

    Completed --> [*]
    Terminated --> [*]

这张图中有一个容易忽略的事实：预算检查必须出现在多个位置。

不能只在进入循环时检查一次，因为：

• 模型调用期间可能消耗大量 Token；
• 工具执行期间可能已经超过 Deadline；
• 并行工具可能同时产生多笔费用；
• 工具结果写回后可能暴露出循环；
• 用户取消可能发生在模型流式输出中间。

---

三、最大步数：最简单，也最容易被误用

3.1 最大步数的形式化定义

设最大步数为 $N_{\max}$，当前已经完成的决策步数为 $n$。则新的决策是否允许开始，可以定义为：

$$n < N_{\max}$$

当 $n = N_{\max}$ 时，不再发起下一次模型决策。

这里必须明确“在什么时候递增”。

推荐流程是：

1. 进入下一次模型决策前，检查 steps_used < max_steps；
2. 为本次决策预留一个步数；
3. 模型调用结束后，将 steps_used 增加 1；
4. 工具执行结束并写回状态；
5. 继续前再次检查。

伪代码如下：

while True:
    if state.steps_used >= budget.max_steps:
        return terminate("max_steps")

    if clock.monotonic() >= budget.deadline:
        return terminate("deadline")

    if state.token_used >= budget.max_tokens:
        return terminate("token_budget")

    if state.cost_used >= budget.max_cost:
        return terminate("cost_budget")

    decision = call_model(state)
    state.steps_used += 1

    if decision.is_final:
        return complete(decision.answer)

    if decision.has_tool_calls:
        results = execute_tools(decision.tool_calls)
        state = update_state(state, decision, results)
        continue

    return terminate("invalid_decision")

3.2 最大步数能解决什么问题

最大步数可以保证：

$$\text{Agent 决策次数} \leq N_{\max}$$

因此，即使模型不断产生工具调用，系统也不会无限发起新的模型决策。

例如：

max_steps = 5

第 1 步：搜索订单
第 2 步：读取退款政策
第 3 步：查询支付状态
第 4 步：再次搜索订单
第 5 步：再次查询支付状态
停止，不再开始第 6 步

它能阻止无限增长，但不能证明任务完成，也不能证明运行没有浪费。

3.3 最大步数不能解决什么问题

最大步数无法识别以下情况：

第 1 步：调用 search("杭州天气")
第 2 步：调用 search("杭州天气")
第 3 步：调用 search("杭州天气")
第 4 步：调用 search("杭州天气")

如果设置 max_steps=10，系统仍然会浪费 10 步后才停止。

最大步数也无法控制单步内部的耗时：

第 1 步：
  模型很快返回
  工具调用等待 20 分钟

这次运行只有一步，却已经违反了交互请求的时延要求。

因此，最大步数是硬上限，不是进度判断器，也不是超时机制。

3.4 最大步数的配置不能脱离任务类型

不同任务需要不同的上限：

任务	常见特征	步数策略
FAQ 查询	通常无需工具或只需一次查询	较小上限
订单客服	查询、校验、执行动作	中等上限
代码修复	编辑、测试、读取错误、重新编辑	较大上限
深度研究	搜索、筛选、分析、交叉验证	依任务复杂度动态分配
浏览器自动化	动作粒度细，步骤多	同时设置动作次数和总 Deadline

对于代码 Agent，测试失败后重新修改可能是有效进展；对于退款 Agent，重复调用支付接口可能是严重风险。因此“最大步数”必须与工具语义、任务成功标准和副作用等级一起设计。

---

四、Deadline：不是“每个请求的 timeout”

4.1 Deadline 的定义

Deadline 是一次 Agent 运行的绝对截止时间。

设运行开始的单调时钟值为 $t_0$，允许的总运行时间为 $D$，则：

$$t_{\text{deadline}} = t_0 + D$$

在任意时刻 $t$，剩余时间为：

$$R(t) = t_{\text{deadline}} - t$$

当：

$$R(t) \leq 0$$

运行必须停止，不能再发起新的模型调用或工具调用。

使用单调时钟而不是墙上时钟，是因为系统时间可能被 NTP 校正、人工修改或容器时间同步影响。Deadline 只关心经过了多少时间，不关心当前日期显示为什么。

4.2 Deadline 与 timeout 的区别

timeout 通常描述一个单独操作最多等待多久：

模型请求 timeout = 30 秒
数据库请求 timeout = 5 秒
搜索接口 timeout = 10 秒

Deadline 描述整个 Agent 运行必须在什么时候结束：

Agent 总 Deadline = 60 秒

如果当前已经过去 55 秒，那么新的工具请求即使配置了 10 秒 timeout，也不能真的再等待 10 秒。正确的单次超时时间应为：

$$\tau_{\text{effective}} = \min(\tau_{\text{configured}}, R(t))$$

例如：

剩余 Deadline：4 秒
工具默认 timeout：10 秒
实际 timeout：4 秒或略小于 4 秒

否则就会出现：

上游 HTTP 请求 60 秒超时
Agent Deadline 30 秒

系统虽然最终会返回，但已经超过了业务允许的总时限。

4.3 Deadline 必须穿透调用栈

一个常见错误是只给最外层协程设置超时：

await asyncio.wait_for(run_agent(), timeout=30)

这可以停止等待，但不一定能停止底层副作用：

• HTTP 请求可能仍在服务器端执行；
• 子进程可能仍未退出；
• 工具可能已经提交了一笔不可撤销的支付；
• 并行任务可能被遗留；
• 重试器可能在外层异常后继续工作。

更可靠的设计是把 deadline 或 remaining_time 作为运行上下文传入每一层：

from dataclasses import dataclass
import time

@dataclass
class RunBudget:
    deadline: float

    def remaining(self) -> float:
        return max(0.0, self.deadline - time.monotonic())

    def ensure_time(self, reserve_seconds: float = 0.0) -> None:
        if self.remaining() <= reserve_seconds:
            raise RuntimeError("deadline_exceeded")

工具调用前：

async def call_search(query: str, budget: RunBudget):
    budget.ensure_time(reserve_seconds=0.2)
    timeout = min(8.0, budget.remaining())
    return await http_search(query, timeout=timeout)

这里的 reserve_seconds 用于保留收尾时间，例如写入运行状态、生成降级响应或释放资源。

4.4 Deadline 的完整算例

设：

总 Deadline：20 秒
模型决策：最多 5 秒
搜索工具：最多 8 秒
数据库工具：最多 4 秒
最终响应收尾：至少预留 1 秒

实际运行：

t=0s：开始运行
t=3s：模型返回，消耗 1 步
t=3s：搜索工具开始，剩余 17 秒
t=10s：搜索返回
t=10s：模型第二次决策开始
t=14s：模型返回
t=14s：数据库工具开始，剩余 6 秒
t=18s：数据库返回
t=18s：预留 1 秒生成最终响应
t=19s：完成

如果数据库在 $t=18s$ 时失败并准备重试，不能直接使用原始的 4 秒 timeout，因为剩余只有 2 秒。运行时应选择：

有效重试 timeout = min(4, 2 - 收尾预留)

如果结果小于零，则不应再重试，而应进入降级或超时终止。

---

五、Token 预算：限制的是模型计算，不只是输出长度

5.1 Token 不是字符数

Token 是模型处理文本的计量单位。它既包括：

• 输入提示词；
• 系统指令；
• 工具定义；
• 对话历史；
• 工具调用参数；
• 工具返回结果；
• 模型输出；
• 某些模型的推理 Token 或隐藏计算 Token；
• 缓存命中或缓存写入部分，具体取决于服务商的计费模型。

因此，Agent 的 Token 消耗通常不是：

用户输入长度 + 最终答案长度

而更接近：

$$T_{\text{total}} = \sum_{i=1}^{n} \left( T_{\text{input},i} + T_{\text{output},i} + T_{\text{reasoning},i} + T_{\text{tool-context},i} \right)$$

工具上下文可能已经包含在输入 Token 中，因此实现统计时不要重复加总；公式是概念分解，实际账单应以供应商返回的 usage 字段为准。

5.2 Agent 中最容易失控的是输入 Token

假设每一步都把完整历史重新发送给模型，且每一步新增 2,000 Token：

第 1 次模型调用：2,000 input tokens
第 2 次模型调用：4,000 input tokens
第 3 次模型调用：6,000 input tokens
第 4 次模型调用：8,000 input tokens
第 5 次模型调用：10,000 input tokens

总输入 Token 为：

$$2000 + 4000 + 6000 + 8000 + 10000 = 30000$$

虽然每一步只新增 2,000 Token，但总输入消耗是 30,000 Token。

如果工具结果每次都包含完整网页、完整日志或完整数据库记录，增长速度会更快：

$$T_{\text{input},i} = T_{\text{固定上下文}} + \sum_{j=1}^{i-1} T_{\text{历史消息},j} + T_{\text{工具结果},i-1}$$

这也是为什么“把所有工具原始结果都塞回上下文”会同时导致 Token、费用、延迟和上下文窗口问题。

5.3 Token 预算应该分层

不建议只定义一个总 Token 数。至少可以拆成：

run_input_token_budget
run_output_token_budget
run_reasoning_token_budget
tool_result_token_budget
final_response_token_budget

例如：

@dataclass
class TokenBudget:
    max_input: int
    max_output: int
    max_total: int
    input_used: int = 0
    output_used: int = 0
    total_used: int = 0

    def can_start_model_call(self) -> bool:
        return self.total_used < self.max_total

但是要注意：在模型调用开始前，通常无法准确知道本次会产生多少输出 Token。因此预算检查分成两类：

1. 准入检查：根据预计输入、最大输出和剩余预算，决定是否允许调用；
2. 结算检查：模型返回后，根据实际 usage 更新账本。

准入检查可以使用：

$$T_{\text{estimated}} = T_{\text{input,estimated}} + T_{\text{max output}}$$

如果：

$$T_{\text{used}} + T_{\text{estimated}} > T_{\max}$$

则可以采取以下动作：

• 压缩历史；
• 截断低价值工具结果；
• 切换小模型；
• 禁止非必要工具；
• 直接生成“预算不足”的降级响应；
• 将任务拆分为可恢复的子任务。

5.4 Token 预算与上下文窗口不是一回事

上下文窗口限制的是单次模型调用能接受多少上下文；Token 预算限制的是整个运行最多消耗多少 Token。

例如：

模型上下文窗口：128,000 Token
Agent 总 Token 预算：40,000 Token

即使单次调用没有超过 128,000 Token，整个 Agent 仍然可能在多次调用后超过 40,000 Token。

反过来：

Agent 总 Token 预算：200,000 Token
模型上下文窗口：32,000 Token

总预算很大，但每次调用仍必须压缩历史，否则单次调用无法执行。

OpenAI 当前 Agents 文档也将会话状态、结果、续接和上下文管理作为运行时的一等问题，而不是把 Agent 只看作一次独立模型调用。(developers.openai.com)

---

六、费用预算：Token 费用只是总费用的一部分

6.1 模型费用的基本公式

设一次模型调用的价格为：

• 输入价格：$p_{\text{in}}$，单位为货币 / Token；
• 输出价格：$p_{\text{out}}$；
• 推理 Token 价格：$p_{\text{reasoning}}$；
• 缓存输入价格：$p_{\text{cached}}$。

则第 $i$ 次模型调用的费用可以表示为：

$$C_{\text{model},i} = T_{\text{in},i}p_{\text{in}} + T_{\text{out},i}p_{\text{out}} + T_{\text{reasoning},i}p_{\text{reasoning}} + T_{\text{cached},i}p_{\text{cached}}$$

总模型费用为：

$$C_{\text{model}} = \sum_i C_{\text{model},i}$$

实际系统应使用供应商响应中的 usage 和当前价格配置结算，不应把价格硬编码在 Agent 逻辑中。价格会随模型、区域、批处理、缓存和服务商政策变化，本文不指定具体模型价格。

6.2 工具费用必须纳入统一预算

完整运行费用应为：

$$C_{\text{run}} = C_{\text{model}} + C_{\text{search}} + C_{\text{database}} + C_{\text{browser}} + C_{\text{code}} + C_{\text{storage}} + C_{\text{network}}$$

例如：

模型调用费用：0.018 元
搜索 API：0.006 元
浏览器执行：0.030 元
代码沙箱：0.012 元
总费用：0.066 元

如果只记录模型费用，账面上会低估真实成本。

此外，工具还可能产生不可直接货币化但必须治理的成本：

• 第三方 API 配额；
• 数据库查询压力；
• 浏览器会话数量；
• 沙箱 CPU 和内存；
• 外部系统写操作；
• 人工审核成本；
• 失败重试引起的重复副作用。

6.3 预估费用与实际费用

一次模型调用存在两个费用值：

estimated_cost：调用前估算
actual_cost：调用后根据 usage 结算

调用前可用上界估算：

$$\hat{C}_i = \hat{T}_{\text{in},i}p_{\text{in}} + T_{\text{max-out},i}p_{\text{out}}$$

调用后使用实际 Token：

$$C_i = T_{\text{actual-in},i}p_{\text{in}} + T_{\text{actual-out},i}p_{\text{out}}$$

如果调用前剩余预算只有 0.01 元，而最坏情况下本次调用可能产生 0.03 元，就不能因为“平均费用通常很低”而放行。预算治理必须对不确定性负责。

6.4 并发会造成预算超卖

假设当前剩余费用预算为 0.05 元，同时准备并发启动三个工具或模型任务：

任务 A 预计最多 0.03 元
任务 B 预计最多 0.03 元
任务 C 预计最多 0.02 元

如果每个任务独立检查：

A：发现剩余 0.05，允许
B：发现剩余 0.05，允许
C：发现剩余 0.05，允许

最终最坏费用是：

$$0.03 + 0.03 + 0.02 = 0.08$$

预算被超卖。

解决方式是原子预留：

$$\text{reserved} + \text{estimate} \leq \text{limit}$$

伪代码：

import asyncio

class CostLedger:
    def __init__(self, limit: float):
        self.limit = limit
        self.reserved = 0.0
        self.actual = 0.0
        self.lock = asyncio.Lock()

    async def reserve(self, estimate: float) -> bool:
        async with self.lock:
            if self.actual + self.reserved + estimate > self.limit:
                return False
            self.reserved += estimate
            return True

    async def settle(self, estimate: float, actual: float) -> None:
        async with self.lock:
            self.reserved -= estimate
            self.actual += actual

调用流程：

estimate = 0.03

if not await ledger.reserve(estimate):
    raise RuntimeError("cost_budget_exceeded")

try:
    result = await call_model()
    actual = calculate_actual_cost(result.usage)
finally:
    await ledger.settle(estimate, actual)

风险在于：如果进程崩溃后没有释放预留费用，账本会出现“幽灵预留”。生产系统需要把预留记录持久化，并通过租约、过期时间或后台对账恢复。

---

七、预算不是一个数字，而是一组可组合的约束

可以把 Agent 预算表示为向量：

$$B = (B_s, B_t, B_c, B_a, B_r)$$

其中：

• $B_s$：最大步数；
• $B_t$：Token 预算；
• $B_c$：费用预算；
• $B_a$：工具动作或调用预算；
• $B_r$：资源预算，例如 CPU、内存、浏览器会话数。

一个动作是否允许执行，不是看单一条件，而是检查：

$$\text{allow}(a) = \text{within\_deadline} \land \text{steps\_available} \land \text{tokens\_available} \land \text{cost\_available} \land \text{tool\_quota\_available} \land \text{policy\_allowed}$$

其中 policy_allowed 很重要：即使还有步数和费用，也不能因为预算充足就允许高风险写操作。

7.1 硬预算、软预算和预警线

预算可以分成三个区间：

0% ～ 70%：正常运行
70% ～ 90%：进入节流或降级准备
90% ～ 100%：只允许收尾或低成本操作
超过 100%：终止

例如 Token 预算为 20,000：

used < 14,000：正常
14,000 <= used < 18,000：压缩上下文、限制搜索
18,000 <= used < 20,000：禁止探索，只允许总结
used >= 20,000：终止或返回部分结果

软预算的作用是避免在预算耗尽时突然失去输出能力。若 Agent 已经找到主要事实，却因为最后一次总结调用没有预算而只返回异常，用户体验和可恢复性都会很差。

7.2 为最终响应预留预算

如果 Agent 只在工具循环中消耗预算，可能出现：

工具阶段：
  Token 已使用 9,800 / 10,000
最终总结：
  预计需要 1,500 Token

此时即使任务事实上已经完成，也没有足够预算生成可读响应。

因此可以定义：

$$B_{\text{working}} = B_{\text{total}} - B_{\text{final}}$$

例如：

总 Token 预算：10,000
最终响应预留：1,500
工作阶段预算：8,500

当工作阶段达到 8,500 Token 时，Agent 应停止探索并进入总结路径。

同样的思想适用于 Deadline：

总 Deadline：30 秒
最终收尾预留：2 秒
工具和模型阶段最多运行：28 秒

---

八、循环检测：识别“重复动作”与“无进展”

8.1 什么是循环

Agent 循环不是简单的“同一个工具被调用两次”。

有些重复是合理的：

第一次读取文件
修改文件
第二次读取文件

第二次读取虽然工具相同，但状态已经改变，具有验证意义。

真正需要检测的是：

在状态没有产生有意义变化的情况下，Agent 重复执行相同或等价的动作，且没有增加完成目标的可能性。

因此循环检测至少需要比较三类信息：

1. 动作是否重复；
2. 观察结果是否重复；
3. 任务状态是否产生进展。

8.2 动作指纹

可以为一次工具调用构造规范化指纹：

$$F_{\text{action}} = \operatorname{hash} ( \text{tool\_name} \Vert \operatorname{canonicalize}(\text{arguments}) )$$

例如下面两个调用应具有相同指纹：

{"tool": "search", "arguments": {"q": "杭州天气", "limit": 10}}

{"arguments": {"limit": 10, "q": "杭州天气"}, "tool": "search"}

因为 JSON 字段顺序不同不应被认为是不同动作。

示例实现：

import hashlib
import json

def action_fingerprint(tool_name: str, arguments: dict) -> str:
    payload = {
        "tool": tool_name,
        "arguments": arguments,
    }
    raw = json.dumps(
        payload,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
    )
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()

8.3 简单的重复动作检测

from collections import deque

class LoopDetector:
    def __init__(self, window_size: int = 6, repeat_threshold: int = 3):
        self.recent = deque(maxlen=window_size)
        self.repeat_threshold = repeat_threshold

    def observe(self, fingerprint: str, progressed: bool) -> bool:
        if progressed:
            self.recent.clear()
            self.recent.append(fingerprint)
            return False

        self.recent.append(fingerprint)
        return self.recent.count(fingerprint) >= self.repeat_threshold

行为示例：

第 1 步：search(A)，有新结果，progressed=True
第 2 步：search(A)，结果相同，progressed=False
第 3 步：search(A)，结果相同，progressed=False
第 4 步：search(A)，结果相同，progressed=False

当阈值为 3 时，第 4 步触发循环终止。

但是这个实现仍然过于简单，因为它无法识别交替循环：

A -> B -> A -> B -> A -> B

也无法识别参数表面变化但语义相同的动作：

search("天气 杭州")
search("杭州 天气")
search("杭州天气")

8.4 状态指纹比动作指纹更重要

可以定义一个任务状态摘要：

$$F_{\text{state}} = \operatorname{hash} ( \text{goal\_facts} \Vert \text{completed\_subtasks} \Vert \text{artifacts} \Vert \text{errors} )$$

如果连续多个步骤满足：

$$F_{\text{state},i} = F_{\text{state},i-1}$$

同时动作发生重复或等价变化，则说明 Agent 没有产生有效进展。

例如：

第 1 步：
  动作：读取 issue #123
  状态：已知 bug 位于 payment.py

第 2 步：
  动作：读取 issue #123
  状态：仍然只知道 bug 位于 payment.py

第 3 步：
  动作：读取 issue #123
  状态：仍然只知道 bug 位于 payment.py

这里不仅动作重复，任务状态也没有变化。

反例：

第 1 步：读取 payment.py
第 2 步：修改 payment.py
第 3 步：运行测试
第 4 步：读取新的测试错误

工具名称可能重复，但状态在持续变化，不能简单判定为循环。

8.5 进展函数

为了形式化“进展”，可以定义一个进展函数：

$$P(S_i) \in \mathbb{R}$$

它表示当前状态距离任务完成的近似程度。若：

$$P(S_{i+1}) - P(S_i) \leq \epsilon$$

持续多个步骤，就可以认为 Agent 进入停滞。

例如代码修复任务可以用：

P(S) =
  通过测试数
  - 未通过测试数
  - 静态检查错误数

如果测试结果如下：

第 1 次：通过 8，失败 4，P=4
第 2 次：通过 8，失败 4，P=4
第 3 次：通过 8，失败 4，P=4

虽然 Agent 可能每次都修改了代码，但没有改善结果，可以触发“无进展终止”。

需要注意，进展函数不是通用真理。研究任务可能短时间内没有新事实，但正在等待搜索结果；客服任务中“确认用户身份”本身可能不增加答案内容，却是必要步骤。因此进展检测应结合任务类型和工具结果，而不是只看一个数值。

8.6 循环检测的几种层级

层级一：完全重复

相同工具 + 相同参数 + 相同上下文

成本低、误报少，适合默认启用。

层级二：窗口内重复

最近 N 步中，同一动作出现 K 次

可以检测间歇性重复，但需要调节窗口和阈值。

层级三：交替循环

A -> B -> A -> B

适合处理“读取状态失败后重复刷新”“搜索后又回到同一搜索”等模式。

层级四：状态停滞

连续 N 步没有新增事实、产物、通过测试或完成子任务

比动作重复更强，但任务适配成本更高。

层级五：语义重复

使用模型或向量相似度判断两个动作是否等价，例如：

查询订单状态
获取订单当前状态

这种方法可能检测出更多循环，但它本身需要额外模型调用或计算，不能在预算紧张的运行中无限使用。

---

九、终止原因必须是结构化数据

不要只返回：

Agent failed

应返回结构化终止信息：

{
  "status": "terminated",
  "reason": "loop_detected",
  "steps_used": 7,
  "max_steps": 12,
  "input_tokens": 8420,
  "output_tokens": 1930,
  "total_tokens": 10350,
  "estimated_cost": 0.071,
  "deadline_ms": 30000,
  "elapsed_ms": 18420,
  "last_action": {
    "tool": "search_orders",
    "arguments_hash": "..."
  },
  "recoverable": true,
  "partial_result_available": true
}

推荐至少区分以下状态：

completed
completed_with_degradation
paused_for_approval
terminated_by_max_steps
terminated_by_deadline
terminated_by_token_budget
terminated_by_cost_budget
terminated_by_loop
failed_tool_error
failed_model_error
cancelled

这些状态不能全部归为 failed，因为恢复策略不同：

状态	是否可恢复	常见处理
paused_for_approval	是	从原状态继续
terminated_by_max_steps	通常是	人工调整计划或继续运行
terminated_by_deadline	视任务而定	返回部分结果或异步继续
terminated_by_loop	通常是	修改参数、换工具或人工介入
terminated_by_cost_budget	是	提升预算或切换模型
failed_tool_error	视错误而定	重试、切换工具或终止
cancelled	是	用户确认后恢复

OpenAI 文档把最大 turn 限制、guardrail 异常和工具错误归为运行时或校验失败，同时把人工审批视为预期的暂停状态；这正说明“暂停”和“失败”不应使用同一个控制分支。(developers.openai.com)

---

十、完整的运行时控制器

下面给出一个不绑定具体模型 SDK 的最小运行时示例。它展示预算、Deadline、工具调用和循环检测如何组合，而不是某个供应商的固定 API。

from __future__ import annotations

import asyncio
import time
from dataclasses import dataclass, field
from typing import Any


class StopRun(Exception):
    def __init__(self, reason: str):
        self.reason = reason
        super().__init__(reason)


@dataclass
class Budget:
    max_steps: int
    max_tokens: int
    max_cost: float
    deadline: float
    final_token_reserve: int = 500
    final_time_reserve: float = 1.0


@dataclass
class Usage:
    tokens: int = 0
    cost: float = 0.0


@dataclass
class RunState:
    steps_used: int = 0
    usage: Usage = field(default_factory=Usage)
    facts: set[str] = field(default_factory=set)
    recent_actions: list[str] = field(default_factory=list)
    history: list[dict[str, Any]] = field(default_factory=list)


def remaining_time(budget: Budget) -> float:
    return max(0.0, budget.deadline - time.monotonic())


def check_budget(state: RunState, budget: Budget) -> None:
    if state.steps_used >= budget.max_steps:
        raise StopRun("max_steps")

    if remaining_time(budget) <= budget.final_time_reserve:
        raise StopRun("deadline")

    if state.usage.tokens >= budget.max_tokens - budget.final_token_reserve:
        raise StopRun("token_budget")

    if state.usage.cost >= budget.max_cost:
        raise StopRun("cost_budget")


def detect_loop(state: RunState, action_key: str, progressed: bool) -> None:
    if progressed:
        state.recent_actions.clear()

    state.recent_actions.append(action_key)
    state.recent_actions = state.recent_actions[-6:]

    # 最近 6 次动作中，同一动作至少出现 3 次且没有进展
    if not progressed and state.recent_actions.count(action_key) >= 3:
        raise StopRun("loop_detected")


async def run_agent(initial_input: str, budget: Budget) -> dict[str, Any]:
    state = RunState()
    state.history.append({"role": "user", "content": initial_input})

    try:
        while True:
            check_budget(state, budget)

            # 真实实现中，这里应调用模型并读取供应商返回的 usage
            decision = await fake_model(state)

            state.steps_used += 1
            state.usage.tokens += decision["tokens"]
            state.usage.cost += decision["cost"]

            if decision["type"] == "final":
                return {
                    "status": "completed",
                    "answer": decision["answer"],
                    "steps_used": state.steps_used,
                    "tokens": state.usage.tokens,
                    "cost": state.usage.cost,
                }

            if decision["type"] != "tool":
                raise StopRun("invalid_decision")

            tool_name = decision["tool"]
            arguments = decision["arguments"]
            action_key = f"{tool_name}:{sorted(arguments.items())}"

            timeout = min(8.0, remaining_time(budget))
            if timeout <= 0:
                raise StopRun("deadline")

            result = await asyncio.wait_for(
                execute_tool(tool_name, arguments),
                timeout=timeout,
            )

            before = len(state.facts)
            state.facts.update(result["facts"])
            progressed = len(state.facts) > before

            detect_loop(state, action_key, progressed)

            state.history.append({
                "role": "tool",
                "tool": tool_name,
                "result": result,
            })

    except StopRun as stop:
        return {
            "status": "terminated",
            "reason": stop.reason,
            "steps_used": state.steps_used,
            "tokens": state.usage.tokens,
            "cost": state.usage.cost,
            "partial_facts": sorted(state.facts),
        }


async def fake_model(state: RunState) -> dict[str, Any]:
    await asyncio.sleep(0.05)

    if len(state.facts) >= 2:
        return {
            "type": "final",
            "answer": "已完成查询。",
            "tokens": 120,
            "cost": 0.002,
        }

    return {
        "type": "tool",
        "tool": "search",
        "arguments": {"q": "example"},
        "tokens": 180,
        "cost": 0.003,
    }


async def execute_tool(tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
    await asyncio.sleep(0.05)
    return {
        "facts": {"事实 A", "事实 B"}
    }


async def main():
    budget = Budget(
        max_steps=5,
        max_tokens=2000,
        max_cost=0.02,
        deadline=time.monotonic() + 5,
    )
    result = await run_agent("查找资料并总结", budget)
    print(result)


if __name__ == "__main__":
    asyncio.run(main())

代码中的关键路径

1. Budget.deadline 使用单调时钟计算绝对截止时间；
2. 每次开始模型调用前执行 check_budget；
3. 模型返回后才增加 steps_used，表示这一轮决策已经真实发生；
4. 模型返回的 usage 应进入 Token 和费用账本；
5. 工具 timeout 取配置值与剩余 Deadline 的较小值；
6. 工具结果写回状态后，通过新增事实判断是否产生进展；
7. 没有进展且同一动作重复达到阈值时，终止运行；
8. 终止响应保留部分事实，便于降级展示或恢复。

示例中的 fake_model、execute_tool 只是可运行的演示替身。真实实现必须使用模型服务返回的实际 usage，而不能使用估算值冒充结算值。

---

十一、工具重试不能绕过 Agent 预算

工具失败后重试通常有两种层级：

Agent 层重试：
  让模型重新判断下一步

工具层重试：
  同一个工具调用自动再次请求

例如：

第 1 步：模型决定调用 search
工具第 1 次尝试：网络超时
工具第 2 次尝试：成功

如果工具重试由运行时透明完成，可以记录：

agent_steps = 1
tool_calls = 1
tool_attempts = 2

但工具重试必须消耗：

• 剩余 Deadline；
• 工具调用配额；
• 外部 API 配额；
• 可能的费用预算。

不能因为“还没有进入下一次模型决策”就认为没有消耗资源。

指数退避也必须受 Deadline 约束。设第 $k$ 次重试等待时间为：

$$b_k = \min(b_{\max}, b_0 2^k + J)$$

其中 $J$ 是抖动值。实际等待必须满足：

$$b_k < R(t) - T_{\text{reserve}}$$

否则应该放弃重试。

不应重试的工具错误

以下错误通常不适合盲目重试：

• 参数校验失败；
• 权限不足；
• 资源不存在；
• 幂等性不明确的写操作；
• 已经确认提交成功但响应丢失；
• 费用预算不足；
• Deadline 已不足以完成收尾。

特别是写操作：

create_refund()
send_email()
place_order()
delete_file()

不能像读取操作一样默认重试。必须使用幂等键、提交状态查询或人工确认，否则一次网络超时可能造成重复副作用。

---

十二、部分结果与降级路径

预算耗尽并不等于所有工作都无效。

例如研究 Agent 已经完成：

已搜索 8 个来源
已提取 5 个关键事实
尚未完成最终交叉验证

如果此时达到 Token 预算，系统可以返回：

status: completed_with_degradation
confidence: low
termination_reason: token_budget
partial_result: ...
missing_checks: ...

但降级不能伪装成完整成功。至少应明确：

• 哪些事实已经确认；
• 哪些事实只是单一来源；
• 哪些步骤没有执行；
• 是否存在未完成的工具调用；
• 用户能否继续运行；
• 继续运行需要什么预算。

一个合理的降级函数可以是：

$$R = \begin{cases} \text{完整答案}, & \text{任务完成}\\ \text{部分答案 + 缺口说明}, & \text{有可靠中间结果}\\ \text{失败原因 + 恢复建议}, & \text{没有可用结果} \end{cases}$$

不要在 Deadline 到期后再启动一次昂贵的“总结模型调用”。如果要生成降级响应，应使用预留的小预算，或者由确定性模板直接拼装。

---

十三、暂停、取消和恢复

13.1 暂停不是终止

人工审批是典型的暂停：

Agent 已经生成：
  “准备向用户退款 500 元”
系统要求人工审批
运行暂停

暂停状态应保存：

run_id
current_agent
history
pending_tool_call
budget_used
deadline_policy
approval_request
state_version

恢复时继续使用原状态，而不是把审批结果当成新的用户请求重新开始。

13.2 Deadline 跨暂停的语义

暂停时有两种合理策略：

策略一：Deadline 继续流逝

适用于：

• 实时客服；
• 浏览器会话；
• 短时交易流程。

如果暂停 30 秒后恢复，而原 Deadline 已到，则直接终止。

策略二：暂停期间冻结运行预算

适用于：

• 人工审核；
• 长时间异步审批；
• 可恢复后台任务。

此时需要记录：

paused_at
resumed_at
paused_duration

并将暂停时间从运行 Deadline 中扣除或单独计算。

两种策略都可以，但必须明确写入运行协议。否则同一个任务在不同恢复时间下会出现不一致行为。

13.3 取消必须可传播

用户取消后，取消信号应传播到：

Agent 主循环
当前模型请求
所有并行工具
重试任务
子 Agent
子进程或沙箱

只取消最外层任务而不取消内部工作，会造成资源泄漏和隐藏费用。

---

十四、观测：没有终止原因，就无法治理成本

每次 Agent 运行至少需要记录以下字段：

run_id
parent_run_id
task_type
model
agent_steps
model_calls
tool_calls
tool_attempts
input_tokens
output_tokens
reasoning_tokens
cached_tokens
estimated_cost
actual_cost
elapsed_ms
deadline_ms
termination_reason
last_tool
loop_detector_hits
partial_result

还应记录每一步的事件：

{
  "run_id": "r-123",
  "step": 3,
  "event": "tool_completed",
  "tool": "search_orders",
  "duration_ms": 842,
  "input_hash": "abc",
  "result_hash": "def",
  "progressed": false,
  "tokens_total": 8200,
  "cost_total": 0.061
}

注意不要把完整用户隐私、支付参数或敏感工具结果直接写入日志。生产观测通常应记录哈希、摘要、字段级脱敏结果和版本号。

OpenAI 的 Agent 运行文档将工具、handoff、审批、流式输出和状态续接都放在同一个运行循环中；其集成与观测文档也把 tracing 作为理解 Agent 运行的重要表面。(developers.openai.com)

14.1 关键指标

终止原因分布

completed：82%
max_steps：4%
deadline：7%
token_budget：3%
cost_budget：1%
loop_detected：2%
tool_error：1%

如果 deadline 突然上升，可能是工具变慢、模型 TTFT 增长或并发排队导致。

步数与成功率

1～2 步：成功率 91%
3～5 步：成功率 88%
6～10 步：成功率 73%
超过 10 步：成功率 41%

这可能说明长运行并没有带来足够的质量收益，也可能说明任务本身需要更强的规划器。不能只看平均步数，应同时观察成功率、成本和用户价值。

Token 与费用的异常点

工具结果 Token 占比过高
输入 Token 随步数二次增长
输出 Token 远低于预留值
失败运行费用高于成功运行

这些指标比单纯的平均 Token 更能定位治理问题。

---

十五、常见误解与反例

误解一：设置 max_steps 就不会死循环

错误原因：最大步数只限制总次数，不判断动作是否重复。

修正方式：

max_steps + 动作指纹 + 状态进展检测

误解二：模型说“完成了”就应该停止

模型输出的自然语言不是可靠的完成信号。应使用可验证条件：

订单状态已更新
测试命令返回成功
文件已写入并通过校验
所有必需字段已填充
审批已通过

完成条件最好由工具结果或确定性检查器确认。

误解三：Token 少，费用就一定低

错误原因：

• 某些模型的输入和输出价格不同；
• 工具和搜索可能独立收费；
• 缓存、推理、批处理有不同计费方式；
• 重试会重复产生费用；
• 并行调用会同时放大成本。

Token 是费用的重要输入，但不是费用本身。

误解四：HTTP timeout 等于 Agent Deadline

错误原因：单个请求 timeout 不限制其他请求、重试、排队、工具内部工作和收尾时间。

修正方式：

总 Deadline
  -> 模型 timeout
  -> 工具 timeout
  -> 重试预算
  -> 并行任务取消
  -> 最终收尾预留

误解五：并发工具可以共享同一个剩余预算而无需锁

错误原因：多个任务可能同时通过预算检查，造成超卖。

修正方式：

原子预留
实际结算
崩溃恢复
租约过期

误解六：重复调用工具一定是循环

错误原因：重复动作可能是验证步骤。

例如：

修改文件
读取文件确认修改
运行测试
再次读取文件定位测试错误

判断循环必须考虑状态是否发生有意义变化。

误解七：终止后重新发起一个新请求即可恢复

错误原因：新请求可能丢失原历史、预算和未完成工具状态。

正确做法是持久化可恢复状态，并区分：

继续同一次运行
开始新的用户回合
从部分结果重新规划

---

十六、推荐的终止决策顺序

一次新的 Agent 决策开始前，可以采用以下顺序：

1. 检查取消信号
2. 检查是否已经完成
3. 检查 Deadline
4. 检查最大步数
5. 检查 Token 预算
6. 检查费用预算
7. 检查工具配额和策略权限
8. 检查循环或停滞状态
9. 原子预留本次调用资源
10. 发起模型调用
11. 结算实际 usage 和费用
12. 执行工具或返回最终结果
13. 写入状态与观测事件

其中“是否已经完成”应优先于“是否还有预算”。如果任务已经满足确定性完成条件，就不应为了生成更多解释而继续运行。

而“预算检查”不能完全放在模型调用前。模型返回后还需要重新检查：

模型调用可能已经耗尽 Token
模型输出可能触发高风险工具
模型输出可能暴露循环
模型调用可能已经越过 Deadline

---

十七、工程上的最低基线

一个可上线的 Agent 运行时，至少应保证以下性质：

有界性

无论模型输出什么，都满足：

$$\text{steps} \leq N_{\max}$$

$$t \leq t_{\text{deadline}}$$

$$C_{\text{run}} \leq C_{\max} + \delta$$

其中 $\delta$ 是不可避免的结算误差、并发在途费用或供应商账单延迟。若系统需要严格不超支，就必须采用更保守的预留策略。

可解释性

每次终止都能回答：

为什么停止？
停止时走了几步？
用了多少 Token？
花了多少钱？
最后一个动作是什么？
有没有部分结果？
还能否恢复？

可恢复性

暂停、超时、预算耗尽和临时工具失败不能都丢失状态。至少应保存：

版本化状态
历史或续接引用
预算账本
最后完成的工具结果
未完成的动作
终止原因

可验证性

完成不能只依赖模型自报。应通过：

工具返回状态
数据库状态
测试结果
Schema 校验
业务规则
人工审批

可降级性

预算接近耗尽时，系统应从“探索模式”切换为：

收集已有事实
停止非必要工具
压缩上下文
生成部分结果
明确说明缺口

Agent 的自动化程度越高，潜在成本和错误累积越高；Anthropic 也明确指出，Agent 的自主性会带来更高成本和复合错误风险，因此需要测试、沙箱和适当的防护机制。(anthropic.com)

---

Agent 的终止机制本质上是在回答一个运行时问题：

在当前状态下，继续运行带来的预期收益，是否仍然大于它的时间、Token、费用和副作用成本？

最大步数提供数量上界，Deadline 提供时间上界，Token 预算提供计算上界，费用预算提供经济上界，循环检测提供进展约束。只有将它们统一到同一个有状态运行时中，Agent 才不会变成一个“偶尔成功、失败时难以解释、成本无法预测”的黑盒循环。

---

系列导航与关联阅读

• 系列入口：Agent 工程完整路线：从运行循环、记忆与协议到安全、评测和生产交付
• 上一篇：Agent 不确定性与拒答：证据不足、置信边界、升级和回退
• 下一篇：Agent 人工介入：确认、澄清、升级、接管、恢复和责任边界
• 延伸：Agent 运行循环：观察、决策、行动、反馈、状态与终止
• 延伸：Agent 延迟与成本：TTFT、步骤、Token、工具耗时、预算和降级

官方资料

• OpenAI Agents Guide
• Anthropic Building Effective Agents

本文依据 Agent、模型、协议与框架官方资料重新梳理；正文、示例与生产清单由 WR BLOG 编写。