1. 为什么 LLM 应用的成本追踪是个绕不开的坎做 LLM 应用的人都有一个共同的体感功能跑通只是第一步账单才是真正的长期考验。你调用一次大模型接口背后消耗的 token 数量、模型单价、缓存命中情况、重试次数、流式返回的分段计费方式这些因素叠加在一起成本曲线往往比预期陡得多。尤其是当应用从单轮问答扩展到多轮对话、Agent 编排、RAG 检索增强、批量离线推理之后成本结构会变得极其复杂靠人工估算基本不可能准确。Opik 这个工具最初吸引我的地方就是它把 LLM 调用链路的追踪和成本统计做在了一起。它不是单纯给你一个仪表盘看总数而是能把每一次调用、每一个 span、每一层嵌套的 token 消耗都记录下来并且支持在数据缺失时做自动补算。这个能力在实际生产环境里非常关键因为很多 LLM 框架在流式返回或者异常中断时并不会完整上报 usage 字段导致账单和实际消耗对不上。这篇文章面向的是已经在做 LLM 应用、或者正准备把原型推向生产的开发者。不管你用的是 OpenAI 兼容接口、还是本地部署的开源模型只要涉及 token 计费或者资源消耗统计这套思路都能直接参考。我会从整体设计、核心细节、实操落地、问题排查四个层面把 Opik 追踪 LLM 成本的完整链路拆开讲清楚包括仪表盘怎么配、自动补算怎么触发、哪些坑我踩过。2. 整体设计思路与方案选型拆解2.1 为什么选 Opik 而不是自己写埋点自己写埋点统计成本听起来简单实际做起来会遇到几个硬骨头。第一是调用链路嵌套问题一个 Agent 可能调用多次 LLM每次 LLM 又可能触发工具调用工具调用再触发新的 LLM 请求这种树状结构用简单的日志很难还原。第二是 usage 字段缺失问题流式接口、部分开源模型的返回体里根本没有 token 统计你得自己估算。第三是成本单价维护问题不同模型、不同版本、不同供应商的价格表一直在变硬编码在代码里迟早会失控。Opik 的设计恰好覆盖了这三个痛点。它用 trace 和 span 的层级结构来还原调用树每个 span 可以独立记录 token 数和成本它内置了模型价格表也支持自定义单价它还提供了自动补算机制当某个 span 没有上报 usage 时可以根据输入输出文本长度做估算。这套组合拳下来比自己从零写一套要省太多事。注意Opik 的价格表更新有延迟如果你用的是新发布的模型或者私有部署模型一定要手动配置单价否则仪表盘上的成本会严重偏低。2.2 成本追踪的核心数据模型理解 Opik 的成本追踪关键要搞清楚它的数据模型。最上层是Trace代表一次完整的用户请求或者一个任务单元。Trace 下面挂Span每个 Span 代表一个具体的操作比如一次 LLM 调用、一次向量检索、一次工具执行。Span 可以嵌套形成树状结构。成本相关的字段主要挂在 Span 上包括model模型标识用来匹配价格表usage.input_tokens输入 token 数usage.output_tokens输出 token 数usage.total_tokens总 token 数cost根据单价和 token 数计算出的成本metadata可以放自定义字段比如缓存命中标记、重试次数这个模型的好处是你可以在任意层级做聚合。比如你想知道某个 Agent 的总成本就把根 Trace 下所有 Span 的 cost 加起来你想知道哪类操作最烧钱就按 Span 的 name 或者 type 分组统计。2.3 仪表盘与自动补算的分工仪表盘负责展示自动补算负责兜底。这两者要配合使用不能偏废。仪表盘的核心指标我建议至少包含这几项总成本趋势、按模型分组的成本占比、按 Span 类型分组的成本分布、平均单次调用成本、token 消耗速率。这些指标能帮你快速定位成本异常比如某个模型突然涨价、某个 Agent 陷入循环导致 token 暴涨。自动补算的触发时机有三个一是 Span 结束时没有 usage 字段二是 usage 字段存在但数值明显异常比如 input_tokens 为 0 但实际有输入三是模型不在价格表中导致 cost 无法计算。补算逻辑要尽量保守宁可估算偏高也不要偏低因为成本低估会让人放松警惕。3. 核心细节解析与实操要点3.1 接入 Opik SDK 的关键配置接入 Opik 的第一步是初始化客户端。Python SDK 的初始化很简单但有几个配置项必须注意import opik opik.configure( api_keyyour-api-key, workspaceyour-workspace, project_namellm-cost-tracking, use_localFalse )project_name建议按环境区分比如llm-cost-tracking-prod和llm-cost-tracking-dev这样仪表盘上不会混在一起。use_local如果设为 True数据会写到本地 SQLite适合开发调试但生产环境一定要用远程模式否则多实例部署时数据会分散。接下来是装饰器的使用。Opik 提供了opik.track装饰器可以自动捕获函数的输入输出和耗时。对于 LLM 调用我建议手动创建 Span因为需要显式传入 model 和 usage 信息from opik import track, Opik track def call_llm(prompt, modelgpt-4o-mini): client Opik() span client.span( namellm_call, input{prompt: prompt}, metadata{model: model} ) # 实际调用逻辑 response llm_client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}] ) span.end( output{response: response.choices[0].message.content}, usage{ input_tokens: response.usage.prompt_tokens, output_tokens: response.usage.completion_tokens } ) return response这里的关键点是span.end()时必须传入 usage否则 Opik 无法自动计算成本。如果你用的是流式接口usage 通常在最后一个 chunk 里需要手动累积。3.2 模型价格表的维护策略Opik 内置的价格表覆盖了主流商业模型但更新频率跟不上模型发布速度。我的做法是维护一份自定义价格表用 YAML 或者 JSON 存启动时加载到 Opik 的配置里custom_pricing { gpt-4o-mini: {input: 0.00015, output: 0.0006}, deepseek-chat: {input: 0.00014, output: 0.00028}, qwen-plus: {input: 0.0004, output: 0.0012} } opik.configure( api_keyyour-api-key, custom_model_pricingcustom_pricing )价格单位要统一我习惯用“每 1K token 的美元价格”因为大部分供应商的报价都是这个单位。如果你的模型是按“每 1M token”报价记得除以 1000。提示价格表变更后需要重启应用才能生效建议把价格表放在配置中心支持热更新避免频繁重启。3.3 自动补算的触发条件与估算逻辑自动补算的核心是估算 token 数。Opik 默认用 tiktoken 做估算但 tiktoken 只支持 OpenAI 系列的 tokenizer对于国产模型或者开源模型估算会有偏差。我的做法是分模型配置 tokenizerdef estimate_tokens(text, model): if model.startswith(gpt): import tiktoken enc tiktoken.encoding_for_model(model) return len(enc.encode(text)) elif model.startswith(qwen): # 用 Qwen 的 tokenizer return len(qwen_tokenizer.encode(text)) else: # 兜底按字符数除以 4 估算 return len(text) // 4补算的触发条件我设了三个阈值usage 缺失、input_tokens 为 0、total_tokens 与 inputoutput 之和不一致。只要命中任意一个就触发补算并在 metadata 里标记auto_calculated: true方便后续审计。补算逻辑要记录原始值和估算值不要直接覆盖。这样如果发现估算偏差大可以回溯调整。3.4 流式返回场景下的成本追踪流式返回是成本追踪最容易出问题的地方。很多 LLM 框架在流式模式下不会返回完整的 usage只在最后一个 chunk 里带一个简略的统计甚至完全不返回。我的处理方式是在流式开始时创建一个 Span状态设为in_progress累积所有 chunk 的文本内容流式结束后如果最后一个 chunk 有 usage直接用如果没有用累积的文本做估算调用span.end()时传入 usage 和stream: true标记span client.span(namellm_stream, metadata{model: model, stream: True}) full_content usage None for chunk in stream: delta chunk.choices[0].delta.content or full_content delta if hasattr(chunk, usage) and chunk.usage: usage chunk.usage if usage is None: usage { input_tokens: estimate_tokens(prompt, model), output_tokens: estimate_tokens(full_content, model) } span.end(output{content: full_content}, usageusage)这个逻辑看起来简单但实际写的时候要注意异常处理。如果流式中途断开span.end()可能不会被调用Span 会一直挂在in_progress状态。我加了一个定时任务扫描超过 5 分钟还是in_progress的 Span强制结束并标记为timeout。4. 实操过程与核心环节实现4.1 环境准备与依赖安装先把基础环境搭起来。Python 版本建议 3.9 以上Opik SDK 对 3.8 的支持已经不太好了。pip install opik pip install tiktoken pip install pyyaml如果你用的是 LangChain 或者 LlamaIndexOpik 有对应的集成包pip install opik-langchain pip install opik-llamaindex安装完成后验证一下版本python -c import opik; print(opik.__version__)我实测下来0.1.x 和 0.2.x 的 API 有差异建议锁定一个稳定版本不要盲目升级。4.2 配置仪表盘的关键指标Opik 的仪表盘是 Web 界面默认提供了一些图表但成本相关的指标需要手动配置。我建议在项目设置里添加这几个自定义面板面板名称指标类型数据来源刷新频率总成本趋势时间序列sum(cost) by day每小时模型成本占比饼图sum(cost) by model每小时Span 类型成本分布柱状图sum(cost) by span.name每小时平均单次调用成本单值avg(cost) per trace实时Token 消耗速率时间序列sum(total_tokens) by hour每小时自动补算比例单值count(auto_calculated) / count(*)每天配置路径在 Opik 的 Dashboard 设置里选择对应的 project然后添加 widget。每个 widget 需要写查询语句Opik 用的是类 SQL 的语法上手不难。注意仪表盘的数据有延迟通常是 1 到 5 分钟。如果你需要实时告警不要依赖仪表盘要用 Opik 的 webhook 或者 API 轮询。4.3 自动补算任务的实现自动补算我做成一个独立的定时任务每 10 分钟跑一次。逻辑是查询最近 1 小时内所有auto_calculated为 false 且 usage 缺失的 Span逐个补算。import opik from datetime import datetime, timedelta def auto_calculate_costs(): client opik.Opik() since datetime.utcnow() - timedelta(hours1) spans client.search_spans( project_namellm-cost-tracking-prod, filter{ created_at: {$gte: since.isoformat()}, usage.input_tokens: {$exists: False} } ) for span in spans: model span.metadata.get(model, unknown) input_text span.input.get(prompt, ) output_text span.output.get(response, ) input_tokens estimate_tokens(input_text, model) output_tokens estimate_tokens(output_text, model) span.update( usage{ input_tokens: input_tokens, output_tokens: output_tokens, total_tokens: input_tokens output_tokens }, metadata{ **span.metadata, auto_calculated: True, calculated_at: datetime.utcnow().isoformat() } )这个任务要注意幂等性避免重复补算。我在 metadata 里加了auto_calculated标记查询时过滤掉已经补算过的。4.4 成本告警的配置光看仪表盘不够成本异常需要主动告警。Opik 支持 webhook我配置了一个简单的告警规则当单日成本超过阈值或者单次 Trace 成本超过阈值时触发 webhook 推送到内部告警系统。alert_rules [ { name: daily_cost_exceeded, condition: sum(cost) by day 50, webhook: https://internal-alert.example.com/opik }, { name: single_trace_cost_exceeded, condition: max(cost) per trace 1, webhook: https://internal-alert.example.com/opik } ]阈值要根据你的业务规模调整。我一开始设了 50 美元一天结果发现正常业务量下经常触发后来调到 200 美元才合理。建议先观察一周的正常波动范围再定阈值。4.5 与现有监控体系的对接Opik 的数据可以通过 API 导出我把它对接到了内部的 Grafana 和 Prometheus。Opik 提供了/api/v1/metrics接口返回 JSON 格式的指标数据Prometheus 的 exporter 可以定时拉取。import requests from prometheus_client import Gauge llm_cost_gauge Gauge(llm_cost_total, Total LLM cost, [model, project]) def collect_metrics(): resp requests.get( https://opik.example.com/api/v1/metrics, headers{Authorization: fBearer {api_key}}, params{project: llm-cost-tracking-prod, window: 1h} ) data resp.json() for item in data[metrics]: llm_cost_gauge.labels( modelitem[model], projectitem[project] ).set(item[cost])这样成本数据就能和现有的业务监控放在一起看方便做关联分析。比如某天成本暴涨同时业务 QPS 也暴涨那可能是流量问题如果 QPS 没变但成本涨了那可能是模型调用逻辑出了问题。5. 常见问题与排查技巧实录5.1 成本数据与实际账单对不上这是最常见的问题原因通常有三个一是价格表没更新二是 token 估算偏差大三是缓存命中没有正确标记。排查步骤我整理成了一张表现象可能原因排查方法解决方案成本偏低 10% 以内价格表略旧对比供应商最新报价更新自定义价格表成本偏低 30% 以上大量 Span 未上报 usage查 auto_calculated 比例检查流式接口的 usage 捕获逻辑成本偏高重试次数被重复计算查 metadata 中的 retry 标记在 Span 上标记 retry聚合时去重成本波动大缓存命中率变化查 cache_hit 标记在 metadata 中记录缓存状态我踩过最坑的一次是流式接口的 usage 捕获。OpenAI 的流式接口默认不返回 usage需要加stream_options{include_usage: True}参数。这个参数不加所有流式调用的 usage 都是空的全靠估算偏差能到 20% 以上。5.2 Span 一直处于 in_progress 状态这个问题通常是因为异常没有被捕获span.end()没有执行。我的解决方案是加一个全局的异常处理器import atexit active_spans [] def cleanup_spans(): for span in active_spans: if span.status in_progress: span.end( metadata{forced_end: True, reason: process_exit} ) atexit.register(cleanup_spans)另外定时任务扫描超时 Span 也很必要。我设的是 5 分钟超过就强制结束并标记timeout。这些 Span 的成本会被单独统计方便排查是哪类操作容易超时。5.3 自动补算的估算偏差过大估算偏差主要来自 tokenizer 不匹配。国产模型的 tokenizer 和 tiktoken 差异很大用 tiktoken 估算 Qwen 的 token 数偏差能到 30%。我的做法是给每个模型配置对应的 tokenizer实在没有的用字符数除以 3.5 做兜底。还有一个技巧是定期校准。每周抽一批有真实 usage 的 Span用真实值除以估算值得到一个校准系数然后把这个系数应用到后续的估算中。我实测下来校准后偏差能控制在 5% 以内。5.4 仪表盘数据延迟导致误判Opik 的仪表盘数据有延迟如果你在压测或者调试时频繁刷新可能会看到旧数据。我的建议是调试时直接用 SDK 的查询接口不要依赖仪表盘。spans client.search_spans( project_namellm-cost-tracking-dev, filter{created_at: {$gte: (datetime.utcnow() - timedelta(minutes5)).isoformat()}} ) total_cost sum(s.cost for s in spans if s.cost) print(f最近 5 分钟成本: {total_cost})这个查询是实时的没有延迟。仪表盘适合看趋势实时查询适合调试。5.5 多实例部署时的数据一致性如果你的应用是多实例部署的每个实例都往 Opik 写数据要注意 project_name 的一致性。我见过有人开发环境用了llm-cost-tracking生产环境用了llm-cost-tracking-prod结果仪表盘上只看到一半数据排查了半天才发现是 project 名不一致。另外如果多个实例同时触发自动补算任务可能会重复补算同一个 Span。我的做法是用分布式锁或者把补算任务单独部署成一个服务只跑一个实例。6. 一些实操心得与后续扩展方向这套方案我跑了大概三个月整体稳定。最大的体会是成本追踪不是配好就完事它是一个持续维护的过程。价格表要跟着模型更新tokenizer 要跟着模型升级告警阈值要跟着业务量调整。我现在的做法是每月做一次成本审计对比 Opik 的数据和供应商账单偏差超过 5% 就排查原因。后续我打算扩展的方向有两个一是把成本数据和业务指标做关联分析比如算出每个用户、每个功能模块的成本这样能更精准地做资源分配二是做成本预测基于历史数据预测未来一周的成本提前发现异常趋势。如果你也在做 LLM 应用的成本追踪建议先从最简单的场景开始把单次调用的成本统计做准再逐步扩展到 Agent 和 RAG 场景。不要一上来就追求全链路覆盖那样很容易在细节上翻车。先把仪表盘跑起来看到数据了再优化补算逻辑这个顺序比较稳妥。