AI 工作流如何做成本监控，避免调用费用失控？

学完这篇，你能给任意一条 AI 工作流装上"成本仪表盘"，做到每次调用花了多少钱一眼可见、超预算前先告警、失控时自动熔断。

第一步：在调用层埋点，把 token 用量记下来

这一步要拿到最原始的计费数据——输入和输出各用了多少 token。所有主流 API 的响应体里都带 usage 字段，把它取出来存好就行。

OpenAI 兼容接口的返回长这样：

{
  "usage": {"prompt_tokens": 812, "completion_tokens": 356, "total_tokens": 1168}
}

如果用的是流式输出（stream=true），很多 SDK 默认不返回 usage，需要显式打开：OpenAI Python SDK 里加 stream_options={"include_usage": True}，Anthropic 则在流结束后读 message_delta 事件里的 usage。

本地预估算的话，装 tiktoken（pip install tiktoken），用 tiktoken.encoding_for_model("gpt-4o").encode(text) 数出 token 数。

注意：别只用 len(text) 估算。中文一个字约 1～2 token，英文一个词约 1.3 token，差一倍就会让预算模型完全失准。

第二步：维护一张单价表，把 token 换算成钱

这一步把 token 变成金额。建一个 JSON 或 YAML 文件，按"模型名 → 输入单价 / 输出单价"记录，单位统一成"元 / 百万 token"。

gpt-4o:        { input: 18.0, output: 72.0 }
gpt-4o-mini:   { input: 1.1,  output: 4.3 }
claude-sonnet: { input: 21.6, output: 108.0 }

每次调用后套公式：

本次成本 = prompt_tokens/1e6*input_price + completion_tokens/1e6*output_price

单价以各家官方定价页为准，建议每季度核对一次。

注意：多模态调用要单独建条目。图片按张或按分辨率档位计费，音频按秒计费，混在一个 token 单价里算会漏掉大头。

第三步：落库 + 看板，让成本能按维度查

这一步把散落的成本数据变成可查询的记录。每次调用写一行到数据库，字段至少包含：

trace_id, workflow_id, user_id, model, prompt_tokens, completion_tokens, cost, created_at

量小用 PostgreSQL 就够；日均百万次以上换 ClickHouse。然后用 Grafana 或 Metabase 连上去，建三个面板：

1. 按天成本趋势（折线图，看有没有异常尖峰）
2. 按 workflow_id 成本排行（柱状图，找出最烧钱的那条流程）
3. 按 user_id 成本 Top 20（发现滥用账号）

如果你用的是 Dify，可以直接进「日志与标注」页面按应用查看 token 消耗；用阿里云百炼则在「费用中心 → 账单明细」按模型筛。但这些平台报表通常有小时级延迟，自建埋点才是实时的。

第四步：设预算阈值，超了立刻告警

这一步让问题在你发现之前先找到你。用 Prometheus + Alertmanager，或者更简单的方案：定时任务每 10 分钟查一次当日累计成本，超过阈值就调飞书/钉钉机器人 webhook。

if today_cost > daily_budget * 0.8:
    send_alert(f"今日 AI 成本已达 {today_cost:.2f} 元，预算 {daily_budget} 元")

阈值建议分两级：80% 触发提醒（还能人工干预），100% 触发熔断。

第五步：加硬熔断，别指望告警及时被人看到

这一步是最后一道闸。在调用前做检查，Redis 计数器最合适：

key = f"cost:{user_id}:{today}"
if float(redis.get(key) or 0) > user_daily_limit:
    raise BudgetExceeded("今日额度已用完")

调用后 redis.incrbyfloat(key, cost) 并 expire(key, 86400)。同时给所有调用强制设 max_tokens（比如单次不超过 2000），这是最便宜的保险——输出长度失控是超支的第一大原因。

第六步：回头优化，这才是真正省钱的环节

监控只是让你看见钱去哪了，省下来还得靠改：

• 模型分级路由：意图识别、分类、格式转换这类任务换小模型，成本能降 90%
• 语义缓存：高频重复问题（客服 FAQ 类）用 Redis 或 GPTCache 缓存答案，命中即零成本
• 提示词瘦身：系统提示词里那些"你是一个专业的……"删掉不影响效果，但每次都计费
• 裁剪上下文：多轮对话不要无脑全量回传，保留最近 5 轮 + 摘要

小结

1. 埋点是地基：从响应的 usage 字段拿 token，流式输出要显式开启用量返回
2. 单价表要维护，多模态单独建条目
3. 成本数据必须落库并可按 workflow / user 维度聚合
4. 告警设两级：80% 提醒、100% 熔断
5. 用 Redis 计数器做硬限额，并给每次调用设 max_tokens
6. 真正省钱靠模型分级、语义缓存、提示词瘦身和上下文裁剪