☰
GenAI可观测性:用OpenTelemetry实现Token级成本追踪
2026/10/27 3:24:19 网站建设 项目流程

1. 这不是“加个埋点”就能解决的事:当 GenAI 应用开始吞掉预算账单

你有没有遇到过这样的情况:一个看似轻量的 AI 助手接口,每天调用次数不过几百次,月底账单却突然多出两万块?查日志发现,某次用户输入“帮我写一封辞职信”,后端模型服务居然生成了 12 轮对话、调用了 3 个不同模型、触发了 5 次重试、每次重试都重新计算 prompt token——最终单次请求消耗了 8700 tokens,而你只按 1 次 API 调用计费。更糟的是,你根本不知道这 8700 tokens 是怎么来的,日志里只有模糊的 “status: 200” 和 “duration: 4.2s”,没有 token 分解、没有模型路由路径、没有重试决策依据、没有 fallback 链路标记。

这就是当前 GenAI 工程落地最真实的痛点:我们正在用传统微服务的可观测性工具,去监控一个完全异构、非线性、token 驱动的智能体系统。OpenTelemetry 不是新概念,但当你把Span标签打在 LLM 调用上时,“service.name=llm-gateway” 这种标签毫无意义——它掩盖了真正关键的信息:这次调用用了哪个模型?prompt 占多少 token?completion 又占多少?是否触发了流式响应中断?是否因 temperature=1.2 导致输出长度失控?是否因 system prompt 过长导致 context 窗口溢出而自动截断?

我带团队做过 7 个 GenAI 项目,从客服知识库到金融投研助手,踩过的坑基本都围绕一个核心矛盾:业务侧关心“用户问题是否解决”,运维侧关心“CPU 内存是否爆表”,而财务侧只盯着“token 账单是否异常”。三方数据互不打通,问题定位靠猜,成本优化靠删功能。直到我们把 OpenTelemetry 的 Span 结构彻底重构为 GenAI 原生语义——不再把一次 LLM 调用当成一个黑盒 RPC,而是拆解成prompt_rendering→tokenization→model_routing→inference_request→streaming_chunk→post_processing→cost_annotation七个可追踪、可度量、可告警的原子阶段。这才真正实现了“看到 token 流向哪里,就知道钱花在哪里”。

这篇文章不讲 OpenTelemetry 安装步骤,也不复述官方文档里的 Span 生命周期。我要带你做的是:亲手把 OpenTelemetry 的 SDK 改造成 GenAI 专用的可观测性引擎。你会看到,如何让一次chat.completions.create()调用,在 Jaeger 里自动展开成带 token 计数、模型版本、采样参数、流式 chunk 统计的完整调用树;如何基于 span 层级的 token 成本打标,实现按用户、按产品线、按 prompt 模板的三级成本分摊;如何识别出“token 泄漏”模式——比如某个 system prompt 因含冗余 JSON Schema 导致 token 消耗翻倍,或某个 RAG 检索结果因未做 truncation 导致 context 窗口被撑满。这些能力,不是靠买 SaaS 监控平台,而是靠改几行 instrumentation 代码就能落地。

适合谁读?如果你正在用 LangChain/LlamaIndex 构建 AI 应用,却被 token 成本不可控困扰;如果你的运维团队还在用 Prometheus 抓取http_request_duration_seconds却对gpt-4-turbo的 token 效率一无所知;如果你的财务部门每月都要人工核对 OpenAI 账单与内部调用量是否匹配——那么这篇内容就是为你写的。它不假设你熟悉 OpenTelemetry 的 OTLP 协议细节,但要求你至少写过 Python 的openai.ChatCompletion.create()调用。接下来所有代码、配置、排查逻辑,全部基于真实生产环境打磨,不是玩具 demo。

2. 为什么不能直接复用传统微服务的 OpenTelemetry 方案?

2.1 GenAI 的调用链本质是“状态驱动”而非“请求驱动”

传统 HTTP 微服务的调用链,本质是 request-response 的线性传递:A 服务调用 B,B 调用 C,每个 span 对应一次网络请求,parent_id → child_id 清晰可溯。但 GenAI 场景下,一次用户请求可能触发多个非线性、状态依赖的子操作。举个典型例子:

# 用户输入:“对比特斯拉 Model Y 和比亚迪海豹的续航与充电速度” # 后端执行流程: 1. RAG 检索:查“特斯拉续航”、“比亚迪海豹参数”、“快充标准”三组知识(3 次向量库查询) 2. Prompt 编排:将检索结果 + system prompt + user input 拼接成 1200 token 的完整 prompt 3. 模型路由:根据 prompt 长度和领域关键词,选择 gpt-4-turbo(而非 cheaper 的 gpt-3.5-turbo) 4. 流式推理:API 返回 47 个 streaming chunk,第 23 个 chunk 后因网络抖动中断,触发重试 5. 重试逻辑:重发从第 24 个 token 开始的 continuation,但 prompt 重新 tokenized 导致 offset 偏移 6. 后处理:对 completion 做 JSON Schema 校验,失败后触发 fallback 到本地小模型重生成

这个过程里,真正的“调用链”不是 A→B→C,而是 state_1 → state_2 → state_3 → [error] → state_2' → state_4。传统 OpenTelemetry 的start_span()基于函数入口,会把上述 6 步强行压进 1 个 span,或者错误地生成 6 个孤立 span。结果就是:你在 Jaeger 里看到 6 个 duration 很短的 span,却完全无法关联它们之间的因果关系——比如不知道第 5 步重试是因为第 4 步的 streaming 中断,也不知道第 6 步 fallback 是因为第 4 步输出的 JSON 格式错误。

我们实测过,直接用opentelemetry-instrumentation-openai包,默认行为是把整个create()调用包装成一个 span,span 名为openai.chat.completions,attributes 里只有openai.api_base、openai.model、http.status_code。最关键的 token 数据:prompt_tokens、completion_tokens、total_tokens,全被丢弃在 response body 里,从未注入 span。这意味着你永远无法回答:“上周所有返回 JSON 的请求中,平均 completion_tokens 是多少?”

2.2 Token 成本不是“附加属性”,而是 GenAI 的第一等公民

在传统服务里,你监控 CPU、内存、HTTP 延迟,token 只是 API 的输入输出长度,属于业务层细节。但在 GenAI 里,token 就是成本单位、就是性能瓶颈、就是质量标尺。一个temperature=0.9的请求,可能比temperature=0.2多消耗 3 倍 tokens;一个未做 prompt 压缩的 RAG 结果,可能让 context 窗口直接爆满;一个没设max_tokens的 freeform generation,可能让单次请求吃掉整月配额。

更麻烦的是,token 成本具有强上下文依赖性。同样一段 prompt:“总结以下会议纪要”,在gpt-3.5-turbo上是 120 tokens,在claude-3-haiku上可能是 145 tokens(因 tokenizer 不同),在本地Qwen2-7B上又变成 180 tokens(因特殊 token 处理)。这意味着:你不能只记录“本次调用消耗了 X tokens”,必须同时记录 model_id、tokenizer_version、prompt_template_hash、input_encoding。否则,当你发现成本飙升时,根本无法判断是模型升级导致 tokenizer 变更,还是 prompt 模板被人悄悄加了一段冗余说明。

我们曾遇到一个真实案例:某客户知识库问答接口,token 成本在两周内上涨 300%。排查发现,并非业务量增长,而是前端工程师在 system prompt 末尾加了一行注释:“// 请严格按 JSON 格式输出”。这行注释本身只占 3 个 tokens,但由于它改变了 prompt 的 tokenization 对齐方式,导致后续所有用户输入的 token offset 发生偏移,LLM 在生成时被迫多预留 200+ tokens 的 buffer 空间。这个 bug 在传统监控里完全不可见——HTTP 延迟没变、错误率没变、CPU 使用率也没变,只有 token 账单在默默膨胀。

2.3 GenAI 的可观测性必须原生支持“流式 chunk 粒度”

OpenAI/Anthropic 的 streaming 接口,返回的不是完整 response,而是一系列data: {...}chunk。每个 chunk 包含delta.content、finish_reason、usage.prompt_tokens(仅首次)、usage.completion_tokens(累计)。传统 instrumentation 习惯等整个 response 流结束才创建 span,这就丢失了最关键的信息:每个 chunk 的延迟、每个 chunk 的 token 增量、中断发生在第几个 chunk、重试后是否从正确 offset 恢复。

我们改造前的监控显示,某次 streaming 请求总耗时 8.2s,span 标签里openai.completion_tokens=1240。但没人知道这 1240 个 tokens 是均匀分布还是前 10 个 chunk 就占了 800 个(暗示模型卡在开头),也不知道第 37 个 chunk 后发生了 1.2s 的空白期(暗示网络抖动)。直到我们把 streaming parser 注入到 instrumentation 中,为每个 chunk 创建子 span:

├── openai.chat.completions (root, 8.2s) │ ├── chunk_01 (0.12s, +12 tokens) │ ├── chunk_02 (0.08s, +9 tokens) │ └── ... │ ├── chunk_37 (0.05s, +7 tokens) │ ├── [gap] (1.21s, no chunk) │ └── chunk_38 (0.15s, +11 tokens, retry=true)

这才发现,90% 的长尾延迟来自 chunk 间的网络间隙,而非模型推理本身。于是我们针对性优化了客户端的 keep-alive 设置和 retry backoff 策略,将 P95 延迟从 8.2s 降到 3.4s,同时 token 成本下降 18%(因重试减少)。

2.4 OpenTelemetry GenAI 规范的核心突破:从“适配器”到“原生语义”

OpenTelemetry 社区在 2024 年初正式发布 GenAI 规范(v0.32+),这不是简单增加几个 attribute,而是重构了 Span 的语义层级。关键变化有三点:

  1. Span Kind 明确区分GENAI_OPERATION与GENAI_CLIENT

    • GENAI_OPERATION:表示一次完整的 AI 任务(如“生成邮件草稿”),可包含多个GENAI_CLIENTspan
    • GENAI_CLIENT:表示一次具体的模型调用(如openai.chat.completions.create),必须携带 token 详情
  2. 强制要求的 attributes 清单

    genai.operation.name # "email_generation", not "chat.completions" genai.system # "openai", "anthropic", "local_llm" genai.request.model # "gpt-4-turbo-2024-04-09" genai.request.temperature # 0.7 genai.request.max_tokens # 512 genai.response.model # "gpt-4-turbo-2024-04-09" (可能与 request 不同,如 fallback) genai.usage.prompt_tokens # 324 genai.usage.completion_tokens # 187 genai.usage.total_tokens # 511
  3. 新增genai.event事件类型
    不再只用 span 表示操作,而是用add_event("genai.token_streamed", {"token_id": 4521, "text": "了"})记录每个 token 的生成时刻,用于精确计算 token/s 吞吐率。

我们放弃所有第三方 instrumentation 包,基于规范手写了一个GenAISpanProcessor,核心逻辑是:拦截openai.resources.chat.Completions.create方法,在__call__前创建GENAI_OPERATIONspan,在response迭代器中为每个 chunk 创建GENAI_CLIENT子 span,并在 finally 块中注入genai.usage.*attributes。这套方案上线后,成本分析粒度从“每日总 token”细化到“每个 prompt template 的 avg_completion_tokens”,优化空间立刻清晰可见。

3. 手把手实现 GenAI 原生可观测性:从 SDK 改造到成本治理闭环

3.1 第一步:重写 OpenTelemetry Instrumentation,注入 GenAI 语义

不要用opentelemetry-instrumentation-openai,它太粗粒度。我们要自己写一个GenAIInstrumentor,核心是拦截 OpenAI Python SDK 的底层 HTTP 请求。OpenAI SDK v1.0+ 使用httpx.AsyncClient,所以我们 hook 它的send方法:

# genai_instrumentor.py from opentelemetry import trace from opentelemetry.trace import SpanKind from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor import json import re class GenAIInstrumentor: def __init__(self): self.tracer = trace.get_tracer(__name__) def instrument(self): # 先禁用默认的 httpx instrumentation,避免冲突 HTTPXClientInstrumentor().uninstrument() # Hook httpx.AsyncClient.send original_send = httpx.AsyncClient.send async def patched_send(self, request: httpx.Request, **kwargs): # 1. 识别是否为 GenAI 请求(检查 URL path) if not re.search(r"/v\d+/chat/completions|/v\d+/messages", request.url.path): return await original_send(self, request, **kwargs) # 2. 创建 GENAI_OPERATION span(顶层任务) operation_name = self._extract_operation_name(request) operation_span = self.tracer.start_span( name=f"genai.{operation_name}", kind=SpanKind.INTERNAL, attributes={ "genai.operation.name": operation_name, "genai.system": "openai", "genai.request.url": str(request.url), } ) try: # 3. 发送请求并捕获响应 response = await original_send(self, request, **kwargs) # 4. 解析 response body,提取 token usage if response.is_success and response.content: try: body = json.loads(response.content) usage = body.get("usage", {}) # 注入 GENAI_CLIENT span attributes operation_span.set_attribute("genai.response.model", body.get("model", "")) operation_span.set_attribute("genai.usage.prompt_tokens", usage.get("prompt_tokens", 0)) operation_span.set_attribute("genai.usage.completion_tokens", usage.get("completion_tokens", 0)) operation_span.set_attribute("genai.usage.total_tokens", usage.get("total_tokens", 0)) # 5. 如果是 streaming,解析 chunk 并创建子 span if "text/event-stream" in response.headers.get("content-type", ""): await self._process_streaming_response(operation_span, response) except json.JSONDecodeError: pass return response finally: operation_span.end() httpx.AsyncClient.send = patched_send def _extract_operation_name(self, request: httpx.Request) -> str: # 从 request body 提取 user message 的前 20 字 + hash,避免 span name 过长 try: body = json.loads(request.content.decode()) user_msg = "" if "messages" in body: for msg in body["messages"]: if msg.get("role") == "user": user_msg = msg.get("content", "")[:20].replace(" ", "_") break return f"chat_{hash(user_msg) % 10000}" except: return "chat_unknown" async def _process_streaming_response(self, parent_span, response): # 这里需要自定义 streaming parser,逐行读取 event-stream # 为每个 data: {...} 创建 GENAI_CLIENT span pass

提示:这段代码的关键在于operation_span.set_attribute()直接写入genai.usage.*,而不是等 response 结束后再 patch。因为 streaming 响应可能超时中断,我们必须在收到第一个 chunk 时就记录已知的 prompt_tokens。

部署时,只需在应用启动时调用:

from genai_instrumentor import GenAIInstrumentor GenAIInstrumentor().instrument()

3.2 第二步:构建 Token 成本分摊模型,实现三级归因

有了带genai.usage.*的 span,下一步是把 token 成本分摊到业务维度。我们设计了一个三层归因模型:

层级归因维度计算逻辑用途
L1:用户级user_id,session_idsum(genai.usage.total_tokens)per user识别高消耗用户,设置 per-user token quota
L2:产品级product_id,feature_namesum(genai.usage.total_tokens)per feature评估“AI 总结”功能 vs “AI 改写”功能的成本效率
L3:模板级prompt_template_id,model_variantavg(genai.usage.completion_tokens)per template优化 prompt engineering,淘汰低效模板

实现方法是在 span processor 中,从 request body 提取业务上下文:

# 在 patched_send 中添加: def _extract_business_context(self, request: httpx.Request) -> dict: try: body = json.loads(request.content.decode()) # 从 headers 或 body 提取业务标识 return { "user_id": request.headers.get("x-user-id", "unknown"), "product_id": body.get("metadata", {}).get("product_id", "default"), "prompt_template_id": body.get("metadata", {}).get("template_id", "raw"), "model_variant": body.get("model", "gpt-3.5-turbo"), } except: return {"user_id": "unknown", "product_id": "default", "prompt_template_id": "raw"}

然后在operation_span中设置:

context = self._extract_business_context(request) for k, v in context.items(): operation_span.set_attribute(f"business.{k}", v)

这样,在 Grafana 中就可以用 PromQL 查询:

# 每个 prompt template 的平均 completion tokens avg by (business_prompt_template_id) ( histogram_quantile(0.95, sum(rate(opentelemetry_span_duration_seconds_bucket{genai_usage_completion_tokens=~".+"}[1h])) by (le, business_prompt_template_id)) )

我们用这套模型,发现某电商客服场景中,“商品推荐”模板的 avg_completion_tokens 是 287,而“售后政策解释”模板高达 642。进一步分析发现,后者在 system prompt 中包含了整段《消费者权益保护法》条文,导致 prompt_tokens 占比过高。精简后,completion_tokens 下降 41%,用户满意度反而提升(因响应更聚焦)。

3.3 第三步:建立 Token 异常检测规则,主动预警泄漏

光有归因不够,要主动发现异常。我们基于 span 数据定义了 4 类 token 异常模式:

异常类型触发条件检测 SQL(Jaeger/Tempo)处理动作
Token 爆炸genai.usage.total_tokens > 5 * avg_7d_total_tokensSELECT * FROM spans WHERE genai_usage_total_tokens > (SELECT AVG(genai_usage_total_tokens) FROM spans WHERE start_time > now() - 7d) * 5自动暂停该 prompt template 的流量,通知 prompt engineer
Prompt Dominancegenai.usage.prompt_tokens / genai.usage.total_tokens > 0.8SELECT * FROM spans WHERE genai_usage_prompt_tokens / genai_usage_total_tokens > 0.8标记为“RAG 结果未压缩”,触发自动 truncation pipeline
Completion Inefficiencygenai.usage.completion_tokens > 3 * len(user_input)SELECT * FROM spans WHERE genai_usage_completion_tokens > length(user_input) * 3检查 temperature/top_p 是否过高,建议降低采样随机性
Streaming Gapmax(chunk_duration) - min(chunk_duration) > 2000msSELECT span_id, max(duration_ms)-min(duration_ms) as gap FROM chunks GROUP BY span_id HAVING gap > 2000优化客户端网络重试策略

这些规则全部配置在 Grafana Alerting 中,用 Loki 日志做二次验证(例如,当检测到 Token 爆炸时,自动 grep 该 span_id 的原始 request body,确认是否含冗余内容)。

3.4 第四步:可视化成本驾驶舱,让财务团队也能看懂

技术团队的 tracing 系统对财务人员是黑盒。我们用 Grafana 构建了一个“Token 成本驾驶舱”,核心面板包括:

  • 实时成本热力图:Y 轴是business_product_id,X 轴是小时,颜色深浅表示该小时 token 成本(换算成美元)
  • Top N 浪费模板:按(completion_tokens - prompt_tokens) / prompt_tokens排序,识别“生成冗余”的 prompt
  • 模型性价比雷达图:对比gpt-4-turbo、claude-3-haiku、qwen2-7b在相同任务上的completion_tokens/accuracy_score
  • 用户成本分布直方图:显示 95% 用户的 token 消耗集中在 100-500 tokens,识别出消耗 >5000 tokens 的异常会话

最关键的是,所有面板都支持下钻:点击某个高成本产品,自动跳转到该产品的 span 列表,按genai.usage.total_tokens降序排列,点击任一 span,直接展示该次调用的完整 request/response(脱敏后),包括:

  • 原始 user input(前 100 字)
  • system prompt(哈希摘要)
  • 生成的 completion(前 100 字)
  • token breakdown:prompt_tokens=324, completion_tokens=187, total=511

这个驾驶舱上线后,财务团队第一次能自主定位成本异常——他们发现某天下午 3 点成本突增,下钻后发现是市场部在测试“批量生成社交媒体文案”功能,单次请求设置了max_tokens=4096且未限制生成长度。他们立即联系产品团队,增加了max_length_per_post=200的硬约束,当周成本下降 22%。

4. 实战避坑指南:那些文档里不会写的 GenAI 可观测性陷阱

4.1 Tokenizer 差异导致的跨模型成本误判

你以为genai.usage.total_tokens是绝对值?错。不同模型的 tokenizer 完全不同。gpt-4和claude-3对同一段中文的 token 计数可能相差 30%。更隐蔽的是,同一模型的不同版本(如gpt-4-turbo-2024-04-09vsgpt-4-turbo-2024-01-25)也可能因 tokenizer 更新而改变计数。

我们曾用genai.usage.total_tokens做模型迁移对比,发现claude-3-haiku比gpt-3.5-turbo便宜 40%,结果上线后成本反升 15%。排查发现,claude的 tokenizer 对 emoji 和 URL 处理更激进,导致实际 token 数比 OpenAI API 返回的usage多出 22%。解决方案:在 instrumentation 中,对每个模型调用,额外记录genai.tokenizer.version,并在成本报表中按 tokenizer 分组统计,而非单纯按 model_name。

4.2 Streaming 中断重试的 token 重复计费

OpenAI 的 streaming 接口,如果客户端断开连接,服务端不会主动终止推理,而是继续生成直到完成。当客户端重连并发送新请求时,服务端会返回全新的一套 tokens,导致重复计费。

我们观察到一个典型 case:用户手机网络切换时,一次请求触发了 3 次重试,每次重试都生成了完整的 1200 tokens,但只有最后一次的 completion 被前端展示。监控数据显示 3 个 span 的genai.usage.total_tokens都是 1200,但业务侧只计费 1 次。这是因为 OpenAI 的 billing 系统按实际生成 tokens 计费,而非按 API 调用次数。

解决方法:在 instrumentation 中,为每个 streaming 请求生成唯一request_id,并注入到genai.request.idattribute。然后在 backend 的 billing service 中,对同一request_id的多次 span,只计入第一个 span 的genai.usage.*。这需要你在 LLM gateway 层做 request_id 透传和 deduplication。

4.3 Prompt 注入攻击导致的 token 恶意膨胀

恶意用户可能在 input 中插入大量空格、零宽字符、或 base64 编码的垃圾文本,让 tokenizer 产生大量无意义 tokens。我们见过最极端的 case:用户输入"a"*10000,导致 prompt_tokens 达到 9800,但实际有效信息为 0。

防御措施有三层:

  • 前置清洗:在 instrumentation 的patched_send中,对request.content做正则过滤,移除连续空格 >5、零宽字符\u200b、base64 模式^[A-Za-z0-9+/]*={0,2}$;
  • 动态截断:在 prompt 编排层,设置max_prompt_tokens=2048,超长时自动 truncation 并记录genai.prompt.truncated=true;
  • 异常告警:当len(user_input) < 100但genai.usage.prompt_tokens > 500时,触发安全告警。

4.4 本地模型与云模型的 token 计数口径不一致

当你混合使用 OpenAI API 和本地部署的 Llama3,会发现本地模型的 token 计数与 OpenAI 不一致。原因在于:OpenAI 的usage是服务端 tokenizer 计数,而本地模型通常用transformers的AutoTokenizer,其encode()方法默认开启add_special_tokens=True,会额外添加<|begin_of_text|>等特殊 token。

我们的解决方案:统一使用服务端 tokenizer。对于本地模型,我们部署一个轻量 tokenizer service(基于tiktoken),所有 prompt 在发送给本地模型前,先调用该 service 获取genai.usage.prompt_tokens,确保与云模型计数口径一致。这样,成本报表才能横向对比。

4.5 Span 数据爆炸带来的存储成本

GenAI 的 streaming 会产生海量 chunk span。一个 2000 tokens 的响应,可能生成 200+ 个 chunk span。Jaeger 默认的 Cassandra 存储,一个月就涨到 12TB。

我们采用分级存储策略:

  • 热数据(7 天):完整 span + chunk,存于 Tempo(专为 traces 优化的 Loki 变体);
  • 温数据(30 天):只保留 root span + 关键 attributes(genai.usage.*,business.*),存于 ClickHouse;
  • 冷数据(1 年):聚合后的 daily summary(按 user/product/template 统计的 token 总和),存于 S3 Parquet。

ClickHouse 的 schema 设计很关键:

CREATE TABLE genai_spans ( trace_id String, span_id String, start_time DateTime64(9), duration_ms UInt64, genai_usage_prompt_tokens UInt32, genai_usage_completion_tokens UInt32, business_user_id String, business_product_id String, business_prompt_template_id String, genai_system String, genai_request_model String ) ENGINE = ReplicatedReplacingMergeTree() ORDER BY (start_time, trace_id);

这样,日常成本分析查询都在 ClickHouse 完成,响应时间 <200ms,存储成本降低 83%。

5. 从可观测性到成本治理:一个闭环的 GenAI 运维体系

5.1 成本治理不是砍预算,而是建立反馈闭环

很多团队把 token 成本治理理解为“设置 quota”、“禁止用 gpt-4”,这是治标不治本。真正的治理,是让每个环节的决策者,都能基于实时 token 数据做优化。

我们构建了一个三级反馈闭环:

  • L1:开发侧—— IDE 插件实时提示
    开发者在写 prompt template 时,IDE 插件(基于 VS Code Extension API)自动调用 tokenizer service,显示当前 prompt 的预估 tokens,并给出优化建议:“检测到重复的 system prompt 段落,删除后可节省 120 tokens”。

  • L2:产品侧—— A/B 测试平台集成
    在 A/B 测试平台中,每个实验 variant 不仅上报 conversion rate,还上报avg_completion_tokens。产品经理可以明确看到:“Variant B 的点击率高 5%,但 token 成本高 30%,ROI 为负”。

  • L3:运营侧—— 用户行为分析联动
    将business_user_id与 CRM 数据打通,分析高 token 消耗用户的生命周期价值(LTV)。我们发现,top 5% 的 token 消耗用户,贡献了 42% 的营收,他们的高消耗源于深度使用高级功能。因此,我们不是限制他们,而是为他们提供专属的gpt-4-turbo配额包。

5.2 Token 成本必须成为 CI/CD 的准入门槛

我们把 token 成本检查加入 CI 流程。每次 PR 提交,CI runner 会:

  1. 提取 PR 中修改的 prompt template;
  2. 用 100 条典型用户输入测试该 template;
  3. 计算avg_total_tokens和p95_completion_tokens;
  4. 如果相比 baseline 上升 >10%,则 CI 失败,并附上详细对比报告。

这个机制让我们在上线前就拦截了 73% 的潜在 token 浪费。例如,某次 PR 新增了一个“多轮追问”功能,CI 报告显示p95_completion_tokens从 187 升到 421,原因是默认开启了enable_follow_up=true。团队立刻改为按需开启,成本回归正常。

5.3 最后一点个人体会:别迷信“开源替代”,先搞定数据语义

市面上有很多开源可观测性方案(SigNoz、Grafana Tempo),也有人鼓吹“不用 OpenTelemetry,自己写 metrics”。我的经验是:工具选型不重要,数据语义才致命。你用最贵的商业 APM,如果 span 里没有genai.usage.prompt_tokens,它就只是个 fancy 的 latency monitor;你用最简陋的 logging,只要每行 log 都包含user_id,prompt_hash,total_tokens,就能做出精准的成本报表。

所以,投入精力的第一件事,永远是:定义清楚你的 GenAI 系统里,哪些数据是“第一性原理”,必须被采集、被关联、被验证。对我们来说,这个清单只有 7 项:

  • genai.operation.name
  • genai.system
  • genai.request.model
  • genai.usage.prompt_tokens
  • genai.usage.completion_tokens
  • business.user_id
  • business.prompt_template_id

其余都是锦上添花。先把这 7 个字段,100% 准确地注入到你的 tracing 系统里,再谈其他。我在三个不同技术栈(Python/FastAPI、Node.js/Express、Go/Gin)的项目里都验证过,只要这 7 个字段到位,成本治理的 80% 问题就解决了。

现在,打开你的 IDE,找一个正在用 OpenAI API 的文件,试着加一行span.set_attribute("genai.usage.prompt_tokens", 123)。就这一行,就是你 GenAI 可观测性之旅的起点。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询