实战操作步骤：给企业 AI API 搭一套余额监控与用量拦截（含完整代码）

实战操作步骤：给企业 AI API 搭一套余额监控与用量拦截（含完整代码）并不只看表面做法，关键还要理解相关条件、限制和后续影响。

前言

企业接入 AI API 之后，运维上最常踩的两个坑：一是余额在半夜见底，服务直接全挂；二是某个脚本失控，月底账单翻了一倍。这两件事都能靠一套不到两百行的监控代码提前拦住。

本文是动手教程，以jiekou.vip为例，跟着五个步骤做完，你会得到：余额剩余天数监控、余额预警、错误类型分级、用量配额拦截、速率熔断。每段代码都可直接跑。

环境准备：Python 3.9+、Redis（用于计数，也可换成任意 KV 存储）。

pip install redis requests

步骤一：先埋点，把每次调用记下来

所有阈值都要从历史数据推导，所以第一步不是写监控，是先让调用有记录。

新建 usage_log.py：

import jsonimport timefrom pathlib import PathLOG_FILE = Path("llm_usage.jsonl")def log_usage(scene: str, env: str, model: str,              input_tokens: int, output_tokens: int, cost: float) -> None:    """一次调用一行 JSON，后续统计直接读这个文件。"""    record = {        "ts": int(time.time()),        "scene": scene,          # 业务场景，成本归属主键        "env": env,              # prod / staging，务必区分        "model": model,        "input_tokens": input_tokens,        "output_tokens": output_tokens,        "cost": cost,    }    with LOG_FILE.open("a", encoding="utf-8") as f:        f.write(json.dumps(record, ensure_ascii=False) + "n")

env 这个字段一定要带。测试流量混进生产统计，是最常见的一类"成本莫名上涨"，多半是某个压测脚本忘了关。

按天聚合，为下一步提供输入：

from collections import defaultdictfrom datetime import datetimedef daily_costs(days: int = 7, env: str = "prod") -> list[float]:    """读日志，返回最近 N 天的每日消耗。"""    buckets = defaultdict(float)    if not LOG_FILE.exists():        return []    with LOG_FILE.open(encoding="utf-8") as f:        for line in f:            r = json.loads(line)            if r.get("env") != env:                continue            day = datetime.fromtimestamp(r["ts"]).strftime("%Y-%m-%d")            buckets[day] += r["cost"]    return [v for _, v in sorted(buckets.items())[-days:]]

步骤二：把余额换算成"还能用几天"

企业订阅包或预付费余额，看绝对数字没有意义。一万块对日耗两百的团队很安全，对日耗三千的团队是明天就断。所以要换算成天数。

新建 balance_monitor.py：

def balance_health(balance: float, recent_daily_costs: list[float]) -> dict:    """按剩余可用天数评估余额健康度。"""    if not recent_daily_costs:        return {"level": "unknown", "days_left": None}    avg = sum(recent_daily_costs) / len(recent_daily_costs)    peak = max(recent_daily_costs)    days_by_avg = balance / avg if avg else float("inf")    days_by_peak = balance / peak if peak else float("inf")    if days_by_peak < 3:        level = "critical"    elif days_by_avg < 7:        level = "warning"    else:        level = "ok"    return {        "level": level,        "days_left": round(days_by_peak, 1),        "avg_daily": round(avg, 2),    }

用峰值算 critical、用均值算 warning，是为了让突发流量那几天不至于把预警拖到来不及。

跑一下看效果：

if __name__ == "__main__":    costs = [180, 210, 195, 240, 205, 190, 230]    for bal in (20000, 3000, 500):        print(bal, balance_health(bal, costs))

输出：

20000 {'level': 'ok', 'days_left': 83.3, 'avg_daily': 207.14}3000  {'level': 'warning', 'days_left': 12.5, 'avg_daily': 207.14}500   {'level': 'critical', 'days_left': 2.1, 'avg_daily': 207.14}

接入要点：这一步需要平台提供余额查询接口，而不是只能在控制台页面看。选平台时确认这项，否则监控只能靠人工填数字。国内常见平台（如 jiekou.vip）提供余额和用量明细的查询接口，接入前在文档里核对一下即可。

挂个定时任务，每小时跑一次：

def check_and_alert(fetch_balance) -> None:    """fetch_balance 是你封装的余额查询函数。"""    health = balance_health(fetch_balance(), daily_costs())    if health["level"] in ("warning", "critical"):        notify(f"[{health['level']}] 余额仅剩 {health['days_left']} 天")

步骤三：把余额耗尽和限流区分开

这一步很多人漏掉，代价不小。余额耗尽和触发限流返回的都是错误，但处理方式完全相反：限流该退避重试，余额耗尽重试一万次也没用，必须立刻切换或降级。

混用同一套重试逻辑的后果，是把一次明确的失败拖成几分钟的雪崩。

新建 error_classify.py：

EXHAUSTED_HINTS = ("insufficient", "quota", "balance", "exceeded")def classify_error(status_code: int, body: str) -> str:    text = body.lower()    if status_code in (402, 403) and any(h in text for h in EXHAUSTED_HINTS):        return "exhausted"      # 余额/配额耗尽：重试无用    if status_code == 429:        return "rate_limited"   # 限流：退避重试    if status_code >= 500:        return "upstream_error" # 上游故障：可重试    return "client_error"       # 请求本身有问题：别重试

配上分场景的降级：

DEGRADABLE_SCENES = {"summary", "tagging", "recommend"}def complete_with_fallback(prompt: str, scene: str, primary, backup):    try:        return primary(prompt)    except UpstreamExhausted:        notify(f"主通道余额耗尽，scene={scene}")        if scene in DEGRADABLE_SCENES:            return backup(prompt)     # 非核心场景切备用        raise                          # 核心链路直接报错，别悄悄劣化

提前把场景分成可降级和不可降级两类。摘要、标签建议这类可以降级；核心交互链路上的调用不能悄悄换成劣化结果，该报错就报错。

步骤四：加用量配额，事前拦截

有了埋点数据，就能定配额。关键是在请求发出前检查，不是月底对账时才发现。

新建 quota.py：

import timeimport redisr = redis.Redis(decode_responses=True)QUOTA = {"summary": 500.0, "chat": 3000.0}      # 每月额度（元）def _key(scope: str) -> str:    return f"cost:{scope}:{time.strftime('%Y%m')}"class QuotaExceeded(Exception):    passdef check_quota(scope: str, est_cost: float) -> None:    used = float(r.get(_key(scope)) or 0)    limit = QUOTA.get(scope, float("inf"))    if used + est_cost > limit:        raise QuotaExceeded(f"{scope}: {used:.2f}/{limit} 已用尽")def record_cost(scope: str, real_cost: float) -> None:    k = _key(scope)    pipe = r.pipeline()    pipe.incrbyfloat(k, real_cost)    pipe.expire(k, 60 * 24 * 3600)    pipe.execute()

检查和记账分成两个函数，因为真实成本要等响应回来读到 usage 才知道。请求前用估算值预检，估宽松点没关系，真实值会在下一次预检时把误差收敛掉。

步骤五：加速率熔断，拦住失控

月度配额拦不住短时间的失控——死循环、重试风暴，这些的特征是速率突然抬高，绝对值可能还远没到月度上限。

追加到 quota.py：

BURST_LIMIT = {"summary": 120, "chat": 600}     # 每分钟调用上限class RateExceeded(Exception):    passdef check_burn_rate(scope: str) -> None:    bucket = f"burn:{scope}:{int(time.time()) // 60}"    calls = r.incr(bucket)    r.expire(bucket, 300)    if calls > BURST_LIMIT.get(scope, 10**9):        raise RateExceeded(f"{scope} 速率超限: {calls}/min")

阈值怎么定：取该场景近两周的每分钟调用量分布，拿 p99 乘三当上限，跑一段时间再往下收。一开始定紧了容易误伤正常流量。

步骤六：串起来，收在一个入口

上面几段代码如果散在各业务仓库里，很快会各自漂移：有的地方升级 SDK 忘了带埋点，有的把配额检查注释掉了因为"本地调试麻烦"。

所有调用走一个统一入口：

def complete(prompt: str, *, scene: str, env: str = "prod", **kw):    check_burn_rate(scene)    check_quota(scene, estimate_cost(prompt, **kw))    resp = client.chat.completions.create(        messages=[{"role": "user", "content": prompt}], **kw    )    cost = actual_cost(resp)    record_cost(scene, cost)    log_usage(scene, env, resp.model,              resp.usage.prompt_tokens, resp.usage.completion_tokens, cost)    return resp

这个入口层还有个附带好处：将来换上游、加多上游容灾、给某个场景灰度新模型，都只改这一处，业务方拿到的接口签名不变。

如果团队规模还不到自建网关的程度，退一步做法是用内部 SDK 包住官方 client，通过私有包分发。

小结

跟着六步做完，你会有一套完整的防护：

1. 埋点：带 scene 和 env，为所有阈值提供依据
2. 余额监控：换算成剩余天数，不看绝对值
3. 错误分级：余额耗尽与限流分开处理，配分场景降级
4. 用量配额：事前拦截，不是事后对账
5. 速率熔断：拦死循环和重试风暴
6. 收口到统一入口：避免各仓库实现漂移

顺序建议别调。先做埋点的原因是它立刻能止血（看清钱花在哪），也为后面的配额提供阈值依据——没有历史分布，配额定紧了误伤业务，定松了形同虚设。