大模型应用全链路可观测性与成本治理实战：从黑盒调用到可归因系统

一、问题的真正起点：为什么传统监控在 LLM 应用上会失效

把一个基于大模型的功能推上生产环境之后，绝大多数团队都会经历同一个心理落差：本地跑通时觉得一切尽在掌握，上线两周后却发现自己对系统一无所知。用户投诉"回答不准"，你打开监控面板，看到的是清一色的 200 状态码和 1.8 秒的平均延迟；财务发来账单，你只知道这个月 API 花费涨了四成，却说不清是哪条业务线、哪个功能、哪个提示词模板贡献了这部分增量。这种"指标全绿、业务告急"的割裂，正是 LLM 应用在工程化阶段遇到的第一个结构性难题。

传统 APM 的假设是：请求成功等于业务成功，延迟和错误率能够刻画服务质量。这个假设在 CRUD 系统里基本成立，因为输入输出都是确定性的结构化数据。但在 LLM 应用里，一次 HTTP 200 的调用完全可能返回一段事实错误、格式错乱、甚至带有明显幻觉的文本。换句话说，技术层面的"成功"与语义层面的"正确"之间没有必然联系。你不去记录输入输出，就永远无法判断这次调用的业务价值是正还是负。

更麻烦的是 LLM 应用的调用结构天然是树状的。一次用户提问可能先触发一次意图分类的小模型调用，再走一次向量检索，再把检索结果拼进提示词交给生成模型，生成结果不满足格式要求又要重试一次，最后可能还要调用一个工具函数做二次校验。这棵调用树里任何一个节点出问题，最终表现都是"答案不好"，但根因可能在上游的检索、在中游的解析、也可能在下游的工具超时。没有结构化的链路追踪，排障就退化成了一场靠运气的猜谜游戏。

二、可观测性的三层数据模型

要把这件事做扎实，需要先建立一个清晰的数据分层。我倾向于把它拆成三层，每层解决不同的归因问题。

第一层是语义化的调用记录（Span）。这里的核心不是"打了多少日志"，而是"记录了什么语义"。一个合格的 LLM Span 至少应该包含：模型标识与版本、完整的输入消息列表、输出的原始文本、输入与输出的 token 数、是否命中缓存、结束原因（正常结束、长度截断还是工具调用）、以及首 token 延迟与总耗时。少任何一项，归因时都会卡住。比如没有记录结束原因，你就无法区分"模型答不出来"和"答案被 max_tokens 截断了"这两种完全不同的故障。

第二层是请求级的关联（Trace）。Span 是点，Trace 是线。通过一个全局 trace_id 把同一次用户请求产生的所有 Span 串起来，并且额外挂上业务维度：用户 ID、会话 ID、租户 ID、功能入口、部署版本号。这一层的价值在于横向切片——当你想知道"新版本上线后，哪个功能的 token 消耗涨得最凶"，只有具备业务维度的 Trace 数据才能回答。

第三层是聚合与洞察。这一层不是简单地把数据画成图，而是要做归因计算。成本需要能按功能模块、按租户、按提示词模板拆分；质量指标需要能按模型版本、按检索策略、按时间段做对比。真正成熟的团队会把这一层做成自动化的异常检测：当某个提示词模板的平均输出 token 突然上涨 40%，或者某类问题的拒答率从 3% 跳到 15%，系统应该主动告警，而不是等人去看报表。

三、最小可用的埋点实现

很多团队卡在"要不要引入商业平台"这一步上。我的建议是先自己写一个最小实现，把数据模型跑通，等确认真实需求再考虑替换。下面这段代码保留了最关键的归因字段，可以直接嵌入现有的调用封装层。

import time
import uuid
import json
from contextlib import contextmanager
from dataclasses import dataclass, field, asdict

@dataclass
class Span:
    trace_id: str
        span_id: str
            parent_id: str | None
                name: str
                    kind: str  # llm / retrieval / tool / chain
                        start_ts: float
                            end_ts: float = 0.0
                                model: str = ""
                                    input_tokens: int = 0
                                        output_tokens: int = 0
                                            cached_tokens: int = 0
                                                first_token_ms: float = 0.0
                                                    finish_reason: str = ""
                                                        status: str = "ok"
                                                            error: str | None = None
                                                                meta: dict = field(default_factory=dict)
class Tracer:
    """极简链路追踪器，生产环境把 sink 换成消息队列即可"""
    def __init__(self, sink, price_table):
            self.sink = sink
                    self.price_table = price_table  # {"model-name": (in_price, out_price)} 单位: 元/千token
                            self._stack = []
    def new_trace(self):
            return uuid.uuid4().hex
    @contextmanager
        def span(self, trace_id, name, kind, model="", meta=None):
                parent = self._stack[-1].span_id if self._stack else None
                        s = Span(
                                    trace_id=trace_id,
                                                span_id=uuid.uuid4().hex,
                                                            parent_id=parent,
                                                                        name=name,
                                                                                    kind=kind,
                                                                                                start_ts=time.time(),
                                                                                                            model=model,
                                                                                                                        meta=meta or {},
                                                                                                                                )
                                                                                                                                        self._stack.append(s)
                                                                                                                                                try:
                                                                                                                                                            yield s
                                                                                                                                                                    except Exception as exc:
                                                                                                                                                                                s.status = "error"
                                                                                                                                                                                            s.error = f"{type(exc).__name__}: {exc}"
                                                                                                                                                                                                        raise
                                                                                                                                                                                                                finally:
                                                                                                                                                                                                                            s.end_ts = time.time()
                                                                                                                                                                                                                                        self._stack.pop()
                                                                                                                                                                                                                                                    self.sink(self._enrich(s))
    def _enrich(self, s: Span):
            row = asdict(s)
                    row["duration_ms"] = round((s.end_ts - s.start_ts) * 1000, 2)
                            if s.kind == "llm" and s.model in self.price_table:
                                        pin, pout = self.price_table[s.model]
                                                    row["cost"] = round(
                                                                    (s.input_tokens - s.cached_tokens) * pin / 1000
                                                                                    + s.cached_tokens * pin * 0.25 / 1000
                                                                                    +                 + s.output_tokens * pout / 1000,
                                                                                    +                 6,
                                                                                    +             )
                                                                                    +         else:
                                                                                    +             row["cost"] = 0.0
                                                                                    +         return row
                                                                                    + ```
这段代码有三个值得说明的设计取舍。第一，缓存 token 单独计价，因为主流模型厂商对命中前缀缓存的输入 token 通常只收原价的 25% 左右，如果不区分，成本核算会系统性高估。第二，父 Span 通过栈式管理自动推导，不需要在业务代码里手动传 parent_id，这对嵌套调用特别友好。第三，异常路径也保证 Span 落库，否则最需要排查的那部分请求恰好会消失在监控里。

## 四、采样策略：不是所有请求都值得全量保存

埋点做完，下一个现实问题是存储成本。一个日均五万次调用的中等规模应用，如果每次都完整保存提示词和输出文本，单月产生的原始数据轻松达到几十 GB。很多团队在这个节点上要么咬牙全存，要么干脆关掉，两者都不对。

更合理的做法是分层采样。**错误请求百分之百记录**，这是最有价值的数据，任何采样都可能漏掉关键样本。**慢请求百分之百记录**，以 P95 或 P99 为阈值，这部分样本是性能优化的主要输入。**正常请求按 1% 到 5% 随机采样**，用于质量趋势分析和效果评估。**特定会话全量记录**，比如正在被投诉的用户、正在做 A/B 实验的流量组，这些场景下样本的完整性比存储成本重要得多。

这里有个容易被忽略的细节：采样必须按 trace 而不是按 span 做。如果按 span 随机采样，一条请求可能只留下检索节点、丢掉生成节点，链路就断了，数据反而比没有更让人迷惑。

## 五、成本归因的四个维度

成本治理的前提是能拆。我一般会要求系统至少支持四个维度的下钻。

| 维度 | 用途 | 常见陷阱 |
| --- | --- | --- |
| 按功能模块 | 判断哪个业务场景最烧钱 | 只统计成功请求，忽略重试导致的重复消耗 |
| 按租户 | 制定配额与计费策略 | 共享缓存池时无法准确分摊 |
| 按模型版本 | 评估升级收益 | 忽略影子流量与灰度流量 |
| 按提示词模板 | 定位低效提示词 | 模板改动频繁导致历史数据不可比 |

实践中，最容易被低估的是"重试成本"。很多系统在失败重试时不会把上一次的消耗计入同一次请求，导致单次请求成本统计偏低，而账单上却真实产生了费用。正确做法是让重试作为子 Span 挂在同一个 trace 下，汇总时自然累加。另一个高频问题是影子流量——为了做模型对比而把请求复制一份发给新模型，这部分消耗如果没有单独打标，会污染主链路的成本数据。

## 六、从可观测到可运营

数据齐全之后，真正的价值才开始显现。我在实际项目里总结出三个立刻见效的用法。

**第一是提示词版本对比**。把提示词模板本身当作可版本化的资产，每次改动都打上版本号，然后对比不同版本的 token 均值、格式合规率、拒答率。很多"感觉变好了"的优化，在数据面前会露出真相——有时新版本只是把答案写得更长，看起来更详细，实际上信息密度下降了。

**第二是缓存命中率运营**。前缀缓存的收益取决于提示词结构：把稳定不变的系统提示词、少量示例、工具定义放在最前面，把易变的用户输入放在最后，命中率能提升一个数量级。这件事只能通过数据验证，靠猜是猜不出来的。

**第三是质量与成本的联合看板**。单看成本会诱导你把模型换成最便宜的那个，单看质量会诱导你无脑用最强模型。把两者放在同一张图上，标出"性价比拐点"，才能做出真正合理的选型决策。我在一个客服场景里做过这样的分析，结论是中等规模模型加一次重排，总成本比大模型直出低 55%，而人工抽检的满意度只差 2 个百分点。

## 七、写在最后

LLM 应用的可观测性不是"加个日志"，而是一套以归因为核心的数据工程。它的难点不在于技术复杂度，而在于是否愿意在项目早期就投入这部分看起来"不产出功能"的基础设施。我的经验是，越早做越省事：等到线上问题频发、账单失控时再补，往往要面对历史数据不可比、埋点侵入业务代码、改造成本高昂等一系列麻烦。

如果只能记住一句话，那就是：**可观测性的目标不是"看见"，而是"敢改"**。当你能在十分钟内说清一次劣化是哪个环节引入的、影响多少用户、多花了多少钱，你才真正拥有了持续迭代这个系统的能力。