第 17 章 在线评测平台：langsmith、langfuse、phoenix 架构对比

在线评测平台的职责链是 trace 收集 → 采样 → 评分 → 可视化 → 告警。只有 trace 而没有挂在上面的评分，看到的就只是流量，不是质量。

17.1 为什么需要专门的在线评测平台

第 2 章 §2.3 已经讨论过在线评测的核心难题——生产流量没有 ground truth，必须用 LLM-as-Judge 实时打分 + 用户隐式反馈 + 抽样人工标注三管齐下。但要把这一套跑成生产系统，还需要回答一系列工程问题：

flowchart TB
  Q[在线评测的工程难题] --> Q1[Trace 怎么收集<br/>侵入业务代码到什么程度]
  Q --> Q2[采样策略<br/>1% 还是 100%]
  Q --> Q3[判分异步 / 同步]
  Q --> Q4[大流量下的存储<br/>每天 GB-TB 级数据]
  Q --> Q5[隐私 / PII 处理]
  Q --> Q6[Dashboard 与告警]
  Q --> Q7[人工标注闭环]

自建一套覆盖以上 7 件事的系统，是以工程团队 × 季度计的投入。这就是为什么从 2023 年起 langsmith / langfuse / phoenix 等专业平台兴起——把这些工程负担集中到平台层。

17.2 三家平台的定位差异

三家平台的位置可以用两个轴粗略定位：

平台	协议	生态绑定
langsmith	商业 SaaS	与 LangChain / LangGraph 深度集成
langfuse	核心 MIT（ee/ 目录另有商业许可）	框架中立，自托管为主
phoenix	Elastic License 2.0（不是 Apache/MIT，不允许作为托管服务转售）	OpenTelemetry 原生，与可观测栈无缝

三家的简短画像：

• langsmith（LangChain 团队商业产品）：与 LangChain / LangGraph 深度耦合，trace 自动捕获，eval / dataset 一体化
• langfuse（开源自托管首选）：核心 MIT（企业功能在 ee/ 下另有许可）、Postgres + ClickHouse 后端、可自部署，集成中立（不绑定 LangChain）
• phoenix（Arize AI 开源）：完全 OpenTelemetry 原生，最大优势是与现有 OTel 基础设施无缝接入

工业团队不应该只看"哪个最强"，而是看"哪个最匹配现有技术栈"。下面分别看三家的工程细节。

17.3 langsmith：trace + eval + dataset 三位一体

17.3.1 数据模型

langsmith 的核心数据单元是 Run（一次完整 LLM 调用，可嵌套）：

from langsmith import Client, traceable, trace

client = Client()

# 装饰器 trace：任意 Python 函数都能挂
@traceable(name="customer-support-rag")
def my_rag(query: str):
    contexts = retriever.invoke(query)
    answer = llm.invoke(prompt.format(query=query, context=contexts))
    return answer

# 手工 trace：trace 是 langsmith 顶层导出的上下文管理器（python/langsmith/run_helpers.py:790），
# 不是 Client 的方法。run_type 默认 "chain"，可以传 "llm" / "tool" / "retriever"
with trace("custom-step", run_type="tool", inputs={"q": q}) as run:
    result = ...
    run.metadata["some-key"] = "some-value"
    run.end(outputs={"r": result})   # end 只是填 outputs：进入上下文时已 POST 建 run，退出时才 PATCH 上报 outputs

每个 Run 自动捕获：

• 输入 / 输出（完整 JSON）
• 子 Run 嵌套（形成 trace tree）
• 模型 / 参数 / token 用量 / 延迟
• 自定义 metadata（user_id / session_id 等）
• 自动关联 dataset 与 evaluation

17.3.2 评测一体化

langsmith 把"在线 trace"和"离线 dataset 评测"拉到同一界面：

from langsmith import evaluate

results = evaluate(
    my_rag,                        # 被测函数——注意签名里有 `/`，target 是位置参数，
                                   # 写成 target=my_rag 会直接 TypeError
    data="customer-support-v1",    # langsmith 上的 dataset 名 / 可迭代 example
    evaluators=[
        faithfulness_evaluator,    # 自定义 grader，逐条样例打分
        cost_evaluator,
    ],
    summary_evaluators=[...],      # 可选：对整个 experiment 汇总打分
    experiment_prefix="prompt-v3",
    max_concurrency=8,             # 默认 0＝不并发（串行跑），传 None 才是不限并发
    num_repetitions=1,             # 同一条样例重复跑几次，用来看方差
)

跑完产出 dashboard：每条样例的 trace + 多个 evaluator 的得分 + 与 baseline 对比。这种"trace、dataset、eval 一体化"是 langsmith 最被认可的工程价值。

17.3.3 Online Evaluator

langsmith 也支持"生产 trace 自动跑评分"，但这套规则是在 Web UI 的 project 设置里配置的（automation rule：过滤条件 + 采样率 + evaluator），Python SDK v0.3.1 里并没有一个 OnlineEvaluator 类——网上不少中文文章凭空造了这个 API，读者不要照抄。

SDK 侧真正提供的是两块拼图：一是采样，LANGSMITH_TRACING_SAMPLING_RATE 环境变量（0-1，Client 在 python/langsmith/client.py:452 读成 self.tracing_sample_rate）控制上报比例；二是回写分数，Client.create_feedback 把任意 grader 的结果挂到具体 run 上：

from langsmith import Client

client = Client()

# 自己在 worker 里拉近期 trace、跑 judge、把分数写回
for run in client.list_runs(project_name="prod-rag", limit=100):
    score, reason = my_faithfulness_judge(run.inputs, run.outputs)
    client.create_feedback(
        run.id,
        key="faithfulness",          # 指标名
        score=score,                 # 0-1 的分
        comment=reason,              # judge 的理由，UI 上可展开
        feedback_source_type="model",  # 标明是模型打的分，区别于人工标注
    )

这条"采样上报 → 异步 judge → create_feedback 回写"的链路，就是第 2 章 §2.3 在线评测在工程上的标准形态；UI 上的 automation rule 只是把这套循环托管掉。

17.4 langfuse：开源自托管首选

17.4.1 架构特点

langfuse 可以自托管。许可证要看清楚：仓库根 LICENSE 写明只有 ee/、web/src/ee/、worker/src/ee/ 之外的内容是 MIT Expat，这三个目录下的企业功能另有商业许可——"MIT 开源"这句话对 langfuse 只对了一半，选型时要确认你要的功能落在哪一侧。以 v3.45.0 为例：托管的 LLM-as-a-judge 评估器在 worker/src/ee/evaluation 下，而 web/src/features/entitlements/constants/entitlements.ts 给免费自托管（oss 档）的权益里没有 model-based-evaluations、data-retention，标注队列数上限为 0——这些要 self-hosted:pro / enterprise 许可或 Cloud 才有。免费自托管想做在线评分，就得自己起 worker 拉 trace、跑 judge、再经 API 写回 score。

v3 的自托管拓扑比很多人想象的重。照仓库根 docker-compose.yml，一套完整部署要起六个容器：

flowchart LR
  App[业务应用] -->|SDK| WEB["langfuse-web<br/>(langfuse/langfuse:3, :3000)"]
  WEB --> PG[(Postgres)]
  WEB --> CH[(ClickHouse)]
  WEB --> R[(Redis<br/>队列 / 缓存)]
  WEB --> S3[(MinIO / S3<br/>事件与媒体文件)]
  R --> W["langfuse-worker<br/>(langfuse/langfuse-worker:3)"]
  W --> CH
  W --> S3
  W --> PG
  style WEB fill:#dbeafe
  style W fill:#dcfce7

工程亮点：

• 数据模型：一次请求是一条 trace，下面挂 observation，类型分 SPAN / GENERATION / EVENT 三种（fern/apis/server/definition/ingestion.yml 的 ObservationType）；评分是独立的 score 对象，带 trace_id、可选 observation_id，来源分 ANNOTATION / API / EVAL（同目录 commons.yml 的 ScoreSource）
• Web / Worker 解耦：业务应用只往 web 推 trace，judge / scoring 在异步 worker 里跑，生产流量与评测开销解耦；两者之间靠 Redis 队列传递
• Postgres + ClickHouse 双库：Postgres 存元数据（用户、配置、dataset）、ClickHouse 存 trace 大流水（列存优化分析查询）
• S3 兼容对象存储是必需项，不是可选项：LANGFUSE_S3_EVENT_UPLOAD_* 与 LANGFUSE_S3_MEDIA_UPLOAD_* 两组配置分别放原始事件和多模态附件，默认的 compose 用 MinIO 顶上
• 官方 SDK 是 Python 与 JS/TS 两种（README 的集成表原话），其他语言只能走 HTTP API 或社区实现——如果你的技术栈是 Go / Java，这一点要提前评估

17.4.2 自托管的工程价值

对中国团队 / 数据合规要求高的团队，自托管几乎是唯一选项——生产 prompt 含 PII 不能上美国 SaaS。langfuse 提供官方 docker-compose.yml，单机起一套用于试用确实简单。

但别把"docker-compose 一键起"等同于"运维门槛低"：上面那张图里的六个组件在生产环境各自需要备份、扩容与监控——ClickHouse 的磁盘增长、Redis 队列积压、对象存储的生命周期策略，都是要人管的。容量能撑多少 trace 完全取决于机器规格、trace 体积、保留期与查询模式，任何"X 核 Y GB 能撑 N 万 trace/天"的说法都必须在你自己的负载上压测得出——本专栏不给这种数字。真实的取舍是：省下 SaaS 订阅费，换来一份持续的运维投入。

17.5 phoenix：OpenTelemetry 原生

17.5.1 OTel 优先的设计哲学

phoenix（Arize AI 开源）选择完全押注 OpenTelemetry：

from openinference.instrumentation.openai import OpenAIInstrumentor
from phoenix.otel import register

# phoenix 接收 OTel trace
tracer_provider = register(project_name="my-rag-app")
OpenAIInstrumentor().instrument(tracer_provider=tracer_provider)

# 之后所有 OpenAI SDK 调用自动 trace 到 phoenix

OTel-first 的工程意义：

• 零侵入：业务调用处不需要改，启动时 register + 装好对应库的 instrumentor 后，这些库的调用自动产生 span
• 统一 stack：与 Datadog / Honeycomb / Jaeger 等 OTel 后端可以并行接入
• 跨语言：OTel 是行业标准，所有主流语言都有支持

17.5.2 trace 的标准 OTel 模型

phoenix 严格遵循 OTel 的 spans + attributes + events 模型，LLM 专属语义用 OpenInference 属性表达（§17.11）：

trace
├── span "rag-pipeline"                       (openinference.span.kind = CHAIN)
│   ├── span "retriever.invoke"               (RETRIEVER)
│   │   ├── attribute: input.value
│   │   ├── attribute: retrieval.documents.0.document.content
│   │   └── event: exception（出错时由 span.record_exception 记录）
│   └── span "llm.generate"                   (LLM)
│       ├── attribute: llm.model_name
│       ├── attribute: llm.token_count.prompt
│       ├── attribute: llm.token_count.completion
│       └── attribute: output.value

这个标准化让 phoenix 的 trace 能用 OTLP 导出给任何 OTel 后端；反过来，phoenix 也能接收别的 OTel 埋点产生的 span，但只有带 OpenInference 属性的 span 才会被识别成 LLM / 检索等语义。OTel 接入并非 phoenix 独有：langfuse v3.45.0 也有 OTel 接收端点（web/src/pages/api/public/otel），会把 gen_ai.* 与 OpenInference 的 llm.model_name、llm.token_count.*、input.value 等映射进自家模型（web/src/features/otel/server/index.ts）；phoenix 的优势在于 OTel 是它唯一的原生数据通道。

17.6 三家平台的工程对比

维度	langsmith	langfuse	phoenix
协议	商业 SaaS	核心 MIT，ee/ 商业许可	Elastic License 2.0
自托管	部分支持（企业版）	docker-compose（web + worker + PG + ClickHouse + Redis + S3 六件套）	docker / pip install arize-phoenix
默认捕获	LangChain 自动	手动 SDK 或 OTel	OTel auto-instrument
评测一体化	一流	中等	一般
开放性	LangChain 生态	中立	OTel 中立
大流量后端	自有云	ClickHouse（+ Postgres 存元数据）	默认 SQLite，可切 Postgres（src/phoenix/db/engines.py）
中国数据合规	否（美国 SaaS）	自托管解决	自托管解决
价格	按席位 + trace 量（见 §17.12.4）	自托管核心免费（ee/ 功能需许可）	自托管免费

17.7 在线 grader 的两种部署模式

flowchart TB
  subgraph Sidecar 模式
    A1[业务应用] -->|每次调用| G1[Grader sidecar]
    G1 -->|实时打分| S1[scoring DB]
  end
  subgraph 异步队列模式
    A2[业务应用] -->|push trace| Q[Trace Queue]
    Q --> W[Grader Worker<br/>异步消费]
    W --> S2[scoring DB]
  end
  style G1 fill:#fee2e2
  style W fill:#dcfce7

两种模式各有取舍：

• Sidecar：评测延迟低（同步可见），但每次业务调用都要等 grader → 生产链路被评测拖慢
• 异步队列：业务零阻塞，但评测结果有几秒到几分钟延迟、不能用于实时决策

工业实践绝大部分用异步队列——除非你的业务真的需要"评测分进入实时回答的决策"（如 best-of-K 选最优）。phoenix / langfuse / langsmith 默认都是异步模式。

17.8 选型决策树

flowchart TD
  Start[在线评测平台选型] --> A{已用 LangChain<br/>/LangGraph?}
  A -->|是| B{有数据出境合规?}
  A -->|否| C{已有 OTel 基础设施?}
  B -->|否| L[langsmith]
  B -->|是| LF1[langfuse 自托管]
  C -->|是| P[phoenix]
  C -->|否| D{需要 evaluator<br/>一体化?}
  D -->|是| LF2[langfuse]
  D -->|否| P2[phoenix]
  style L fill:#dbeafe
  style LF1 fill:#dcfce7
  style LF2 fill:#dcfce7
  style P fill:#fef3c7
  style P2 fill:#fef3c7

5 个分支的具体选型逻辑：

• LangChain 重度 + 无合规约束 → langsmith（一体化最强）
• LangChain 重度 + 数据不出境 → langfuse 自托管（兼容性优先）
• 已有 OTel → phoenix（与现有可观测栈无缝）
• 一般场景需要 evaluator 一体化 → langfuse（托管 judge 属 ee/ 功能，免费自托管需自建评分 worker，见 §17.4.1）
• 一般场景只要 trace → phoenix

切换成本：langsmith ↔ langfuse 相对低（都是装饰器 + SDK 的接入形态），切到 phoenix 要改成 OTel 埋点，工作量更大；具体多久取决于埋点散布的范围。所以最初选型很重要。

17.9 隐私与合规：三个工程要点

无论选哪家平台，PII / 合规处理都是工程团队自己的责任：

17.9.1 PII 自动遮罩

在 SDK 层做 PII 检测 + 遮罩：

import re
from langsmith import traceable

def redact_pii(text: str) -> str:
    # 不用 \b：Python 3 里汉字也算 \w，"手机13800138000" 这种数字紧贴汉字的写法 \b 匹配不到
    text = re.sub(r"(?<!\d)\d{11}(?!\d)", "[PHONE]", text)          # 手机号
    text = re.sub(r"(?<!\d)\d{17}[\dXx](?!\d)", "[ID]", text)       # 身份证
    text = re.sub(r"\b[\w.-]+@[\w.-]+\.\w+\b", "[EMAIL]", text)
    return text

# 在 trace 推送前调用。注意三家 SDK 都没有一个叫 create_trace 的通用方法：
# langsmith 是 @traceable 包住的函数返回值 / `trace(...)` 上下文管理器，
# langfuse 是 @observe，phoenix 是 OTel span——脱敏要做在**喂给它们之前**。
@traceable(name="customer-support-rag")
def my_rag(query: str):
    answer = pipeline(query)
    return redact_pii(answer)          # 出参脱敏

safe_query = redact_pii(user_query)    # 入参脱敏后再进被 trace 的函数
my_rag(safe_query)

这一步必须在数据离开应用进程之前完成——一旦上传到云平台再脱敏，已经为时已晚。

17.9.2 数据保留期

GDPR / CCPA / 中国个人信息保护法都有"数据最小保留"原则。三家的保留期能力差别很大，且随版本和套餐变动（SaaS 侧常常与订阅档位绑定；自托管侧则可能要你自己写清理任务，例如 langfuse v3.45.0 的数据保留功能 data-retention 就不在免费自托管权益里，见 §17.4.1）——别假设有一个统一的默认值，上线前去各自的文档确认你那套部署到底能不能自动过期、默认留多久，然后显式配置。

17.9.3 用户撤回 / 数据删除

GDPR 的"被遗忘权"落到 trace 上，就是按 user_id 找出并删除该用户的 trace。平台一般提供按 ID 删除的接口（如 langfuse v3.45.0 的 DELETE /api/public/traces/{traceId} 与批量的 DELETE /api/public/traces，见 fern/apis/server/definition/trace.yml），但"按用户找得到"的前提是应用层确保 trace 都带 user_id 标签——这是设计 instrumentation 时就要考虑的事。

17.10 一份完整的在线评测落地方案

整合本章方法成一份新项目的工程清单：

[ ] Week 1: 选型 + POC
    [ ] 按 §17.8 决策树选 1 个平台
    [ ] 起一个 dev 环境跑通 trace 收集
    [ ] 在 100 条样例上验证 evaluator 工作

[ ] Week 2: 集成主链路
    [ ] SDK 集成到生产代码
    [ ] PII 遮罩 (§17.9.1)
    [ ] 设 1% 采样率 → 观察存储成本

[ ] Week 3: Evaluator
    [ ] 部署 Faithfulness / Hallucination Rate 在线 grader
    [ ] 配 dashboard
    [ ] 设阈值 + Slack 告警

[ ] Week 4: Hard Case Mining
    [ ] 接 thumbs-up/down 反馈
    [ ] 每周 review 低分 trace, 反哺离线集
    [ ] 与第 3 章 §3.6 闭环

[ ] Month 2: 元评测 + 优化
    [ ] 排进 §8.8.2 的元评测季度仪式
    [ ] 调整采样率、grader 选型

这份清单按周排期，走完一遍约一个月（排期为示意，按团队实际调整）。Week 1-2 是最关键的——选错平台或集成不彻底，后面所有工作都被卡住。

17.11 trace 标准与 schema

17.11.1 OpenInference：跨平台 trace 标准的雏形

OpenTelemetry 定义了"通用 trace 协议"，但 LLM 应用的 trace 需要更多专属语义——prompt、completion、token usage、tool call 这些字段。OTel 本身不规定如何用 attribute 表达它们。

OpenInference 是 Arize 团队推动的一份"LLM 专属 OTel 语义约定"。它定义了一组标准 attribute：

llm.input_messages.0.message.role     = "user"
llm.input_messages.0.message.content  = "..."
llm.output_messages.0.message.role    = "assistant"
llm.output_messages.0.message.content = "..."
llm.model_name                    = "gpt-4o"
llm.token_count.prompt            = 234
llm.token_count.completion        = 156
llm.invocation_parameters         = '{"temperature": 0.7, "max_tokens": 500}'
retrieval.documents.0.document.content = "..."
retrieval.documents.0.document.score   = 0.91
llm.tools.0.tool.json_schema      = "{...}"   # 提供给模型的工具定义挂在 LLM span 上
tool.name                         = "search_database"   # TOOL span 自身的工具名

这套约定的工程价值：埋点与平台解耦。同一份 OTel 数据可以同时导出给多个后端；phoenix 原生识别这些属性，langfuse v3.45.0 的 OTel 端点也映射了其中的模型名、token 用量、input/output（§17.5.2）；其他平台识别到什么程度，以各自文档为准。

OpenInference 为 LangChain、LlamaIndex、OpenAI SDK、Anthropic SDK 等主流库分别提供了 auto-instrumentation 包（openinference-instrumentation-*），装上就能把这些库的调用自动翻译成 OpenInference span——注意是外挂的 instrumentation 包，不是这些 SDK "内置"了 OpenInference。这意味着工程团队不必绑定单一平台——埋点用 OpenInference 标准、平台层可替换。

工业团队即使现在用 langsmith，也建议把内部 trace 抽象层做成 OpenInference 兼容——为未来切换 / 多平台并行留口子。

17.11.2 OpenInference Semantic Conventions 字段速查表

OpenInference 是 Arize（Phoenix 团队）推动、建立在 OTel 之上的 LLM trace 约定（github.com/Arize-ai/openinference），它在 OTel span 的 attribute 层定义了 LLM 专属字段。下面这批属性是工程实务里最常写到的（常量名以你装的那版 openinference-semantic-conventions 为准，本表按 Phoenix 7.5.1 声明的下限 openinference-semantic-conventions>=0.1.12（pyproject.toml）装 0.1.12 核过；带 .<i>. 的是列表按下标展平成多个 attribute）（"常用"一列标的是经验上几乎每条 LLM span 都会带的字段，不是规范意义上的必填——OpenInference 真正强制的只有 openinference.span.kind）：

字段名	类型	常用	工程含义
openinference.span.kind	string	✅	LLM / RETRIEVER / EMBEDDING / TOOL / AGENT / CHAIN / RERANKER / GUARDRAIL / EVALUATOR / UNKNOWN
llm.model_name	string	✅	gpt-4o / claude-opus-4 / qwen-max
llm.provider	string	✅	托管方：openai / anthropic / azure / google / aws 等
llm.system	string	-	模型产品系：openai / anthropic / vertexai 等（Azure 上的 OpenAI 模型即 provider=azure、system=openai）
llm.input_messages.<i>.message.role / .content	展平列表	✅	多轮输入消息
llm.output_messages.<i>.message.role / .content	展平列表	✅	模型回应消息
llm.token_count.prompt	int	✅	prompt tokens
llm.token_count.completion	int	✅	completion tokens
llm.token_count.total	int	✅	total tokens
llm.invocation_parameters	JSON 字符串	-	{temperature, top_p, max_tokens}
llm.tools.<i>.tool.json_schema	展平列表（JSON 字符串）	-	提供给模型的工具定义
llm.function_call	JSON 字符串	-	旧式 function calling 的调用记录
input.value / output.value	string	✅	通用 input/output（非 LLM span 用）
retrieval.documents.<i>.document.id / .content / .score	展平列表	-	RAG 检索结果
retrieval.documents 所在 span 的 input.value	string	-	RETRIEVER span 的 query 原文按惯例走通用 input 字段
embedding.embeddings.<i>.embedding.text / .vector	展平列表	-	向量化结果
tool.name / tool.parameters	string / JSON 字符串	-	TOOL span 的工具名 + 参数定义
session.id / user.id	string	-	会话与用户标识
metadata	JSON 字符串	-	自定义业务字段
tag.tags	array	-	用于分类的多标签

OpenInference 与 OTel GenAI 语义约定不是一回事。 OTel 官方另有一套 GenAI 语义约定，属性全部以 gen_ai. 开头：模型名是 gen_ai.request.model / gen_ai.response.model（对应上表的 llm.model_name），token 用量是 gen_ai.usage.input_tokens / gen_ai.usage.output_tokens（对应 llm.token_count.prompt / .completion），提供方是 gen_ai.provider.name（旧名 gen_ai.system，已弃用；大致对应上表的 llm.provider）。这套约定自 v1.42.0（2026-06-12）起从 semantic-conventions 主仓库迁到独立的 open-telemetry/semantic-conventions-genai 仓库维护，截至 2026-09-25 仍标注为 Development（未 stable），口径与《Harness Engineering》专栏第 19 章 §19.4 一致。Phoenix 7.5.1 按 OpenInference 解析 span；langfuse v3.45.0 的 OTel 端点两套都认（§17.5.2）。新项目若要同时照顾两边，可以两套属性都写，并把 semconv 版本一起上报。

工程实务：团队同时用多个平台时，OpenInference / GenAI 这类公开约定是现成的"中间语"——属性按约定写，接收端就不必逐个平台定制解析。这是 §17.13.5 "垂直锁定"的逃生方案之一。

from openinference.semconv.trace import SpanAttributes
from opentelemetry import trace

tracer = trace.get_tracer(__name__)
# prompt / response 是你这次调用的实际输入与输出
with tracer.start_as_current_span("llm_call") as span:
    span.set_attribute(SpanAttributes.OPENINFERENCE_SPAN_KIND, "LLM")
    span.set_attribute(SpanAttributes.LLM_MODEL_NAME, "gpt-4o")
    span.set_attribute(SpanAttributes.LLM_PROVIDER, "openai")
    span.set_attribute(SpanAttributes.LLM_TOKEN_COUNT_PROMPT, 124)
    span.set_attribute(SpanAttributes.LLM_TOKEN_COUNT_COMPLETION, 89)
    span.set_attribute(SpanAttributes.INPUT_VALUE, prompt)
    span.set_attribute(SpanAttributes.OUTPUT_VALUE, response)

10 行代码完成"标准化的 LLM trace"——埋点不绑定任何平台，换后端时只需改导出目的地。这是 LLM 可观测的"长期资产"：trace 都按公开约定写入，将来换平台时埋点层基本不用动，迁移成本主要落在历史数据与看板上。

17.11.3 常被遗忘的细节：trace 的 schema 演化

trace 数据有一个看似无关紧要、实际影响深远的工程问题：schema 演化。

随着业务发展，你的 trace 字段会扩展——新增 user_id、新增 session_id、新增 LLM provider 标签等。这些字段一旦写入数据库，再想 query 历史数据就会遇到：

查询 6 个月前的数据 → 当时的 trace 没有 provider 字段 → query 失败 / 默认值不对

三家平台的应对：

• langsmith：metadata 字段自由，每个 Run 自带任意 dict，无 schema 约束。但代价是无法跨时间统一 query
• langfuse：用 ClickHouse 的 Map 类型存自定义字段（metadata Map(LowCardinality(String), String)，见 packages/shared/clickhouse/migrations/clustered/0001_traces.up.sql），加字段不需要 migrate。但 query 性能比定字段稍差
• phoenix：依赖 OTel attributes，OTel 标准本身允许任意字段 — 但跨平台兼容时需要遵循 OpenInference 命名约定（详见 §17.11.1）

工业实践：

1. 把核心字段（user_id、session_id、provider、model_name、cost、latency）从第一天就规范化
2. 业务相关字段放 business_metadata，约定可演化
3. 每季度审查一次 trace schema，标记 deprecated 字段 6 个月后清理

这种"schema 治理"是评测平台运维的隐藏工作量——很多团队上线时不重视，半年后被"想做新分析却查不到老数据"卡住。

17.11.4 不容易做对的工程话题：trace 数据的"语义切片"

trace 数据的价值不在于"全部存下来"，而在于"能按业务相关维度切片"。常被忽视的"语义切片"维度：

• 按 user cohort：付费用户 vs 免费用户的评测分数差异
• 按 session 长度：短 session vs 长 session 的失败模式差异
• 按 referrer：从 SEO / 直接访问 / API 调用三类来源的不同质量
• 按时段：高峰 / 低峰时段的延迟与质量差异
• 按 model version：不同模型版本的实际表现对比

每个切片都对应一种"业务洞察"。比如发现"长 session 的 Faithfulness 更低" → 提示模型在长上下文上有问题；"高峰时段 Latency 飙升" → 提示需要扩容。

工程实务：trace 入库时把这些切片维度作为 metadata 存储。Dashboard 支持按任一维度切片观察。这种"多维分析能力"是 trace 系统的核心价值——远比"trace 数据存下来"本身重要。

17.12 三家平台横向对照

17.12.1 三家平台的 evaluator 部署模式对比

一个常被忽视的差异——三家平台怎么把 grader 部署到生产。这决定了开发者迭代评测的便利度：

平台	部署模式	改动 grader 后多久生效	离线 / 在线 grader 共用度
langsmith	UI 上配 automation rule + Python 侧 run_evaluator 函数	立即（push 即生效）	高（同一函数）
langfuse	UI 配置托管 LLM-as-a-judge（v3.45.0 属 ee/ 功能）；或自建 worker 经 API 写 score	立即	中（托管 judge 用 prompt 模板配置，与离线代码是两套）
phoenix	Python 侧 phoenix.evals（llm_classify / run_evals）批量跑 span，再用 Client.log_evaluations 回写	下一次跑批	中（离线 run_experiment 的 evaluator 与回写 span 的评估是两条路径）

langsmith 在这点上有明显优势——它的 evaluator 就是普通 Python 函数，装饰一下就行。 装饰器定义在 python/langsmith/evaluation/evaluator.py:387，导出名是 run_evaluator。

要强调的是：没有叫 @evaluator 的装饰器，网上不少示例把这个名字写错了， 照抄会直接 NameError。

下面是自己写一个 grader 的样子：

from langsmith.evaluation import run_evaluator

@run_evaluator
def faithfulness(run, example):
    answer = run.outputs["answer"]
    context = run.inputs["context"]
    score = call_judge(answer, context)
    return {"score": score, "key": "faithfulness"}

同一个函数既可以在离线 evaluate(...) 里用，也可以在自己的异步 worker 里跑完用 create_feedback 回写到生产 run 上（§17.3.3）——开发者不需要写两套判分逻辑。要强调的是：Python SDK 里没有 OnlineEvaluator 这个类，UI 上的 automation rule 才是托管版。

langfuse 的托管 judge 在 UI 里配 prompt 模板与过滤条件，免费自托管则要自己写评分 worker（§17.4.1）。phoenix 的在线评估是"拉 span → 跑 phoenix.evals → log_evaluations 回写"的批处理（packages/phoenix-evals/src/phoenix/evals/classify.py、src/phoenix/session/client.py:271），与 OTel 采集链路解耦，改 grader 不需要重启服务，但要自己调度跑批。

工业团队选型时，把"evaluator 迭代节奏"列入决策树：

• 团队对 prompt / grader 高频迭代（每周改动）→ langsmith
• 节奏中等 + 自托管要求 → langfuse
• evaluator 相对稳定 + 强 OTel 栈 → phoenix

17.12.2 三家平台的 dataset / experiment / human feedback 集成模式

平台不止做 trace + grader，还要把 dataset / experiment / human feedback 三个工件统一管理。这是把"在线评测"和"离线评测"打通的最后一公里。

flowchart LR
  Trace[在线 trace] -->|标注| HF[Human Feedback]
  HF -->|入数据集| DS[Dataset]
  DS -->|跑实验| Exp[Experiment]
  Exp -->|对比| Trace
  style HF fill:#fef3c7
  style DS fill:#dcfce7
  style Exp fill:#dbeafe

三家的支持差异：

• langsmith：dataset / experiment / feedback 三件事一体化，UI 流畅，跨工件关联做得最好
• langfuse：三件事都支持但 UI 相对独立，跨工件跳转需要点几次
• phoenix：7.5.1 已有 datasets 与 run_experiment（src/phoenix/experiments/functions.py:86），但工作流更偏 Python notebook 驱动

工业团队的判断：如果你的工作流是 "线上 trace → 选低分 trace → 标注 → 入集 → 离线实验 → 上线" 这个循环，langsmith 的体验明显胜出。其他两家适合工作流相对独立的场景。

17.12.3 三家平台的"接入代码"对照

整合本章方法学，给一份"同一个 RAG 应用接入三家平台的代码差异"对照：

LangSmith：

from langsmith import traceable

@traceable(name="rag-pipeline")
def my_rag(query: str):
    contexts = retriever.invoke(query)
    answer = llm.invoke(f"Q: {query}\nContext: {contexts}")
    return answer
# 自动 trace 到 LangSmith dashboard

Langfuse：

# langfuse Python SDK 2.x 写法，与 v3.45.0 仓库 README 一致；新版 SDK 导入路径有变化，以所装版本文档为准
from langfuse.decorators import observe

@observe(name="rag-pipeline")
def my_rag(query: str):
    contexts = retriever.invoke(query)
    answer = llm.invoke(f"Q: {query}\nContext: {contexts}")
    return answer
# 自动 trace 到 Langfuse

Phoenix（OTel 原生）：

from openinference.instrumentation.openai import OpenAIInstrumentor
from phoenix.otel import register

tracer_provider = register(project_name="my-rag-app")
OpenAIInstrumentor().instrument(tracer_provider=tracer_provider)

# 普通函数, 无装饰器
def my_rag(query: str):
    contexts = retriever.invoke(query)
    answer = llm.invoke(f"Q: {query}\nContext: {contexts}")
    return answer
# OTel auto-instrument 自动 trace

接入差异对照表：

维度	LangSmith	Langfuse	Phoenix
API 风格	@traceable	@observe	OTel auto-instrument
侵入度	中（要装饰器）	中（要装饰器）	低（零代码）
配置复杂度	低（只要 API key）	低（只要 API key）	中（要 register OTel）
跨语言支持	Python / JS	Python / JS/TS（官方只有这两种，其他语言走 HTTP API）	OTel 全语言
学习曲线	平	平	略陡（OTel 概念）

工程团队的实务：

• 已有 OTel 栈 → Phoenix 零侵入接入
• 全 Python / JS 团队 → LangSmith / Langfuse 装饰器最简
• 多语言项目 → Phoenix / OTel（Langfuse 官方 SDK 只覆盖 Python 与 JS/TS）

17.12.4 三家平台的计价结构（2026-09-25 核对）

SaaS 定价改得比源码还勤，任何写死在文章里的价格表都会很快过期，本专栏只给计价结构和官方定价页链接，具体数字请当场去查：

平台	计价结构	官方定价页
LangSmith	商业 SaaS：按席位订阅 + 超出基础额度后按 trace 量计费，有免费档	https://www.langchain.com/pricing-langsmith
Langfuse	二选一：自托管完全免费（自付服务器与运维），或用 Langfuse Cloud 按用量单元（units）分档订阅（免费档每月含 50k units）；企业功能（ee/ 目录）需商业许可	https://langfuse.com/pricing
Phoenix	自托管免费（Elastic License 2.0），托管与企业支持走 Arize 商业版报价	https://phoenix.arize.com/pricing/

算总成本时别只看订阅费——自托管这一侧要把服务器、存储增长、以及持续的运维人力折进去，这往往才是拐点的主导项：

工程团队的判断：

• 大流量：SaaS 的用量计费随 trace 量上涨，自托管的增量主要是存储与计算——量到某个点后自托管可能划算，但拐点取决于你的固定运维成本（§17.13.4 的测算就是反例）
• 小流量：Cloud 更省工程时间，免费额度往往就够用
• 中国合规：自托管唯一选项
• 海外创业团队：商业 Cloud 起步最快

建议：打开上面三个定价页，按你团队当前 trace 量算一次实际成本，并把自托管的运维人力也折算进去。30 分钟做一次真实计算，比"凭直觉觉得贵 / 便宜"或抄别人的价格表靠谱得多。

17.12.5 选型 cheatsheet：3 分钟做决定

给一份 3 分钟做决定的 cheatsheet：

你的情况	选谁	主要理由
已经用 LangChain / LangGraph + 海外 SaaS 可接受	langsmith	一体化最优、零搭建
中国数据合规 + 团队懂 docker	langfuse	核心开源、可自托管
已有 OTel 基础设施（Datadog / Honeycomb / Grafana）	phoenix	OTel 原生、与现有栈无缝
个人开发者 / 极小团队	langfuse 自托管 + 单机，或 Cloud 免费档	零授权费起步
大企业 / 强合规 / 数据敏感	langfuse 私有部署	完全自主 + 合规
学术研究 / 论文实验	phoenix + Python notebook	灵活、与 ML 工具链接合
多 LLM provider 组合（OpenAI/Claude/本地）	phoenix（OTel 中立）	不绑生态

无论选哪家，最重要的是从第一天就把 trace 接上——晚做几个月会损失大量 hard case 数据。这种"先有再说"的迭代节奏比"选完美工具"更重要。

17.13 自建还是买

17.13.1 自建 vs 用平台：一个被低估的技术债

很多团队的初期会想"先自己搭一套简单 trace 系统，等以后再迁移到 langsmith / langfuse"。这种判断在大多数情况下是错的——自建的"简单 trace"会迅速演化成不可维护的技术债。

自建的演化轨迹通常是：

Week 1: 一个 print + 写 jsonl 文件
Week 4: 加 SQLite, 简单 SQL 查询
Week 12: SQLite 扛不住了, 上 Postgres
Month 6: 想做 trace 嵌套 → 重构数据模型
Month 9: 想加 grader → 写一套 evaluator 抽象
Month 12: 想加 dashboard → 接 Grafana, 维护成本巨大
Year 2: 想加自动告警 → ... 重写一半组件
Year 3: 团队成员变动, 没人懂这套自建系统了

每一步都是合理的演化，但回头看，团队相当于花 2 年时间重新发明了 langsmith / langfuse。同样这 2 年时间，如果直接用平台、把团队精力放在业务评测上，产出通常高得多。

例外情况——值得自建的场景：

• 极端规模：量级远超现成平台的常规部署，需要定制存储与查询
• 极端合规：政务 / 军工等场景，连开源平台的引入都受限
• 公司有专职可观测性团队：自建有人维护，能持续投入

如果不在以上三种情况，直接用平台是几乎所有团队的最优选择。这个判断在 2026 年比 2023 年更明确——三家平台都已成熟，自建的相对优势在缩小。

17.13.2 在线评测平台的"开源 vs 商业"选型经济学

工业团队选 langsmith（商业）还是 langfuse / phoenix（开源）的决策不只看技术，更看经济学。一份示意测算（席位单价取 LangSmith Plus 档官方价 $39/席位/月，2026-09-25 核对；其余均为示意值）：

商业平台 langsmith 年度成本（10 人团队 / 月调用 100 万 trace）：

seat: 10 人 × $39/月 = $4,680/年
trace 量超额: $1-3K/年
高级功能（dataset / eval API）: 包含
合计: $6-8K / 年 ≈ ¥4.5-6 万/年

开源 langfuse 自托管年度成本（同等规模）：

服务器（1 台 4C16G + 200GB）: ¥1.5-2 万/年
工程师维护时间: 0.2 FTE × ¥30 万 = ¥6 万/年
LLM-judge API 调用费: ¥3-6 万/年
合计: ¥10-14 万/年

在这组示意参数下商业版反而更便宜。原因：开源的"免费"是有代价的——服务器 + 工程师维护时间。这个成本在小团队规模下比商业 SaaS 还高。

转折方向：团队规模与 trace 量越大，按 seat / trace 量计费的商业版越贵，自托管的"固定成本"越可能占优——但拐点在哪要代入自家数字（§17.13.4 的测算里拐点甚至不存在）。

工程团队的判断（分档阈值为示意）：

• < 30 人 / 月 trace < 500 万 → 商业版
• 30-100 人 / 月 trace 500 万 - 5000 万 → 看合规要求
• 100+ 人 / 月 trace > 5000 万 → 开源自托管

中国数据合规场景例外——任何规模都建议开源自托管，因为"数据出境"风险比节省的钱重要得多。

17.13.3 在线评测平台的"投资 vs 自建"决策框架

给一份"投资 SaaS 平台 vs 自建"的决策框架：

flowchart TD
  Start[需要在线评测] --> A{规模 / 合规要求?}
  A -->|低 + 海外可接受| SaaS[商业 SaaS<br/>langsmith]
  A -->|低 + 国内合规| OS[开源自托管<br/>langfuse / phoenix]
  A -->|高 + 极敏感| Build[完全自建]
  Build --> Cost[投入: 1-2 人 × 6 月]
  OS --> Cost2[投入: 0.3 人 × 持续]
  SaaS --> Cost3[投入: $5-50K/年]
  style SaaS fill:#dcfce7
  style OS fill:#fef3c7
  style Build fill:#fee2e2

图中投入量为示意值。多数团队应该选 SaaS 或开源自托管。完全自建只在极特殊场景值得（参见 §17.13.1 自建技术债演化）。

17.13.4 "在线评测平台自建 vs SaaS"的成本拐点测算

§17.13.2 概念性讨论了开源 vs 商业的选择，本节给出量化的成本拐点——按月 trace 量算"什么时候自建更划算"。

import math
from dataclasses import dataclass

@dataclass
class PlatformCostComparison:
    monthly_trace_volume: int
    saas_monthly_usd: float
    self_hosted_monthly_usd: float
    breakeven_volume: int
    recommendation: str
    rationale: str

class TracePlatformCostCalculator:
    """计算 SaaS vs 自建在不同 trace 量下的成本拐点"""

    # SaaS 价格：示意档位，不是任何平台的实际价目（实际价目见 §17.12.4 的官方定价页，
    # 例如 Langfuse Cloud 免费档每月含 50k units）
    SAAS_TIERS = [
        (50_000, 39),         # < 50k traces: $39/mo
        (1_000_000, 199),     # 50k-1M: $199/mo
        (10_000_000, 999),    # 1M-10M: $999/mo
        (100_000_000, 4999),  # 10M-100M: $4999/mo
    ]

    # 自建固定成本：基础设施 + 1 名 SRE 0.3 FTE
    SELF_HOSTED_FIXED_USD = 600 + 5_000   # 服务器 $600 + 0.3 FTE $5000
    SELF_HOSTED_PER_MILLION_VAR_USD = 50  # 每百万 trace 增量存储 + 计算

    def saas_price(self, volume: int) -> float:
        for tier_max, price in self.SAAS_TIERS:
            if volume <= tier_max:
                return price
        return self.SAAS_TIERS[-1][1] * (volume / self.SAAS_TIERS[-1][0])

    def self_hosted_cost(self, volume: int) -> float:
        return (self.SELF_HOSTED_FIXED_USD +
                (volume / 1_000_000) * self.SELF_HOSTED_PER_MILLION_VAR_USD)

    def find_breakeven(self) -> int:
        """二分搜索找拐点"""
        lo, hi = 1000, 100_000_000
        while lo < hi:
            mid = (lo + hi) // 2
            if self.saas_price(mid) > self.self_hosted_cost(mid):
                hi = mid
            else:
                lo = mid + 1
        return lo

    def analyze(self, monthly_volume: int) -> PlatformCostComparison:
        saas = self.saas_price(monthly_volume)
        sh = self.self_hosted_cost(monthly_volume)
        breakeven = self.find_breakeven()

        if monthly_volume < breakeven:
            rec = "saas"
            rationale = f"流量 < {breakeven:,} 拐点，SaaS 更便宜"
        else:
            rec = "self_hosted"
            rationale = f"流量 ≥ {breakeven:,}，自建省 ${saas - sh:.0f}/月"

        return PlatformCostComparison(
            monthly_trace_volume=monthly_volume,
            saas_monthly_usd=round(saas, 2),
            self_hosted_monthly_usd=round(sh, 2),
            breakeven_volume=breakeven,
            recommendation=rec,
            rationale=rationale,
        )

flowchart LR
  V[月度 trace 量] --> CALC[Cost Calculator]
  CALC --> S[SaaS 价]
  CALC --> H[自建价]
  S --> CMP{<br/>哪个便宜?<br/>}
  H --> CMP
  CMP -->|"这套参数下：恒成立"| SA[选 SaaS]
  CMP -->|"需自己代入固定成本重算"| SH[选自建]
  CMP --> BE[breakeven 点]
  BE -. "本模型下不存在（见下文）" .-> INFO[拐点信息]
  style SA fill:#e8f5e9
  style SH fill:#fff3e0

把上面这段代码原样跑一遍（价目为代码里的示意档位），得到的是这张表：

月度 trace 量	SaaS 月费	自建月费	月度差（SaaS − 自建）	谁更便宜
10k	$39	$5,601	-$5,562	SaaS
100k	$199	$5,605	-$5,406	SaaS
1M	$199	$5,650	-$5,451	SaaS
10M	$999	$6,100	-$5,101	SaaS
12M（刚跳档，差距收窄）	$4,999	$6,200	-$1,201	SaaS
50M	$4,999	$8,100	-$3,101	SaaS
100M	$4,999	$10,600	-$5,601	SaaS
500M	$24,995	$30,600	-$5,605	SaaS

关键洞察，和多数人预期相反：这个模型里根本不存在"自建拐点"。 每一行的月度差都是负的——SaaS 更便宜；差距最小的是刚过 10M 处（SaaS 刚跳进 $4,999 那档，10,000,001 条时差 -$1,101，12M 时 -$1,201），之后又重新拉开。原因在参数本身：顶档之上 SaaS 按 4999 × volume / 1e8 线性外推，折合每百万 $49.99，而自建的变动成本是每百万 $50，再加 $5,600 的固定底座——SaaS 的单位成本永远比自建低一点点，曲线不会相交。把量拉到 100 亿 trace/月，SaaS 仍便宜 $5,700。

顺带一个必须点破的实现坑：find_breakeven() 用二分在 [1000, 100_000_000] 里找"第一个 SaaS 比自建贵的点"，但这个区间里压根没有这样的点——循环从不收窄 hi，lo 一路涨到上界，函数于是把搜索上界 100,000,000 当成拐点返回。analyze() 拿它一比，100M 及以上就报 recommendation="self_hosted"，rationale 还会印出"自建省 $-5,601/月"这种负数。二分找交点前必须先验两端符号相反，否则拿回来的不是答案，是区间端点。这也是"自建一定更省"这类直觉如何被一段看着挺专业的代码盖章的真实样本。

所以按纯价格，这套参数下的结论只有一句：SaaS 一直更便宜。工程实务的选型经验也要照着改：

1. < 10M trace/月：SaaS，没有任何犹豫余地，自建 $5,600 的固定底座压不住
2. 10M-100M：仍是 SaaS，只是差距在刚过 10M 处收窄到千把美元——真要自建，得靠价格以外的理由
3. > 100M：先别信"自建省钱"，把自己的固定成本（SRE 工时、ClickHouse 集群、备份与合规审计）代进 SELF_HOSTED_FIXED_USD / SELF_HOSTED_PER_MILLION_VAR_USD 重算——拐点存不存在完全取决于这两个数，不取决于流量
4. 任何量级：合规、数据出境、vendor lock-in 这三条才是自建的真实驱动力，价格不是

不仅看价格还要看：

• 合规线：某些行业（金融 / 政府）必自建，价格不在考量
• 未来 12 月增长：流量会快速增长 → 早自建避免迁移成本
• 团队 SRE 能力：没 SRE 自建 = 没 trace 平台

补充一点：拐点是每个团队用自己的固定成本算出来的，不要指望官方文档替你确认。

读者把 TracePlatformCostCalculator 接入团队"季度成本审计"之前，先把 find_breakeven() 的两端符号检查补上、再把两个自建成本常量换成自家真实数字——修好之后它才能给出"现在该 saas 还是 self-hosted"的明确答案；不修，它只会把搜索上界当拐点还给你。

17.13.5 不容忽视的工程问题：trace 平台的"垂直锁定"

任何 SaaS 平台都有"vendor lock-in"风险——锁定后切换成本高。trace 平台特别明显：

• LangSmith trace 数据用 LangSmith 自家的 Run 格式（0.3.1 的 Python SDK 不产出 OTel span；JS 侧的 js/src/vercel.ts 是把 Vercel AI SDK 的 OTel span 转成 Run 上报）
• Langfuse 用自家 schema，v3.45.0 起也能经 OTel 端点接收 span（§17.5.2）
• Phoenix 用 OTel + OpenInference 约定

锁定程度：LangSmith > Langfuse > Phoenix（OTel 中立度反序）

工程修法（避免锁定）：

1. 抽象层封装：业务代码不直接调 LangSmith API，过一层自家封装
2. OpenInference 标准：trace 元数据用 OpenInference 命名约定
3. 数据导出能力：定期把 trace 数据导出到自家数据仓库（备份 + 防锁定）
4. 二次开发不写死：自定义 evaluator 用通用接口，不绑定平台特性

这些工程动作能让团队在未来切换平台时，从"大面积重写"降为"局部适配"。看似繁琐的抽象层，是评测体系长期可维护的工程保险。

17.14 采样、成本与保留策略

17.14.1 Trace 存储成本优化的工程套路

第 2 章 §2.3.4 给过采样策略的数学。这里补充存储层的具体优化，因为生产规模下 trace 存储成本可能比 LLM API 调用费还高：

flowchart TB
  Raw[每天原始 trace] --> S{优化层}
  S --> S1[1. 字段裁剪<br/>大 prompt content 不全存]
  S --> S2[2. 分级保留<br/>失败 trace 90 天, 成功 7 天]
  S --> S3[3. 列存压缩<br/>ClickHouse / Parquet]
  S --> S4[4. 冷热分层<br/>近 7 天 SSD, 老的 S3]
  S --> S5[5. 采样存储<br/>关键路径 100%, 长尾 1%]
  S1 --> Save[降本]
  S2 --> Save
  S3 --> Save
  S4 --> Save
  S5 --> Save
  style Save fill:#dcfce7

每条具体做法：

1. 字段裁剪：保留 metadata、user_id、scores，把超长 prompt content 用 hash + S3 引用代替（除非 trace 被标记为 hard case）
2. 分级保留：失败 / 低 score 的 trace 保留 90 天供根因分析；正常 trace 7 天后归档
3. 列存压缩：ClickHouse / Parquet 按列存储 + zstd 压缩，重复度高的文本字段压缩效果明显（langfuse 的 observations 表就给 input / output 列配了 CODEC(ZSTD(3))，见 packages/shared/clickhouse/migrations/clustered/0002_observations.up.sql）
4. 冷热分层：近期（7 天内）SSD 高速访问；30+ 天的归档到 S3 / OSS
5. 关键路径全采样、长尾低采样：高价值场景（账户、支付）100%、长尾 1%

这五条组合能把 trace 存储成本压下一大截，具体比例取决于你的 trace 体积与保留期，要用自己的账单验证。

17.14.2 Trace 采样的"分层抽样"算法

§17.3.3 提到了在线 trace 采样（LANGSMITH_TRACING_SAMPLING_RATE 那一档是均匀采样），但未深入算法。生产环境 1 亿日 trace 量级时，纯均匀采样会让"罕见但重要"的 case（如 user feedback 差评、tool call 失败）淹没在海量正常流量里。下面是一份分层采样算法实现，思路与各家 APM 的 tail-based sampling 一致：

import random
import hashlib
from dataclasses import dataclass
from typing import Callable
from collections import defaultdict

@dataclass
class TraceMetadata:
    trace_id: str
    user_feedback: str | None
    error: bool
    tool_call_failed: bool
    latency_ms: int
    is_high_value_user: bool

@dataclass
class SamplingDecision:
    sampled: bool
    reason: str
    stratum: str

class StratifiedTraceSampler:
    """分层采样：保证罕见高价值 trace 满采，常态流量降采"""

    STRATA_RATES = {
        "user_negative_feedback": 1.0,
        "error": 1.0,
        "tool_call_failed": 1.0,
        "high_value_user": 0.5,
        "long_latency_p99": 0.3,
        "normal": 0.01,
    }

    def __init__(self, base_rate: float = 0.01,
                 latency_p99_ms: int = 5000):
        self.base_rate = base_rate
        self.latency_p99_ms = latency_p99_ms
        self.stratum_counts = defaultdict(int)

    def _stratify(self, m: TraceMetadata) -> str:
        if m.user_feedback == "negative":
            return "user_negative_feedback"
        if m.error:
            return "error"
        if m.tool_call_failed:
            return "tool_call_failed"
        if m.latency_ms > self.latency_p99_ms:
            return "long_latency_p99"
        if m.is_high_value_user:
            return "high_value_user"
        return "normal"

    def _deterministic_dice(self, trace_id: str) -> float:
        """用 trace_id 的 hash 决定采样命中——同 trace 多次决策一致"""
        h = hashlib.md5(trace_id.encode()).hexdigest()[:16]
        return int(h, 16) / float(2 ** 64)

    def decide(self, m: TraceMetadata) -> SamplingDecision:
        stratum = self._stratify(m)
        rate = (self.base_rate if stratum == "normal"
                else self.STRATA_RATES[stratum])
        dice = self._deterministic_dice(m.trace_id)
        sampled = dice < rate
        if sampled:
            self.stratum_counts[stratum] += 1
        return SamplingDecision(
            sampled=sampled,
            reason=f"stratum={stratum} rate={rate} dice={dice:.4f}",
            stratum=stratum,
        )

    def report(self) -> dict:
        total = sum(self.stratum_counts.values())
        return {
            "total_sampled": total,
            "by_stratum": dict(self.stratum_counts),
            "stratum_pct": {k: round(v / max(total, 1) * 100, 2)
                            for k, v in self.stratum_counts.items()},
        }

flowchart TB
  T[trace 入栈] --> Q1{user_feedback = negative?}
  Q1 -->|是| S1[100% 采]
  Q1 -->|否| Q2{error?}
  Q2 -->|是| S1
  Q2 -->|否| Q3{tool_call 失败?}
  Q3 -->|是| S1
  Q3 -->|否| Q4{p99 长尾?}
  Q4 -->|是| S2[30% 采]
  Q4 -->|否| Q5{高价值用户?}
  Q5 -->|是| S3[50% 采]
  Q5 -->|否| S4[1% 采]
  style S1 fill:#ffebee
  style S4 fill:#e8f5e9

工程实务的 4 个关键设计：

• 重要 stratum 100% 采：差评 / 异常 / tool 失败一条不漏
• 延迟长尾 30%：保证慢请求得到充足分析样本
• 高价值用户 50%：付费 / VIP 用户的体验问题不能被均值掩盖
• deterministic dice：用 trace_id 的 hash 决定采样命中——保证"同一 trace 在重试时仍命中或仍不命中"，不会丢失上下文

数学上的好处（以下各层流量为示意值）：1 亿日 trace 量级 + 1% 基准采样率 + 6 个 strata，总采样数约 100 万条/天出头。其中：

• 错误 / 差评类（约 1-2 万）100% 采样 → 完整捕获
• 高价值用户 trace（约 3 万）50% 采样
• 长尾延迟（约 5 万）30% 采样
• 普通流量 1% 采样 → 平均 100 万条

按这组数算：纯均匀 1% 采样时，这约 10 万条罕见 case 只能采到约 1,000 条，占样本约 0.1%；分层后采到约 4.5 万条，占样本约 4%——罕见 case 的绝对条数多了几十倍。样本主体仍是普通流量，所以 dashboard 要按 stratum 分开看，别混在一起算均值。

17.14.3 在线评测平台的"trace 数据保留策略"——成本与价值的平衡

trace 数据无限保留的成本是天文数字。下面给出工业团队的"分层保留 + 自动归档"策略：

import time
from dataclasses import dataclass
from datetime import datetime, timedelta
from enum import Enum
from typing import Iterable

class TraceImportance(Enum):
    HOT = "hot"          # 热数据：最近 7 天，全量
    WARM = "warm"        # 温数据：8-30 天，仅采样
    COLD = "cold"        # 冷数据：1-12 月，仅 metadata
    ARCHIVED = "archived"  # 归档：12 月+，对象存储
    PURGED = "purged"     # 已删

@dataclass
class TraceRetentionDecision:
    trace_id: str
    age_days: int
    classification: TraceImportance
    storage_tier: str       # "hot_db" | "cold_db" | "s3_glacier"
    estimated_size_mb: float
    cost_per_month_usd: float

class TraceRetentionPolicy:
    """trace 数据分层保留策略"""

    HOT_DAYS = 7
    WARM_DAYS = 30
    COLD_DAYS = 365

    # 不同层成本（每 GB 每月，示意单价，按你所用云厂商的定价页替换）
    COST_HOT = 0.40        # SSD / Postgres
    COST_COLD = 0.05       # HDD / 压缩
    COST_ARCHIVED = 0.005  # S3 Glacier

    # 重要性提升器：高价值 trace 多保留
    IMPORTANCE_BOOSTERS = {
        "user_thumbs_down": 365,    # 差评 trace 至少留 1 年
        "safety_incident": 365 * 7, # 安全事件 7 年（合规）
        "training_candidate": 180,  # 用作微调的 6 个月
    }

    def classify(self, trace: dict) -> TraceRetentionDecision:
        age_days = (datetime.now() - datetime.fromisoformat(
            trace["created_at"]
        )).days

        # importance booster
        retention_floor = 0
        for tag, days in self.IMPORTANCE_BOOSTERS.items():
            if trace.get(tag):
                retention_floor = max(retention_floor, days)

        effective_days = age_days - retention_floor   # < 0 强制保留

        if effective_days < self.HOT_DAYS:
            cls, tier = TraceImportance.HOT, "hot_db"
            cost = (trace.get("size_kb", 1) / 1024 / 1024) * self.COST_HOT
        elif effective_days < self.WARM_DAYS:
            cls, tier = TraceImportance.WARM, "hot_db"
            cost = (trace.get("size_kb", 1) / 1024 / 1024) * self.COST_HOT
        elif effective_days < self.COLD_DAYS:
            cls, tier = TraceImportance.COLD, "cold_db"
            cost = (trace.get("size_kb", 1) / 1024 / 1024) * self.COST_COLD
        else:
            cls, tier = TraceImportance.ARCHIVED, "s3_glacier"
            cost = (trace.get("size_kb", 1) / 1024 / 1024) * self.COST_ARCHIVED

        return TraceRetentionDecision(
            trace_id=trace["trace_id"],
            age_days=age_days,
            classification=cls,
            storage_tier=tier,
            estimated_size_mb=round(trace.get("size_kb", 1) / 1024, 2),
            cost_per_month_usd=round(cost, 6),
        )

    def estimate_total_cost(self, traces: list[dict]) -> dict:
        decisions = [self.classify(t) for t in traces]
        from collections import Counter
        by_tier = Counter(d.classification.value for d in decisions)
        total_cost = sum(d.cost_per_month_usd for d in decisions)
        return {
            "total_traces": len(decisions),
            "by_classification": dict(by_tier),
            "monthly_storage_usd": round(total_cost, 2),
            "annual_storage_usd": round(total_cost * 12, 2),
        }

flowchart LR
  T[新 trace] --> AGE{age?}
  AGE -->|< 7d| HOT[hot_db SSD<br/>$0.40/GB-mo]
  AGE -->|7-30d| WARM[hot_db<br/>仅采样保留]
  AGE -->|30-365d| COLD[cold_db<br/>$0.05/GB-mo]
  AGE -->|> 365d| ARC[S3 Glacier<br/>$0.005/GB-mo]
  ARC -->|> 7yr 非合规| PUR[purge]
  T --> BOOST{重要 tag?}
  BOOST -->|safety_incident| FLOOR1[强制 7 年]
  BOOST -->|user_thumbs_down| FLOOR2[强制 1 年]
  style HOT fill:#ffebee
  style ARC fill:#e8f5e9
  style PUR fill:#fff3e0

工程实务的 4 条保留策略经验：

数据类型	保留期	存储层	成本
普通 trace	30d	hot → cold	中等
用户差评 trace	1 年	hot → cold	略高
安全事件 trace	7 年	hot → cold → archived	合规要求
训练候选 trace	6 月	hot 层	高（频繁访问）

具体例子（示意数据，需用你自己的数据复算）：1 亿 trace/月（每 trace 平均 5KB）：

阶段	数据量	成本
7 天 hot	23M 条 = 115 GB	$46/月
30 天 cold	70M 条 = 350 GB	$17.5/月
1 年 archived	1B 条 = 5 TB	$25/月
7 年安全（仅安全事件 trace，示意按 0.1% 计）	84M 条 = 420 GB	$2.1/月
总计	--	约 $91/月 ≈ $1.1k/年

对比"无差别全留 hot 1 年"：1.2B trace × 5KB ≈ 6 TB × $0.40/GB-mo ≈ $2.4k/月 ── 同样 1 年窗口内分层只要约 $89/月（前三行之和），节省 96% 左右；连 7 年安全档一起算（约 $91/月）也仍在 96% 左右——安全档只收安全事件 trace，体量很小。

3 类常见 retention 错误：

错误	现象	修法
所有数据全留 hot	月度存储账单随保留期线性膨胀	分层 + 自动迁移
用户差评 trace 30 天就删	hard case mining 无源	加 importance booster
没 archived 层	合规审计要 1 年前的 trace 找不到	加 S3 Glacier 归档

研究背景：

• AWS Glacier / GCP Coldline 是分层存储行业标杆
• LangSmith / Langfuse 都有类似 retention 设置，但默认值不一定符合业务

把 TraceRetentionPolicy 接进 trace 平台的 cron job 每天扫描、迁移后，降幅用自己的账单核对。注意代码里的 importance booster 会让被标记的 trace 在保底期内一直留在 hot 层（effective_days < 0），如果只想"留得久"而不想"留在贵的层"，要把保底期与分层判断拆开。

17.14.4 在线评测的"成本优化层级"——按用户分层评测，把 grader 钱花在刀刃上

线上 100% 流量全跑 grader，而 grader 若用与业务同档的模型，评测成本就与推理成本同一量级，月度账单可能接近翻倍。但真正需要每次评分的只是"高 stake / 异常 / 高价值用户"这一小部分流量；常规请求逐条评分的边际收益很低，抽样看趋势就够。本节给读者一份"按用户分层"在线评测成本优化方案——关键流量全量评分，同时把 grader 成本砍掉大半。

graph LR
    A[在线 trace 100% 流量] --> B{分层路由}
    B --> C[L1: 高 stake]
    B --> D[L2: 异常]
    B --> E[L3: 高价值用户]
    B --> F[L4: 常规]
    C --> G[100% grader<br/>10% 流量]
    D --> H[100% grader<br/>5% 流量]
    E --> I[50% grader<br/>15% 流量]
    F --> J[1% 抽样<br/>70% 流量]
    G & H & I & J --> K[聚合分析]
    K --> L[L1/L2 真实信号]
    K --> M[L3/L4 统计趋势]
    L & M --> N[决策入口]

4 层 × 流量占比 × 评测覆盖率 × 成本节省（流量占比为示意值）：

层	触发条件	流量占比	评测覆盖率	实际占评测成本	业务价值
L1 高 stake	退款 / 法律 / 医疗 query	10%	100%	43%	critical
L2 异常	latency > p95 / refusal / hallucination 嫌疑	5%	100%	22%	high
L3 高价值用户	VIP / 企业付费 / lifetime > $1K	15%	50%	32%	medium
L4 常规	其余	70%	1% 抽样	3%	trend only

合计：流量 100% 跑 grader → 成本记为 1；分层后 ≈ 0.10×1.0 + 0.05×1.0 + 0.15×0.5 + 0.70×0.01 = 0.232 → 节省约 77%（"实际占评测成本"一列即各层成本除以 0.232）。

配套实现：分层在线评测路由器：

import random
from dataclasses import dataclass, field
from datetime import datetime
from typing import Literal

UserTier = Literal["L1_high_stake", "L2_anomaly", "L3_high_value", "L4_normal"]

@dataclass
class TraceContext:
    trace_id: str
    user_id: str
    user_lifetime_value_usd: float = 0.0
    is_vip: bool = False
    query_text: str = ""
    latency_ms: int = 0
    p95_latency_ms: int = 1500
    has_refusal_marker: bool = False

@dataclass
class TieredOnlineGraderRouter:
    high_stake_keywords: tuple[str, ...] = (
        "退款", "法律", "诉讼", "医疗", "诊断", "金融", "投资",
    )
    high_value_lifetime_threshold: float = 1000.0
    l4_sampling_rate: float = 0.01
    l3_sampling_rate: float = 0.50

    def classify(self, ctx: TraceContext) -> UserTier:
        if any(kw in ctx.query_text for kw in self.high_stake_keywords):
            return "L1_high_stake"
        if (ctx.latency_ms > ctx.p95_latency_ms or ctx.has_refusal_marker):
            return "L2_anomaly"
        if ctx.is_vip or ctx.user_lifetime_value_usd >= self.high_value_lifetime_threshold:
            return "L3_high_value"
        return "L4_normal"

    def should_grade(self, tier: UserTier, rng: random.Random | None = None) -> bool:
        rng = rng or random
        if tier in ("L1_high_stake", "L2_anomaly"):
            return True
        if tier == "L3_high_value":
            return rng.random() < self.l3_sampling_rate
        return rng.random() < self.l4_sampling_rate

    def route(self, ctx: TraceContext) -> dict:
        tier = self.classify(ctx)
        return {
            "trace_id": ctx.trace_id,
            "tier": tier,
            "should_grade": self.should_grade(tier),
            "ts": datetime.now().isoformat(),
        }

    def estimate_monthly_savings(self, monthly_traces: int,
                                  cost_per_grader_call_usd: float = 0.005) -> dict:
        # 经验流量分布
        L1, L2, L3, L4 = 0.10, 0.05, 0.15, 0.70
        baseline_cost = monthly_traces * cost_per_grader_call_usd
        actual_calls = monthly_traces * (
            L1 * 1.0 + L2 * 1.0 + L3 * self.l3_sampling_rate + L4 * self.l4_sampling_rate
        )
        actual_cost = actual_calls * cost_per_grader_call_usd
        return {
            "monthly_traces": monthly_traces,
            "baseline_cost_usd": baseline_cost,
            "tiered_cost_usd": actual_cost,
            "savings_usd": baseline_cost - actual_cost,
            "savings_pct": (baseline_cost - actual_cost) / baseline_cost * 100,
            "actual_grader_call_count": int(actual_calls),
        }

举例（示意数据，需用你自己的数据复算）：某 100 万 trace/月平台：

• baseline_cost = 1M × $0.005 = $5000/月
• 接入路由：L1+L2 (15%) 全跑 + L3 (15%) 50% + L4 (70%) 1%
• actual = 1M × 0.232 = 232K calls = $1,160/月
• savings = $3,840/月 ≈ $46K/年
• 重要的是：L1/L2 高 stake 流量 100% 评测，没有任何信号丢失
• L3/L4 仍有统计趋势可看

17.14.5 团队的 trace 系统总成本估算

汇总本章方法，给一份中等规模团队（月调用量 500 万 trace）trace 系统年度总成本的示意拆分（各项金额均为示意值，只用来说明该算哪些项）：

项目	年成本
平台 SaaS 费 / 自托管资源	¥50-200 万
存储成本（90 天热 + 1 年冷）	¥10-30 万
Online judge LLM 调用费	¥30-60 万
Engineer 维护时间（1 FTE × 30%）	¥30-50 万
合规审计 / PII 处理工具	¥5-20 万
总计	¥125-360 万/年

这个数字看起来吓人，但要和"评测体系缺失导致的事故损失"放在一起比。中等规模团队应把 trace 系统视为"必要基础设施"而非"可选锦上添花"。

成本优化方向：

• 采样率从 100% 降到 1-5%：是最大的杠杆，judge 调用费按采样率同比例下降
• 平台从商业 SaaS 切到开源自托管：省订阅费，但要把运维人力折回去算（§17.13.4）
• judge 模型从 Claude Sonnet 降到 Haiku 或本地模型：按两档单价差省钱，先用元评测确认小模型判分质量够用

这些优化叠加后总成本能明显下降，落到多少用自己的账单算。

17.14.6 trace 数据导出与归档的完整脚本

整合本章方法学，给一份"trace 数据每月归档到 S3"的完整脚本：

# trace_archiver.py
import boto3
import gzip
import json
from datetime import datetime, timedelta, timezone
from pathlib import Path
from langfuse import Langfuse   # 以 langfuse Python SDK 2.x 为准

class TraceArchiver:
    def __init__(self, langfuse_client, s3_bucket: str, retention_days: int = 30):
        self.lf = langfuse_client
        self.s3 = boto3.client("s3")
        self.bucket = s3_bucket
        self.retention = retention_days

    def archive_month(self, year: int, month: int):
        """归档某月的所有 trace 到 S3 并 PII 脱敏"""
        start = datetime(year, month, 1)
        end = (start + timedelta(days=32)).replace(day=1)

        # GET /api/public/traces 的 limit 被 zod 卡在 100（publicApiPaginationZod，
        # packages/shared/src/utils/zod.ts:52-61），写 1_000_000 会被直接判非法；
        # 返回体是 {data, meta}，要翻页取 .data，不能对响应对象本身做迭代
        traces = []
        page = 1
        while True:
            resp = self.lf.fetch_traces(
                from_timestamp=start, to_timestamp=end, page=page, limit=100,
            )
            traces.extend(resp.data)
            if page >= resp.meta.total_pages:
                break
            page += 1

        # PII 脱敏
        scrubbed = [self._scrub_pii(t.dict()) for t in traces]   # dict 里的时间字段是 datetime

        # gzip 压缩
        archive_path = Path(f"/tmp/traces_{year}_{month:02d}.jsonl.gz")
        with gzip.open(archive_path, "wt", encoding="utf-8") as f:
            for trace in scrubbed:
                f.write(json.dumps(trace, ensure_ascii=False, default=str) + "\n")

        # 上传 S3
        s3_key = f"traces/{year}/{month:02d}/traces.jsonl.gz"
        self.s3.upload_file(
            str(archive_path), self.bucket, s3_key,
            ExtraArgs={"StorageClass": "GLACIER_IR"},
        )

        # 删除归档过的 hot trace（节省热存储）。SDK 2.x 的 Langfuse 类没有 delete_trace，
        # 要走底层 API 客户端（DELETE /api/public/traces/{traceId}）；timestamp 带时区，
        # 所以 now 也要带时区，否则相减会 TypeError
        for trace in traces:
            if (datetime.now(timezone.utc) - trace.timestamp).days > self.retention:
                self.lf.client.trace.delete(trace.id)

        return {
            "archived": len(scrubbed),
            "s3_path": f"s3://{self.bucket}/{s3_key}",
            "size_mb": archive_path.stat().st_size / 1024 / 1024,
        }

    def _scrub_pii(self, trace: dict) -> dict:
        """PII 脱敏"""
        import re
        text = json.dumps(trace, ensure_ascii=False, default=str)   # default=str 处理 datetime
        text = re.sub(r"(?<!\d)\d{11}(?!\d)", "[PHONE]", text)
        text = re.sub(r"(?<!\d)\d{17}[\dXx](?!\d)", "[ID]", text)
        text = re.sub(r"\b[\w.-]+@[\w.-]+\.\w+\b", "[EMAIL]", text)
        return json.loads(text)

约 70 行代码完成 trace 归档的完整生命周期（月度归档 / PII 脱敏 / gzip 压缩 / S3 GLACIER 冷存储 / 删除热 trace）。冷存储与本地 SSD 的单价差在云厂商定价页上是公开的，且各家各区不同——按你自己用的云现场查一次再定保留期，别用记忆里的倍数。

17.15 隐私与合规

17.15.1 工程问题：Trace 的"原数据" vs "脱敏后数据"

trace 系统有一个看似细节但影响深远的工程决策——保留原数据还是脱敏后数据？

两套方案：

方案 A：原数据 + 访问层脱敏

trace 入库时: 完整原文（含 PII）
访问 API 时: 按调用者权限动态脱敏

方案 B：入库前脱敏

trace 入库时: 已经脱敏（PII 替换为 token）
访问 API 时: 直接返回（无 PII）

权衡：

维度	方案 A	方案 B
原始还原能力	强（必要时管理员能看完整）	弱（已经丢失原文）
隐私风险	高（数据泄露 = PII 泄露）	低（泄露的也是脱敏后）
合规配合	弱（GDPR 数据删除请求难处理）	强（已无 PII）
调试便利	高	中（脱敏可能影响排查）

国内多数团队选方案 B 因为合规风险更低；海外团队部分选 A 因为能保留更多调试能力。

工程实务：关键合规场景必须方案 B（医疗 / 金融 / 政务）；非高合规场景方案 A + 强访问控制可接受。这个决策一旦做了很难反悔——半年后想从 A 切到 B，要重新处理已存的几 TB trace 数据，工程量极大。

17.15.2 trace 平台的"PII / 敏感信息脱敏"工程实践

§17.15.1 提出了"PII 脱敏"的概念，本节给出生产级的工程实现——这是 trace 平台合规上线的硬约束（GDPR / CCPA / 个保法）。

import re
from dataclasses import dataclass
from typing import Iterable, Pattern, Callable

@dataclass
class RedactionRule:
    name: str
    pattern: Pattern
    replacement: str
    severity: str   # "high" | "medium" | "low"

@dataclass
class RedactionReport:
    original_length: int
    redacted_length: int
    redactions_count: int
    by_rule: dict[str, int]

class TraceRedactor:
    """trace 写入平台前的 PII 脱敏管线"""

    DEFAULT_RULES = [
        # 高敏：身份证 / 银行卡 / 手机
        RedactionRule(
            "cn_id_card",
            re.compile(r"(?<!\d)(?:\d{17}[\dXx]|\d{15})(?!\d)"),
            "[ID_CARD_REDACTED]",
            "high",
        ),
        RedactionRule(
            "bank_card",
            re.compile(r"(?<!\d)(?:\d{16}|\d{19})(?!\d)"),
            "[CARD_REDACTED]",
            "high",
        ),
        RedactionRule(
            "cn_mobile",
            re.compile(r"(?<!\d)1[3-9]\d{9}(?!\d)"),
            "[PHONE_REDACTED]",
            "high",
        ),
        RedactionRule(
            "email",
            re.compile(r"\b[\w._%+-]+@[\w.-]+\.[A-Za-z]{2,}\b"),
            "[EMAIL_REDACTED]",
            "medium",
        ),
        RedactionRule(
            "ipv4",
            re.compile(r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b"),
            "[IP_REDACTED]",
            "low",
        ),
        # 中敏：地址
        RedactionRule(
            "cn_address",
            re.compile(r"[一-龥]{2,}(省|市|区|县|路|街|号|楼|室)[一-龥\d-]+"),
            "[ADDRESS_REDACTED]",
            "medium",
        ),
        # 低敏：可能是自然人姓名
        RedactionRule(
            "cn_name",
            re.compile(r"(?<!\w)[张王李赵刘陈杨黄周吴徐孙朱马胡郭林何高梁郑罗宋谢唐韩冯邓曹彭曾肖田董袁潘于蒋蔡余杜叶程苏魏吕丁任沈姚卢姜崔钟谭陆汪范金石廖贾夏韦付方白邹孟熊秦邱江尹薛闫段雷侯龙史陶黎贺顾毛郝龚邵万钱严覃武戴莫孔]{1}[一-龥]{1,2}(?!\w)"),
            "[NAME_REDACTED]",
            "low",
        ),
    ]

    def __init__(self, custom_rules: list[RedactionRule] = None,
                 enabled_severities: set[str] = None):
        self.rules = list(self.DEFAULT_RULES) + (custom_rules or [])
        self.enabled = enabled_severities or {"high", "medium", "low"}

    def redact(self, text: str) -> tuple[str, RedactionReport]:
        original_len = len(text)
        by_rule = {}
        redacted = text
        for rule in self.rules:
            if rule.severity not in self.enabled:
                continue
            matches = rule.pattern.findall(redacted)
            if matches:
                by_rule[rule.name] = len(matches)
                redacted = rule.pattern.sub(rule.replacement, redacted)
        return redacted, RedactionReport(
            original_length=original_len,
            redacted_length=len(redacted),
            redactions_count=sum(by_rule.values()),
            by_rule=by_rule,
        )

    def redact_trace(self, trace: dict) -> tuple[dict, list[RedactionReport]]:
        """递归脱敏 trace 的 input / output / metadata 字段"""
        reports = []
        cleaned = {}
        for k, v in trace.items():
            if isinstance(v, str):
                rv, report = self.redact(v)
                cleaned[k] = rv
                if report.redactions_count > 0:
                    reports.append(report)
            elif isinstance(v, dict):
                rv, sub_reports = self.redact_trace(v)
                cleaned[k] = rv
                reports.extend(sub_reports)
            elif isinstance(v, list):
                cleaned[k] = [self.redact(item)[0] if isinstance(item, str)
                               else item for item in v]
            else:
                cleaned[k] = v
        return cleaned, reports

flowchart LR
  T[trace 入栈] --> R[TraceRedactor]
  R --> H1{高敏匹配?}
  H1 -->|身份证 / 银行卡 / 手机| RED[替换 ##_REDACTED##]
  R --> M1{中敏?}
  M1 -->|email / 地址| RED
  R --> L1{低敏?}
  L1 -->|姓名 / IP| RED
  RED --> CLEAN[clean trace]
  CLEAN --> TP[trace 平台]
  R --> RP[RedactionReport]
  RP --> AUD[审计日志]
  style RED fill:#fff3e0
  style CLEAN fill:#e8f5e9

工程实务的 5 条 PII 脱敏准则：

1. 必须在写入 trace 平台之前脱敏：一旦泄漏到第三方 SaaS，违法成本不可逆
2. 保留 redaction 报告：合规审计需"何时何地脱敏了多少条 PII"
3. enabled_severities 可配置：开发环境关 low、生产开 all
4. 定期审计：抽样 trace 看是否有遗漏的新 PII 模式（如新身份证规则）
5. 失败兜底：脱敏 fail 时整条 trace 标记 dropped 而非半 redacted

合规法规对照：

法规	关键条款	本工具如何满足
中国个保法第 4、73 条	匿名化处理后的信息不属于个人信息；去标识化后仍属个人信息	正则脱敏一般只达到去标识化，不能据此免除第 13 条的处理依据（如同意）；要按匿名化对待须确保无法识别且不能复原
GDPR Art. 17	删除权	redaction 报告可索引到 trace_id 删除
GDPR Art. 25	data protection by design	脱敏在设计层而非事后
HIPAA §164.514	de-identification standard	18 类 PHI 必脱敏（医疗场景需扩展规则）

具体例子：把客服 chatbot 的 trace 文本"我是张三，手机 13800138000，订单地址北京市朝阳区某路 100 号"喂给上面的 TraceRedactor().redact()，实跑输出是 "我是张三，手机 [PHONE_REDACTED]，[ADDRESS_REDACTED] 100 号"。这正好暴露了正则脱敏的两个坑：姓名规则的 (?<!\w) 在 Python 3 里把汉字也当 \w，名字前紧贴"我是"时就匹配不上；地址规则是贪婪的，把"订单地址"一起吞掉，却漏了"100 号"。所以准则 4 的"定期审计"不是可选项，姓名、地址这类自由文本最好交给 NER 模型（如下面提到的 Presidio）而不是正则。

研究背景：

• Microsoft Presidio (github.com/microsoft/presidio) 是 PII 脱敏的开源标杆
• AWS Comprehend / GCP DLP 是云厂商商业方案

部署 TraceRedactor 后，trace 平台从"PII 雷区"变成"合规可控的可观测基础设施"——任何审计、安全、合规问题都能从 redaction 报告快速答复。这是 LLM 应用大规模上线的合规底线。

17.15.3 在线评测的"日志保留期 vs GDPR 删除请求"协同——用户行权时如何不丢评测信号

GDPR 第 17 条与中国《个保法》第 47 条都赋予用户删除权（right to erasure）；GDPR 第 12(3) 条要求在收到请求后一个月内答复处理结果（必要时可再延长两个月），个保法第 50 条要求及时受理和处理。但在线评测系统的 trace 库中包含大量与该用户关联的评测数据 — 直接删除会让历史评测信号断裂、聚合分数失真。本节给读者一份「合规删除 + 评测信号保留」工程方案，兼顾法律合规与评测连续性。

graph LR
    A[用户提删除请求] --> B{trace 处理策略}
    B --> C[1. 直接删除 trace]
    C --> D[评测时序断裂<br/>历史均值变化]
    B --> E[2. 匿名化 + 保留]
    E --> F[user_id → hash<br/>评测信号保留]
    B --> G[3. 聚合后删除]
    G --> H[只保留 metric 聚合<br/>个人 trace 删除]
    F & H --> I[合规 + 评测可用]
    I --> J[GDPR audit 通过]
    I --> K[历史 metric 时序连续]

3 类删除策略 × 合规度 × 评测连续性：

策略	GDPR 合规	评测信号	实现复杂度	推荐
直接删除	✓ 完全	✗ 时序断裂	低	不推荐
匿名化保留	△ 下面代码的加盐哈希只是假名化，GDPR 下仍属个人数据（Recital 26）；还要删掉 query/response 原文、销毁盐值或映射，才可能达到匿名化	✓ 完整保留	中	一般推荐（先过法务）
聚合后删除	✓ 完全	△ 仅聚合分	高	强合规场景

配套实现：GDPR 删除请求处理 + 评测信号保护：

import hashlib
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Literal

DeletionStrategy = Literal["hard_delete", "anonymize_retain", "aggregate_then_delete"]

@dataclass
class TraceRecord:
    trace_id: str
    user_id: str
    timestamp: datetime
    query: str
    response: str
    eval_score: float | None = None

@dataclass
class GDPRDeletionRequest:
    request_id: str
    user_id: str
    received_at: datetime
    sla_deadline: datetime  # 按 GDPR 第 12(3) 条，收到请求后一个月
    chosen_strategy: DeletionStrategy

@dataclass
class GDPRComplianceProcessor:
    trace_db: list[TraceRecord] = field(default_factory=list)
    aggregated_metrics: dict[str, list[float]] = field(default_factory=dict)
    anonymized_count: int = 0
    deleted_count: int = 0

    def hash_user_id(self, user_id: str, salt: str = "evals_salt") -> str:
        return "anon_" + hashlib.sha256((user_id + salt).encode()).hexdigest()[:16]

    def hard_delete(self, user_id: str) -> int:
        before = len(self.trace_db)
        self.trace_db = [t for t in self.trace_db if t.user_id != user_id]
        self.deleted_count += before - len(self.trace_db)
        return before - len(self.trace_db)

    def anonymize_retain(self, user_id: str) -> int:
        anon_id = self.hash_user_id(user_id)
        affected = 0
        for t in self.trace_db:
            if t.user_id == user_id:
                t.user_id = anon_id
                t.query = "[anonymized]"     # query 含 PII，匿名化
                t.response = "[anonymized]"
                affected += 1
        self.anonymized_count += affected
        return affected

    def aggregate_then_delete(self, user_id: str) -> dict:
        user_traces = [t for t in self.trace_db if t.user_id == user_id]
        if user_traces:
            scores = [t.eval_score for t in user_traces if t.eval_score is not None]
            if scores:
                avg = sum(scores) / len(scores)
                key = f"avg_score_{datetime.now().strftime('%Y-%m')}"
                # 把这部分用户的均值加入聚合，不再保留个人 trace
                self.aggregated_metrics.setdefault(key, []).append(avg)
        return {"aggregated_count": len(user_traces),
                "deleted_count": self.hard_delete(user_id)}

    def handle_request(self, req: GDPRDeletionRequest) -> dict:
        if req.chosen_strategy == "hard_delete":
            n = self.hard_delete(req.user_id)
            return {"strategy": "hard_delete", "affected_traces": n,
                    "eval_signal_lost": True}
        if req.chosen_strategy == "anonymize_retain":
            n = self.anonymize_retain(req.user_id)
            return {"strategy": "anonymize_retain", "affected_traces": n,
                    "eval_signal_lost": False}
        return {"strategy": "aggregate_then_delete",
                **self.aggregate_then_delete(req.user_id),
                "eval_signal_lost": False}

    def sla_status(self, req: GDPRDeletionRequest) -> dict:
        days_left = (req.sla_deadline - datetime.now()).days
        return {
            "request_id": req.request_id,
            "days_until_sla": days_left,
            "sla_violated": days_left < 0,
        }

    def quarterly_compliance_audit(self,
                                   handled_requests: list[GDPRDeletionRequest]) -> dict:
        violations = [r for r in handled_requests
                      if (datetime.now() - r.received_at).days > 30]
        return {
            "total_requests_handled": len(handled_requests),
            "sla_violations": len(violations),
            "avg_days_to_handle": sum(
                (datetime.now() - r.received_at).days for r in handled_requests
            ) / max(len(handled_requests), 1),
            "anonymized_total": self.anonymized_count,
            "hard_deleted_total": self.deleted_count,
            "audit_grade": ("A" if not violations
                           else "B" if len(violations) <= 2
                           else "C"),
        }

举例（示意数据，需用你自己的数据复算）：某团队收到 3 起删除请求：

• request 1（普通用户）→ anonymize_retain：保留评测信号 + 满足合规
• request 2（强合规场景如医疗）→ aggregate_then_delete：个人 trace 删除，aggregated metric 保留
• request 3（明确要求 hard delete）→ hard_delete：完全删除（接受评测信号丢失）
• evaluator 看时序图：anonymized 部分仍贡献统计，hard_deleted 部分 ratio < 1%（不影响整体趋势）

避免「评测时序数据被 GDPR 大范围删除导致历史 metric 全部失真」与「合规处理拖过法定时限」两个极端。

配套行业研究背景：

• "GDPR right to erasure" Art 17 + 中国《个保法》第 47 条
• "Anonymization vs pseudonymization" 来自 ISO/IEC 27559

注意 GDPRComplianceProcessor 只按请求上标注的策略执行删除，策略由法务 / 业务预先判定，代码不替你选；时限 SLA 也要单独跟踪。

17.15.4 在线评测平台的"多区域"考量

国际化 LLM 应用上线时还有一个工程考量——多区域 trace 平台部署。具体挑战：

• 数据主权：欧盟用户的 trace 必须留在欧盟（GDPR）
• 延迟：trace 推送跨大洲会明显增加上报延迟
• 法律责任：不同司法管辖区的法律责任不同
• 监管合规：EU AI Act / NIST RMF / 中国合规各有差异

工程修法：

• 每区域独立 trace 平台：欧盟用 EU 实例、美国用 US 实例、亚太用 APAC 实例
• 数据不跨境：每区域的 trace 数据本地存储 + 处理
• 合规配置：每区域单独配置数据保留期 / 用户撤回流程
• 统一 dashboard：跨区域指标可聚合，但原始数据不汇聚

工业实务：跨国 LLM 应用必须从第一天就考虑多区域架构。等用户量增长后再补多区域，成本要高得多。这是国际化 LLM 应用工程的"长期规划"项。

17.16 从 trace 到改进

17.16.1 工程实战：trace 数据如何驱动 prompt 优化

把 trace 数据用起来不只是"看 dashboard"。下面是一个具体的 prompt 优化工作流：

flowchart TB
  T[1k+ 生产 trace] --> S{按 score 分桶}
  S -->|低分| L[低分 trace 池]
  S -->|高分| H[高分 trace 池]
  L --> A1[抽取共性失败模式<br/>用 LLM 聚类]
  H --> A2[抽取共性成功模式]
  A1 --> Hyp[假设: prompt 在哪些场景失败]
  A2 --> Hyp
  Hyp --> NewP[改写 prompt v2]
  NewP --> Test[在原 trace 池上回放<br/>看新 prompt 表现]
  Test --> Compare[对比 v1 / v2 指标]
  Compare -->|v2 更好| Deploy[上线]
  Compare -->|不如 v1| Iter[再迭代]
  style Hyp fill:#fef3c7
  style Deploy fill:#dcfce7

关键步骤：

1. 按 score 分桶：langsmith / langfuse 都支持按 score / feedback 筛选 trace
2. LLM 聚类：用 LLM 把低分 trace 的失败原因聚成 5-10 类（自动归纳"为什么挂了"）
3. 假设驱动：基于聚类结果提出 prompt 改进假设
4. 回放验证：在原 trace 池上用新 prompt 重跑（不需要新拉用户流量）
5. 指标对比：v1 vs v2 的 paired comparison

这种"trace 驱动 prompt 优化"的工作流，比"工程师拍脑袋改 prompt"高效得多。它把"prompt engineering" 从"猜"升级成"基于数据的工程优化"——这正是评测体系给团队带来的能力升级。

17.16.2 "在线 trace → 黄金集"自动 mining 流水线

§3.6 提出 Hard Case Mining，§17.10 给了 trace 平台。本节把两者衔接起来——一份能直接跑的 mining 流水线：从 trace 平台抓出"明显 trace 失败"的样本，自动转成黄金集草稿，提 PR 给评测工程师 review。

import asyncio
import json
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import Iterable, Callable, Awaitable
from pathlib import Path

@dataclass
class HardCaseCandidate:
    source_trace_id: str
    user_query: str
    bot_answer: str
    failure_signal: str
    severity: float
    suggested_label: str
    suggested_expected: str

class TraceToGoldenSetMiner:
    """从在线 trace 中自动挖掘 hard case，输出黄金集 PR draft"""

    SIGNAL_WEIGHTS = {
        "user_thumbs_down": 1.0,
        "judge_score_low": 0.8,
        "tool_call_error": 0.7,
        "long_latency_p99": 0.4,
        "user_followup_repeated": 0.6,
    }

    def __init__(self, trace_fetcher: Callable[[datetime, datetime], Awaitable[list[dict]]],
                 dedup_set: set[str], judge_fn: Callable):
        self.fetch = trace_fetcher
        self.dedup = dedup_set
        self.judge = judge_fn

    def _score_severity(self, trace: dict) -> tuple[float, str]:
        signals = []
        score = 0.0
        if trace.get("user_feedback") == "negative":
            signals.append("user_thumbs_down")
            score += self.SIGNAL_WEIGHTS["user_thumbs_down"]
        if trace.get("judge_score", 1.0) < 0.6:
            signals.append("judge_score_low")
            score += self.SIGNAL_WEIGHTS["judge_score_low"]
        if any(t.get("error") for t in trace.get("tool_calls", [])):
            signals.append("tool_call_error")
            score += self.SIGNAL_WEIGHTS["tool_call_error"]
        if trace.get("latency_ms", 0) > 5000:
            signals.append("long_latency_p99")
            score += self.SIGNAL_WEIGHTS["long_latency_p99"]
        if trace.get("user_followup_similar", False):
            signals.append("user_followup_repeated")
            score += self.SIGNAL_WEIGHTS["user_followup_repeated"]
        return min(score, 1.0), "+".join(signals) or "none"

    def _question_hash(self, q: str) -> str:
        import hashlib
        return hashlib.md5(q.lower().strip().encode()).hexdigest()[:12]

    async def _suggest_expected(self, query: str, bot_answer: str) -> str:
        """用 reasoning judge 拟一个 expected answer 草稿"""
        prompt = (f"Query: {query}\nBot answer (failed): {bot_answer}\n"
                  "Write a 2-sentence ideal answer that the bot SHOULD have given.")
        return await self.judge(prompt)

    async def mine(self, days_back: int = 7,
                   min_severity: float = 0.6) -> list[HardCaseCandidate]:
        end = datetime.now()
        start = end - timedelta(days=days_back)
        traces = await self.fetch(start, end)
        candidates = []
        for trace in traces:
            sev, signal = self._score_severity(trace)
            if sev < min_severity:
                continue
            qhash = self._question_hash(trace["user_query"])
            if qhash in self.dedup:
                continue
            self.dedup.add(qhash)
            expected = await self._suggest_expected(
                trace["user_query"], trace["bot_answer"])
            candidates.append(HardCaseCandidate(
                source_trace_id=trace["trace_id"],
                user_query=trace["user_query"],
                bot_answer=trace["bot_answer"],
                failure_signal=signal,
                severity=round(sev, 3),
                suggested_label=signal.split("+")[0],
                suggested_expected=expected,
            ))
        return sorted(candidates, key=lambda c: -c.severity)

    def emit_pr_draft(self, candidates: list[HardCaseCandidate],
                       out_path: Path):
        records = [{
            "id": f"mined-{c.source_trace_id[:8]}",
            "input": c.user_query,
            "expected": c.suggested_expected,
            "tags": ["mined-from-prod", c.suggested_label],
            "severity": c.severity,
            "source_trace": c.source_trace_id,
            "needs_human_review": True,
        } for c in candidates]
        out_path.write_text("\n".join(json.dumps(r, ensure_ascii=False)
                                       for r in records))

flowchart LR
  T[在线 trace 平台] -->|fetch 7 天| FT[trace list]
  FT --> SC[severity 评分]
  SC -->|< 0.6| DROP[丢弃]
  SC -->|≥ 0.6| DDUP{已存在?}
  DDUP -->|是| DROP
  DDUP -->|否| EXP[reasoning judge<br/>拟 expected answer]
  EXP --> CAND[HardCaseCandidate]
  CAND --> EMIT[输出 jsonl + 提 PR]
  EMIT --> REVIEW[评测工程师 review]
  REVIEW --> GOLD[合入黄金集]
  style EXP fill:#e3f2fd
  style REVIEW fill:#fff3e0
  style GOLD fill:#e8f5e9

工程实务的 5 条 mining 规则：

• 每周跑一次：cron 触发，输出当周 hard case PR
• 去重必做：用 query 的 md5 做幂等，防止"同一 query 反复入集"
• 严重度阈值 0.6：太低引入噪音，太高漏掉真问题
• expected 是草稿不是定稿：标 needs_human_review: True，必须人工 review 后才能合入
• 来源 trace_id 必留：以后需回溯"这条 case 当初为何加入"

具体效果（示意数据，需用你自己的数据复算）：在 1 亿日 trace 量级团队，每周 mining 出 30-50 条候选 → review 通过约 60-70% → 月增 80-130 条黄金集。半年下来黄金集从 500 长到约 1000-1300，且新增部分全部来自真实失败。

17.16.3 "User Feedback → Trace 关联"的最小集成实现

§17.1 与 §17.10 都把用户反馈列为 trace 的核心信号源。但工业实务中的痛点是"用户点了拇指 → 找不到对应 trace"——前端 UI 与后端 trace 的 ID 关联缺失，反馈数据沦为孤岛。下面是这个工程粘合层的最小实现：

import asyncio
import time
import uuid
from dataclasses import dataclass, asdict
from typing import Callable, Awaitable
from datetime import datetime, timezone

@dataclass
class FeedbackEvent:
    feedback_id: str
    trace_id: str
    user_id: str
    feedback_type: str   # "thumbs_up" | "thumbs_down" | "rating" | "comment"
    value: int | str
    comment: str | None
    user_session_id: str
    submitted_at: str
    client_app: str

class FeedbackTraceLinker:
    """前端反馈 → 后端 trace 的关联粘合层"""

    def __init__(self, trace_platform_client,
                 trace_id_resolver: Callable[[str, str], Awaitable[str]]):
        self.platform = trace_platform_client
        self.resolve = trace_id_resolver

    async def receive_feedback(self, request: dict) -> FeedbackEvent:
        # 1. 反馈先入库（不阻塞用户）
        event = FeedbackEvent(
            feedback_id=str(uuid.uuid4()),
            trace_id=request.get("trace_id", ""),
            user_id=request["user_id"],
            feedback_type=request["type"],
            value=request["value"],
            comment=request.get("comment"),
            user_session_id=request["session_id"],
            submitted_at=datetime.now(timezone.utc).isoformat(),  # 带时区，否则 .timestamp() 会按本地时区解读
            client_app=request.get("app", "web"),
        )

        # 2. 若前端没传 trace_id，反向查（用 session + 时间窗）
        if not event.trace_id:
            event.trace_id = await self.resolve(
                request["session_id"],
                request.get("estimated_response_time", time.time())
            )

        # 3. 写回 trace 平台（让 trace 自带反馈分）
        if event.trace_id:
            await self._enrich_trace(event)
        return event

    async def _enrich_trace(self, event: FeedbackEvent):
        score = {"thumbs_up": 1.0, "thumbs_down": 0.0,
                 "rating": event.value / 5 if isinstance(event.value, int) else None}.get(
            event.feedback_type, None)
        if score is None:
            return
        await self.platform.score(
            trace_id=event.trace_id,
            name=f"user.{event.feedback_type}",
            value=score,
            comment=event.comment,
            metadata={"feedback_id": event.feedback_id,
                      "user_id": event.user_id,
                      "session": event.user_session_id},
        )

    async def join_with_traces(self, feedbacks: list[FeedbackEvent],
                                lookup_window_min: int = 30):
        """批处理：把零散 feedback 与 trace 做 batch join"""
        from collections import defaultdict
        by_session = defaultdict(list)
        for f in feedbacks:
            by_session[f.user_session_id].append(f)
        results = []
        for session_id, events in by_session.items():
            traces = await self.platform.fetch_traces_by_session(
                session_id, window_min=lookup_window_min)
            for event in events:
                matched = self._match_by_time(event, traces)
                if matched:
                    event.trace_id = matched["trace_id"]
                    await self._enrich_trace(event)
                results.append(event)
        return results

    def _match_by_time(self, event: FeedbackEvent,
                        traces: list[dict]) -> dict | None:
        """同 session 内最近的 trace（5 秒内）"""
        ev_time = datetime.fromisoformat(event.submitted_at).timestamp()
        candidates = [t for t in traces
                      if abs(t["completed_at"] - ev_time) < 5]
        return min(candidates, key=lambda t: abs(t["completed_at"] - ev_time)) \
            if candidates else None

flowchart TB
  UF[用户点拇指] -->|带 trace_id 的请求| API[/feedback API/]
  UF2[用户填评分但前端漏传 trace_id] -->|带 session_id 的请求| API
  API --> EV[FeedbackEvent 入库]
  EV -->|trace_id 已知| EN[enrich trace 直接挂分]
  EV -->|trace_id 缺失| RES[resolver: session + time 反查]
  RES -->|找到| EN
  RES -->|超时未找到| ORPH[孤儿反馈库]
  EN --> TP[trace 平台]
  TP --> DB[评测 dashboard]
  style ORPH fill:#fff3e0
  style EN fill:#e8f5e9

工程实务的 4 条粘合层设计：

1. 前端必带 trace_id——后端响应时把 trace_id 一起返给前端，前端反馈时回传。这是最便宜稳定的方式
2. session_id + 时间窗反查作为兜底——前端有时漏传或失效，后端用 session + 时间窗匹配。代码里的 5 秒只适合"答完立刻点拇指"的场景；用户读完长回答再反馈往往晚得多，窗口要按自家反馈延迟的分布放宽，并改成"取提交时间之前最近一条 trace"
3. 先入库再 enrich——用户提交反馈瞬间 ack，trace 关联异步处理（避免反馈接口受 trace 平台延迟影响）
4. 孤儿反馈不要丢——找不到 trace 的反馈也保留在独立表，每周人工抽查 → 推动前端改埋点

具体例子（示意数据，需用你自己的数据复算）：1000 万日 trace 中约 0.5% 用户主动反馈（5 万条）。粘合层成功率应 ≥ 95%——剩 5% 孤儿反馈集中分析能找到前端埋点漏洞，每月修 2-3 个可让粘合率涨到 98%+。

研究背景：

• LangSmith 的 Client.create_feedback（§17.3.3）与 langfuse 的 score（POST /api/public/scores，来源记为 API）都是"把用户反馈写回 trace"的现成落点

部署该粘合层后能解锁评测体系的 3 个关键能力：

1. dashboard 上每条 trace 旁有"用户反馈分"chip
2. mining 流水线（§17.16.2）能直接消费反馈信号
3. 元评测（§8.6）的"calibration"有了真实人工 anchor

把它视为"在线评测平台的最后一公里"——评测平台、trace、用户反馈在此交汇。

17.16.4 在线评测的"trace 反向归因调试器"——用户投诉 → 找到具体那一次坏调用

在线评测最大的价值不只是"统计趋势"，而是"反向定位"——当用户投诉"昨天下午 3 点钟，bot 说了奇怪的话"时，能否在数十万条 trace 里精准找到那条对话，定位到具体那次模型调用，看到 prompt + retrieval + tool call 的完整快照？本节给读者一份"反向归因调试器"工程方案。

graph LR
    A[用户投诉] --> B[输入归因线索]
    B --> C[user_id]
    B --> D[时间窗口]
    B --> E[关键词]
    B --> F[或 share_url]
    C & D & E & F --> G[多字段索引检索]
    G --> H[候选 trace 列表]
    H --> I{用户确认}
    I --> J[展开完整 trace]
    J --> K[模型 prompt 快照]
    J --> L[retrieval chunks]
    J --> M[tool call 详情]
    J --> N[历史 turn]
    K & L & M & N --> O[根因分析]
    O --> P[反馈到对应评测维度]

反向归因 5 类输入线索 + 检索难度对照（命中率为示意值）：

输入线索	索引字段	检索难度	命中率	适用场景
user_id + 时间窗	(user_id, ts)	简单	99%	用户登录账户
模糊时间 + 关键词	full-text on input/output	中	70-85%	截图但无 user_id
share_url 短码	share_id 直查	简单	100%	用户主动分享反馈
trace_id（开发者复制）	直查	简单	100%	内部 dogfood
仅截图（无线索）	图像中提取关键句 → full-text	难	30-50%	客服场景

配套实现：trace 反向归因调试器：

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Optional

@dataclass
class TraceAttributionQuery:
    user_id: Optional[str] = None
    ts_window: Optional[tuple[datetime, datetime]] = None
    keywords: list[str] = field(default_factory=list)
    share_url_token: Optional[str] = None
    trace_id: Optional[str] = None

@dataclass
class TraceRecord:
    trace_id: str
    user_id: str
    ts: datetime
    user_input: str
    final_output: str
    retrieval_chunks: list[str]
    tool_calls: list[dict]
    history_turns: list[dict]
    share_token: Optional[str] = None

@dataclass
class TraceAttributionDebugger:
    trace_index: list[TraceRecord]

    def search(self, q: TraceAttributionQuery) -> list[TraceRecord]:
        if q.trace_id:
            return [t for t in self.trace_index if t.trace_id == q.trace_id]
        if q.share_url_token:
            return [t for t in self.trace_index if t.share_token == q.share_url_token]
        candidates = self.trace_index
        if q.user_id:
            candidates = [t for t in candidates if t.user_id == q.user_id]
        if q.ts_window:
            start, end = q.ts_window
            candidates = [t for t in candidates if start <= t.ts <= end]
        if q.keywords:
            kw = [k.lower() for k in q.keywords]
            candidates = [t for t in candidates
                          if any(k in (t.user_input + t.final_output).lower() for k in kw)]
        return sorted(candidates, key=lambda t: t.ts, reverse=True)[:50]

    def expand_full_context(self, trace: TraceRecord) -> dict:
        return {
            "trace_id": trace.trace_id,
            "ts": trace.ts.isoformat(),
            "user": trace.user_id,
            "user_input": trace.user_input,
            "model_output": trace.final_output,
            "retrieval_chunks_count": len(trace.retrieval_chunks),
            "retrieval_preview": [c[:80] for c in trace.retrieval_chunks[:3]],
            "tool_calls": trace.tool_calls,
            "history_count": len(trace.history_turns),
        }

    def diagnose_failure_class(self, trace: TraceRecord) -> str:
        """根据 trace 特征自动初判失败类别"""
        out_low = trace.final_output.lower()
        if any(k in out_low for k in ["sorry", "抱歉", "无法回答", "不知道"]):
            return "over_refusal（可能误拒）"
        if not trace.retrieval_chunks:
            return "no_retrieval（RAG 完全未触发）"
        if len(trace.retrieval_chunks) > 0 and len(trace.final_output) < 20:
            return "underutilized_context（context 拉了但答案过短）"
        if any(t.get("error") for t in trace.tool_calls):
            return "tool_error（tool 调用失败）"
        return "needs_human_review（无明显标记，建议人工 review）"

举例（示意数据，需用你自己的数据复算）：客户经理收到投诉"我们 VIP 客户 A 昨天下午 3 点 bot 给了错误的退款金额"：

• query = TraceAttributionQuery(user_id="vip_A_001", ts_window=(2026-04-27 14:00, 2026-04-27 16:00), keywords=["退款", "金额"])
• search → 8 条候选，按 ts 降序排
• 与客户确认"是 14:23 那条"
• expand_full_context → 看到 retrieval_chunks 拉了 3 个旧政策 chunk + tool_call 计算金额参数错位
• diagnose_failure_class → "needs_human_review"
• 根因：retriever 抓了 6 个月前已废止的退款政策 chunk → 反馈到 §13.14.2 freshness 评测维度

17.16.5 常被低估的能力：Replay

trace replay 是指把过去某条 trace 的输入"重放"过当前最新的 LLM / prompt / chain，看输出是否变化。平台的 API 都能把历史 trace 的输入拉出来，重放本身可以像下面这样自己写。

工程价值：

1. 回归测试：模型版本升级时，把昨天 1000 条 trace 重放一遍，对比指标变化
2. prompt 改动验证：改了 prompt 后用历史 trace 验证有没有破坏过去能解决的问题
3. A/B 验证：不需要等真实用户，用历史 trace 模拟新 prompt 的效果

from statistics import mean
from langsmith import Client

client = Client()
# 注意 filter 用的是 LangSmith 自己的函数式 DSL（见 client.list_runs 的 docstring 示例），
# 不是 SQL——写成 "created_at >= '...'" 是无效的
old_traces = client.list_runs(
    project_name="my-app",
    filter='gt(start_time, "2026-04-01T00:00:00Z")',
    is_root=True,          # 只要根 run，避免把子 span 也拉下来
    limit=1000,
)

new_results = []
for trace in old_traces:
    new_output = my_new_chain.invoke(trace.inputs)
    new_results.append({
        "input": trace.inputs,
        "old_output": trace.outputs,
        "new_output": new_output,
    })

# 用同一个 grader 评估新旧输出
old_scores = [grader(r["input"], r["old_output"]) for r in new_results]
new_scores = [grader(r["input"], r["new_output"]) for r in new_results]

# 对比
print(f"Old avg: {mean(old_scores):.3f}")
print(f"New avg: {mean(new_scores):.3f}")
print(f"Improved cases: {sum(n > o for o, n in zip(old_scores, new_scores))}")

这种"用历史数据评测未来变更"的能力，让 prompt / model 升级几乎不需要等真实用户验证——开发阶段就能拿到定量证据。这是评测体系给迭代速度的核心红利之一。

工业团队的实践：每次 prompt 改动 PR 都跑一遍"过去 7 天 trace 的 replay"，得到指标变化报告、附在 PR 评论里。

17.17 实时评测与告警

17.17.1 在线评测的"streaming evaluation"工程模式——逐 token 评测

新一代 LLM 应用是 streaming 输出（OpenAI / Claude 等 API 都支持流式输出），而评测多在 final response 完成后跑。下面给出 streaming evaluation 工程方案——边生成边评测：

import asyncio
from dataclasses import dataclass, field
from typing import Iterable, AsyncIterator

@dataclass
class StreamingEvalResult:
    response_chunks: list[str]
    early_warning_triggered: bool   # 中途触发警报
    early_termination_at_token: int | None
    final_score: float
    latency_to_first_warning_ms: int | None

class StreamingEvaluator:
    """边生成边评测——降低尾部风险"""

    EARLY_TERM_THRESHOLDS = {
        "fabrication_likely": 0.8,
        "harm_detected": 0.9,
        "deviation_from_topic": 0.7,
    }

    def __init__(self, fast_classifier, llm_streaming_call):
        self.classifier = fast_classifier   # 小模型快速分类
        self.llm = llm_streaming_call

    async def evaluate_stream(self, query: str,
                               check_every_tokens: int = 50
                               ) -> StreamingEvalResult:
        chunks = []
        early_warning = False
        early_term = None
        first_warn_ms = None

        async for chunk in self.llm(query):
            chunks.append(chunk)
            full_so_far = "".join(chunks)
            estimated_tokens = len(full_so_far) // 4

            # 每 N tokens 跑一次快速 check
            if estimated_tokens % check_every_tokens < 5:
                check_result = await self.classifier(query, full_so_far)
                for risk_type, threshold in self.EARLY_TERM_THRESHOLDS.items():
                    if check_result.get(risk_type, 0) > threshold:
                        early_warning = True
                        early_term = estimated_tokens
                        first_warn_ms = check_result.get("elapsed_ms", 0)
                        # 中断生成 + 替换为安全回应
                        return StreamingEvalResult(
                            response_chunks=chunks,
                            early_warning_triggered=True,
                            early_termination_at_token=early_term,
                            final_score=0.0,
                            latency_to_first_warning_ms=first_warn_ms,
                        )

        # 流结束：跑最终评测
        final_score = await self.classifier(query,
                                              "".join(chunks))
        return StreamingEvalResult(
            response_chunks=chunks,
            early_warning_triggered=False,
            early_termination_at_token=None,
            final_score=final_score.get("overall", 0.0),
            latency_to_first_warning_ms=None,
        )

flowchart LR
  Q[user query] --> S[LLM streaming output]
  S --> CK[每 50 token check]
  CK --> R{有 risk?}
  R -->|fabrication > 0.8| TERM[早终止 + 安全回应]
  R -->|harm > 0.9| TERM
  R -->|无| CONT[继续生成]
  CONT --> S
  S --> END[完整结束]
  END --> FE[final evaluation]
  style TERM fill:#ffebee
  style FE fill:#e8f5e9

工程实务的 4 类 streaming eval 优势：

维度	传统 batch eval	Streaming eval
用户感知延迟	LLM 全输出后才评	边输出边评
危险内容拦截	全输出后再 sanitize	中途即可中断
评测成本	1 次 / response	多次 micro-eval
工程复杂度	简单	中等

具体例子（示意数据，需用你自己的数据复算）：客服 chatbot 流式输出：

• 用户问 "怎么破解别人 wifi 密码"
• LLM 流式开始："要破解 wifi 你需要..."
• token 50 时 fast classifier 检测 "harm > 0.95" → 早终止
• 切换为 "抱歉我无法帮助破解他人设备" → 用户体验仍 OK

vs 传统 batch eval：完整生成后才发现危险 → moderation 替换 → 用户看到长延迟。

3 类 streaming eval 工程难点：

难点	现象	修法
检测延迟	每 50 token 跑 LLM 慢	用小模型 / 规则快速预判
假阳早终止	误判中断好回答	threshold 谨慎 + multi-class 投票
chunk 边界处理	流被切断 UX 差	统一改"安全回应"模板

17.17.2 在线评测的"实时告警"工程实现——分钟级发现问题

§17.10 的 Week 3 提到阈值告警，但具体实现需要支持"分钟级响应"——下面给出生产级的实时告警工程方案：

import asyncio
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Iterable, Callable, Awaitable

@dataclass
class RealtimeAlert:
    alert_id: str
    triggered_at: str
    metric_name: str
    current_value: float
    threshold: float
    severity: str
    affected_traces: list[str]
    suggested_action: str

class RealtimeMetricMonitor:
    """对 trace 流的实时指标监控（sliding window）"""

    def __init__(self, window_seconds: int = 60,
                 alert_threshold_overrides: dict = None):
        self.window_sec = window_seconds
        self.events: dict[str, deque] = {}
        self.thresholds = {
            "thumbs_down_rate": 0.10,         # > 10% 触发
            "p95_latency_ms": 3000,           # > 3s
            "safety_violation_rate": 0.005,    # > 0.5%
            "tool_call_error_rate": 0.05,
            "hallucination_rate": 0.10,
            **(alert_threshold_overrides or {}),
        }

    def ingest_event(self, metric_name: str, value: float):
        """实时摄入一个事件"""
        now = datetime.now().timestamp()
        if metric_name not in self.events:
            self.events[metric_name] = deque()
        self.events[metric_name].append((now, value))
        # 清掉超过 window 的旧事件
        while (self.events[metric_name] and
                self.events[metric_name][0][0] < now - self.window_sec):
            self.events[metric_name].popleft()

    def check_alert(self, metric_name: str) -> RealtimeAlert | None:
        if metric_name not in self.events:
            return None
        events = self.events[metric_name]
        if len(events) < 10:
            return None   # 样本不够

        values = [v for _, v in events]
        # rate 类用 mean，latency 用 p95
        if "rate" in metric_name:
            current = sum(values) / len(values)
        else:
            current = sorted(values)[int(0.95 * len(values))]

        threshold = self.thresholds.get(metric_name)
        if not threshold:
            return None

        if current > threshold:
            severity = ("critical" if current > threshold * 2
                        else "warning")
            return RealtimeAlert(
                alert_id=f"alert-{metric_name}-{int(datetime.now().timestamp())}",
                triggered_at=datetime.now().isoformat(),
                metric_name=metric_name,
                current_value=round(current, 4),
                threshold=threshold,
                severity=severity,
                affected_traces=[],   # 需关联 trace_id
                suggested_action=self._suggest(metric_name, severity),
            )
        return None

    def _suggest(self, metric: str, severity: str) -> str:
        if severity == "critical":
            return f"立即冻结 deploy + page oncall（{metric} 超阈 ≥ 2x）"
        return f"通知 Slack #evals + 1 小时内调查"

flowchart LR
  T[trace 流] --> ING[ingest_event]
  ING --> W[60s sliding window]
  W --> CALC[实时算 p95 / rate]
  CALC --> CMP{超阈?}
  CMP -->|是 + 2x| CRIT[critical → page]
  CMP -->|是 + 1x| WARN[warning → Slack]
  CMP -->|否| OK[正常]
  CRIT --> FREEZE[冻结 deploy]
  WARN --> SLA[1h 内查]
  style CRIT fill:#ffebee
  style WARN fill:#fff3e0
  style OK fill:#e8f5e9

工程实务的 4 类实时告警维度：

metric	window	阈值	severity
safety_violation_rate	60s	> 0.5%	critical (page)
thumbs_down_rate	60s	> 10%	warning
p95 latency	60s	> 3s	warning
tool_call_error_rate	60s	> 5%	warning

具体例子（示意数据，需用你自己的数据复算）：某客服 chatbot 周一 14:32 触发 safety_violation_rate=1.2%（窗口 60s 收 8 条违规 / 650 条流量）：

• alert 触发 → page oncall（PagerDuty）
• alert 12 秒内到达手机
• oncall 5 分钟内 ack + freeze deploy
• 调查发现：周末 merge 的 prompt 改动让 jailbreak 成功率上升
• 30 分钟内 rollback
• 总耗时 < 35 分钟（vs 没实时告警的"用户骂上微博才发现"路径，节省天级）

3 类实时告警常见错误：

错误	现象	修法
window 太短	噪声触发频繁	至少 60s
全用 mean	latency 受异常拉高	用 p95
没 severity 分级	全 page → oncall fatigue	分 warning / critical

17.17.3 在线评测的"feature flag × 在线 grader"协同——A/B 实验下评测分数怎么读

LLM 应用上线 prompt 改动 / 模型升级时几乎都用 feature flag 灰度（10% 流量先试）。但如果在线 grader 把灰度组和对照组混在一起跑均值，得到的分数完全失真——一个 90 分组 + 一个 70 分组混在一起 = 80 分，看上去"还行"，实际有一组掉得很厉害。本节给读者一份 feature flag × 在线 grader 协同的工程方案。

graph LR
    A[新 prompt / 新模型] --> B[Feature Flag 灰度]
    B --> C[10% 流量 → 实验组]
    B --> D[90% 流量 → 对照组]
    C & D --> E[在线 grader]
    E --> F[按 flag 分组打分]
    F --> G[实验组 Faithfulness]
    F --> H[对照组 Faithfulness]
    G & H --> I{分组对照}
    I --> J[A/B 显著差异检验]
    J -->|实验组好| K[逐步放量到 100%]
    J -->|对照组好| L[回滚 + 复盘]
    J -->|无差异| M[继续灰度收数据]

A/B 评测 4 个工程关键决策：

决策	错误做法	正确做法	后果
trace 是否标 flag	不标，混合统计	每条 trace 必须带 experiment_arm 标签	不标=分数失真
样本量计算	拍脑袋"跑一周"	按 §4.9.5 MDD 算 N	太少→统计不显著
置信区间	看绝对分对比	bootstrap CI + p-value	误判随机抖动
灰度策略	一次 10% 直接上	1% → 5% → 25% → 100%	风险可控

配套实现：feature-flag aware 在线 grader：

import statistics
import math
from dataclasses import dataclass, field
from datetime import datetime
from typing import Literal

ExperimentArm = Literal["control", "treatment"]

@dataclass
class TaggedEvalRecord:
    trace_id: str
    timestamp: datetime
    experiment_id: str
    arm: ExperimentArm
    metric_name: str
    score: float
    user_id: str | None = None

@dataclass
class ABEvalAnalyzer:
    records: list[TaggedEvalRecord] = field(default_factory=list)

    def by_arm(self, experiment_id: str, metric: str) -> dict[str, list[float]]:
        scores: dict[str, list[float]] = {"control": [], "treatment": []}
        for r in self.records:
            if r.experiment_id == experiment_id and r.metric_name == metric:
                scores[r.arm].append(r.score)
        return scores

    def descriptive_stats(self, scores: list[float]) -> dict:
        if not scores:
            return {"n": 0, "mean": None, "std": None}
        return {
            "n": len(scores),
            "mean": statistics.mean(scores),
            "std": statistics.stdev(scores) if len(scores) >= 2 else 0.0,
            "stderr": (statistics.stdev(scores) / math.sqrt(len(scores))
                      if len(scores) >= 2 else 0.0),
        }

    def two_sample_t_test(self, a: list[float], b: list[float]) -> dict:
        """简化两样本 t 检验，给出 t 统计 + 近似 p-value"""
        if len(a) < 2 or len(b) < 2:
            return {"valid": False, "reason": "样本量太小"}
        ma, mb = statistics.mean(a), statistics.mean(b)
        sa, sb = statistics.stdev(a), statistics.stdev(b)
        na, nb = len(a), len(b)
        se = math.sqrt(sa**2 / na + sb**2 / nb)
        if se == 0:
            return {"valid": False, "reason": "无变异"}
        t = (mb - ma) / se
        # 近似 p-value：|t| > 1.96 → p < 0.05 / |t| > 2.58 → p < 0.01
        if abs(t) > 2.58:
            p_approx = "< 0.01"
        elif abs(t) > 1.96:
            p_approx = "< 0.05"
        else:
            p_approx = "> 0.05"
        return {
            "valid": True, "t": round(t, 3), "p_approx": p_approx,
            "delta_mean": mb - ma,
            "significant": abs(t) > 1.96,
        }

    def decide(self, experiment_id: str, metric: str,
               min_significant_delta: float = 0.02) -> dict:
        scores = self.by_arm(experiment_id, metric)
        ctrl = self.descriptive_stats(scores["control"])
        treat = self.descriptive_stats(scores["treatment"])
        test = self.two_sample_t_test(scores["control"], scores["treatment"])
        if not test.get("valid"):
            return {"verdict": "insufficient_data", **test}
        if test["significant"] and test["delta_mean"] >= min_significant_delta:
            verdict = "promote_treatment"
        elif test["significant"] and test["delta_mean"] <= -min_significant_delta:
            verdict = "rollback_treatment"
        else:
            verdict = "no_significant_difference_continue"
        return {
            "verdict": verdict,
            "control": ctrl, "treatment": treat,
            "test": test,
            "recommendation": {
                "promote_treatment": "下一灰度阶段：从 10% 提到 25%",
                "rollback_treatment": "立即回滚 + 复盘 prompt diff",
                "no_significant_difference_continue": "继续收 1 周数据再决策",
                "insufficient_data": "样本量不足，至少需 N=200 双臂",
            }[verdict],
        }

举例（示意数据，需用你自己的数据复算）：某团队上线新版 prompt，10% 灰度 1 周后跑分析：

• control n=8500, mean Faithfulness 0.78
• treatment n=940, mean Faithfulness 0.83
• t = 4.21, p < 0.01，delta = +0.05
• decide → "promote_treatment"，下一阶段 25%

如果直接看混合分数 = (0.78 × 8500 + 0.83 × 940) / 9440 = 0.785，老板看着以为"涨了 0.005"，根本不会做出"promote 到 25%"的决策——丢失关键决策信号。

配套行业研究背景：

• "Online controlled experiments" 见 Kohavi et al. 2009《Controlled experiments on the web: survey and practical guide》（Data Mining and Knowledge Discovery）
• "Significance testing for ML" 来自 Bruce & Bruce《Practical Statistics for Data Scientists》（O'Reilly）

17.17.4 "trace 平台 SLO + 自身可靠性"——在线评测平台自己也需要监控

在线评测平台是"评测者"，但它自己也是一个 service——它自己的可靠性失控会让所有依赖它的评测系统瘫痪。下面给出 trace 平台自身的 SLO 设计：

from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Iterable

@dataclass
class TracePlatformSLO:
    metric: str
    objective: float
    measurement_window: str
    consequence_if_breach: str

class TracePlatformReliabilityMonitor:
    """trace 平台自身的可靠性监控"""

    SLO_DEFINITIONS = [
        TracePlatformSLO(
            "ingestion_success_rate", 0.999,
            "rolling 24h",
            "trace 丢失 → 评测失真 → freeze deploy",
        ),
        TracePlatformSLO(
            "query_p95_latency_ms", 500,
            "rolling 1h",
            "dashboard 慢 → 工程师不用 → adoption 跌",
        ),
        TracePlatformSLO(
            "storage_capacity_used_pct", 0.85,
            "rolling 7d",
            "存储满 → 新 trace 写失败 → 评测断流",
        ),
        TracePlatformSLO(
            "uptime_pct", 0.999,
            "rolling 30d",
            "平台挂 → 评测 + 监控全瘫痪",
        ),
        TracePlatformSLO(
            "data_loss_rate", 0.0001,
            "rolling 7d",
            "丢数据 → 历史评测不可信 → meta-eval 信号失真",
        ),
    ]

    def assess_health(self, current_metrics: dict) -> dict:
        breaches = []
        warnings = []
        for slo in self.SLO_DEFINITIONS:
            current = current_metrics.get(slo.metric)
            if current is None:
                continue
            # 越高越好的两项；其余（延迟、容量、丢数率）越低越好
            if slo.metric in ("ingestion_success_rate", "uptime_pct"):
                if current < slo.objective:
                    breaches.append(slo)
                elif current < slo.objective + (1 - slo.objective) / 2:
                    warnings.append(slo)   # 已用掉一半以上的错误预算
            else:
                if current > slo.objective:
                    breaches.append(slo)
                elif current > slo.objective * 0.95:
                    warnings.append(slo)

        if breaches:
            status = "critical"
        elif warnings:
            status = "degraded"
        else:
            status = "healthy"

        return {
            "status": status,
            "breached_slo": [s.metric for s in breaches],
            "warning_slo": [s.metric for s in warnings],
            "consequences": [b.consequence_if_breach for b in breaches],
            "deploy_decision": "FREEZE" if breaches else "ALLOW",
        }

flowchart LR
  TP[Trace 平台自身] --> M[5 大 SLO 监控]
  M --> S1[ingestion success ≥ 99.9%]
  M --> S2[query p95 < 500ms]
  M --> S3[storage < 85%]
  M --> S4[uptime ≥ 99.9%]
  M --> S5[data loss < 0.01%]
  S1 --> H{综合 health}
  S2 --> H
  S3 --> H
  S4 --> H
  S5 --> H
  H -->|breach| FRZ[freeze deploy<br/>page SRE]
  H -->|warning| ATT[加强观察]
  H -->|all green| OK[正常]
  style FRZ fill:#ffebee
  style OK fill:#e8f5e9

工程实务的 5 条平台 SLO 经验：

SLO	健康范围	breach 后果
ingestion success	≥ 99.9%	trace 丢失 → 评测信号缺失
query p95 latency	< 500ms	dashboard 慢 → adoption 跌
storage capacity	< 85% used	满则新数据丢
uptime	≥ 99.9%	评测系统全瘫
data loss	< 0.01%	历史评测不可信

具体例子（示意数据，需用你自己的数据复算）：某团队 trace 平台 6 个月运营：

月	主要事件	SLO 状态
1	Langfuse 升级回滚一次	1 次 critical
3	存储满 90% 紧急扩容	1 次 critical
5	query 延迟 700ms 持续	warning 一周
6	全部 healthy	月度 SLO 全绿

教训：第 5 月 query 慢 → 加索引修好。说明 trace 平台和任何 service 一样需要 SRE 治理。

3 类常见 trace 平台失控：

现象	原因	修法
突然 query 慢	数据增长但没加索引	季度 capacity planning
升级后丢数据	没 backup / 无回滚	升级前必备份
oncall 没人响应	没 set rotation	trace 平台必入 oncall 列表

研究背景：

• Google《Site Reliability Engineering》一书的 Service Level Objectives 一章：SLO 与错误预算的基本方法

17.18 成熟度与产品决策

17.18.1 "在线评测平台是否成功"的判断信号

给一份"团队的在线评测平台是否真的在工作"的诊断信号：

□ 平台每天有人主动看（不只是"配了告警"）
□ 月度有 5+ 个 trace 因为告警被深入分析
□ 季度有指标因为漂移而调整
□ 在线评测的发现能反哺到离线集（每月 ≥ 5 条）
□ 工程师 review PR 时会看 trace 链接
□ PM 在做产品决策时会引用在线评测数据
□ 合规审计能看到完整的 trace 历史

7 项全过 = 平台在生产中真正发挥价值。3-5 项过 = 平台在用但没充分用。0-2 项过 = 平台只是摆设、需要重新审视投入。

工业实务：把这份信号清单作为月度自查项。如果发现平台沦为摆设 → 立即审视 owner / 流程 / 文化。技术上做到不难，"真的有人用"才是难点。

17.18.2 工程现实：trace 平台的"成熟周期"

给在线评测平台的"成熟周期"判断（年限划分是经验性的示意，不是统计结论）：

平台年龄	状态	投入风险
0-1 年	early adopter	高（API 频繁 break）
1-2 年	早期生态	中（核心功能稳定）
2-3 年	主流采用	低（feature 完善）
3-5 年	成熟	极低（行业事实标准）
5+ 年	维护期	风险升高（可能停滞或被替代）

按这个时间表：

• LangSmith（2023 公开）
• Langfuse（2023 开源）
• Phoenix（2023 开源）

到 2026 年三家都满 3 年左右，处在"主流采用"到"成熟"之间。按这张示意表，当下 3 家都处于风险较低的阶段；但每年关注一次成熟度变化，几年后可能有新的"事实标准"出现，要预留切换的工程灵活性（参见 §17.13.5 vendor lock-in 防御）。

17.18.3 trace 平台与 RUM (Real User Monitoring) 的协同

成熟团队会把 LLM trace 平台与传统 RUM（Real User Monitoring）系统对接。RUM（如 Datadog RUM、Sentry）跟踪前端用户行为，trace 跟踪 LLM 调用——两者协同能形成完整的用户体验视图：

RUM 数据: 用户停留时长 / 点击 / 滚动 / 跳出
+ Trace 数据: LLM 调用 / 评测分数 / 失败 case
= 完整用户画像

具体集成：

• 用户的 RUM session_id 关联到该用户产生的所有 LLM trace
• 跳出率高的 session → 关联的 LLM trace 是否质量低
• 用户拇指向下 → 对应的 LLM trace 标记为 hard case

这种 "RUM + LLM trace" 的协同让评测体系从"LLM 单点"扩展到"用户体验全景"。也是 LLM 应用从"功能完整"走向"用户体验优化"的关键工程基础。

工业实务：RUM 数据 + LLM trace 的关联在 LangSmith / Langfuse 等平台还需要自己接（默认不集成）。需要在 SDK 集成时显式传递 user / session id 关联。这是评测平台的下一个工程演进方向——从"LLM-only" 升级到"产品体验全栈"。

17.18.4 "评测信号到产品决策的链路"——评测结果如何驱动产品迭代

评测体系的最终价值在于"驱动产品改进"——但很多团队评测分数躺在 dashboard 里没人用。下面给出"评测 → 产品决策"的工程化链路：

from dataclasses import dataclass, field
from datetime import datetime
from typing import Iterable

@dataclass
class EvalToDecisionLinkage:
    eval_signal_id: str
    metric: str
    delta_pp: float
    product_decisions_triggered: list[str]
    eng_actions_taken: list[str]
    user_impact_observed: str
    cycle_time_days: float

class EvalDecisionTracker:
    """跟踪评测信号到产品决策的端到端链路"""

    def __init__(self):
        self.linkages: list[EvalToDecisionLinkage] = []

    def record(self, signal: dict, decisions: list[str],
                actions: list[str], impact: str = "TBD"):
        self.linkages.append(EvalToDecisionLinkage(
            eval_signal_id=signal["id"],
            metric=signal["metric"],
            delta_pp=signal["delta_pp"],
            product_decisions_triggered=decisions,
            eng_actions_taken=actions,
            user_impact_observed=impact,
            cycle_time_days=signal.get("cycle_time", 0),
        ))

    def quarterly_audit(self) -> dict:
        """评测信号有多少真正变成了产品决策"""
        n = len(self.linkages)
        actionable = sum(1 for l in self.linkages
                         if l.product_decisions_triggered
                         or l.eng_actions_taken)
        avg_cycle = sum(l.cycle_time_days for l in self.linkages) / max(n, 1)
        return {
            "total_signals": n,
            "actionable_signals": actionable,
            "actionable_pct": round(actionable / max(n, 1), 3),
            "avg_signal_to_action_days": round(avg_cycle, 1),
            "signals_without_action": [l.eval_signal_id
                                          for l in self.linkages
                                          if not l.eng_actions_taken][:10],
        }

flowchart LR
  E[评测信号产生] --> T[Tracker 记录]
  T --> Q1{触发产品决策?}
  T --> Q2{触发工程行动?}
  Q1 -->|是| PD[产品决策列表]
  Q1 -->|否| ORP[孤儿信号]
  Q2 -->|是| EA[工程行动列表]
  Q2 -->|否| ORP
  PD --> IM[用户影响观察]
  EA --> IM
  IM --> AUD[季度 audit]
  AUD --> R{actionable_pct ≥ 60%?}
  R -->|是| OK[评测体系健康]
  R -->|否| CONCERN[评测脱离产品]
  style ORP fill:#fff3e0
  style OK fill:#e8f5e9
  style CONCERN fill:#ffebee

工程实务的 5 类信号 → 决策链路（cycle days 为示意值）：

信号类型	典型决策 / 行动	cycle days
Faithfulness 跌 5pp	rollback prompt	1-3
Safety 违规	立即冻结 + 修	< 1
用户差评涨	加 hard case + 改 prompt	5-10
Cost 涨 30%	切便宜 model	10-20
长尾 fabrication	加 retriever 评测	14-30

具体例子（示意数据，需用你自己的数据复算）：某团队季度 actionable_pct = 35%——说明 65% 评测信号没人响应：

诊断：

• 太多噪声信号（flaky）
• dashboard 没人盯
• 无 owner 跟进

修法：清 flaky case (§18.13.1) + 接 PR comment bot (§18.11.2) + 强制每周 evals review。

3 季度后 actionable_pct 涨到 78%。

3 类常见"评测脱离产品"现象：

现象	原因	修法
dashboard 全绿但用户骂	评测集与生产脱节	§3.11.4 KL audit
评测红但工程师漠视	信号无 owner	必加 oncall
改 prompt 不看评测	评测嵌入流程不深	§18.10.1 quality gate 强制

17.19 跨专栏关联

• 本专栏第 2 章 §2.3 在线评测概念：本章是其工程实现
• 本专栏第 3 章 §3.6 Hard Case Mining：本章 trace + 用户反馈是其数据来源
• 本专栏第 6 章 LLM-as-Judge：本章 in-platform grader 是其在生产的部署形态
• 本专栏第 8 章 Meta-Eval：本章 §17.10 清单里排进的元评测季度仪式（§8.8.2）
• 本专栏第 18 章 CI Quality Gate：在线告警与离线 CI 的协同
• 《LangChain 设计与实现》专栏第 12 章 回调与可观测性：LangSmith 的接入点就在回调体系上（LangChainTracer 每个运行开始时 POST、结束时 PATCH，与 §17.3.1 trace 上下文管理器的上报方式一致）
• 《Harness Engineering》专栏第 19 章 可观测性与调试：Agent 侧 trace 的统一设计

17.20 本章小结

• 在线评测平台解决 7 类工程难题：trace 收集、采样、judge 异步、大流量存储、PII、dashboard、人工闭环
• 三家代表平台定位差异明确：langsmith（商业 + LangChain）/ langfuse（开源 + 中立）/ phoenix（开源 + OTel 原生）
• 数据模型：langsmith 的 Run、langfuse 的 trace + observation、phoenix 的 OTel spans——形态不同但语义对齐
• 在线 grader 工业普遍用异步队列模式，避免阻塞业务链路
• 选型 6 维度（合规 / 生态 / OTel / 一体化 / 大流量 / 价格）+ 5 分支决策树
• PII 自动遮罩 + 数据保留期 + 用户撤回 = 不可省的合规 3 件套
• 4 周从零到生产的工程清单

最后一章——CI Quality Gate，把评测变成开发流程的硬门禁。