原文出处：Per-Run Spending Controller with the Responses API 原作者：OpenAI · 许可证：MIT License 中文译本由诸葛AI学院整理，仅供学习参考，版权归原作者与 OpenAI 所有。

用 Responses API 搭建每次运行的花费控制器

使用 Responses API 的智能体（agent）为了完成一个任务，可能会调用模型好几次。组织和项目级别的花费上限（spending limit）能管住整体用量，却没法告诉你：这个任务还付得起下一次请求吗？

给每次运行（run）单独设一笔预算（budget）。每次发起模型请求之前，先数一数输入令牌（token）的数量，把它的响应最坏情况下可能花掉的钱预先扣留。等响应回来，记录实际花费，并把没用完的钱退回预算。一旦某个请求会超出剩余余额，就在发出之前停下来。

举个例子：工单 #4821 在问订单 ORD-42 什么时候能到。这个订单已发货，预计周五送达。给这张工单 0.02 美元的预算。第一次回复花了 0.01 美元。再来一次回复最多可能花 0.0146 美元。预算只剩 0.01 美元，应用就此停止。

这套控制器处理的是默认处理层级（default service tier）上的同步、非流式 Responses API 请求。它的预算只覆盖模型令牌费用，托管工具和其他费用不在其中。下文的单价、令牌数、模型名和限额都是编的示例，不是 OpenAI 现行价格、真实模型限制，也不构成对你最终账单的任何保证。

估算一次 Responses API 请求可能花多少钱

示例单价以"每百万令牌多少美元"计：

把应用侧限制设为每次请求 10,000 个输入令牌、250 个输出令牌。这些数字只是示例配置，不是真实模型的上限。只有当所选模型对缓存写入单独计价时，缓存写入单价才适用。

工单 #4821 用了 1,200 个输入令牌。为了不把响应成本估低，假设每个输入令牌都按最高的输入单价计算。在这里，单独计价的缓存写入是最贵的：

text (1,200 x $8.00 + 250 x $20.00) / 1,000,000 = $0.0146

用 Python 内置的 Fraction（分数）类型来保持美元金额精确无损。在发出真实请求之前，先核对模型、默认处理层级、令牌限额和价格。verified_at 日期记录的是你何时核对过这些价格，并不能证明它们现在仍然有效。参见现行 API 定价和模型文档。

让每次运行不超出预算

定义模型和价格

需要 Python 3.12 或更高版本，以及 openai>=2.28.0。把模型、默认处理层级、令牌单价和请求限额配置在一起。使用固定的模型 ID，API 会在 response.model 里原样返回它。本示例不支持模型别名。只有当模型不对缓存写入单独计价时，才把 cache_write_usd_per_million 设为 None。这些示例单价只能配合离线客户端使用。

```python from dataclasses import dataclass, field from datetime import date from fractions import Fraction from threading import Lock from types import SimpleNamespace from typing import Any from openai.types.responses import Response

@dataclass(frozen=True) class RateCard: model: str service_tier: str input_usd_per_million: Fraction cached_usd_per_million: Fraction cache_write_usd_per_million: Fraction | None output_usd_per_million: Fraction max_input_tokens: int max_output_tokens: int example_only: bool = True verified: bool = False verified_at: date | None = None

def __post_init__(self) -> None:
    if not all(type(value) is str and value.strip()
               for value in (self.model, self.service_tier)):
        raise ValueError("Model and service tier must be nonempty strings")
    if self.service_tier != "default":
        raise ValueError("Only the default service tier is supported")
    for name in (
        "input_usd_per_million", "cached_usd_per_million",
        "output_usd_per_million",
    ):
        value = getattr(self, name)
        if type(value) is not Fraction or value <= 0:
            raise ValueError(f"{name} price must be a positive exact dollar amount")
    if self.cache_write_usd_per_million is not None and (
        type(self.cache_write_usd_per_million) is not Fraction
        or self.cache_write_usd_per_million <= 0
    ):
        raise ValueError("Cache-write price must be a positive exact dollar amount")
    if type(self.max_input_tokens) is not int or self.max_input_tokens <= 0:
        raise ValueError("Maximum input tokens must be a positive integer")
    if type(self.max_output_tokens) is not int or self.max_output_tokens < 16:
        raise ValueError("Maximum output tokens must be an integer of at least 16")
    if type(self.example_only) is not bool or type(self.verified) is not bool:
        raise ValueError("Pricing verification flags must be boolean")
    if self.example_only and self.verified:
        raise ValueError("Example pricing cannot be marked verified")
    if self.verified_at is not None and type(self.verified_at) is not date:
        raise ValueError("Pricing verification date must be a date")
    if self.verified and self.verified_at is None:
        raise ValueError("Record when you checked the model pricing")
    if self.example_only and self.verified_at is not None:
        raise ValueError("Example pricing cannot have a verification date")
    if not self.example_only and self.model == "example-model":
        raise ValueError("Replace the example model with your verified model")

EXAMPLE_RATE_CARD = RateCard( model="example-model", service_tier="default", input_usd_per_million=Fraction("4.00"), cached_usd_per_million=Fraction("2.00"), cache_write_usd_per_million=Fraction("8.00"), output_usd_per_million=Fraction("20.00"), max_input_tokens=10_000, max_output_tokens=250, example_only=True, ) ```

跟踪这次运行的预算

RunBudget 同时跟踪已经花掉的钱和临时扣留的钱。它内部的锁（lock）能防止同一个 Python 进程里的两个请求扣同一笔钱。当缓存写入有单独单价时，响应必须包含缓存写入的令牌数。

```python class BudgetExceeded(RuntimeError): pass

class UncertainCharge(RuntimeError): pass

@dataclass class RunBudget: maximum: Fraction spent: Fraction = field(default_factory=Fraction, init=False) pending: Fraction = field(default_factory=Fraction, init=False) blocked: bool = field(default=False, init=False) _holds: dict[object, Fraction] = field(default_factory=dict, init=False, repr=False) _lock: Lock = field(default_factory=Lock, init=False, repr=False)

def __post_init__(self) -> None:
    if type(self.maximum) is not Fraction or self.maximum <= 0:
        raise ValueError("Budget must be a positive exact dollar amount")

def ensure_active(self, minimum: Fraction) -> None:
    if type(minimum) is not Fraction or minimum < 0:
        raise ValueError("Minimum must be a nonnegative exact dollar amount")
    with self._lock:
        if self.blocked or self.spent + self.pending + minimum > self.maximum:
            raise BudgetExceeded("The remaining run budget is insufficient")

def reserve(self, amount: Fraction) -> object:
    if type(amount) is not Fraction or amount <= 0:
        raise ValueError("Reservation must be a positive exact dollar amount")
    with self._lock:
        if self.blocked or self.spent + self.pending + amount > self.maximum:
            raise BudgetExceeded("The remaining run budget is insufficient")
        handle = object()
        self._holds[handle] = amount
        self.pending += amount
        return handle

def settle(self, handle: object, actual: Fraction) -> None:
    if type(handle) is not object or type(actual) is not Fraction or actual < 0:
        raise ValueError("Invalid spend settlement")
    with self._lock:
        held = self._holds.get(handle)
        if held is None or held > self.pending:
            raise ValueError("Reservation is unknown or already settled")
        del self._holds[handle]
        self.pending -= held
        self.spent += actual
        if actual > held:
            self.blocked = True
            raise UncertainCharge("Actual spend exceeded the amount reserved")

def block(self) -> None:
    with self._lock:
        self.blocked = True

def _tokens(value: Any, name: str) -> int: if type(value) is not int or value < 0: raise UncertainCharge(f"Invalid {name} token count") return value

def actual_cost(usage: Any, rates: RateCard) -> Fraction: if usage is None: raise UncertainCharge("Token usage is missing") details = getattr(usage, "input_tokens_details", None) if details is None: raise UncertainCharge("Input token details are missing") input_tokens = _tokens(getattr(usage, "input_tokens", None), "input") output_tokens = _tokens(getattr(usage, "output_tokens", None), "output") total_tokens = _tokens(getattr(usage, "total_tokens", None), "total") if total_tokens != input_tokens + output_tokens: raise UncertainCharge("Total tokens do not match input and output") cached = _tokens(getattr(details, "cached_tokens", None), "cached") if rates.cache_write_usd_per_million is not None: if not hasattr(details, "cache_write_tokens"): raise UncertainCharge("Cache-write token accounting is missing") written = _tokens(details.cache_write_tokens, "cache-write") else: observed = _tokens(getattr(details, "cache_write_tokens", 0), "cache-write") if observed: raise UncertainCharge("Cache writes require a verified cache-write price") written = 0 ordinary = input_tokens - cached - written if ( input_tokens > rates.max_input_tokens or output_tokens > rates.max_output_tokens or ordinary < 0 ): raise UncertainCharge("Usage exceeds the configured request bounds") return ( ordinary * rates.input_usd_per_million + cached * rates.cached_usd_per_million + written * (rates.cache_write_usd_per_million or Fraction()) + output_tokens * rates.output_usd_per_million ) / 1_000_000 ```

每次请求前先查预算

数令牌和生成响应时，要用同一个 model 和同一个 input。指令、工具 schema、图片、文件和会话历史同样消耗输入令牌。如果你加了其中任何一项，就把相同的受支持字段一并传给这两个请求。max_output_tokens、service_tier 和 store 只随响应请求发送。

```python def response_with_budget( client: Any, prompt: str, budget: RunBudget, rates: RateCard, , allow_example: bool = False, ) -> Response | SimpleNamespace: if type(prompt) is not str or not prompt.strip(): raise ValueError("Only nonempty text prompts are supported") if type(allow_example) is not bool: raise ValueError("Example authorization must be a boolean") if (rates.example_only or not rates.verified) and not ( allow_example and type(client) is OfflineClient ): raise ValueError("API requests require explicitly verified pricing") if client.max_retries != 0: raise ValueError("Initialize the OpenAI client with max_retries=0") budget.ensure_active( rates.max_output_tokens * rates.output_usd_per_million / 1_000_000 ) request = {"model": rates.model, "input": prompt} count = _tokens(client.responses.input_tokens.count(request).input_tokens, "input") if count > rates.max_input_tokens: raise BudgetExceeded("Request exceeds the configured input limit") worst_input_price = max( price for price in ( rates.input_usd_per_million, rates.cached_usd_per_million, rates.cache_write_usd_per_million, ) if price is not None ) reservation = budget.reserve( ( count * worst_input_price + rates.max_output_tokens * rates.output_usd_per_million ) / 1_000_000 ) try: response = client.responses.create( *request, max_output_tokens=rates.max_output_tokens, service_tier=rates.service_tier, store=False, ) if response.model != rates.model: raise UncertainCharge("Response used an unexpected model") if response.service_tier != rates.service_tier: raise UncertainCharge("Unexpected service tier") cost = actual_cost(response.usage, rates) except BaseException: # 提交之后请求被中断，也可能照样产生费用。 budget.block() raise if response.status != "completed": budget.block() budget.settle(reservation, cost) if response.status != "completed": raise UncertainCharge(f"Response ended with status: {response.status}") return response

def format_dollars(amount: Fraction) -> str: return "$" + f"{amount:.9f}".rstrip("0").rstrip(".") ```

response_with_budget 返回完整的响应，包括 response.output_text 和 response.output。它会核对 API 用的是配置里的模型和处理层级。如果一个请求被中断、或者它的成本无法确认，这次运行就永久停止，已扣留的预算也不再可用。如果未完成的响应报出了自己的用量，控制器会记录这笔成本，并让这次运行保持停止状态。

要换成真实客户端，用 from openai import OpenAI 导入 OpenAI，核对当前价格，并设置 OPENAI_API_KEY。然后用 client = OpenAI(max_retries=0, timeout=60.0) 创建客户端。请求可能在客户端超时之后仍在运行，所以要保留它的扣留额度。

示例：给一张工单设预算

工单 #4821 的第一次响应报出 1,200 个输入令牌：400 个缓存令牌，500 个单独计价的缓存写入令牌，剩下 300 个是普通输入令牌。它还报出 200 个输出令牌：

text (300 x $4.00 + 400 x $2.00 + 500 x $8.00 + 200 x $20.00) / 1,000,000 = $0.0100

第一次响应花了 0.01 美元，还剩 0.01 美元。下一次响应可能花 0.0146 美元，所以控制器在发出下一个模型请求之前就停了下来。

把上面的三个代码块和下面的离线示例按顺序存为 controller.py。安装 SDK 后运行：

bash python -m pip install "openai>=2.28.0" python controller.py

工单和订单都是虚构的，离线客户端不会发出任何网络请求：

```python class OfflineClient: slots = ("model_calls",) max_retries = 0

def __init__(self) -> None:
    self.model_calls = 0

@property
def responses(self) -> "OfflineClient":
    return self

@property
def input_tokens(self) -> "OfflineClient":
    return self

def count(self, **_: Any) -> SimpleNamespace:
    return SimpleNamespace(input_tokens=1200)

def create(self, **request: Any) -> SimpleNamespace:
    self.model_calls += 1
    return SimpleNamespace(
        model=request["model"], service_tier="default", status="completed",
        output_text="Order ORD-42 has shipped and should arrive Friday.",
        usage=SimpleNamespace(
            input_tokens=1200, output_tokens=200, total_tokens=1400,
            input_tokens_details=SimpleNamespace(
                cached_tokens=400, cache_write_tokens=500
            ),
        ),
    )

if name == "main": client = OfflineClient() budget = RunBudget(Fraction("0.02")) ticket = "Ticket #4821: order ORD-42 shipped and arrives Friday." print(response_with_budget( client, ticket + " Draft a support reply.", budget, EXAMPLE_RATE_CARD, allow_example=True, ).output_text) try: response_with_budget( client, ticket + " Write a second version.", budget, EXAMPLE_RATE_CARD, allow_example=True, ) except BudgetExceeded: print("Ticket #4821 stopped: the next step would exceed its budget.") print("Spent: " + format_dollars(budget.spent)) print("Reserved: " + format_dollars(budget.pending)) print(f"Model calls: {client.model_calls}") ```

预期输出：

text Order ORD-42 has shipped and should arrive Friday. Ticket #4821 stopped: the next step would exceed its budget. Spent: $0.01 Reserved: $0 Model calls: 1

局限与其他成本

普通输入、缓存输入、单独计价的缓存写入和输出，每一项只算一次钱。当模型不对缓存写入单独计价时，非缓存输入按普通输入单价收费。出现意料之外的正数缓存写入令牌数会让运行停止。请查看模型的提示词缓存指南。推理令牌（reasoning token）已经包含在输出总数里了。

那把锁只保护单个 Python 进程。当多个工作进程共享一笔预算时，要用一个共享存储，把"检查预算"和"扣留预算"合成一个原子操作，避免它们扣同一笔钱。如果一个请求的最终成本未知，就放弃这次运行并保留其扣留额度。开启重试之前，先看Python SDK 的重试行为。

托管工具会产生单独的费用。比如网络搜索，可能按每次调用收费，相关的模型令牌也另外计费。预算还不包括：存储、非默认处理层级、地区差异价格、长上下文费率、流式输出、后台请求、服务端托管的智能体运行，以及账户特定的费用。后台请求最初可能返回 queued 或 in_progress，需要另设规则来跟踪最终用量。设置 store=False 并不保证零数据保留（Zero Data Retention），参见数据控制。

项目花费上限管的是项目总花费，且可能不会立即生效。告警不会拦截请求。Costs API 参考给出的是每日汇总，它无法判断某一个任务能否付得起下一笔钱。