AI全栈知识15：AI应用的可观测性 - Token监控与成本控制
写在前面
你维护一个普通微服务，监控很简单：QPS、延迟、错误率、CPU、内存。这些指标稳定可预测，每次请求消耗的资源基本一致。
AI服务不一样。同一个接口，用户问"你好"可能消耗100个Token，问"帮我写一份完整的K8s部署方案"可能消耗5000个Token。每次请求的成本完全不可预测。
如果不做好监控，你可能遇到：
- 月底收到一张天价账单，不知道钱花在哪
- 某个用户在刷量，Token一直在烧但没人发现
- Agent陷入死循环，一个请求消耗了几万Token
- 用户投诉"AI好慢"，但你看CPU利用率才30%
这篇帮你建立AI服务的完整可观测性体系。
AI服务跟普通服务的监控区别
普通微服务的监控
指标固定、成本可预测： - QPS：每秒处理多少请求 - 延迟：每个请求多少毫秒 - 错误率：多少请求报错 - 资源：CPU/内存使用率 特点：每次请求消耗的资源差不多AI服务多出来的监控
指标波动大、成本不可预测： - Token消耗：每次请求不一样（100~10000+） - 首Token延迟：用户等多久看到第一个字 - 生成速度：每秒输出多少Token - 成本（元）：直接跟钱挂钩 - 调用轮次：Agent跑了几轮 - 模型选择：用了哪个模型（大小模型成本不同）| 维度 | 普通微服务 | AI服务 |
|---|---|---|
| 成本模型 | 固定（服务器费/月） | 变动（按Token计费） |
| 延迟构成 | 单一（处理时间） | 复合（首Token+生成） |
| 请求成本差异 | 几乎一样 | 差100倍（简单问题vs复杂问题） |
| 失败模式 | 崩溃/超时 | 额外有：死循环/Token爆炸/模型拒绝 |
| 成本归因 | 按服务分摊 | 需按团队/功能/用户归因 |
核心指标一：Token消耗
为什么Token是最重要的指标
Token = 钱。调用大模型API按Token计费：
通义千问qwen-turbo： 输入：0.003元/千Token 输出：0.006元/千Token GPT-4o： 输入：0.01元/千Token 输出：0.03元/千Token一个请求如果消耗5000 Token（输入3000+输出2000），用GPT-4o就花了0.09元。看着不多，100个用户每人每天10次 = 90元/天 = 2700元/月。
如果有个Agent死循环消耗了50000 Token，一次请求就花掉将近1元。不监控根本发现不了。
需要记录的Token指标
# 每次请求记录token_metrics={"request_id":"req-xxx","user_id":"user-001","function":"customer_service",# 哪个功能"model":"qwen-turbo",# 用了哪个模型"prompt_tokens":1500,# 输入Token"completion_tokens":800,# 输出Token"total_tokens":2300,# 总Token"cost_yuan":0.0093,# 换算成钱"rounds":3,# Agent跑了几轮}Prometheus打点
fromprometheus_clientimportCounter,Histogram# Token总量TOKENS_TOTAL=Counter('ai_tokens_total','Total tokens consumed',['model','function','token_type']# 按模型、功能、输入/输出分类)# 每次请求的Token数分布TOKENS_PER_REQUEST=Histogram('ai_tokens_per_request','Tokens per request',['model','function'],buckets=[100,500,1000,2000,5000,10000,20000,50000])# 使用时TOKENS_TOTAL.labels(model='qwen-turbo',function='customer_service',token_type='prompt').inc(1500)TOKENS_TOTAL.labels(model='qwen-turbo',function='customer_service',token_type='completion').inc(800)TOKENS_PER_REQUEST.labels(model='qwen-turbo',function='customer_service').observe(2300)核心指标二：延迟（TTFT + 生成速度）
两段延迟
AI服务的延迟跟普通服务不一样，分两段：
用户发送问题 ↓ [等待...] ← TTFT（Time To First Token）：首Token延迟 ↓ 用户在这段时间里看到的是空白/loading 开始输出第一个字 ↓ [持续输出中...] ← 生成速度：每秒输出多少Token ↓ 用户看到文字在一个一个往外蹦 输出完毕 ← 总延迟（End-to-End Latency）为什么要分开看
| 指标 | 影响什么 | 用户感受 |
|---|---|---|
| TTFT | 用户等多久开始看到回答 | TTFT>3秒用户就觉得卡了 |
| 生成速度 | 文字蹦出来的快慢 | <10 tokens/s像打字机，>30基本感知不到 |
| 总延迟 | 完整回答出来要多久 | Agent多轮可能几十秒 |
用户体验主要看TTFT。哪怕总回答需要10秒，只要0.5秒就开始出字，用户感觉就很快（流式输出的优势）。
Prometheus打点
TTFT=Histogram('ai_time_to_first_token_seconds','Time to first token',['model','function'],buckets=[0.1,0.3,0.5,1.0,2.0,3.0,5.0,10.0])GENERATION_SPEED=Histogram('ai_generation_tokens_per_second','Token generation speed',['model'],buckets=[5,10,20,30,50,100])E2E_LATENCY=Histogram('ai_request_duration_seconds','End-to-end request latency',['model','function'],buckets=[1,2,5,10,20,30,60])核心指标三：成本归因
问题
月底Token账单来了，总共花了5000元。老板问：钱花在哪了？哪个团队花的？哪个功能最烧钱？
如果没有成本归因，你只能说"总共花了5000"。有了归因，你能说：
本月Token成本归因： 客服机器人：3200元（64%）← 最大头 内部知识搜索：1100元（22%） 运维Agent：500元（10%） 测试/开发调试：200元（4%）怎么做
在每次API调用时打上标签（团队、功能、环境）：
classCostTracker:def__init__(self):self.records=[]defrecord(self,team,function,model,tokens,cost):self.records.append({"timestamp":datetime.now(),"team":team,# 哪个团队"function":function,# 哪个功能"model":model,# 什么模型"tokens":tokens,# 消耗多少Token"cost":cost,# 花了多少钱})defdaily_report(self):"""生成每日成本报告"""# 按团队/功能汇总...Prometheus Label设计
# 关键：通过label区分来源COST_YUAN=Counter('ai_cost_yuan_total','Total cost in yuan',['team','function','model','environment'])# 使用时COST_YUAN.labels(team='customer_service',function='chat_bot',model='qwen-turbo',environment='production').inc(0.009)Grafana里就能按team筛选，看每个团队的花费趋势。
核心指标四：调用链追踪
为什么需要
Agent可能调了5轮LLM、3个工具，最后给了一个错误答案。不记录调用链，你不知道是第几轮出的问题。
一次Agent请求的完整调用链
{"trace_id":"trace-abc123","request_id":"req-001","user_input":"帮我查查order-service为什么报错","total_duration_ms":8500,"total_tokens":4500,"total_cost_yuan":0.018,"rounds":[{"round":1,"action":"llm_call","model":"qwen-turbo","prompt_tokens":800,"completion_tokens":50,"duration_ms":1200,"decision":"call tool: get_unhealthy_pods"},{"round":2,"action":"tool_call","tool":"get_unhealthy_pods","duration_ms":300,"result":"order-service-xxx: CrashLoopBackOff"},{"round":3,"action":"llm_call","model":"qwen-turbo","prompt_tokens":1200,"completion_tokens":60,"duration_ms":1500,"decision":"call tool: get_pod_logs"},{"round":4,"action":"tool_call","tool":"get_pod_logs","duration_ms":500,"result":"OutOfMemoryError..."},{"round":5,"action":"llm_call","model":"qwen-turbo","prompt_tokens":1800,"completion_tokens":200,"duration_ms":2000,"decision":"final_answer"}]}有了这个，出问题时能精确定位：第几轮、调了什么、结果是什么、Token花在哪了。
接入方式
可以用OpenTelemetry或自己写日志。核心是每次LLM调用和工具调用都带上同一个trace_id：
importuuidclassRequestTracer:def__init__(self):self.trace_id=str(uuid.uuid4())self.rounds=[]deflog_llm_call(self,round_num,model,prompt_tokens,completion_tokens,duration_ms,decision):self.rounds.append({"round":round_num,"action":"llm_call","model":model,"prompt_tokens":prompt_tokens,"completion_tokens":completion_tokens,"duration_ms":duration_ms,"decision":decision,})deflog_tool_call(self,round_num,tool_name,duration_ms,result_preview):self.rounds.append({"round":round_num,"action":"tool_call","tool":tool_name,"duration_ms":duration_ms,"result":result_preview[:200],# 只存前200字})核心指标五：异常检测
AI服务独有的异常场景
| 异常 | 现象 | 原因 | 告警条件 |
|---|---|---|---|
| Token暴涨 | 单次请求>20000 Token | Agent死循环/用户输入超长文本 | 单次>阈值 |
| 成本突增 | 今日花费已超预算80% | 流量突增/被刷量/模型选错了 | 日成本>阈值 |
| 延迟飙升 | TTFT从0.5s变成5s | 模型过载/并发太高 | P99>阈值 |
| 拒绝率升高 | 模型频繁返回无法回答 | 安全策略误杀/Prompt有问题 | 拒绝率>10% |
| 空回答 | 模型返回空字符串 | 模型异常/Token用完 | 空回答率>5% |
| 重复调用 | 同一用户短时间大量请求 | 恶意刷量/前端bug | 每分钟>20次 |
告警规则（Prometheus AlertManager）
groups:-name:ai_service_alertsrules:# Token单次暴涨-alert:HighTokensPerRequestexpr:ai_tokens_per_request>20000for:0mlabels:severity:warningannotations:summary:"单次请求Token异常高"# 日成本超预算-alert:DailyCostExceededexpr:sum(increase(ai_cost_yuan_total[24h]))>200for:5mlabels:severity:criticalannotations:summary:"今日AI成本已超200元"# TTFT延迟飙升-alert:HighTTFTexpr:histogram_quantile(0.99,ai_time_to_first_token_seconds)>5for:5mlabels:severity:warningannotations:summary:"首Token延迟P99超过5秒"# 用户刷量-alert:UserRateLimitexpr:sum(rate(ai_tokens_total[1m])) by (user_id)>50000for:2mlabels:severity:warningannotations:summary:"某用户Token消耗速率异常"Grafana面板设计
推荐面板布局
第一行：概览 - 今日总Token消耗 + 今日总成本（大数字显示） - 当前QPS - 当前活跃用户数 第二行：延迟 - TTFT P50 / P95 / P99 趋势图 - 生成速度 (tokens/s) 趋势图 - 请求总延迟分布 第三行：Token详情 - Token消耗趋势（按功能/团队分色） - 每次请求Token分布（直方图） - 输入Token vs 输出Token比例 第四行：成本 - 每日成本趋势（柱状图） - 成本按团队归因（饼图） - 成本按模型归因（饼图） - 预算消耗进度条 第五行：异常 - 单次高Token请求列表 - 错误/拒绝率趋势 - Agent平均轮次趋势PromQL示例
# 今日总Token sum(increase(ai_tokens_total[24h])) # 今日总成本（元） sum(increase(ai_cost_yuan_total[24h])) # TTFT P99 histogram_quantile(0.99, rate(ai_time_to_first_token_seconds_bucket[5m])) # 按团队的Token消耗速率 sum(rate(ai_tokens_total[5m])) by (team) # Agent平均轮次 avg(ai_agent_rounds_per_request)成本控制实战
预算管理
classBudgetManager:def__init__(self,daily_budget_yuan=200,monthly_budget_yuan=5000):self.daily_budget=daily_budget_yuan self.monthly_budget=monthly_budget_yuan self.daily_spent=0self.monthly_spent=0defcheck_budget(self,estimated_cost):"""请求前检查预算"""ifself.daily_spent+estimated_cost>self.daily_budget:returnFalse,"今日预算已用完"ifself.monthly_spent+estimated_cost>self.monthly_budget:returnFalse,"本月预算已用完"returnTrue,"OK"defon_request_complete(self,actual_cost):"""请求完成后记录花费"""self.daily_spent+=actual_cost self.monthly_spent+=actual_cost# 达到80%告警ifself.daily_spent>self.daily_budget*0.8:send_alert("今日预算已使用80%")省钱策略
| 策略 | 做法 | 节省比例 |
|---|---|---|
| 大小模型路由 | 简单问题用小模型，复杂问题用大模型 | 30-50% |
| Prompt精简 | 去掉冗余的System Prompt | 10-20% |
| 缓存 | 相同问题直接返回缓存结果 | 视场景20-80% |
| 限流 | 每用户每分钟限制请求次数 | 防止异常消耗 |
| 设Token上限 | max_tokens限制输出长度 | 防止超长回答 |
| 选择合适模型 | 不是所有场景都需要最贵的模型 | 50-90% |
缓存示例
importhashlibclassResponseCache:def__init__(self,ttl_seconds=3600):self.cache={}self.ttl=ttl_secondsdefget_cache_key(self,model,messages):"""相同的输入生成相同的key"""content=f"{model}:{str(messages)}"returnhashlib.md5(content.encode()).hexdigest()defget(self,model,messages):key=self.get_cache_key(model,messages)ifkeyinself.cache:entry=self.cache[key]iftime.time()-entry["time"]<self.ttl:returnentry["response"]# 命中缓存，省TokenreturnNonedefset(self,model,messages,response):key=self.get_cache_key(model,messages)self.cache[key]={"response":response,"time":time.time()}完整监控架构
AI应用（打点） │ ├── Prometheus指标（Token/延迟/成本/错误率） │ ↓ │ Prometheus Server │ ↓ │ Grafana（面板展示） │ ↓ │ AlertManager（告警通知） │ ├── 调用链日志（每轮详情） │ ↓ │ ELK / Loki（存储查询） │ ↓ │ Kibana / Grafana（搜索回溯） │ └── 成本报表 ↓ 定时任务（每日/每周汇总） ↓ 发送到钉钉/邮件面试怎么说
如果被问"AI服务怎么做监控"：
"AI服务跟普通微服务最大的区别是每次请求成本不固定，所以监控要重点关注Token消耗和成本。
我的监控体系分五个维度：
- Token消耗：按模型、功能、团队打标签，能做成本归因
- 延迟：分首Token延迟（TTFT）和总延迟，用户体验主要看TTFT
- 成本：每日预算管控，达到80%告警，达到100%限流
- 调用链：Agent的每轮LLM调用和工具调用都记录，出问题能回溯
- 异常检测：单次Token暴涨、用户刷量、死循环都有告警规则
技术栈用Prometheus打点 + Grafana面板 + AlertManager告警，调用链用ELK或Loki。
还有一些省钱手段：大小模型路由（简单问题用小模型）、响应缓存、Prompt精简。我们落地后Token成本降了约40%。"
延伸思考
| 问题 | 答案 |
|---|---|
| 自部署vLLM也需要监控Token吗 | 需要。虽然不按Token付费，但Token数影响延迟和吞吐，是容量规划的依据 |
| 怎么估算一次请求的成本 | 输入Token × 输入单价 + 输出Token × 输出单价。大部分API返回usage字段 |
| 监控数据存多久 | 明细日志保留7-30天，聚合指标保留3-6个月做趋势分析 |
| 小公司需要做这么完整吗 | 不用。先做Token总量+日成本+TTFT这三个最核心的，其他慢慢补 |
| OpenTelemetry支持AI监控吗 | 社区在推LLM Observability标准，LangSmith和LangFuse是专门做AI调用链的工具 |
小结
本篇核心收获：
- AI服务监控核心差异：每次请求成本不固定，Token=钱
- 五大监控维度：Token消耗、延迟（TTFT+总延迟）、成本归因、调用链、异常检测
- TTFT比总延迟更重要：用户体验看首字输出速度
- 成本归因必须做：按团队/功能打标签，知道钱花在哪了
- 异常检测防止烧钱：Token暴涨、死循环、刷量都要有告警
- 省钱手段：大小模型路由+缓存+Prompt精简+限流
下一篇预告
AI全栈知识16：AI应用架构设计 - 从单体到平台
下一篇进入架构设计：
- OneAPI网关层设计
- 应用层（RAG/Agent）的分层
- 模型层（vLLM）的部署策略
- 从单应用到AI平台的演进路径
参考链接
- Prometheus Client Python
- OpenTelemetry LLM Observability
- LangFuse（AI调用链追踪）
- LangSmith（LangChain官方追踪）
- vLLM Metrics文档
