基于OpenTelemetry实现LLM应用Token级可观测性与成本归因

1. 项目概述为什么我们需要Token级的可观测性如果你正在构建或维护一个基于大语言模型LLM的应用无论是内部的智能客服、代码助手还是对外的AI产品那么下面这个场景你一定不陌生这个月的API账单又爆了比上个月高了30%。你打开后台只能看到一堆调用次数和总费用但具体是哪个用户的哪次对话、哪个功能模块消耗了这么多Token不知道。某个用户的查询突然返回了乱码或者耗时极长你想复现问题却发现日志里只有简单的成功/失败记录至于LLM内部到底“思考”了什么生成了多少Token在哪个环节卡住了还是一头雾水。更头疼的是产品经理跑来问“我们给VIP用户提供的那个高级摘要功能单次调用成本到底是多少值不值得收费”你手头根本没有细粒度的数据来回答。这就是当前LLM应用开发中一个普遍存在的“黑盒”困境。我们就像在驾驶一架没有仪表盘的飞机只知道油料预算在快速消耗却不知道每个引擎功能/用户的实时工况。传统的应用监控Metrics、Logs在面对LLM这种以非结构化文本交互为核心、按Token计费的新范式时显得力不从心。Token级可观测性就是为了给这架飞机装上精密的仪表盘。它不仅仅是记录一次API调用而是要深入到每一次LLM交互的“原子”层面——追踪每一个输入Prompt和输出Completion的Token数量、内容、耗时并将这些数据与具体的业务实体如用户ID、会话ID、功能模块关联起来。从工程角度看实现Token级可观测性需要打通从数据采集、链路追踪到成本归因的完整闭环。这不仅仅是运维的范畴更直接关系到产品的成本控制、体验优化和商业化策略。今天我就结合自己在一线落地这套系统的经验从Trace采集开始一直聊到如何实现精准的Cost Attribution成本归因把其中的核心思路、技术选型、实操细节以及踩过的坑毫无保留地分享给你。2. 核心需求与价值解析不止于“看见”在动手搭建之前我们必须想清楚投入精力做Token级可观测性到底要解决哪些具体问题它的价值体现在哪里我将其归纳为以下四个核心层面这也是我们技术方案设计的出发点。2.1 精细化成本管控与商业洞察这是最直接、最刚性的需求。LLM API的计费模式通常是按Token消耗量阶梯计价动辄每月数万甚至数十万美元的账单如果无法拆解就是一笔糊涂账。成本分摊Showback/Chargeback在多租户SaaS产品或内部多个部门共用LLM服务的场景下你需要准确地将API成本分摊到具体的客户、部门或项目上。Token级追踪能告诉你客户A在本月因为使用了“长文档总结”功能消耗了500万Token对应成本为X美元。功能级ROI分析产品中的每一个AI功能如翻译、润色、摘要的成本效益如何通过归因你可以计算出单次摘要的平均Token消耗和成本从而评估该功能是否值得收费或者如何优化提示词Prompt以降低成本。异常消耗预警与审计突然出现单次对话消耗数万Token的“异常查询”可能是提示词注入Prompt Injection攻击也可能是代码Bug导致循环调用。实时监控Token消耗速率和分布能帮你快速定位并止损。2.2 性能瓶颈诊断与体验优化LLM应用的性能瓶颈非常特殊它可能不在你的代码而在模型本身、网络延迟或提示词设计。端到端链路分解一次用户请求可能包含意图识别小模型→ 知识库检索RAG→ 主LLM调用 → 后处理。Token级Trace可以清晰展示每个阶段的耗时、输入输出Token数。你会发现瓶颈可能出在检索了过多无关内容导致输入Token激增而非LLM生成本身。生成过程可视化对于流式输出Streaming场景你可以追踪到每个Token返回的延迟绘制出Token-by-Token的延迟曲线从而识别出网络波动或模型服务端的不稳定。长上下文Long Context优化当使用支持128K甚至更长上下文的模型时每次都将全部历史对话作为输入成本极高。通过Trace分析你可以统计不同会话长度的实际Token消耗与效果为设计更智能的上下文窗口压缩或总结策略提供数据依据。2.3 提示词工程与模型效果评估提示词Prompt的设计是LLM应用的核心但其效果评估往往依赖人工抽查。Token级观测性提供了量化的评估手段。A/B测试数据支撑对同一功能设计两套不同的提示词A和B通过对比两者在平均消耗Token数、任务完成率、耗时等指标上的差异可以科学地评估哪种提示词更高效、更经济。Token消耗模式分析分析发现某个提示词总是导致模型生成大量“思考过程”Chain-of-ThoughtToken但最终答案质量提升有限。这时你就可以优化提示词引导模型用更简洁的方式输出。评估幻觉Hallucination与冗余通过对比输入上下文和输出内容结合后续的用户反馈如“踩/赞”可以间接评估模型输出质量与Token消耗的关系。2.4 安全与合规审计在一些对安全、合规要求严格的领域如金融、医疗需要对AI的决策过程进行追溯。全链路审计追踪当AI给出了一个有问题的建议时你需要能完整复现当时的输入用户问题、检索到的知识、模型输出以及内部的推理步骤。Token级的Trace记录了所有这些信息满足审计要求。敏感信息监控可以配置规则对输入/输出中可能包含的特定敏感Token模式如身份证号、银行卡号模板进行监控和告警。3. 技术架构选型为什么是OpenTelemetry明确了需求接下来就是技术选型。我们的目标是构建一个标准化、低侵入、可扩展的观测体系。在众多方案中OpenTelemetryOTel成为了不二之选。它不是一个具体的工具而是一套云原生观测性的标准提供了统一的API、SDK和采集器Collector。下面我详细解释为什么OTel适合LLM场景以及我们如何扩展它。3.1 OpenTelemetry的核心优势厂商中立与标准化OTel定义了观测数据的通用标准Trace, Metric, Log。使用OTel采集数据后你可以自由地将数据导出到任何支持OTel的后端如Jaeger、Zipkin、Prometheus、Grafana Tempo或商业化的Datadog、New Relic等。这避免了被某个特定厂商锁定的风险。低侵入性与自动注入通过OTel提供的自动插桩Auto-instrumentation库对于使用流行框架如LangChain、LlamaIndex、FastAPI的应用你只需添加几行配置代码就能自动捕获HTTP请求、数据库调用等Span。对于LLM调用我们需要进行手动插桩但OTel统一的API让这个过程非常规范。强大的上下文传播Context Propagation这是实现链路追踪Trace的关键。OTel能够自动将Trace上下文Trace ID, Span ID通过HTTP头等方式在服务间传递。对于LLM应用这意味着你可以将一个用户请求从Web入口到最后的LLM API调用完整地串联成一条追踪链路。灵活的采集与处理管道OTel Collector可以作为一个独立的代理接收来自应用的数据并进行过滤、转换、丰富例如添加业务标签和批量导出。这让我们可以在数据上报前就完成很多重要的预处理工作。3.2 针对LLM场景的架构扩展标准的OTel主要面向传统的微服务追踪对于LLM特有的“Token”和“成本”概念需要进行扩展。我们的架构核心是创建自定义的Span和Attribute。整体数据流如下[你的LLM应用] --(携带OTel上下文)-- [LLM SDK如OpenAI] --(返回Token数等)-- [你的应用] | V [你的应用] --(创建LLM Span记录Token数、模型、成本等属性)-- [OTel SDK] -- [OTel Collector] -- [观测后端如JaegerGrafana]关键在于我们在调用LLM API的前后使用OTel SDK手动创建了一个Span。在这个Span中我们记录了LLM调用的详细信息作为属性Attributesllm.model: 模型名称如 “gpt-4-turbo-preview”llm.provider: 供应商如 “openai”, “anthropic”llm.prompt_tokens: 提示词Token数llm.completion_tokens: 补全Token数llm.total_tokens: 总Token数llm.estimated_cost: 估算成本根据Token数和模型单价计算llm.function_call: 触发的函数调用名称如果使用Function Callinguser.id: 用户ID从业务上下文获取session.id: 会话IDfeature.flag: 功能标识如 “document_summary_v2”这样每一个LLM调用在追踪系统中都成为一个富含信息的Span并且通过Trace ID与其他业务Span关联。注意模型单价可能变动且不同供应商、不同区域的定价可能不同。llm.estimated_cost属性最好在Collector端通过查找表动态计算而不是写死在应用代码中以提高灵活性。3.3 与其他方案的对比你可能会考虑其他方案比如各云厂商自家的监控如Azure AI Studio的监控、AWS Bedrock的CloudWatch日志。它们深度集成开箱即用但锁定性强数据难以导出进行自定义分析且无法统一追踪你自有业务代码的链路。自行打日志最灵活但也是最原始的方式。你需要自己设计日志格式、聚合系统、解析查询。在跨服务、异步调用场景下手动关联日志非常痛苦难以构建完整的调用链。商业LLM可观测性平台如Weights Biases、Arize、LangSmith等。它们功能强大专门为LLM设计提供了提示词管理、评估、追踪等一站式服务。但通常价格昂贵且可能与你现有的技术栈集成度不够。我们的选择基于OTel自建在成本可控、灵活性高、与现有技术栈融合度好之间取得了最佳平衡特别适合有一定工程能力、需要深度定制的团队。4. 工程落地从代码插桩到数据可视化理论说再多不如一行代码。接下来我将以Python应用使用FastAPI和OpenAI SDK为例分步拆解如何实现这套系统。我们假设你已经有一个基本的FastAPI应用。4.1 第一步环境准备与基础插桩首先安装必要的OTel库和导出器。我们使用控制台输出和Jaeger作为后端来演示。pip install opentelemetry-api opentelemetry-sdk opentelemetry-instrumentation-fastapi opentelemetry-instrumentation-requests opentelemetry-exporter-jaeger-thrift然后在你的应用启动文件如main.py中初始化OTelfrom opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter from opentelemetry.exporter.jaeger.thrift import JaegerExporter from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor from opentelemetry.instrumentation.requests import RequestsInstrumentor import fastapi # 1. 设置全局的TracerProvider trace.set_tracer_provider(TracerProvider()) # 2. 创建导出器这里用控制台和Jaeger举例 console_exporter ConsoleSpanExporter() jaeger_exporter JaegerExporter( agent_host_namelocalhost, # Jaeger agent地址 agent_port6831, ) # 3. 创建Span处理器并添加到TracerProvider trace.get_tracer_provider().add_span_processor( BatchSpanProcessor(console_exporter) ) trace.get_tracer_provider().add_span_processor( BatchSpanProcessor(jaeger_exporter) ) # 4. 自动插桩FastAPI和requests库用于HTTP调用包括OpenAI API RequestsInstrumentor().instrument() app fastapi.FastAPI() FastAPIInstrumentor.instrument_app(app) # 获取一个Tracer tracer trace.get_tracer(__name__)现在你的FastAPI应用的所有HTTP端点Endpoint和通过requests库发起的调用包括OpenAI SDK底层的调用都会被自动追踪。但这还远远不够我们还需要手动增强LLM调用的Span。4.2 第二步封装LLM客户端实现Token级追踪我们不希望在每个调用LLM的地方都写一遍重复的追踪代码。最佳实践是封装一个自己的LLM客户端或在调用前后使用装饰器/中间件。这里展示一个封装OpenAI客户端的示例import openai from opentelemetry import trace from typing import Dict, Any, Optional import asyncio # 如果是异步调用 class TracedOpenAIClient: def __init__(self, api_key: str, default_model: str gpt-3.5-turbo): self.client openai.OpenAI(api_keyapi_key) self.default_model default_model self.tracer trace.get_tracer(__name__) # 一个简单的模型单价查找表示例需定期更新 self.model_pricing { gpt-3.5-turbo-0125: {prompt: 0.0005, completion: 0.0015}, # $ per 1K tokens gpt-4-turbo-preview: {prompt: 0.01, completion: 0.03}, # ... 其他模型 } def _calculate_cost(self, model: str, prompt_tokens: int, completion_tokens: int) - float: 计算预估成本美元 if model not in self.model_pricing: return 0.0 pricing self.model_pricing[model] cost (prompt_tokens / 1000) * pricing[prompt] (completion_tokens / 1000) * pricing[completion] return round(cost, 6) def chat_completion(self, messages: list, model: Optional[str] None, **kwargs) - Dict[str, Any]: 封装聊天补全调用并创建追踪Span model model or self.default_model with self.tracer.start_as_current_span(llm.chat_completion) as span: # 在Span中记录调用参数注意避免记录完整消息以防隐私泄露 span.set_attribute(llm.provider, openai) span.set_attribute(llm.model, model) span.set_attribute(llm.messages_count, len(messages)) # 可以记录系统提示词的角色但过滤具体内容 system_prompts [msg for msg in messages if msg.get(role) system] span.set_attribute(llm.system_prompts_count, len(system_prompts)) # 执行实际的LLM调用 response self.client.chat.completions.create( modelmodel, messagesmessages, **kwargs ) # 从响应中提取Token使用情况 usage response.usage prompt_tokens usage.prompt_tokens completion_tokens usage.completion_tokens total_tokens usage.total_tokens # 将Token详情和成本记录为Span属性 span.set_attribute(llm.prompt_tokens, prompt_tokens) span.set_attribute(llm.completion_tokens, completion_tokens) span.set_attribute(llm.total_tokens, total_tokens) estimated_cost self._calculate_cost(model, prompt_tokens, completion_tokens) span.set_attribute(llm.estimated_cost_usd, estimated_cost) # 如果有工具调用也记录下来 if response.choices[0].message.tool_calls: tool_names [tc.function.name for tc in response.choices[0].message.tool_calls] span.set_attribute(llm.tool_calls, ,.join(tool_names)) # 将业务上下文如用户ID从当前Span的父级继承或手动设置 # 假设我们通过OTel的Baggage或自定义中间件设置了user.id # span.set_attribute(user.id, get_current_user_id()) return { content: response.choices[0].message.content, raw_response: response, usage: usage, estimated_cost: estimated_cost } # 使用示例 traced_llm TracedOpenAIClient(api_keyyour-api-key) result traced_llm.chat_completion( messages[{role: user, content: 请用一句话解释量子计算}], modelgpt-3.5-turbo ) print(result[content]) print(f本次调用消耗Token: {result[usage].total_tokens}, 预估成本: ${result[estimated_cost]})这个封装类做了几件关键事创建了一个名为llm.chat_completion的Span。在调用前记录了模型、消息数等上下文。在调用后从OpenAI响应中提取usage字段并将Token数量和计算出的成本设置为Span的属性。安全地处理了可能包含敏感信息的消息内容只记录元数据不记录具体内容。4.3 第三步关联业务上下文Cost Attribution的关键孤立的LLM Span价值有限我们必须把它和具体的业务场景关联起来。这需要通过OTel的上下文传播来实现。通常我们在请求入口如FastAPI中间件就创建根Span并注入业务标识。from opentelemetry import baggage, trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.context import Context import uuid app.middleware(http) async def add_correlation_context(request: fastapi.Request, call_next): # 从请求头或认证信息中提取业务标识 user_id request.headers.get(X-User-ID, anonymous) session_id request.headers.get(X-Session-ID, str(uuid.uuid4())) feature_flag request.query_params.get(feature, default) # 创建一个携带业务Baggage的上下文 ctx baggage.set_baggage(user.id, user_id) ctx baggage.set_baggage(session.id, session_id, contextctx) ctx baggage.set_baggage(feature.flag, feature_flag, contextctx) # 将业务上下文设置为当前上下文 token trace.set_tracer_provider(trace.get_tracer_provider()).force_flush() # 注意实际中需要更精细的上下文管理这里为简化示例 # 通常使用 trace.use_span() 或类似机制 response await call_next(request) return response然后在你的LLM封装客户端或任何需要记录业务属性的地方从Baggage中取出这些值并设置到Span上with self.tracer.start_as_current_span(llm.chat_completion) as span: # ... 之前的代码 ... # 从Baggage中获取业务上下文 current_baggage baggage.get_baggage() if current_baggage: user_id current_baggage.get(user.id) if user_id: span.set_attribute(user.id, user_id) # 同样设置 session.id, feature.flag 等 span.set_attribute(session.id, current_baggage.get(session.id)) span.set_attribute(feature.flag, current_baggage.get(feature.flag))这样最终在Jaeger或Grafana Tempo中查看Trace时你就能清晰地看到用户Auser.id123在会话XYZ中使用了“高级摘要”功能feature.flagpremium_summary触发了一次LLM调用消耗了850个Token成本约为$0.00425。4.4 第四步使用OTel Collector进行数据增强与导出将数据直接从应用发送到后端如Jaeger可以工作但使用OTel Collector作为中间层能带来巨大好处。Collector可以运行在应用同一Kubernetes集群的Sidecar模式或作为一个中心化服务。核心优势解耦与缓冲应用无需关心后端地址Collector负责重试和缓冲提高可靠性。数据预处理在Collector中你可以使用Attributes Processor或Transform Processor来丰富Span数据。例如根据llm.model和llm.total_tokens动态计算成本并添加llm.cost_usd属性。这样成本计算逻辑可以统一在Collector中管理无需修改所有应用代码。路由与过滤可以将高成本的Trace路由到不同的存储或分析管道。统一导出Collector可以将数据同时导出到多个目的地比如Trace到TempoMetrics到PrometheusLogs到Loki。一个简单的Collector配置 (otel-collector-config.yaml) 示例如下receivers: otlp: protocols: grpc: endpoint: 0.0.0.0:4317 http: endpoint: 0.0.0.0:4318 processors: batch: # 批量处理优化性能 attributes/llm-cost: # 属性处理器计算LLM成本 actions: - key: llm.estimated_cost_usd action: upsert # 这里可以使用更复杂的查找表从环境变量或配置中心读取 value: | if llm.model gpt-4-turbo-preview { (llm.prompt_tokens / 1000) * 0.01 (llm.completion_tokens / 1000) * 0.03 } else if llm.model gpt-3.5-turbo-0125 { (llm.prompt_tokens / 1000) * 0.0005 (llm.completion_tokens / 1000) * 0.0015 } else { 0 } exporters: jaeger: endpoint: jaeger-all-in-one:14250 tls: insecure: true prometheusremotewrite: # 也可以将Token数作为指标导出 endpoint: http://prometheus:9090/api/v1/write headers: x-scope-orgid: tenant1 service: pipelines: traces: receivers: [otlp] processors: [batch, attributes/llm-cost] exporters: [jaeger] metrics: receivers: [otlp] processors: [batch] exporters: [prometheusremotewrite]4.5 第五步可视化、告警与成本分析数据收集齐全后最后一步是让它产生价值。这里通常结合使用Grafana。链路追踪可视化Grafana Tempo/Jaeger在Grafana中配置Tempo或Jaeger数据源。你可以通过{span_namellm.chat_completion}这样的查询来过滤出所有LLM调用。点击任意一个Trace可以查看完整的调用链以及每个Span包括LLM Span的详细属性如Token数、成本、用户ID。技巧利用Tempo的TraceQL可以执行更强大的查询例如{ resource.service.namemy-llm-app span.attributes.llm.estimated_cost_usd 0.1 }找出所有单次成本超过10美分的LLM调用。指标监控与告警Grafana Prometheus通过OTel Collector我们可以将LLM Span的属性如llm.total_tokens作为指标Metrics导出到Prometheus。但更常见的做法是在应用层或Collector层进行聚合。例如你可以写一个简单的后台作业定期从Trace存储中聚合数据生成如下指标llm_token_consumption_total{model, user_id, feature}按模型、用户、功能聚合的总Token消耗。llm_api_cost_total{model, user_id}按模型、用户聚合的总成本。llm_request_duration_seconds_bucket{model, status}LLM API调用耗时的直方图。在Grafana中为这些指标创建仪表盘并设置告警规则。例如“当llm_token_consumption_total在5分钟内环比增长超过200%时触发告警”这可能意味着有异常流量或攻击。成本归因分析Grafana 自定义查询这是最终目标。你可以在Grafana中创建专门的“成本分析”看板。按用户/租户排名一个SQL查询如果Trace数据存储在支持SQL的后端如Tempo with Parquet或Prometheus查询列出过去24小时消耗成本最高的前10个用户。按功能模块分解一个饼图展示不同feature.flag的成本占比。成本趋势按天/周绘制总成本曲线并与业务量用户活跃度、API调用次数叠加分析其相关性。效率指标计算“每美元生成的Token数”或“每次对话平均成本”作为提示词和模型选型优化的核心KPI。5. 实战中的挑战与解决方案在实际落地过程中我们遇到了不少挑战这里分享几个典型的“坑”和我们的应对策略。5.1 挑战一异步与并发调用下的上下文丢失LLM应用大量使用异步编程以提高吞吐。在异步任务中OTel的上下文Context可能会丢失导致新创建的Span无法关联到正确的Trace上。解决方案显式传递上下文在创建异步任务时手动捕获当前上下文并传递。import asyncio from opentelemetry import trace from opentelemetry.context import Context async def call_llm_async(messages): # 在异步函数开始时恢复上下文 token trace.set_tracer_provider(trace.get_tracer_provider()).force_flush() with tracer.start_as_current_span(async_llm_call): # ... 调用LLM ... pass # 在主函数中 current_ctx trace.get_current_span().get_span_context() asyncio.create_task(call_llm_async_with_context(messages, current_ctx))使用支持异步的框架确保你使用的OTel SDK和插桩库支持你的异步框架如asyncio,anyio。对于复杂的场景考虑使用opentelemetry-instrumentation-aiohttp-client等专门的库。利用框架集成像FastAPI、Django等框架的OTel自动插桩通常已经处理了异步上下文传播问题优先使用它们。5.2 挑战二流式响应Streaming的追踪OpenAI等API支持流式返回Token。如果简单地在开始和结束时记录一个Span你会丢失中间每个Chunk的到达时间信息无法分析生成过程的延迟分布。解决方案创建子Span为整个流式调用创建一个父Span然后为接收到的每个数据块Chunk创建一个短暂的子Span记录其到达时间戳和包含的Token数如果API返回。with tracer.start_as_current_span(llm.chat_completion.stream) as parent_span: stream client.chat.completions.create(modelmodel, messagesmessages, streamTrue) token_count 0 start_time time.time() for chunk in stream: with tracer.start_as_current_span(llm.chunk.receive, parentparent_span) as chunk_span: delta chunk.choices[0].delta if delta.content: token_count estimate_token_count(delta.content) # 需要估算 chunk_span.set_attribute(llm.chunk.arrival_offset_ms, (time.time() - start_time)*1000) chunk_span.set_attribute(llm.estimated_tokens_so_far, token_count) # 流结束后在父Span中记录总Token数如果API在最后返回usage聚合指标即使不创建子Span也可以在客户端聚合流式响应的延迟指标例如“首Token时间”Time to First Token, TTFT和“Token间延迟”Inter-token Latency并将这些作为指标Metrics上报。5.3 挑战三隐私、安全与数据采样Trace数据可能包含敏感信息用户输入、模型输出、内部提示词模板。全量采集和存储所有数据不仅成本高还有隐私和安全风险。解决方案敏感信息过滤Redaction在OTel SDK端或Collector端使用Processor过滤掉Span属性中的敏感字段。例如将llm.prompt属性的值替换为[REDACTED]或仅保留其哈希值。# 在Collector配置中 processors: attributes/redact: actions: - key: llm.prompt action: delete # 或 action: upsert, value: [REDACTED]尾部采样Tail Sampling这是OTel Collector的一个强大功能。你可以配置采样策略例如基于错误采样所有出错的Tracestatus.code ERROR100%保留。基于成本采样所有高成本的Traceattributes.llm.estimated_cost_usd 1100%保留。基于概率采样对其他Trace进行低概率采样如1%。 这样你既能捕获所有异常和高价值事件又大幅降低了存储成本。processors: tail_sampling: policies: [ { name: high-cost-policy, type: numeric_attribute, numeric_attribute: {key: llm.estimated_cost_usd, min_value: 1.0} }, { name: error-policy, type: status_code, status_code: {status_codes: [ERROR]} }, { name: probabilistic-policy, type: probabilistic, probabilistic: {sampling_percentage: 1} } ]数据保留策略在存储层面如Jaeger、Tempo设置合理的TTL生存时间自动删除旧数据。5.4 挑战四多模型供应商与复杂工作流的统一追踪一个应用可能同时调用OpenAI、Anthropic、Azure OpenAI甚至本地部署的模型。工作流可能涉及链式调用LangChain、代理Agent决策等。解决方案抽象统一的客户端接口像之前封装的TracedOpenAIClient一样为每个供应商都创建一个带追踪的封装客户端。它们都实现同一个基础接口确保追踪数据格式的一致性如属性键名llm.model,llm.provider。利用LangChain等框架的回调CallbacksLangChain内置了回调系统可以钩入LLM调用的各个生命周期。你可以编写一个OpenTelemetryCallbackHandler在on_llm_start,on_llm_end等事件中创建和结束OTel Span并记录Token使用情况。这是集成现有LangChain应用最优雅的方式。工作流级别的Span对于复杂的Agent或Chain创建一个更高层级的Span如agent.run或workflow.summarize_document将内部所有的LLM调用、工具调用作为其子Span。这样你可以从宏观上分析整个工作流的成本和性能。6. 总结与展望构建数据驱动的LLM应用运营实现Token级的可观测性不是一个一蹴而就的项目而是一个需要持续迭代的工程实践。它从最初帮助你看清“钱花在哪了”逐渐演变为支撑你进行提示词优化、模型选型、架构设计、产品定价和异常风控的核心数据基础设施。从我实际落地的经验来看最大的回报不是节省了多少API费用虽然这很直接而是整个团队对LLM应用运行状态建立了数据驱动的共同认知。工程师能快速定位性能瓶颈产品经理能基于成本数据做出功能决策运营人员能识别高价值用户和异常行为。最后再分享两个小技巧从小处着手不必一开始就追求全链路、全功能的完美方案。可以从最核心、成本最高的一个LLM调用点开始插桩先实现基本的Token计数和成本估算看到价值后再逐步推广到其他模块。将成本作为核心SLO像关注应用延迟和错误率一样将“每次调用平均Token成本”或“每用户日均成本”纳入你的服务等级目标SLO中。设立基线监控异常这能从根本上推动团队关注效率。LLM应用的浪潮才刚刚开始可观测性是我们驾驭这股浪潮的罗盘。希望这篇长文能为你点亮前行的路让你在构建下一代AI应用时心中有“数”脚下有路。