LLM 应用的Token级可观测性：从Trace 采集到 CostAttribution 的工程落地

引子：凌晨 3 点的账单告警

上个月的一个周二凌晨，我们的 LLM 应用收到了费用异常告警——过去 24 小时的 API 调用成本比上周同期高了 47%。

值班同学打开 Grafana，看到了熟悉的"总 Token 消耗量"曲线确实有一个陡峭的上升。但当他试图回答"是谁用的"“用了什么模型”"是哪个功能触发的"这三个问题时，发现手头的监控只能看到聚合数字，没有任何一条 trace 能把一次调用的 input tokens、output tokens、model name、user id、feature flag 串起来。

最终花了 4 个小时翻日志、对时间戳、手动关联才定位到一个新上线的 RAG 功能在处理长文档时没有做 chunk size 限制，导致单次请求的 input tokens 飙到了 120K。

如果我们有 Token 级的可观测性，这个排查应该在 5 分钟内完成。

这篇文章就是关于如何建设这套能力的。不是讲"为什么要监控"的道理，也不是讲"怎么省钱"的策略，而是聚焦一个具体的工程问题：如何让你的 LLM 应用具备 Token 粒度的 Trace 能力和 Cost Attribution 能力。

一、Token 级可观测性到底在解决什么问题

在传统的 Web 应用中，可观测性的基本单位是Request：一次 HTTP 请求的延迟、状态码、错误信息。但在 LLM 应用中，这个粒度不够了。

一次 LLM 调用的核心度量单位是Token。Token 数量直接决定了：

• 成本：几乎所有 LLM API 都按 token 计费，且 input/output 价格不同
• 延迟：output token 数量直接影响 TTFT（Time To First Token）和总生成时间
• 质量：context window 的使用率影响模型的注意力分配
• 配额：rate limit 通常以 TPM（Tokens Per Minute）为单位

所以我们需要一个新的观测维度：每一次 LLM 调用消耗了多少 token、属于谁、为了什么目的、花了多少钱。

这就是 Token 级可观测性的定义。它不是传统 APM 的简单扩展，而是 LLM 应用特有的工程需求。

与传统 APM 的关键差异

理解了这些差异，你就知道为什么直接把 Datadog/Prometheus 的传统指标套用到 LLM 应用上会水土不服。

二、OpenTelemetry GenAI Semantic Conventions：标准化的基石

2025 年底，OpenTelemetry 的 GenAI Semantic Conventions 从 experimental 升级到 stable。这意味着我们终于有了一个厂商无关的标准来描述 LLM 调用的 trace 数据。

核心 Span 属性

一次 LLM Chat Completion 调用的标准 span 包含以下关键属性：

# OpenTelemetry GenAI Semantic Conventions (stable)span.set_attributes({# 系统标识"gen_ai.system":"openai",# 或 anthropic, google, etc."gen_ai.request.model":"gpt-4o",# Token 用量（核心！）"gen_ai.usage.input_tokens":1250,"gen_ai.usage.output_tokens":380,# 请求参数"gen_ai.request.temperature":0.7,"gen_ai.request.max_tokens":1024,# 响应元数据"gen_ai.response.finish_reasons":["stop"],"gen_ai.response.id":"chatcmpl-abc123",})

这几个gen_ai.usage.*属性就是 Token 级可观测性的数据基础。有了它们，我们就可以在 Jaeger/Grafana/Langfuse 中按 token 维度进行聚合、过滤和告警。

Span 层级结构

一个真实的 LLM 应用调用链通常长这样：

[Root Span] handle_user_query (2.3s) ├── [Span] llm.chat gpt-4o (1.8s) │ ├── input_tokens: 2,400 │ └── output_tokens: 850 ├── [Span] embedding text-embedding-3-small (0.12s) │ └── input_tokens: 380 └── [Span] llm.chat gpt-4o-mini (0.3s) # 二次调用（如 summary） ├── input_tokens: 900 └── output_tokens: 120

这种层级结构让我们可以精确地知道：一次用户请求总共消耗了多少 token，其中多少用于主推理、多少用于 embedding、多少用于后处理。

自动 Instrumentation vs 手动埋点

目前主流的 Python/TypeScript LLM SDK 都提供了 OpenTelemetry 自动 instrumentation：

# 方式一：自动 instrumentation（推荐起步）fromopenllmetryimportinit init()# 自动 patch openai, anthropic, langchain 等# 方式二：手动埋点（精细控制）fromopentelemetryimporttrace tracer=trace.get_tracer("my-llm-app")asyncdefcall_llm(prompt:str,user_id:str):withtracer.start_as_current_span("llm.chat",attributes={"gen_ai.system":"openai","gen_ai.request.model":"gpt-4o","app.user_id":user_id,# 自定义业务属性"app.feature":"document_summary",# 功能标识})asspan:response=awaitclient.chat.completions.create(...)# 回填 token 用量span.set_attribute("gen_ai.usage.input_tokens",response.usage.prompt_tokens)span.set_attribute("gen_ai.usage.output_tokens",response.usage.completion_tokens)returnresponse

我的建议是两者结合：用自动 instrumentation 覆盖标准 LLM 调用，用手动埋点补充业务上下文（user_id、tenant_id、feature flag）。没有业务上下文的 token trace，在做 cost attribution 时会非常痛苦。

三、Streaming 场景的 Token 计数难题

如果你的 LLM 应用使用了 streaming（绝大多数生产应用都会用），你会发现一个棘手的问题：在 SSE streaming 模式下，你拿不到准确的 token 计数。

问题本质

OpenAI 的 streaming response 中，每个 chunk 只包含 delta content，不包含 usage 信息。虽然最新的 API 支持在最后一个 chunk 中返回stream_options.include_usage=true，但这有两个限制：

1. 并非所有厂商都支持
2. 如果 stream 中途断开（网络超时、客户端取消），你永远拿不到最终的 usage

三种解决方案

方案 A：依赖 API 返回的 usage（最准确）

# OpenAI streaming with usageresponse=awaitclient.chat.completions.create(model="gpt-4o",messages=messages,stream=True,stream_options={"include_usage":True}# 关键参数)final_usage=Noneasyncforchunkinresponse:ifchunk.usage:final_usage=chunk.usage# 最后一个 chunk 包含 usageiffinal_usage:span.set_attribute("gen_ai.usage.input_tokens",final_usage.prompt_tokens)span.set_attribute("gen_ai.usage.output_tokens",final_usage.completion_tokens)else:# stream 中断，降级到估算span.set_attribute("gen_ai.usage.output_tokens",estimated_tokens)span.set_attribute("app.token_count.source","estimated")

方案 B：本地 Tokenizer 实时计数（最通用）

importtiktokenclassStreamingTokenCounter:"""实时计算 streaming 输出的 token 数"""def__init__(self,model:str):self.encoding=tiktoken.encoding_for_model(model)self._buffer=""self._token_count=0deffeed(self,delta_content:str)->int:"""喂入一个 chunk，返回累计 token 数"""self._buffer+=delta_content# 注意：不能逐字符 tokenize，需要累积后重新计算# 因为 BPE tokenizer 的边界可能在 chunk 中间new_count=len(self.encoding.encode(self._buffer))delta=new_count-self._token_count self._token_count=new_countreturnself._token_count@propertydeftotal_tokens(self)->int:returnself._token_count

这里有一个容易踩的坑：BPE tokenizer 不是字符级别的。你不能对每个 chunk 单独 tokenize 然后累加，因为 token 边界可能跨越两个 chunk。正确做法是累积所有文本后重新 tokenize。

方案 C：API Proxy 层统一采集（推荐生产方案）

如果你有自己的 LLM Gateway / API Proxy，可以在 proxy 层同时拿到 request body（包含 input）和完整的 streaming response（拼接后的 output），在 proxy 侧做 token 计数和 trace 上报。这种方式对业务代码零侵入：

// LLM Gateway 中间件伪代码app.post("/v1/chat/completions",async(req,res)=>{conststartTime=Date.now();constinputTokens=countTokens(req.body.messages,req.body.model);// 转发请求并收集完整响应constupstreamResponse=awaitforwardToUpstream(req);if(req.body.stream){// Streaming: 透传 chunks，同时累积内容letoutputContent="";constpassthrough=newTransformStream({transform(chunk,controller){outputContent+=extractDeltaContent(chunk);controller.enqueue(chunk);// 透传给客户端},flush(){// Stream 结束后上报 traceconstoutputTokens=countTokens(outputContent,req.body.model);reportTokenTrace({userId:req.headers["x-user-id"],model:req.body.model,inputTokens,outputTokens,latencyMs:Date.now()-startTime,feature:req.headers["x-feature"],});}});upstreamResponse.body.pipeThrough(passthrough).pipeTo(res.writable);}else{// Non-streaming: 直接从 response 取 usageconstoutputTokens=upstreamResponse.usage?.completion_tokens??countTokens(upstreamResponse.choices[0].message.content,req.body.model);reportTokenTrace({/* ... */});res.json(upstreamResponse);}});

生产建议：优先用方案 C（Proxy 层采集）作为主力，方案 A 作为校验，方案 B 作为 fallback。三者互补，确保 token 计数的覆盖率接近 100%。

四、Cost Attribution：把 Token 变成钱

有了 token trace 数据，下一步是把它转化为可操作的 cost insight。这需要一个Cost Attribution 模型。

四维归因框架

我们在实践中总结出了四个归因维度：

Token Cost = f(User, Tenant, Feature, Model)

实现：从 Trace 到 Cost Dashboard

# 成本归因处理器classCostAttributionProcessor:"""将 token trace 转化为成本记录"""# 模型定价表（应定期同步厂商最新价格）PRICING={"gpt-4o":{"input":2.50,"output":10.00,"cached_input":1.25},"gpt-4o-mini":{"input":0.15,"output":0.60,"cached_input":0.075},"claude-sonnet-4-20250514":{"input":3.00,"output":15.00,"cached_input":0.30},"text-embedding-3-small":{"input":0.02,"output":0},}defprocess_trace(self,trace:dict)->dict:model=trace["model"]pricing=self.PRICING.get(model)ifnotpricing:logger.warning(f"Unknown model pricing:{model}")returnNoneinput_cost=trace["input_tokens"]*pricing["input"]/1_000_000output_cost=trace["output_tokens"]*pricing["output"]/1_000_000# Cached token 折扣处理cached_cost=0iftrace.get("cached_tokens"):cached_cost=trace["cached_tokens"]*pricing.get("cached_input",pricing["input"])/1_000_000input_cost-=trace["cached_tokens"]*pricing["input"]/1_000_000total_cost=input_cost+output_cost+cached_costreturn{"timestamp":trace["timestamp"],"user_id":trace.get("user_id"),"tenant_id":trace.get("tenant_id"),"feature":trace.get("feature","unknown"),"model":model,"input_tokens":trace["input_tokens"],"output_tokens":trace["output_tokens"],"cached_tokens":trace.get("cached_tokens",0),"cost_usd":round(total_cost,6),"latency_ms":trace.get("latency_ms"),}

实时估算 vs 异步对账

在生产环境中，我们采用了双轨制：

1. 实时估算：在 API Proxy 层，用本地 tokenizer 即时计算 token 数并乘以单价，写入 Redis 滑动窗口计数器。用于实时告警和配额控制。精度约 ±5%。

2. 异步对账：每小时从 trace backend（如 ClickHouse）拉取完整 trace 数据，用厂商 API 返回的精确 usage 重新计算成本，修正实时估算的偏差。用于日报、周报和计费。

这种设计的好处是：实时路径保证低延迟（不影响请求处理），异步路径保证高精度（不丢钱）。

五、生产环境的三个关键权衡

1. 采样策略：全量还是抽样？

Token trace 本身也是有成本的。如果你的应用日均 100 万次 LLM 调用，每次 trace 写入 ClickHouse 约 500 bytes，那一天就是 500MB 的 trace 数据。不算大，但如果加上 prompt/completion 原文存储，量级就完全不同了。

我们的采样策略：

classTraceSampler:defshould_sample(self,trace_context:dict)->bool:# 错误请求：100% 采集iftrace_context.get("error"):returnTrue# 高延迟请求（>10s）：100% 采集iftrace_context.get("latency_ms",0)>10000:returnTrue# 高 token 消耗（>50K）：100% 采集total_tokens=trace_context.get("input_tokens",0)+trace_context.get("output_tokens",0)iftotal_tokens>50000:returnTrue# 常规请求：10% 采样returnrandom.random()<0.1

核心原则：异常全采，正常抽样。这样既控制了存储成本，又确保出了问题时有完整的 trace 可查。

2. Token 计数精度：够用就行

tiktoken 和厂商 API 返回的 token 数并不总是完全一致。原因包括：

• 不同版本的 tokenizer 实现有细微差异
• 某些厂商使用修改版的 BPE
• system prompt 的 token 计算方式可能不同

务实的做法：以厂商 API 返回的 usage 为准（ground truth），本地 tokenizer 仅用于 streaming 中断时的 fallback 估算。在 cost dashboard 中标注数据来源（api_returnedvsestimated），让使用者知道数据的可信度。

3. 隐私合规：trace 里记不记原文？

这是一个需要在安全和可观测性之间做取舍的问题。

分级策略：

在生产环境中，我们只记录 token 数量和元数据（model、user_id、feature、latency），不记录 prompt 和 completion 的原文。如果需要调试具体问题，通过 feature flag 临时开启特定用户的详细 trace。

六、落地路线图：从零到可用的四个阶段

如果你的团队还没有 Token 级可观测性，这里是一个渐进式的落地路线：

Stage 1：基础 Trace（1-2 周）

• 接入 OpenTelemetry GenAI instrumentation
• 采集 input_tokens / output_tokens / model / latency
• 写入现有 trace backend（Jaeger/Tempo）
• 验收标准：能在 Jaeger 中看到每次 LLM 调用的 token 用量

Stage 2：业务上下文注入（1 周）

• 在 trace 中添加 user_id / tenant_id / feature 等业务属性
• 建立 Cost Attribution 的数据管道
• 搭建基础的 Grafana Dashboard
• 验收标准：能按用户/功能/模型维度查看 token 消耗和成本

Stage 3：告警与配额（1-2 周）

• 配置基于 token 消耗的异常告警
• 实现租户级/用户级 token 配额控制
• 接入实时估算 + 异步对账的双轨成本计算
• 验收标准：token 消耗异常能在 5 分钟内触发告警

Stage 4：深度分析（持续迭代）

• Token 消耗趋势分析与预测
• 模型 cost/quality 对比分析
• Prompt Caching 命中率监控
• 与业务指标（转化率、留存率）的关联分析
• 验收标准：能用数据驱动模型选型和功能优化决策

七、写在最后

回到开头那个凌晨 3 点的故事。在我们建设完 Token 级可观测性之后，类似的异常排查变成了这样：

1. 收到告警 → 打开 Grafana Token Cost Dashboard
2. 按 feature 维度下钻 → 发现document_summary功能的成本占比从 5% 飙升到 35%
3. 按 user 维度再下钻 → 确认不是个别用户的问题，是该功能本身的逻辑变更
4. 查看该功能的 trace 详情 → 发现平均 input tokens 从 3K 涨到了 80K
5. 定位到根因 → 新版本的 chunking 逻辑去掉了 max_chunk_size 限制

全程 4 分钟。

Token 级可观测性不是一个锦上添花的能力，它是 LLM 应用进入生产环境的入场券。就像你不会让一个 Web 应用在没有 APM 的情况下上线一样，你也不应该让一个 LLM 应用在没有 Token Trace 的情况下裸奔。

希望这篇文章能帮你在自己的项目中落地这套能力。如果你已经在做了，欢迎在评论区分享你的实践经验。

参考资料

• OpenTelemetry GenAI Semantic Conventions: https://opentelemetry.io/docs/specs/semconv/gen-ai/
• Langfuse Documentation: https://langfuse.com/docs
• Arize Phoenix: https://docs.arize.com/phoenix
• tiktoken: https://github.com/openai/tiktoken