个人开发者如何设置API成本预警？从Token统计到硬限额的完整方案

平台支出提醒和硬限额          ↓请求前预估Token和费用          ↓请求后记录真实Token          ↓项目、用户和单次请求限额

max_output_tokens=800

预计最高费用=输入Token ÷ 1,000,000 × 输入单价+最大输出Token ÷ 1,000,000 × 输出单价

项目用户IDAPI Key或服务账号模型输入Token输出Token单次费用请求时间请求是否成功

personal-chatbotblog-generatordocument-summarytest-scripts

Project Settings→ Limits→ Spend→ Edit spend limit

50%80%95%100%

pip install openai requests

export OPENAI_API_KEY="你的API Key"export OPENAI_MODEL="gpt-5.6"export MONTHLY_BUDGET_USD="20"export INPUT_PRICE_PER_1M="填写当前模型输入价格"export OUTPUT_PRICE_PER_1M="填写当前模型输出价格"export MAX_OUTPUT_TOKENS="800"export ALERT_WEBHOOK_URL=""

from __future__ import annotationsimport osimport sqlite3from datetime import datetime, timezonefrom decimal import Decimalfrom typing import Anyimport requestsfrom openai import OpenAIDB_PATH = os.getenv("COST_DB_PATH", "api_cost.db")MODEL = os.getenv("OPENAI_MODEL", "gpt-5.6")MONTHLY_BUDGET = Decimal(    os.getenv("MONTHLY_BUDGET_USD", "20"))INPUT_PRICE = Decimal(    os.environ["INPUT_PRICE_PER_1M"])OUTPUT_PRICE = Decimal(    os.environ["OUTPUT_PRICE_PER_1M"])MAX_OUTPUT_TOKENS = int(    os.getenv("MAX_OUTPUT_TOKENS", "800"))ALERT_LEVELS = (    Decimal("0.50"),    Decimal("0.80"),    Decimal("0.95"),)ALERT_WEBHOOK_URL = os.getenv("ALERT_WEBHOOK_URL", "")client = OpenAI()def init_db() -> None:    with sqlite3.connect(DB_PATH) as conn:        conn.execute(            """            CREATE TABLE IF NOT EXISTS api_usage (                id INTEGER PRIMARY KEY AUTOINCREMENT,                created_at TEXT NOT NULL,                model TEXT NOT NULL,                input_tokens INTEGER NOT NULL,                output_tokens INTEGER NOT NULL,                estimated_cost_usd TEXT NOT NULL            )            """        )def current_month() -> str:    return datetime.now(timezone.utc).strftime("%Y-%m")def get_monthly_spend() -> Decimal:    month = current_month()    with sqlite3.connect(DB_PATH) as conn:        rows = conn.execute(            """            SELECT estimated_cost_usd            FROM api_usage            WHERE substr(created_at, 1, 7) = ?            """,            (month,),        ).fetchall()    return sum(        (Decimal(row[0]) for row in rows),        start=Decimal("0"),    )def estimate_cost(    input_tokens: int,    output_tokens: int,) -> Decimal:    cost = (        Decimal(input_tokens) * INPUT_PRICE        + Decimal(output_tokens) * OUTPUT_PRICE    ) / Decimal("1000000")    return cost.quantize(Decimal("0.000001"))def save_usage(    input_tokens: int,    output_tokens: int,    cost: Decimal,) -> None:    created_at = datetime.now(timezone.utc).isoformat()    with sqlite3.connect(DB_PATH) as conn:        conn.execute(            """            INSERT INTO api_usage (                created_at,                model,                input_tokens,                output_tokens,                estimated_cost_usd            )            VALUES (?, ?, ?, ?, ?)            """,            (                created_at,                MODEL,                input_tokens,                output_tokens,                str(cost),            ),        )def send_alert(message: str) -> None:    print(f"[API COST ALERT] {message}")    if not ALERT_WEBHOOK_URL:        return    try:        # 不同通知平台的JSON格式可能不同，        # 飞书、企业微信、Slack需要按各自格式修改。        requests.post(            ALERT_WEBHOOK_URL,            json={"text": message},            timeout=5,        ).raise_for_status()    except requests.RequestException as exc:        print(f"Webhook发送失败：{exc}")def check_alerts(    old_spend: Decimal,    new_spend: Decimal,) -> None:    for level in ALERT_LEVELS:        threshold = MONTHLY_BUDGET * level        if old_spend < threshold <= new_spend:            percent = int(level * 100)            send_alert(                f"本月API费用已达到预算的{percent}%："                f"{MONTHLY_BUDGET}"            )def get_value(    obj: Any,    name: str,    default: int = 0,) -> int:    if obj is None:        return default    if isinstance(obj, dict):        return int(obj.get(name, default) or default)    return int(getattr(obj, name, default) or default)def guarded_generate(prompt: str) -> str:    old_spend = get_monthly_spend()    # 请求前准确计算输入Token    token_result = client.responses.input_tokens.count(        model=MODEL,        input=prompt,    )    input_tokens = token_result.input_tokens    # 按最大输出Token计算本次请求的最坏费用    maximum_request_cost = estimate_cost(        input_tokens=input_tokens,        output_tokens=MAX_OUTPUT_TOKENS,    )    if old_spend + maximum_request_cost > MONTHLY_BUDGET:        raise RuntimeError(            "本次请求可能超过月度API预算，已停止调用。"            f"当前费用：{maximum_request_cost}，"            f"预算：{get_monthly_spend()}")    except Exception as exc:        print(f"调用失败：{exc}")

{    "provider": "openai",    "model": "模型名称",    "input_tokens": 1000,    "output_tokens": 300,    "cost_usd": "0.004500",    "user_id": "user_123",    "project": "blog-generator"}

月度预算：20美元每日预算：2美元单用户每日预算：0.20美元单次请求最高费用：0.05美元

单次请求是否超限        ↓用户今日是否超限        ↓项目今日是否超限        ↓项目本月是否超限        ↓发送API请求

while True:    try:        call_api()        break    except Exception:        continue

import timefor attempt in range(3):    try:        call_api()        break    except Exception:        if attempt == 2:            raise        time.sleep(2 ** attempt)

浏览器或APP    ↓自己的后端服务    ↓验证用户和额度    ↓调用模型API

简单分类 → 低成本模型普通问答 → 常规模型复杂推理 → 高能力模型

OpenAI及模型API开发工具订阅测试环境正式项目其他SaaS软件

代码侧预算拦截        ↓API平台硬限额        ↓付款卡额度控制