使用 Responses API 构建每次运行的支出控制器

技术背景

在 AI 应用中，API 调用成本不可忽视，尤其是生产环境。开发者常面临预算失控、成本估算不准等问题。Responses API 按 token 计费，且价格因输入/输出、缓存等而异。手动估算成本容易出错，且无法在请求前阻止超支。因此，需要一个自动化的支出控制器，在每次请求前预留预算，请求后精确结算，并在超支时阻止后续请求。

核心原理

核心原理是预算预留与结算机制。每次请求前，根据输入 token 数（通过计数 API 获取）和最大输出 token 数，按最高可能价格预留资金。请求完成后，根据实际用量精确计算成本并结算。若实际成本超过预留，则标记预算阻塞，防止后续请求。使用 Fraction 类型避免浮点误差，确保金额精确。

架构分析

系统由三个核心组件组成：RateCard 定义模型价格和限制；RunBudget 管理预算状态，包括已花费、预留和阻塞标志；response_with_budget 函数协调流程，先检查预算，预留资金，发送请求，结算成本。使用锁保证线程安全，支持单进程内并发。离线客户端用于测试，真实客户端需配置 API 密钥。

完整实战教程

估算 Responses API 请求的可能成本

示例价格以每百万 token 的美元数表示：

| Token 类型 | 示例价格（美元/百万 token） |

| --- | ---: |

| 普通输入 | $4.00 |

| 缓存输入 | $2.00 |

| 缓存写入 | $8.00 |

| 输出 | $20.00 |

将应用程序限制设置为每次请求 10,000 个输入 token 和 250 个输出 token。这些数字是示例设置，并非真实模型的限制。仅当所选模型单独对缓存写入计费时，缓存写入价格才适用。

工单 #4821 使用了 1,200 个输入 token。为避免低估响应成本，假设每个输入 token 都使用最高的输入价格。此处，单独计费的缓存写入是最昂贵的：

(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。将模型、默认处理层级、令牌价格和请求限制一起配置。使用 API 在 response.model 中返回的固定模型 ID，此示例不支持模型别名。仅当模型不单独计费缓存写入时，才将 cache_write_usd_per_million 设置为 None。示例价格仅适用于离线客户端。

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 跟踪已花费的资金和暂时预留的资金。它的锁防止同一 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

在每次请求前检查预算

在计算 token 和生成响应时，使用相同的 model 和 input。指令、工具模式、图像、文件和对话历史也会消耗输入 token。如果你添加了其中任何一项，请将相同的受支持字段传递给两个请求。仅在响应请求中发送 max_output_tokens、service_tier 和 store。

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:
        # An interruption after submission may still incur a charge.
        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 个输出令牌：

(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 并运行：

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

票券和订单是虚构的，离线客户端不会发送任何网络请求：

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}")

预期输出：

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

限制与其他成本

普通输入、缓存输入、单独计费的缓存写入和输出只计算一次。当模型不单独对缓存写入计费时，按普通费率对非缓存输入收费。意外的正缓存写入计数会停止运行。请查看模型的提示缓存指南。推理令牌已包含在输出总数中。

锁保护单个 Python 进程。当多个工作进程共享预算时，使用共享存储，在一次操作中检查和预留预算，这样它们就不能预留相同的资金。如果请求的最终成本未知，则放弃该运行并保留其预留。在启用重试之前，请查看 Python SDK 重试行为。

托管工具可能会产生额外费用。例如，网络搜索可能对每次调用和相关模型令牌收费。预算还不包括存储、非默认处理层级、区域价格、长上下文费率、流式传输、后台请求、服务器管理的代理运行以及账户特定费用。后台请求可能首先返回 queued 或 in_progress，因此需要单独的规则来跟踪最终使用情况。设置 store=False 并不保证零数据保留；请参阅数据控制。

项目支出限制涵盖项目总支出，可能不会立即生效。警报不会停止请求。成本 API 参考显示每日总计；它无法告诉你某个任务是否能负担得起下一个请求。

常见问题

常见错误包括：未设置 max_retries=0 导致重试可能重复计费；未验证价格导致使用过期价格；缓存写入 token 未处理导致成本低估；请求中断后未阻塞预算导致后续超支。排查方法：检查响应中的 usage 字段，确保 total_tokens 等于 input+output；验证模型和 service_tier 是否匹配；使用离线客户端测试流程。

企业应用扩展

生产环境优化：使用共享存储（如 Redis）实现多进程预算同步；考虑流式响应和后台请求的计费差异；集成项目级支出限制，但注意其延迟；定期更新价格表，并验证价格有效性；扩展支持非默认服务层级和区域价格。

总结

本文介绍如何为 OpenAI Responses API 构建精确的每次运行支出控制器，通过预算预留、精确成本核算和异常处理，确保每次请求不超预算。核心是使用 Fraction 精确计算成本，通过 RateCard 定义价格，RunBudget 管理预留与结算，并在请求前检查预算。文章提供完整代码、离线测试示例及生产环境注意事项，适合有编程基础的开发者学习 AI 工程实践。

• --

内容来源与验证

• 内容等级：PREMIUM_A
• 质量等级：A
• 难度：中级
• 验证等级：V1
• 原文作者：alexey-openai
• 原文日期：2026-08-17
• 来源许可：MIT
• 来源版本：a071870150350455d9aa9cc423c4cb25a4f44bb0
• 查看原文