问题比你想的更普遍

LLM API 的计费有个根本性的不对称：调用前你不知道要花多少钱。按理说，

• 传统 API成本可预期。固定费用，
• 大模型 API同一请求可能因 prompt 长度、程序消息复杂度、输出长度而相差数十倍。
• 常见爆炸场景
• Agent 工具调用循环: 工具返回异常导致无限重试，几分钟内把预算烧光。
• Prompt 膨胀: 对话历史无窗口截断，成本呈指数级增长。
• 多租户噪音租户: 某租户被滥用或并发脚本导致总账单飙升，但被其他租户淹没。
• 未设置 max_tokens: 模型输出超长。没有上限，成本不受控制，说起来，

成本异常检测的三层架构

从工程角度。LLM 成本监控可以分三层。其实，根据团队规模与复杂度选取合适方案：

Layer	适用场景
网站级	小团队、直接使用模型网站 Projects 分账。无需改业务代码，
应用级	单体应用、轻量监控，不想引入额外基础设施。
网关级	微服务、多团队、需要统一观测面板的场景。老实说，

多层组合可同时获得实时信号与日级对账功能。详细说说每一层实现与告警规则。

Layer ：网关级监控

至于主要思路。在应用代码和 LLM API 之间插入轻量代理，拦截所有请求/响应，提取 usage 信息并推送至 Promeus。

代理实现

# llm_gateway.py
import time,asyncio,httpx
from promeus_client import Counter,Histogram,start_http_server
TOKEN_TOTAL = Counter(
"llm_tokens_total","Total tokens consumed",)
COST_TOTAL = Counter(
"llm_cost_usd_total","Total estimated cost in USD",)
REQUEST_DURATION = Histogram(
"llm_request_duration_seconds","LLM API request latency",)
MODEL_PRICES = {
# 示例价格表
"deepseek-v3": {"input": 5/1_000_000,"output": 10/1_000_000},}
def estimate_cost -> float:
prices = MODEL_PRICES.get
return prices * prompt_tokens + prices * completion_tokens
async def proxy_llm_request(payload: dict,tenant_id: str,feature: str,upstream_url: str = "https://api.deepseek.com/v1/chat/completions"。api_key: str = "") -> dict:
model = payload.get
start = time.monotonic
async with httpx.AsyncClient as client:
resp = await client.post(
upstream_url,json=payload,headers={"Authorization": f"Bearer {api_key}","Content-Type": "application/json"}
)
resp.raise_for_status
data = resp.json
duration = time.monotonic - start
REQUEST_DURATION.labels.observe
usage = data.get
prompt_tokens = usage.get
completion_tokens= usage.get
TOKEN_TOTAL.labels(model=model,tenant_id=tenant_id,feature=feature,token_type="prompt").inc
TOKEN_TOTAL.labels(model=model,tenant_id=tenant_id,feature=feature,token_type="completion").inc
cost = estimate_cost
COST_TOTAL.labels(model=model,tenant_id=tenant_id,feature=feature).inc
return data
提示：在开启服务前执行 `start_http_server` 开启 Promeus 抓取端点。

Grafana 告警规则

# grafana-alerts.yaml
groups这方面，- name: llm_cost_anomaly
说到rules，# Rule1：5 分钟 token 消耗速率超过历史基线的3倍
- alert: LLMTokenBurnRateAnomaly
至于expr，|
rate>
avg_over_time) * 3
for的观点是。2m
说到labels，severity: warning
annotations:
summary: 'LLM token 消耗异常 '
description: |
Feature {{ $labels.feature }} 的 tenant {{ $labels.tenant_id }} 在过去 {{ $value | humanize }} 分钟内
token 消耗速率是过去1小时均值的 {{ $value | humanize }} 倍。# Rule2：单次请求输出过多
- alert: LLMSingleRequestOveruse
expr这方面，increase>?，for :0m
说到labels，severity : warning
annotations:
summary : 'LLM 单次请求输出 token 超过阈值'
description :
'{{ $labels.feature }} 在最近一分钟内 completion token 增量超过阈值。请检查是否设置了 max_token。'
# Rule3：单租户成本占比异常
- alert : LLMTenantCostDominance
expr :
rate /
ignoring group_left sum without)>?，for :5m
labels :
severity : critical
annotations :
summary : '单租户 LLM 成本占比过高'
description : '租户 {{ $labels.tenant_id }} 在过去10分钟内占总 LLM 成本的百分之 {{ |humanize}}，可能存在滥用。'
将上述 YAML 导入 Grafana 后即可在 “Alerting” 页面查看实时告警。

Layer ：应用级轻量监控

适用于不想部署完整监控栈，只需单文件即可运行的场景。主要思路是包装 SDK 并写入本地数据库，接下来周期性检索数据做 z‑score 异常检测。

SDK 包装器示例

# cost_tracker.py
import sqlite3,time,json,toml,humanize
from datetime import datetime。timedelta
from typing import Optional
MODEL_PRICES_CNY={
# RMB / 百万 token 示例价表
'deepseek-v3':,}
class CostTrackingLLMClient:
def __init__:
self.base_url=biz_url;self.api_key=api_key;self.db_path=db_path;self._init_db
def _init_db:
with sqlite3.connect as c:
c.execute('''CREATE TABLE IF NOT EXISTS cost_log(
ts REAL NOT NULL,tenant TEXT NOT NULL,feature TEXT NOT NULL,model TEXT NOT NULL,prompt INT,completion INT,cost_cny REAL。latency_ms INT )''')
c.execute')
c.execute')
def _estimate_cost->float:
in_p,out_p=MODEL_PRICES_CNY.get)
return +)/1000000
def _log:
with sqlite3.connect as c:c.execute',args)
async def chat(self,messages:list,m:model='deepseek-v3',tnt:str='default',feat:str='unknown',max_tok:int=None):
import httpx;话说回来，start=time.time
payload={'model': m,'messages':messages,'max_token':max_tok}
async with httpx.AsyncClient as cli:
r=await cli.post(f'{self.base_url}/chat/completions'。
json=payload,jsonify=True if hasattr else False,# 为兼容老版 httpx 写法略去...
headers={'Authorization': f'Bearer {self.api_key}','Content-Type':'application/json'})
r.raise_for_status;data=r.json
latency=int-start)*1000);us=data.get
cp_prompt=int);cp_comp=int)
cost=self._estimate_cost
self._log,tnt,feat。m,c_prompt_cp_prompt,c_completion_cp_comp,cost,latency);return data
将此类嵌入业务逻辑后每次调用即自动记录成本信息。

滑动窗口异常检测器示例

# anomaly_detector.py
import sqlite3,time,statistics,json,requeests,schedule。detaillogging as log
class CostAnomalyDetector:
def __init__:
self.db=basedb
def _query_sum->float:
sql='SELECT COALESCE,0) FROM cost_log WHERE ts>?AND ts完整实现请参考原文中的 `CostAnomalyDetector` 类；老实说，主要是对比当前 N 分钟窗口与历史同周期均值。通过 z‑score 判断是否超阈值，从而触发飞书或邮件通知。

利用模型厂商提供的 Usage API 做日级对账和趋势告警。其实，不需要改业务代码，但粒度最粗，只能发现“今天比昨天贵很多”的情况。

通用日级成本对比脚本示例

# usage_poller.py
import httpx,asyncio,datetime as dt
async def fetch_daily_usage:
end_date=datetime.date.today
start_date=end_date-dt.timedelta
async with httpx.AsyncClient as cli:
resp=await cli.get(f'{bizurl}/organization/usage'。params={'start_date':start_date.isoformat,'end_date':end_date.isoformat},headers={'Authorization':f'Bearer {api_key}'})
resp.raise_for_status;return resp.json.get
async def check_spike:
logsawait fetch_daily_usage
...
只需配置 cron job 每晚运行一次即可得到每日费用报表及超支告警。

四类异常的处置建议 & 根因修复表格

推荐处置 根因修复 Warning记录日志。飞书通知后强制设置 #max_token#/#set_limit# 校验返回长度或加 #max_output_length# Warning检查客户端重试风暴，在入口加 rate limiting；评估错误率是否异常调整重试策略并加入熔断器--/>Critical立即暂停该 Agent 的 key，加 max_turns/max_cost 熔断器维护 Agent 层预算管理—如下 BudgetedAgent 示例--/>Critical临时限速该租户；通知客户引入 per‑tenant quota 完善多租户隔离策略和配额管控--/>

*BudgetedAgent 示例*: 为每个 Agent 创建一个预算限制，在每一步计算消耗并提前终止。以防止无限循环吞噬预算.

python class BudgetedAgent: """带成本熔断器的 Agent 包装""" def __init__): self.budget_cny ​ ​budget_cny       ​    ​spent_cny ​ ​int      turns ​int      client=COST_TRACKING_CLIENT async def step: if self.spent_cny>=self.budget_cny or self.turns>=MAX_TURNS:return raise RuntimeError resp=self.client.chat;cost=_calc，…

从实验数据来看，三种方案的检测延迟对比

异常类型	告警等级
单请求输出过大
全局频率骤增
Agent 成本滚雪球
单租户占比 >%

*误报率基于真实流量测试得出；延迟以首次触发时间为准，*

方案	告警延迟	误报率	部署复杂度
LPM+Grafana — 分钟 <30 %低
SQlite+定时检测 - 分钟 <15 %中等
User‑API 拉取- 分钟≥50 %最低 td valign=”top”≥50 min td valign=”top”≤10 % td valign=""低

几个容易踩的坑 & 避免方式：

• PROM counter 重置导致误报："rate" 会看到负增长再归零。可使用 #increase,或让 counter 永久递增且不手动清零.

价格表不及时更新："硬编码价格会偏差累积";建议使用外部 YAML/JSON 配置。并通过热 reload 或容器重启更新.

只监控成本而忽略 Token 数："cost/token ≈ price"，但 price 波动大；建议先监测 raw tokens 再算成本.

MIS streaming 模式："stream=True 时缺少 final usage"，可在 stream 完成后手动发送一次 “usage event”。如果无法获取，可禁用 streaming 或通过 SDK 抽象统一处理.

SQlite 数据膨胀："cost_log 表会变慢"。定期执行 VACUUM 与保留最近 N 天的数据即可保持查询性能.

• 刚起步项目——先做 SQLite+z‑score；按理说，两小时内即可发现问题。毫无额外基础设施负担.

已有监控栈——直接上 Prom/Grafana；三条 alert rule 足矣，覆盖全局频繁和单轮消耗两类场景.

小团队直接使用模型网站——Usage API 拉取 + 比率告警；几乎零开发成本，却能把日常开销锁定在视野之中.

1. 强制设定 #max_token## —— 最简单的一次性消耗防护.

Agent 层维护预算熔断器 —— 循环内部检查。更早拦截潜在滥用.

多租户 per‑tenant Token 跟踪 —— 防止坏客户淹没整体指标，让每个使用者都透明可见.

Maturity note: LLM 成本观测仍处于快速迭代阶段，但以上几种方案已足够应对当下最常见且破坏性的“费用炸弹”。按理说，无论选择哪一层，都请记得：

• 持续审查 “price map”。不要让估算漂移影响决策. '价格变更就等价于重新估算'.

`

。

---