AI 应用的可观测性不是画监控大屏：把「一次回答怎么来的」变成可回放、可回归的测试证据

把接口调用链 Trace 搬到 AI 应用：给 retrieval / prompt / llm / tool / postprocess 每段都留可断言的中间产物把接口调用链 Trace 搬到 AI 应用：给 retrieval / prompt / llm / tool / postprocess 每段都留可断言的中间产物

一个 AI 应用给出了错答案，你打开日志想查为什么，却只看到孤零零一行「模型返回：xxx」。检索到底召回了哪些段落？Prompt 最后拼成了什么样、有没有超长？中间调了哪几个工具、入参出参是什么？每步花了多少时间、烧了多少 token——全都没有记录。

结果就是三连问都答不上来：错了没法归因，改完没法验证，回归更无从谈起。这篇以质量审计报告体讲一件具体的事：用 OpenTelemetry 把一条 AI 请求的全链路 Trace 补全，让「模型为什么这么答」变成一份可回放、可断言的测试证据。

一、审计结论摘要

审计对象是这套 AI 应用的可观测性与可测试性，三条发现按风险从高到低。

第一，日志只记最终输出，链路中间态全丢。出错时无法回答「错在检索、Prompt、模型还是工具」，归因全靠猜，属阻断级。

第二，中间产物没有落成可断言的结构化数据。召回文档 id、Prompt 长度、工具入参这些关键量从来没被记录，也就没法写「召回不得为空」「Prompt 不得超长」这类断言。

第三，改完无法验证、回归无从谈起。因为没有任何一条用例能重放一次请求、比对每一段的中间产物，改动是否引入退化只能靠人肉抽查。

建议动作三条：用 OpenTelemetry 给 retrieval / prompt_build / llm_call / tool_call / postprocess 五段各埋一个 span 并记录关键属性、把 span 导出成结构化 jsonl、用 pytest 对这份 Trace 写断言让它可回归。第四节给可运行实现，第五节给 Trace 分段清单。

二、先厘清目的：可观测性是为了做测试证据，不是为了大屏

先给结论：AI 应用的可观测性，价值不在监控大屏好不好看，在于它能不能把「一次回答是怎么来的」变成一份可回放、可断言的测试证据。这两件事看着像，其实差得远。

大屏关心的是聚合指标——平均延迟、错误率、QPS，它回答「系统整体健不健康」。测试证据关心的是单条请求的每一段中间产物——这一次召回了哪几篇、Prompt 拼成了什么、工具返回了什么，它回答「这一次为什么答成这样」。前者是运维视角，后者是质量视角，本篇只谈后者。

把它锚回你熟悉的东西：这就是接口测试里的调用链 Trace。过去 Trace 记录的是一次请求穿过哪些微服务、每一跳的耗时和返回；现在同一条链路上多了 AI 特有的几段——检索、Prompt 组装、模型调用、工具调用、后处理。原理完全一样：每一段都留下可断言的中间产物，出错时顺着链路一段段往回看，就能定位到底哪一段坏了。区别只是，被测对象从确定性的服务返回，变成了非确定性的模型输出，所以更要靠中间态来归因。

三、一条 AI 请求该拆成哪几段

结论：把一次 AI 请求拆成五个可埋点的阶段，每段记它能回答「为什么这么答」的关键属性。

retrieval（检索）：记录召回的文档 id 列表、召回条数、相似度分数——它回答「模型看到的信息对不对」。prompt_build（Prompt 组装）：记录最终 Prompt 的字符/token 长度、模板版本、是否超长被截断——它回答「喂给模型的输入长什么样」。llm_call（模型调用）：记录模型名与版本、输入输出 token 数、耗时——它回答「模型这一跳发生了什么」。tool_call（工具调用）：记录工具名、入参、出参、是否报错——它回答「模型有没有拿到正确的外部结果」。postprocess（后处理）：记录是否命中过滤、是否改写、最终输出长度——它回答「模型原始输出到用户之间被动了什么」。

这五段串起来，就是一条完整的、可回放的证据链。下面是可运行实现。

四、可运行实现：OpenTelemetry 埋 span + 导出 jsonl

实现分两段。第一段用 OpenTelemetry 给五段各埋一个 span、记录关键属性，并用自定义 exporter 把 span 落成 jsonl；第二段用 pytest 对这份 Trace 写断言。

"""
trace_ai.py —— 用 OpenTelemetry 给一条 AI 请求埋全链路 span，导出成 jsonl
运行：python trace_ai.py   （生成 trace.jsonl）
真实项目里把 fake_* 换成你的检索/模型/工具调用即可。
"""
import json
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SpanExporter, SimpleSpanProcessor


class JsonlExporter(SpanExporter):
    """把每个 span 连同它的关键属性落成一行 json，作为可回放的测试证据。"""
    def __init__(self, path="trace.jsonl"):
        self.path = path

    def export(self, spans):
        with open(self.path, "a", encoding="utf-8") as f:
            for s in spans:
                f.write(json.dumps({
                    "name": s.name,
                    "attrs": dict(s.attributes),
                    "duration_ms": round((s.end_time - s.start_time) / 1e6, 2),
                }, ensure_ascii=False) + "\n")

    def shutdown(self):
        pass


provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(JsonlExporter()))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("ai.request")


def handle_query(question: str):
    """一条 AI 请求：五个阶段各埋一个 span，每段记能归因的关键属性。"""
    with tracer.start_as_current_span("ai_request") as root:
        root.set_attribute("question.len", len(question))

        with tracer.start_as_current_span("retrieval") as sp:
            docs = fake_retrieve(question)           # 你的检索
            sp.set_attribute("retrieval.doc_ids", [d["id"] for d in docs])
            sp.set_attribute("retrieval.count", len(docs))

        with tracer.start_as_current_span("prompt_build") as sp:
            prompt = fake_build_prompt(question, docs)
            sp.set_attribute("prompt.char_len", len(prompt))
            sp.set_attribute("prompt.template_version", "v3")
            sp.set_attribute("prompt.truncated", len(prompt) > 8000)

        with tracer.start_as_current_span("llm_call") as sp:
            raw = fake_llm(prompt)
            sp.set_attribute("llm.model", "demo-model")
            sp.set_attribute("llm.model_version", "2026-08")
            sp.set_attribute("llm.out_tokens", len(raw) // 4)

        with tracer.start_as_current_span("tool_call") as sp:
            result = fake_tool(raw)
            sp.set_attribute("tool.name", "calc_refund")
            sp.set_attribute("tool.ok", result.get("ok", False))

        with tracer.start_as_current_span("postprocess") as sp:
            final = fake_postprocess(result)
            sp.set_attribute("post.filtered", False)
            sp.set_attribute("post.final_len", len(final))
        return final


# —— 下面是桩，真实项目替换为实际实现 ——
def fake_retrieve(q): return [{"id": "doc_1"}, {"id": "doc_2"}]
def fake_build_prompt(q, docs): return f"根据{[d['id'] for d in docs]}回答:{q}"
def fake_llm(prompt): return "退款金额为 99 元"
def fake_tool(raw): return {"ok": True, "value": 99}
def fake_postprocess(result): return f"退款金额：{result['value']} 元"


if __name__ == "__main__":
    handle_query("我的订单能退多少钱")
    provider.force_flush()
    print("已导出 trace.jsonl")

为什么这么写：把五个阶段做成嵌套 span、都挂在 ai_request 这个根 span 下，是因为回放时要能一眼看出「这些段属于同一次请求」，父子关系就是这条证据链的骨架。每个 span 只记「能用来归因和断言」的属性——retrieval 记 doc_ids 和 count、prompt 记长度和是否截断、llm 记模型版本和 token、tool 记入参出参和是否成功——是因为 Trace 不是记得越全越好，记多了噪声大、也难断言，关键是把「为什么这么答」的几个决定量钉住。自定义 JsonlExporter 把 span 落成结构化 jsonl 而不是发到监控后端，是因为本篇要的是「可被 pytest 读取、可断言、可版本化比对」的测试证据，不是一张实时大屏。踩过的坑有两个：一是 SimpleSpanProcessor 是同步导出，测试里用它才能保证 handle_query 返回时 jsonl 已落盘；生产环境要换 BatchSpanProcessor，但那样测试里就得显式 force_flush，否则读到空文件、断言假绿。二是 prompt.truncated 这类布尔属性一定要在埋点时就记下来，事后想从 Prompt 长度反推「当时到底截没截断」是推不出来的——归因证据必须在现场留。

五、把 Trace 变成会失败的断言

结论：Trace 光记下来还不够，要能用 pytest 对它写断言，它才从「日志」升级成「测试证据」。

"""
test_trace.py —— 对导出的 Trace 写断言，让「为什么这么答」可回归
运行：先 python trace_ai.py 生成 trace.jsonl，再 pytest -q test_trace.py
"""
import json
import pytest


def load_trace(path="trace.jsonl"):
    with open(path, encoding="utf-8") as f:
        return {json.loads(l)["name"]: json.loads(l) for l in f if l.strip()}


@pytest.fixture(scope="module")
def spans():
    return load_trace()


def test_retrieval_not_empty(spans):
    """召回不得为空——空召回意味着模型是在凭空编。"""
    assert spans["retrieval"]["attrs"]["retrieval.count"] > 0


def test_prompt_not_truncated(spans):
    """Prompt 不得被截断——截断会悄悄丢掉关键上下文。"""
    assert spans["prompt_build"]["attrs"]["prompt.truncated"] is False


def test_tool_call_succeeded(spans):
    """工具调用必须成功——工具失败却继续答，就是幻觉高发点。"""
    assert spans["tool_call"]["attrs"]["tool.ok"] is True


def test_model_version_pinned(spans):
    """模型版本必须锁定在期望值——版本漂了，答案漂了要能归因到这一跳。"""
    assert spans["llm_call"]["attrs"]["llm.model_version"] == "2026-08"

为什么这么写：每条断言都对应一个「会让答案变错、但从最终输出看不出来」的中间态——召回为空、Prompt 被截断、工具失败却硬答、模型版本悄悄换了。这正是全链路 Trace 的价值：最终答案对的时候这些断言全绿，一旦某段坏了，你不用去猜，断言直接告诉你是哪一段。把它们写成 pytest 用例，就能进回归——每次改检索、改 Prompt 模板、换模型，都重放一批请求、比对每段中间产物，退化当场暴露。踩过的坑：断言要钉在「结构化的中间属性」上，别去断言最终那段自然语言文本等不等于某句话；模型输出是非确定性的，对文本做等值断言只会假红一片，而 retrieval.count、prompt.truncated、tool.ok 这些中间量是稳定的，才是可回归的抓手。

六、审计清单：一条 Trace 作为测试证据，该查什么

再补一张两种做法的对照，说明为什么「只记最终输出」不够：

这五项没有一项需要额外预算，只是把「记一行模型返回」升级成「给链路每段埋点、导出、断言」。反过来说，一份只记最终输出、连召回条数和 Prompt 长度都没留的 AI 应用日志，它出问题时给不出任何归因依据，也就撑不起一次像样的质量审计。

AI 应用的可观测性，真正的验收标准不是大屏有多漂亮，而是「随便挑一条错误回答，你能不能顺着 Trace 一段段指出它到底坏在哪」。

记不住「怎么答出来的」，就没资格说「测过了」——Trace 不是日志的装饰，是回答的可回放证据。

你们的 AI 应用出错时，日志能回答「检索召回了什么、Prompt 拼成了什么样」吗？评论区聊聊。