LLM 可观测性与评测平台全景指南：Langfuse、LangSmith、Braintrust 怎么选

一、背景：为什么传统监控管不住大模型应用

做过 RAG 问答、Agent 助手的团队都遇到过这类灵异事件：同样的提示词，今天答得好好的，明天换了个模型版本就开始胡说八道；HTTP 状态码全是 200，延迟也正常，但用户投诉回答驴唇不对马嘴；Agent 调用了十几个工具、烧了几千 token，最后交出一个自信满满的错误答案——日志里却看不出任何异常。

传统 APM（应用性能监控）盯的是 CPU、内存、接口延迟、错误率这些”机器指标”。但 LLM 应用的故障大量发生在语义层：检索召回了错误的文档、提示词被截断、工具参数传错了、模型出现了幻觉。这些在 APM 眼里全部”正常”。

LLM 可观测性与评测平台要补的正是这块空白：把 LLM 管线里的每一个 span（提示词、补全、检索、工具调用、token 数、延迟、花费）全部记录下来，再用自动化评估器给输出质量打分。到 2026 年，这类工具已经从”可选项”变成了生产级 AI 的核心基建：LLM 可观测性平台市场规模 2026 年约 26.9 亿美元，预计 2030 年达到 92.6 亿美元；LangChain 对 1300 多名从业者的调查显示，57% 的人已经在生产环境运行 Agent，其中 89% 接入了可观测性，但只有约一半做了离线评测——质量仍是上线最大的拦路虎（32% 受访者这么认为）。

本教程带你从零建立完整认知：市场四类玩家怎么选、OpenTelemetry 语义约定怎么落地、CI 回归评测怎么做、线上失败案例如何回流成测试集，最后用 Langfuse 自建方案做一次完整实战。

二、原理：先理解三层能力模型

选型之前必须先理解：所谓”LLM 可观测性平台”，其实是三层能力的组合，不同厂商的强项各不相同。

2.1 第一层：追踪（Tracing）

追踪回答的是”发生了什么”。一次用户提问进来，平台把它拆成一棵 span 树：

trace（一次问答请求）
├── span: 路由/意图识别（输入、输出、延迟）
├── span: 向量检索（query、top-k、召回文档 id、分数）
│   └── span: embedding 调用（模型、token 数）
├── span: 提示词组装（模板版本、最终 prompt 全文）
├── span: LLM 生成（模型、温度、输入输出全文、token、费用）
└── span: 工具调用 × N（函数名、参数、返回值）

关键设计点：

全量记录 prompt 与 completion：没有这两样，线上问题根本无法复现。这是 LLM 追踪和传统链路追踪最大的区别。
** token / 费用归因**：每个 span 挂 token 数，trace 级别汇总成单次请求成本，才能做成本优化。
采样与脱敏：全量记录意味着巨大的存储开销和隐私风险，生产环境一般按比例采样，并对 PII 字段做脱敏或屏蔽。

2.2 第二层：评测（Evaluation）

评测回答的是”答得好不好”。分两种：

离线评测（Offline Eval）：准备一个带标准答案的测试集，每次改提示词、换模型、调检索参数后跑一遍，对比指标变化。这是 CI 回归的基础。
在线评测（Online Eval）：对线上真实流量抽样打分（LLM-as-a-judge、人工标注、用户点赞/点踩），发现生产环境的新问题。

常用评估器（evaluator）类型：

类型	原理	适用场景
规则/启发式	正则、长度、格式校验、JSON 合法性	格式约束类任务
Embedding 相似度	生成答案与参考答案的向量相似度	开放问答粗筛
LLM-as-a-judge	用强模型按 rubric 打分	语气、相关性、幻觉检测
Agent 轨迹评分	检查工具调用序列是否合理	Agent 任务
人工标注	人审	高风险场景终裁

LLM-as-a-judge 的核心坑是”自利偏见”（judge 偏爱自己家族模型的输出）和分数不稳定，后面”常见坑”一节细讲。

2.3 第三层：生产监控与实验（Monitoring & Experimentation）

质量看板：评分分布、幻觉率、拒答率、token 成本随时间的曲线，模型版本一上线就能看出来有没有变差。
告警：错误率、延迟 p99、费用突增、评分下滑触发通知。
提示词/数据集版本管理：把 prompt 模板、测试集、评估配置都版本化，才能做可复现的对比实验（A/B 两个 prompt 跑同一测试集）。

2.4 市场四类玩家

AI 原生开源派（Langfuse 为代表）：从 LLM 场景长出来，trace 数据模型贴合 prompt/completion/retrieval，自建友好，被 ClickHouse 收购后在分析性能上有加持。适合想自主可控、数据不出内网的团队。
生态绑定派（LangSmith 为代表）：与 LangChain/LangGraph 深度集成，在自家生态里埋点几乎零成本，评测与部署流水线顺滑。重度 LangChain 用户首选，但中立性弱一些。
评测优先派（Braintrust 为代表）：从评测切入，数据集管理、批量打分、回归对比体验最好，”先写测试集再上线”理念的团队会很喜欢。
传统可观测延伸派（Arize 等）：从 ML/APM 监控延伸过来，强项是生产指标、漂移检测、告警体系，适合已经有成熟 SRE 体系、要把 LLM 纳入统一监控的大厂。

选型口诀：生态深就用生态绑定的，要自建就用开源的，评测驱动就用评测优先的，已有监控大盘就用传统延伸派。

2.5 OpenTelemetry 语义约定：为什么重要

各家 SDK 字段各异会导致”换平台 = 重埋点”。OpenTelemetry 的 GenAI 语义约定（gen_ai.* 系列属性：gen_ai.system、gen_ai.request.model、gen_ai.usage.input_tokens 等）正在统一 LLM span 的字段命名。遵循它的好处：埋点一次，多后端可接；将来从自建切到商业平台迁移成本低。本教程实战代码会显式使用 OTel 兼容的 span 命名。

三、环境准备

3.1 方案与依赖

实战采用 Langfuse 自建（Docker）+ Python SDK 方案，另配一个最小的 RAG 应用作为被观测对象。需要：

Docker 与 Docker Compose（跑 Langfuse + Postgres + ClickHouse）
Python 3.10+
一个 OpenAI 兼容的模型 API（OpenAI、DeepSeek、通义千问均可，代码里用环境变量切换）

# 1. 拉起 Langfuse 自建栈
mkdir llm-obs && cd llm-obs
curl -O https://raw.githubusercontent.com/langfuse/langfuse/main/docker-compose.yml
docker compose up -d

# 2. 打开 http://localhost:3000，注册账号，创建一个项目，
#    拿到 Public Key / Secret Key

# 3. Python 环境
pip install langfuse openai faiss-cpu tiktoken pandas datasets

3.2 环境变量

export LANGFUSE_PUBLIC_KEY="pk-lf-xxx"
export LANGFUSE_SECRET_KEY="sk-lf-xxx"
export LANGFUSE_HOST="http://localhost:3000"
export LLM_API_KEY="你的模型key"
export LLM_BASE_URL="https://api.openai.com/v1"   # 兼容渠道自行替换
export LLM_MODEL="gpt-4o-mini"

四、分步实战

我们要观测的对象是一个最小 RAG 问答服务：读本地文档 → 切分 → 向量检索 → 组装 prompt → LLM 生成。先让它”裸奔”，再一层层加上追踪、评测、回归、回流。

4.1 步骤一：写一个最小 RAG 应用（被观测对象）

# app.py —— 最小 RAG（后续所有观测都围着它转）
import os
from openai import OpenAI

client = OpenAI(api_key=os.environ["LLM_API_KEY"],
                base_url=os.environ.get("LLM_BASE_URL"))

DOCS = [
    ("退货政策", "本店支持7天无理由退货，15天内质量问题免费换新，退款在3个工作日内原路返回。"),
    ("运费规则", "满99元免运费，未满收取6元运费，偏远地区加收8元，新疆西藏不包邮。"),
    ("会员权益", "会员享95折，生日月88折，积分100抵1元，每年赠送2张免邮券。"),
    ("发票说明", "支持电子普通发票与增值税专用发票，开票内容为明细，需下单时备注。"),
]

# 极简检索：关键词命中计分（真实项目换成向量检索，不影响观测方法）
def retrieve(query: str, top_k: int = 2):
    scored = sorted(DOCS, key=lambda d: sum(1 for ch in query if ch in d[1]), reverse=True)
    return [{"title": t, "content": c} for t, c in scored[:top_k]]

def answer(query: str) -> str:
    hits = retrieve(query)
    context = "\n".join(f"【{h['title']}】{h['content']}" for h in hits)
    prompt = f"你是客服助手，只能依据以下资料回答，资料不足时说不知道。\n资料：\n{context}\n\n用户问题：{query}"
    resp = client.chat.completions.create(
        model=os.environ.get("LLM_MODEL", "gpt-4o-mini"),
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
    )
    return resp.choices[0].message.content

if __name__ == "__main__":
    print(answer("满100块包邮吗？"))

4.2 步骤二：接入 Langfuse 追踪（核心：span 树 + token 费用）

# trace_app.py —— 给 RAG 加上完整追踪
import os, time
from langfuse import Langfuse
from openai import OpenAI
from app import retrieve

langfuse = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
    host=os.environ["LANGFUSE_HOST"],
)
client = OpenAI(api_key=os.environ["LLM_API_KEY"],
                base_url=os.environ.get("LLM_BASE_URL"))
MODEL = os.environ.get("LLM_MODEL", "gpt-4o-mini")

def answer_traced(query: str, session_id: str = "demo") -> str:
    trace = langfuse.trace(name="rag-answer", session_id=session_id,
                           input=query, metadata={"model": MODEL})
    # --- retrieval span（OTel 风格命名） ---
    t0 = time.time()
    hits = retrieve(query)
    trace.span(name="retrieval", input=query,
               output={"hits": hits},
               metadata={"top_k": len(hits),
                         "latency_ms": round((time.time() - t0) * 1000, 1)})
    # --- prompt 组装 span（含模板版本，可复现） ---
    context = "\n".join(f"【{h['title']}】{h['content']}" for h in hits)
    template_v = "prompt/v3"
    prompt = f"你是客服助手，只能依据以下资料回答，资料不足时说不知道。\n资料：\n{context}\n\n用户问题：{query}"
    trace.span(name="prompt.build", input={"template": template_v},
               output={"prompt": prompt})
    # --- LLM 生成 span（记录全文 + token + 费用归因） ---
    gen = trace.generation(name="llm.generate", model=MODEL,
                           model_parameters={"temperature": 0},
                           input=[{"role": "user", "content": prompt}])
    resp = client.chat.completions.create(
        model=MODEL, messages=[{"role": "user", "content": prompt}], temperature=0)
    text = resp.choices[0].message.content
    usage = resp.usage
    gen.end(output=text,
            usage_details={"input": usage.prompt_tokens,
                           "output": usage.completion_tokens,
                           "total": usage.total_tokens})
    trace.update(output=text)
    langfuse.flush()
    return text

if __name__ == "__main__":
    for q in ["满100块包邮吗？", "会员生日有什么优惠？", "能开专票吗？"]:
        print("Q:", q)
        print("A:", answer_traced(q), "\n")

跑完后去 Langfuse 界面验证三件事：每条 trace 能点开看到检索命中了哪两篇文档；generation 里有 prompt 全文、回答全文、token 数；按 session 或时间能汇总出单次问答成本。

4.3 步骤三：定义评估器（规则 + LLM-as-a-judge 双轨）

# evaluators.py —— 可复用的评估器
import os, json
from openai import OpenAI

client = OpenAI(api_key=os.environ["LLM_API_KEY"],
                base_url=os.environ.get("LLM_BASE_URL"))
JUDGE_MODEL = os.environ.get("LLM_MODEL", "gpt-4o-mini")

def eval_grounded(answer: str, contexts: list[str]) -> dict:
    """规则评估：答案关键断言是否能在检索上下文中找到（极简版接地性检查）。"""
    import re
    nums = re.findall(r"\d+", answer)
    ctx = "".join(contexts)
    unsupported = [n for n in nums if n not in ctx]
    score = 1.0 if not unsupported else max(0.0, 1 - 0.25 * len(unsupported))
    return {"score": score, "unsupported_numbers": unsupported}

JUDGE_PROMPT = """你是问答质量评审。只依据【资料】判断【回答】。
从三个维度打分（0~1）并给出理由，以JSON返回：
{{"faithfulness": x, "relevance": x, "no_hallucination": x, "reason": "..."}}
【资料】：{context}
【用户问题】：{question}
【回答】：{answer}"""

def eval_llm_judge(question: str, answer: str, context: str) -> dict:
    resp = client.chat.completions.create(
        model=JUDGE_MODEL,
        messages=[{"role": "user",
                   "content": JUDGE_PROMPT.format(context=context,
                                                 question=question, answer=answer)}],
        temperature=0, response_format={"type": "json_object"},
    )
    return json.loads(resp.choices[0].message.content)

4.4 步骤四：建测试集 + 跑离线评测（CI 回归的基础）

# offline_eval.py —— 离线评测集与批量打分
from trace_app import answer_traced, langfuse
from app import retrieve
from evaluators import eval_grounded, eval_llm_judge

# 测试集：每个用例含参考答案与判定要点（落地时至少 50~200 条）
DATASET = [
    {"q": "满100块包邮吗？", "ref": "满99元免运费，所以100元包邮。",
     "must_contain": ["99"]},
    {"q": "会员生日有什么优惠？", "ref": "生日月88折。",
     "must_contain": ["88"]},
    {"q": "能开专票吗？", "ref": "支持增值税专用发票。",
     "must_contain": ["专用发票"]},
    {"q": "新疆包邮吗？", "ref": "新疆不包邮，偏远地区加收。",
     "must_contain": ["不包邮", "加收"]},
    {"q": "退款几天到账？", "ref": "3个工作日内原路返回。",
     "must_contain": ["3个工作日"]},
]

def run_eval(dataset_name: str = "rag-baseline-v1"):
    ds = langfuse.create_dataset(name=dataset_name)
    for item in DATASET:
        ds.create_item(input={"q": item["q"]},
                       expected_output={"ref": item["ref"],
                                        "must": item["must_contain"]})
    results = []
    for item in DATASET:
        q = item["q"]
        ans = answer_traced(q, session_id=f"eval-{dataset_name}")
        hits = retrieve(q)
        ctx = "\n".join(h["content"] for h in hits)
        rule = eval_grounded(ans, [h["content"] for h in hits])
        judge = eval_llm_judge(q, ans, ctx)
        passed = all(m in ans for m in item["must_contain"])
        results.append({"q": q, "answer": ans, "rule": rule["score"],
                        "faith": judge["faithfulness"], "passed": passed})
    ok = sum(1 for r in results if r["passed"])
    print(f"通过 {ok}/{len(results)}")
    for r in results:
        print(f"[{'PASS' if r['passed'] else 'FAIL'}] {r['q']} "
              f"rule={r['rule']} faith={r['faith']}")
    return results

if __name__ == "__main__":
    run_eval()

把 offline_eval.py 接进 CI（GitHub Actions 示例）：每次改 prompt 模板或升级模型，流水线自动跑测试集，通过率 < 阈值（比如 0.9） 则阻断合并。这就是”CI 回归”。

# .github/workflows/rag-eval.yml
name: rag-eval
on: [pull_request]
jobs:
  eval:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: pip install langfuse openai
      - run: python offline_eval.py
        env:
          LANGFUSE_PUBLIC_KEY: ${{ secrets.LANGFUSE_PUBLIC_KEY }}
          LANGFUSE_SECRET_KEY: ${{ secrets.LANGFUSE_SECRET_KEY }}
          LANGFUSE_HOST: ${{ secrets.LANGFUSE_HOST }}
          LLM_API_KEY: ${{ secrets.LLM_API_KEY }}

4.5 步骤五：线上失败回流测试集（闭环的核心）

线上用户点”踩”、judge 打低分、人工抽查发现的坏案例，要一键回流成测试集的新用例：

# reflux.py —— 失败案例回流
from langfuse import Langfuse
import os

langfuse = Langfuse(public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
                    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
                    host=os.environ["LANGFUSE_HOST"])

def reflux_bad_cases(score_threshold: float = 0.5,
                     dataset_name: str = "rag-baseline-v1"):
    """把线上低分 trace 回流为测试集用例（示意：按 score 过滤）。"""
    ds = langfuse.get_dataset(dataset_name)
    # 实际做法：查询线上 trace 中 judge 分数 < 阈值或用户点踩的样本，
    # 经人工确认参考答案后追加为新 item
    ds.create_item(
        input={"q": "偏远地区运费到底怎么收？"},   # 线上真实失败问题
        expected_output={"ref": "偏远地区加收8元，新疆西藏不包邮。",
                         "must": ["加收", "不包邮"]},
        metadata={"source": "prod-feedback", "date": "2026-08-10"})
    print("已回流 1 条失败案例，下次 CI 会覆盖它。")

if __name__ == "__main__":
    reflux_bad_cases()

完整闭环就是：线上追踪 → 抽样评分 → 失败人工确认 → 回流测试集 → CI 回归守住下限。每一次线上事故都沉淀为永久的回归用例，测试集越滚越大，模型越换越稳。

4.6 步骤六：成本与质量看板（生产监控）

在 Langfuse 里建两个 dashboard：

质量：faithfulness 均值/分布、拒答率、用户点踩率，按模型版本分组对比；
成本：单请求 token p50/p99、日总费用、检索命中率 vs 生成长度（检索越准，生成越短越便宜）。

告警建议：faithfulness 日均值跌超 0.1 发告警；单日费用超预算 120% 发告警；p99 延迟超阈值发告警。

五、常见坑

只记元数据不记全文：很多团队为了省存储只记 token 数和延迟，出问题时无法复现。正确做法是全文记录 + 采样 + PII 脱敏，而不是不记。
Judge 用自家小模型：拿被测模型自己当裁判，自利偏见严重。Judge 至少要和被测模型同级或更强，且定期用人审校准 judge 与人的一致性（agreement rate）。
测试集太小且全是 happy path：5 条用例的回归等于没有。测试集要覆盖：边界数字（99 元门槛）、否定问法（”新疆包邮吗”期望否定回答）、资料缺失（期望说不知道）、多跳组合（”会员在偏远地区买99元包邮吗”）。
改 prompt 不跑回归：prompt 是代码，改 prompt 就要跑 CI。把 prompt 模板版本化（prompt/v3），每次变更走评测流水线。
采样率拍脑袋：流量大时全量存 trace 费用爆炸。建议生产按 5%~10% 采样，但错误和低分样本 100% 保留（tail-based sampling）。
忽略检索质量：RAG 答得差，七成是检索的锅。给 retrieval span 单独打分（召回文档是否包含答案），检索坏了先修检索，别调提示词。
费用归因缺失：Agent 场景一个请求可能调十几次模型，没有 span 级 token 归因，月底账单一脸懵。每个 generation 必须挂 token 用量。

六、总结

LLM 应用的故障在语义层，传统 APM 看不见，必须用”追踪 + 评测 + 监控”三层能力覆盖。
四类平台按需选：生态深用 LangSmith 类，自建用 Langfuse 类，评测驱动用 Braintrust 类，已有 SRE 大盘用 Arize 类。
埋点遵循 OpenTelemetry GenAI 语义，换平台不重埋。
工程闭环四件套：全量追踪 → 双轨评估（规则+judge）→ CI 回归门禁 → 线上失败回流测试集。
从今天起给你的 RAG/ Agent 接上追踪，攒下第一批 trace 和测试集——这是后续一切优化的地基。

参考资料：Langfuse / LangSmith / Braintrust / Arize 官方文档与 LangChain State of Agent Engineering 调查。点击阅读原文

# AI教程 # 教程

文章版权归作者所有，未经允许请勿转载。

暂无评论

暂无评论...