前面的系列已经介绍了模型调用、上下文、生成参数、错误处理以及成本基础。把这些知识放进真实应用时，还缺一个经常被忽略的边界：一次请求到底允许消耗多少资源。本文只聚焦一个核心问题——在发出模型请求前，用普通 Python 代码实现一个可解释的预算闸门，阻止超出次数、Token 或费用上限的调用。示例不连接任何厂商 API，因此可以离线运行和验证。

为什么要在请求前检查预算

模型调用通常不是普通的本地函数。它可能产生按输入和输出分别计费的 Token 消耗，也可能因为重试、批处理或用户重复点击而快速累积请求量。仅在响应返回后统计，无法阻止已经发生的超预算调用。

预算闸门的职责很单一：接收本次请求的估算值和累计用量，判断是否允许继续。它不负责选择模型、不负责执行网络请求，也不负责决定业务内容。这样的边界便于测试：同样的输入应得到稳定的“允许”或“拒绝”结果。

需要注意，下面的 Token 估算只是工程上的粗略预估，不等同于某个模型服务的官方计费结果。不同模型的分词规则不同；正式计费应以对应服务的官方用量字段和价格文档为准。闸门的价值在于提前设定保守上限，而不是伪造精确账单。

先定义输入和规则

我们把预算拆成三类：单次请求最多估算多少 Token、整个运行最多允许多少 Token、最多允许多少次请求，以及累计费用上限。实际项目也可以只启用其中一部分。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from dataclasses import dataclass


@dataclass(frozen=True)
class Budget:
    max_requests: int
    max_input_tokens: int
    max_total_tokens: int
    max_cost: float
    input_price_per_million: float
    output_price_per_million: float


@dataclass
class Usage:
    requests: int = 0
    input_tokens: int = 0
    output_tokens: int = 0
    cost: float = 0.0

Budget 是不可变配置，避免程序运行中悄悄修改上限；Usage 保存已经消耗的资源。费用使用浮点数只适合这个演示。若要做严格财务核算，应改用 decimal.Decimal，并按服务商公布的单位精度处理。

实现一个保守的 Token 估算器

在没有真实 API 响应之前，可以先用字符数做预估。中文、英文、代码和标点的分词比例并不相同，所以这里故意取一个偏保守的上限：每四个字符至少按一个 Token 估算，并向上取整。

1
2
3
4
5
6
7
8
from math import ceil


def estimate_tokens(text: str) -> int:
    """仅用于发请求前的保守估算，不代表服务商的精确分词结果。"""
    if not text:
        return 0
    return ceil(len(text) / 4)

这个函数的优点是透明、无依赖、容易替换。接入具体服务后，可以把它替换为该模型对应的 tokenizer；预算闸门的接口不需要改变。

让闸门返回可解释的决定

拒绝调用时，调用方需要知道原因，而不是只得到一个布尔值。下面的实现按请求次数、单次输入 Token、累计 Token 和预计费用依次检查，并返回第一条失败原因。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def estimate_cost(input_tokens: int, output_tokens: int, budget: Budget) -> float:
    input_cost = input_tokens / 1_000_000 * budget.input_price_per_million
    output_cost = output_tokens / 1_000_000 * budget.output_price_per_million
    return input_cost + output_cost


def check_budget(
    prompt: str,
    expected_output_tokens: int,
    usage: Usage,
    budget: Budget,
) -> tuple[bool, str, int, float]:
    input_tokens = estimate_tokens(prompt)
    request_cost = estimate_cost(input_tokens, expected_output_tokens, budget)

    if usage.requests + 1 > budget.max_requests:
        return False, "超过请求次数上限", input_tokens, request_cost
    if input_tokens > budget.max_input_tokens:
        return False, "单次输入 Token 估算值超限", input_tokens, request_cost
    if usage.input_tokens + input_tokens + expected_output_tokens > budget.max_total_tokens:
        return False, "累计 Token 预算不足", input_tokens, request_cost
    if usage.cost + request_cost > budget.max_cost:
        return False, "累计费用预算不足", input_tokens, request_cost
    return True, "允许调用", input_tokens, request_cost


def record_usage(
    usage: Usage,
    input_tokens: int,
    output_tokens: int,
    request_cost: float,
) -> None:
    usage.requests += 1
    usage.input_tokens += input_tokens
    usage.output_tokens += output_tokens
    usage.cost += request_cost

expected_output_tokens 必须来自应用自己的上限或生成参数，而不能假设模型一定只返回很短的内容。检查通过后，只有在请求真正成功时才调用 record_usage；如果网络超时且无法确认服务端是否已处理，应根据服务商的幂等和计费规则设计更谨慎的记账策略。

运行一个完整的离线示例

下面的程序模拟两次调用。它不会访问网络，也不会伪造模型响应，只演示闸门如何决定和记录用量。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
budget = Budget(
    max_requests=2,
    max_input_tokens=80,
    max_total_tokens=180,
    max_cost=0.01,
    input_price_per_million=1.0,
    output_price_per_million=2.0,
)
usage = Usage()

for prompt in ["请把这段会议记录整理成三个行动项。", "请再给出一份详细的风险说明。"]:
    allowed, reason, input_tokens, request_cost = check_budget(
        prompt, expected_output_tokens=60, usage=usage, budget=budget
    )
    print(f"allowed={allowed} reason={reason} input_tokens={input_tokens}")
    if not allowed:
        continue

    # 这里替换成真实的模型调用；示例用预设值模拟成功后的用量记录。
    actual_output_tokens = 45
    actual_cost = estimate_cost(input_tokens, actual_output_tokens, budget)
    record_usage(usage, input_tokens, actual_output_tokens, actual_cost)

print(usage)

运行方式是把代码保存为 budget_gate.py 后执行 python budget_gate.py。输出中的具体 Token 数由字符串长度决定，程序会明确显示每次是否放行以及累计 Usage。示例只记录“成功返回”的预设输出 Token；真实代码应使用 API 响应中的实际用量字段更新，而不是继续使用估算值。

常见问题

为什么不直接把预算检查写在调用函数里

可以封装，但不要把检查、网络请求、重试和业务提示词全部揉成一个函数。独立的闸门可以单元测试，也可以在命令行、队列消费者和 Web 服务中复用。调用层只需要遵守“先检查，成功后记账”的协议。

估算太保守会不会浪费预算

会，所以应把估算视为安全边界，而不是最终账单。上线前用实际响应中的 Token 数据校准估算比例；宁可在开发阶段预留余量，也不要把上限设得刚好贴边。

重试应该如何计数

重试不是免费的“同一次调用”。如果服务端可能已经处理请求，重试可能再次消耗 Token。最简单的策略是每一次实际发出的请求都占用一次次数预算；更复杂的系统还应为重试设置独立上限，并记录请求 ID。

多个进程如何共享预算

本文的 Usage 只在单进程内有效。多个 Worker 同时运行时，必须把用量放进具有原子更新能力的共享存储，或者由集中式队列统一发放额度。仅靠进程内变量会导致每个进程都以为自己还有完整预算。

小结

预算闸门不是计费系统，却是 AI 应用的重要安全边界。本文用四步完成了一个可运行的最小实现：用数据类描述预算和用量，用保守规则估算输入 Token，在请求前检查次数、Token 与费用，成功后记录实际用量。下一步接入具体模型服务时，只需替换请求和用量读取部分，同时保留这层独立的约束逻辑，就能让成本控制从“事后统计”变成“请求前决策”。