用Token账本监控免费模型额度，避免生产环境突现429

你的免费模型端点有配额。没人告诉你什么时候用完。一次请求成功，下一个返回 429。没有警告，没有倒计时。

数据是存在的。它藏在响应里。usage 对象报告 token 计数。限流头报告剩余容量。大多数集成把这两个都丢弃了。

本教程构建一个 token 账本。它计量每一次调用。它预测配额何时耗尽。四个阶段，一小时。每个阶段以验证步骤结束。

Disclosure: This article was prepared as part of MonkeyCode's product outreach. I used MonkeyCode's free model access and free server option for testing. The pipeline works with any OpenAI-compatible endpoint.

Stage 1: Capture the usage object

每个 OpenAI 兼容的响应都带有一个 usage 对象。

{
  "usage": {
    "prompt_tokens": 42,
    "completion_tokens": 87,
    "total_tokens": 129
  }
}

一些网关也添加了头信息。X-RateLimit-Remaining 和 X-RateLimit-Reset 是常见的。并非每个端点都发送它们。解析两者，使用存在的那个。

下面的包装函数返回文本和元数据。永远不要丢弃元数据。

import json
import time
import urllib.request

def call_and_measure(base_url, api_key, model, messages):
    body = json.dumps({
        "model": model,
        "messages": messages,
        "temperature": 0
    }).encode()
    req = urllib.request.Request(
        base_url + "/chat/completions",
        body,
        {"Content-Type": "application/json",
         "Authorization": "Bearer " + api_key}
    )
    start = time.monotonic()
    with urllib.request.urlopen(req, timeout=60) as resp:
        data = json.load(resp)
        headers = dict(resp.headers)
    elapsed_ms = (time.monotonic() - start) * 1000
    usage = data.get("usage", {})
    return {
        "text": data["choices"][0]["message"]["content"],
        "usage": usage,
        "headers": headers,
        "latency_ms": round(elapsed_ms, 1),
        "ts": time.time()
    }

Verify Stage 1: Run one call. Print the returned dict. Confirm usage.total_tokens is present. Confirm the headers dict exists, even if empty.

Stage 2: Persist to SQLite

内存中的字典不是账本。账本要在重启后存活。SQLite 是 Python 自带的。零依赖。

创建一张表。保持窄表设计。

CREATE TABLE IF NOT EXISTS calls (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    ts REAL NOT NULL,
    model TEXT NOT NULL,
    prompt_tokens INTEGER NOT NULL,
    completion_tokens INTEGER NOT NULL,
    total_tokens INTEGER NOT NULL,
    latency_ms REAL NOT NULL,
    rate_limit_remaining TEXT,
    rate_limit_reset TEXT
);

每次调用插入一行。使用事务。免费服务器会在写入中途挂掉。事务保持账本完整。

不同网关之间头信息的 key 大小写不一致。读取前先规范化。

import sqlite3

def header_value(headers, name):
    lower = name.lower()
    for k, v in headers.items():
        if k.lower() == lower:
            return v
    return None

def insert_call(db_path, row):
    with sqlite3.connect(db_path) as conn:
        conn.execute(
            """INSERT INTO calls
               (ts, model, prompt_tokens, completion_tokens,
                total_tokens, latency_ms, rate_limit_remaining,
                rate_limit_reset)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
            (row["ts"], row["model"],
             row["usage"].get("prompt_tokens", 0),
             row["usage"].get("completion_tokens", 0),
             row["usage"].get("total_tokens", 0),
             row["latency_ms"],
             header_value(row["headers"], "X-RateLimit-Remaining"),
             header_value(row["headers"], "X-RateLimit-Reset"))
        )

Verify Stage 2: Make three calls. Query the table. Confirm three rows. Confirm the token columns match the printed usage objects.

Stage 3: Project the wall

没有预测的账本只是一本日记。预测回答一个问题：配额何时耗尽？

预测是线性的。取最近一小时的调用。计算每小时 token 数。用剩余配额除以该速率。

在配置文件中设置你的配额。脚本无法猜测。不同的免费层级有不同的限额。

import json
import sqlite3
import time

QUOTA = json.load(open("quota.json"))
# {"monthly_tokens": 1000000, "reset_day": 1}

def project_exhaustion(db_path):
    with sqlite3.connect(db_path) as conn:
        row = conn.execute(
            """SELECT MIN(ts), MAX(ts),
                      SUM(total_tokens)
               FROM calls
               WHERE ts > ?""",
            (time.time() - 3600,)
        ).fetchone()
    start_ts, end_ts, tokens = row
    if not tokens or end_ts == start_ts:
        return None
    hours = (end_ts - start_ts) / 3600
    tokens_per_hour = tokens / hours
    remaining = QUOTA["monthly_tokens"]
    hours_left = remaining / tokens_per_hour
    return time.time() + hours_left * 3600

线性预测不是唯一选项。

从最近一小时的速率开始。当你的流量形态证明有更优方案时再升级。

Verify Stage 3: Run the projection. Confirm the output moves after a burst of calls. Confirm it recovers after a quiet hour.

Stage 4: Alert before the wall

预测会漂移。告警捕获漂移。每小时检查一次预测。当配额墙移近到不足 24 小时时告警。

写一个标记文件。这是在免费服务器重启后仍能存活的简单告警机制。

def check_wall(db_path, alert_path, threshold_hours=24):
    eta = project_exhaustion(db_path)
    if eta is None:
        return
    hours_left = (eta - time.time()) / 3600
    if hours_left < threshold_hours:
        with open(alert_path, "w") as f:
            f.write(json.dumps({"eta": eta, "hours_left": hours_left}))
    else:
        import os
        if os.path.exists(alert_path):
            os.remove(alert_path)

把它接入 cron。每三十分钟一次足矣。免费服务器会杀死长进程。cron 会重启检查。

*/30 * * * * cd /opt/token-ledger && python check.py >> check.log 2>&1

Verify Stage 4: Temporarily set threshold_hours to 9999. Confirm the marker file appears. Revert. Confirm it clears.

Caveats

The usage object is only as honest as the provider. Some endpoints omit it. Some report only prompt tokens. The ledger records zeros. The projection degrades into a guess.

Rate-limit headers are advisory. Their format changes between gateways. Treat them as hints, not contracts.

Linear projection assumes steady consumption. A batch job on Monday breaks the model. The alert fires late. Review the ledger weekly. Look at the shape, not just the number.

Who should not use this

Teams with irregular traffic should not trust the linear forecast. Add a weighted average or a manual review step first.

Teams that need per-feature cost attribution need more than one table. Add a feature column. Group by it. The ledger is a foundation, not a full cost system.

The free tier is a black box

The response headers are the key. Meter every call. Project the wall. Alert before it hits.

The pipeline is about forty lines. It runs on a free server. It turns a surprise 429 into a scheduled event.

MonkeyCode's free model access and free server option gave me the infrastructure for this. The ledger is the part you build. Build it before the wall finds you.