Agent 多轮循环运行预算：max_turns、墙钟 deadline、token/费用上限与取消传播

生产 Agent 的一次用户请求通常不是单次 completion，而是 「模型 →（可选）tool_calls → 回灌 → 再模型」 的多轮循环，直到模型不再发起工具、用户点 Stop、或平台强制收尾。单工具超时、结果字节预算、并行信号量只能约束一步；若缺少 run 级预算 与 取消传播，仍会出现：模型反复「再试一次」烧光配额、用户已离开浏览器但 shell 进程组仍在跑、或墙钟超时后只停了 HTTP 响应却未杀下游。本文只讨论 单次 run（一次用户触发的完整循环） 的预算与取消，不重复各工具自身的 SSRF、路径边界、进程组杀与回灌截断设计。

问题背景：单步控件挡不住「无限环」

典型数据流：

1. 客户端发起 run（HTTP/SSE/WebSocket），携带会话 id、用户消息与可选 run_id。
2. 编排器进入循环：调用聊天 / Responses API → 解析是否含 tool_calls / tool_use → 执行工具 → 写回 tool 结果 → 再请求模型。
3. 结束条件本应是：模型给出最终文本且无工具调用、或显式 finish/end_turn、或预算耗尽、或取消。

在缺少 run 级控件时，下列问题会叠加：

• 轮次无界：模型对失败工具反复重试、或在「搜索 — 阅读 — 再搜索」上抖动，一轮对话可产生数十次 completion 与数百次 tool 调用。OWASP GenAI 将过度代理（Excessive Agency）列为独立风险面：权限与步数未限时，危害随循环放大。
• 墙钟与用户预期脱节：UI 显示「思考中」90s 后用户刷新页面；若服务端只以连接存活为生命周期，Worker 可能继续跑完整个长任务并继续计费。
• token / 费用与单结果预算正交：max_result_tokens 限制的是单次 tool 回灌；整次 run 的 prompt+completion 累计仍可线性增长（历史消息膨胀、每轮再附 tool 结果）。
• 取消不是布尔开关：用户 Stop、网关 idle timeout、run deadline、单 tool timeout、提供商 429 重试退避，五者时间轴不同。只 break 循环而不取消飞行中的 HTTP / 子进程，会留下孤儿副作用（半写文件、未提交 SQL、未回收浏览器页）。
• 流式与非流式路径分叉：SSE 上客户端断开可被服务端感知（写失败 / req.aborted）；纯异步「提交后轮询」的 run 没有天然 disconnect，必须依赖显式 cancel API 与 lease 心跳。

因此应把一次 run 建模为带预算与协作取消的状态机，而不是 while True：

accept(run)
  → 初始化 RunBudget + CancelToken
  → loop:
      if budget_exhausted or cancel_requested → finalize(reason)
      model_call(stream, linked to token)
      if tool_calls:
          execute_tools(linked to token, per-call timeout ≤ remaining wall)
          append results (经既有回灌预算/脱敏)
      else:
          finalize(completed)
  → persist terminal status; 释放租约与下游资源

可落地实现：四级预算、状态机与取消顺序

1. 四级预算：轮次、墙钟、token、费用

不要只设一个「最大步数」。四类上限正交，任一项触发即进入 受控收尾（见下节），而不是继续循环：

说明：

• turn 的定义要写死：推荐「一次模型 HTTP 请求 = 1 turn」，并行 tool_calls 不额外增加 turn，但计入 max_tool_calls_per_run。
• token 以提供商 usage 为准：流式结束后读响应中的 usage（或厂商等价字段）；本地 tokenizer 只作预检与截断，不作为计费唯一真相。
• 费用表版本化：模型单价变更时，预算比较必须带 price_book_version，审计日志保留当次单价快照。

Python 示意（预算结构与检查）：

from __future__ import annotations

import time
from dataclasses import dataclass, field


@dataclass
class RunBudget:
    max_model_turns: int = 16
    max_tool_calls: int = 64
    run_deadline_s: float = 120.0
    max_input_tokens: int = 500_000
    max_output_tokens: int = 64_000
    max_cost_micros: int = 50_000  # 例：5 分 USD = 50_000 微美元，按业务改
    # 单价：micros per token（示例占位，生产来自配置中心）
    input_price_micros: int = 2
    output_price_micros: int = 8

    model_turns: int = 0
    tool_calls: int = 0
    input_tokens: int = 0
    output_tokens: int = 0
    cost_micros: int = 0
    started_mono: float = field(default_factory=time.monotonic)

    def remaining_wall_s(self) -> float:
        return self.run_deadline_s - (time.monotonic() - self.started_mono)

    def reason_if_exhausted(self) -> str | None:
        if self.model_turns >= self.max_model_turns:
            return "max_model_turns"
        if self.tool_calls >= self.max_tool_calls:
            return "max_tool_calls"
        if self.remaining_wall_s() <= 0:
            return "run_deadline"
        if self.input_tokens >= self.max_input_tokens:
            return "max_input_tokens"
        if self.output_tokens >= self.max_output_tokens:
            return "max_output_tokens"
        if self.cost_micros >= self.max_cost_micros:
            return "max_cost"
        return None

    def note_usage(self, *, input_tokens: int, output_tokens: int) -> None:
        self.input_tokens += input_tokens
        self.output_tokens += output_tokens
        self.cost_micros += (
            input_tokens * self.input_price_micros
            + output_tokens * self.output_price_micros
        )

2. 循环骨架：先检预算，再绑定剩余墙钟

每轮进入模型调用前：

1. reason = budget.reason_if_exhausted()；非空则 finalize(reason)，不再发模型请求。
2. 为本轮设置 per_model_timeout_s = min(provider_default, budget.remaining_wall_s() - safety_margin_s)。
3. 工具批：per_call_timeout_s = min(tool_default, budget.remaining_wall_s() - safety_margin_s)；若剩余墙钟已小于 min_useful_tool_s（如 2s），直接跳过执行并回灌 skipped: run_deadline_near。

import asyncio
from typing import Any, Callable, Awaitable

safety_margin_s = 1.5
min_useful_tool_s = 2.0

async def agent_run(
    *,
    budget: RunBudget,
    cancel: asyncio.Event,  # 用户 Stop / 断连 时 set
    call_model: Callable[..., Awaitable[dict[str, Any]]],
    run_tools: Callable[..., Awaitable[list[dict[str, Any]]]],
    on_event: Callable[[dict[str, Any]], Awaitable[None]],
) -> dict[str, Any]:
    messages: list[dict[str, Any]] = []  # 已含系统与用户消息
    terminal = "completed"

    while True:
        if cancel.is_set():
            terminal = "cancelled"
            break
        reason = budget.reason_if_exhausted()
        if reason:
            terminal = reason
            break

        rem = budget.remaining_wall_s()
        if rem <= safety_margin_s:
            terminal = "run_deadline"
            break

        budget.model_turns += 1
        try:
            model_out = await asyncio.wait_for(
                call_model(messages, cancel=cancel),
                timeout=max(0.1, rem - safety_margin_s),
            )
        except asyncio.TimeoutError:
            terminal = "run_deadline"
            break
        except asyncio.CancelledError:
            terminal = "cancelled"
            raise

        budget.note_usage(
            input_tokens=int(model_out.get("input_tokens", 0)),
            output_tokens=int(model_out.get("output_tokens", 0)),
        )
        await on_event({"type": "model_done", "turn": budget.model_turns})

        calls = model_out.get("tool_calls") or []
        if not calls:
            messages.append(model_out["assistant_message"])
            terminal = "completed"
            break

        # 工具批：受 run 级 tool 上限与剩余墙钟约束
        room = budget.max_tool_calls - budget.tool_calls
        if room <= 0:
            terminal = "max_tool_calls"
            break
        if len(calls) > room:
            # 超出部分不执行，回灌明确错误，避免模型以为调用成功
            overflow = calls[room:]
            calls = calls[:room]
        else:
            overflow = []

        rem = budget.remaining_wall_s()
        if rem < min_useful_tool_s:
            terminal = "run_deadline"
            break

        budget.tool_calls += len(calls)
        tool_results = await run_tools(
            calls,
            cancel=cancel,
            per_call_timeout_s=min(60.0, rem - safety_margin_s),
        )
        for c in overflow:
            tool_results.append({
                "tool_call_id": c["id"],
                "is_error": True,
                "content": "rejected: max_tool_calls_per_run exceeded",
            })
        # 此处接既有：结果字节/token 预算、脱敏、按 tool_call_id 写回
        messages.extend(model_out["assistant_message_with_calls"])
        messages.extend(format_tool_messages(tool_results))

    await on_event({"type": "run_terminal", "status": terminal, "budget": snapshot(budget)})
    return {"status": terminal, "messages": messages}

要点：

• 预算检查在循环顶部与昂贵操作前各做一次，避免「已超限仍发起最后一次长 completion」。
• overflow tool_calls 必须回灌错误结果：提供商要求每个 tool_call_id 有对应结果；静默丢弃会导致下一轮上下文非法或模型重复调用。
• safety_margin_s：为取消传播、span 导出与状态落库留出时间，避免 deadline 踩点时来不及杀子进程。

3. 取消传播顺序：从边缘到副作用源

取消源应归一为单一 CancelToken / asyncio.Event / context.Context + AbortSignal，避免「UI 停了、Worker 不知」：

传播顺序（推荐）：

1. 置 cancel 并拒绝新的模型 turn / 新的 tool 准入。
2. 取消飞行中的模型流：中止对提供商的 HTTP/2 流（SDK 的 abort / 关闭 body）；停止向客户端写 token delta。
3. 取消飞行中的 tool 批：对每条 in-flight 调用触发既有终止路径 ——HTTP 工具 AbortSignal、shell 进程组 SIGTERM→SIGKILL、SQL pg_cancel_backend / 连接 close、浏览器关 page。
4. 持久化：已完成的 tool 结果与部分 assistant 文本应落库（便于审计与「继续」）；未完成的 call 写 status=cancelled 占位结果（若仍计划把状态暴露给后续 turn）。
5. 向客户端发终端事件：event: run_terminal + status + 预算快照；然后关闭 SSE。

伪代码（与并行 tool 批衔接）：

async def run_tools(calls, *, cancel: asyncio.Event, per_call_timeout_s: float):
    # 与并行调度文一致：all_settled 或 deadline_all_settled
    # 每个 handler 必须接受 cancel / AbortSignal
    ...

async def watch_disconnect(request, cancel: asyncio.Event):
    while not cancel.is_set():
        if await request.is_disconnected():
            cancel.set()
            return
        await asyncio.sleep(0.5)

4. 终端状态与对模型 / 用户的可见语义

对用户可见文案应稳定、可行动（例如「已达最大工具调用次数 64」），不要把内部主机名与栈轨迹塞进 UI；对模型若开启「预算耗尽后再要一句总结」，必须再扣一轮 turn—— 更稳妥的是不再调用模型，由模板生成系统摘要。

5. 配置示例（网关 YAML）

agent_run:
  max_model_turns: 16
  max_tool_calls_per_run: 64
  run_deadline_s: 120
  max_input_tokens_per_run: 500000
  max_output_tokens_per_run: 64000
  max_cost_micros: 50000
  max_wall_without_progress_s: 90
  safety_margin_s: 1.5
  min_useful_tool_s: 2.0
  cancel_on_disconnect: true   # 交互式 SSE 默认 true；异步 job 默认 false
  on_budget_exhausted: stop    # stop | optional_summary_turn（慎用，会再耗 turn）
  persist_partial_results: true
tool_defaults:
  per_call_timeout_s: 60       # 实际 min(该值, remaining_wall - safety_margin)
model_defaults:
  # 单次 completion 上限，仍受 run 级 output 累计约束
  max_completion_tokens: 4096

异步 Job（消息队列 / SKIP LOCKED 租约）额外要求：

• cancel 写在 DB / 缓存的 runs.cancel_requested_at，Worker 每 tool 边界与模型调用前轮询（或 LISTEN/NOTIFY）。
• 租约 locked_until 心跳必须短于 run_deadline 的可接受漂移；Worker 死亡后任务回 pending，不得在无幂等键时自动重放写工具。

风险与边界

预算耗尽 ≠ 事务回滚。 停止循环不会撤销已执行的 DROP、已推送的 webhook、已发送的邮件。写工具必须继续依赖幂等键、审批门与补偿；run 预算只限制继续造成伤害的速率与长度。

cancel_on_disconnect 的产品语义。 移动网络闪断会被误判为取消。交互式场景可对 disconnect 做短宽限（如 5～15s 内允许同一 run_id 的 SSE 续挂再订阅）；超时仍无订阅者再真正 cancel。后台「生成报告」类任务应默认不随断连取消，并提供独立取消按钮。

嵌套超时必须单调。 推荐不等式：run_deadline ≥ 单 turn 模型超时 + 单批 tool 墙钟 + safety_margin，且 单 tool timeout ≤ 批墙钟 ≤ 剩余 run 墙钟。外层 25s、内层 shell 60s 会导致网关已 504 而子进程仍在跑 —— 必须在 disconnect/504 路径上触发进程组杀（见 shell 工具专文）。

与并行 tool 批、结果预算的分工。 并行信号量限制瞬时飞行数；max_tool_calls_per_run 限制累计次数；max_result_tokens 限制单次回灌体积。三者缺一都会在不同轴上被打穿。

本地 token 预估误差。 多模态、缓存命中、厂商特殊计费（缓存读写分价）会使本地估算与账单偏离。费用熔断应以提供商 usage 与账单字段为准；预估只用于「是否还值得再开一轮」的启发式。

协作取消的响应时延。 同步阻塞在无超时的 DNS/read() 上时，仅靠 asyncio.Event 不够，必须在 I/O 层绑定 deadline 或在独立线程池中可中断的包装。取消是尽力而为；最终靠 cgroup / 连接池强制回收兜底。

安全边界不因「快停」而放宽。 取消路径上的错误日志仍可能含密钥与绝对路径，需与回灌脱敏同一套规则；status 事件对外暴露预算字段时避免泄露租户计费细节给其他租户。

不要用「再给模型一次总结 turn」默认吞掉超限。 额外 turn 本身可能再次触发工具（若未把 tools 从请求中摘掉）。若需要总结：tool_choice: none / 等价开关 + 更短 max_completion_tokens，并计入预算。

参考来源

• OWASP Gen AI Security Project — LLM06:2025 Excessive Agency
• OpenAI — API Reference：Chat Completions（流式响应与 usage）
• OpenAI — Function calling
• Anthropic — Tool use with Claude
• Anthropic — Streaming Messages
• MDN — AbortController
• MDN — ReadableStream 与取消
• Python 3 — asyncio Task cancellation
• Python 3 — asyncio.timeout / wait_for
• Model Context Protocol — Tools（tools/call、执行错误与协议错误）
• LangGraph — Graph recursion limit（编排层步数上限的一种实现参考）
• PostgreSQL — SELECT FOR UPDATE SKIP LOCKED（异步 run 租约队列常用原语）

systems