第6章 Dataset 与 CI 回归 — Langfuse 自建实战

Dataset 是什么

一个 dataset 是一组带"正确答案"的测试用例,每条 item 结构:

{
  "input":            {...},   // 喂给你 LLM 应用的东西(通常是 user query)
  "expected_output":  {...},   // 期望输出(可以是自由文本、结构化、甚至 rubric)
  "metadata":         {...}    // 标签, 分组, 难度, source 等
}

它承担的角色就是传统软件里的测试夹具——每次改 prompt / 换模型 / 改 RAG,都在这组固定 case 上跑,对比前后分数,防止回退。

创建 dataset 的三个来源

① 手工挑几十条(冷启动)

from langfuse import Langfuse

lf = Langfuse()

lf.create_dataset(
    name="customer-bot-regression-v1",
    description="客服 bot 核心回归集, 每次上线前必跑",
    metadata={"owner": "qa-team", "env": "staging"},
)

for q, a in [
    ("我的订单 2024001 什么时候到", "根据物流信息预计 3 日送达"),
    ("怎么退款", "在订单页点申请退款, 3-5 个工作日到账"),
    # ... 再补 30-50 条
]:
    lf.create_dataset_item(
        dataset_name="customer-bot-regression-v1",
        input={"query": q},
        expected_output={"answer": a},
    )

② 从生产 trace 圈(最常用)

UI 上任意 trace 详情页都有 "Add to dataset" 按钮——看到一条表现好的 / 表现差的 trace,点一下就进 dataset。质量差的那些加进 dataset 特别有用:原本没 cover 的边界 case,下次不再翻车。

批量从 trace 圈也支持:

# 把最近 7 天 score < 0.5 的 trace 全挑进 regression 集
traces = lf.get_traces(
    from_timestamp=seven_days_ago,
    limit=100,
)
for t in traces.data:
    if any(s.value < 0.5 for s in (t.scores or [])):
        lf.create_dataset_item(
            dataset_name="customer-bot-regression-v1",
            input=t.input,
            expected_output=t.output,  # 先用原输出占位, QA 再修正
            source_trace_id=t.id,      # 关键: 保留溯源
            metadata={"reason": "low-score", "imported_at": today()},
        )

低分 trace 别急着当金标
生产低分 trace 的 expected_output 八成也是错的。推荐流程:先打上 metadata 标记成"待审"放入 dataset,然后让 QA 人在 UI 里修正 expected_output。Langfuse 的 human annotation 队列就是干这个的(下一章细讲)。

③ 合成(LLM 生成)

冷启动期没有足够 trace,让 GPT-4 按 schema 生成一批,再人工抽查 10% 修正:

prompt = """生成 20 条客服咨询问题与标准答案, JSON 数组, 每条 {query, answer},
覆盖订单查询 / 物流 / 退款 / 售后 / 投诉 5 个类别。"""

items = json.loads(openai.chat.completions.create(...).choices[0].message.content)
for it in items:
    lf.create_dataset_item(
        dataset_name="customer-bot-synthetic-v1",
        input={"query": it["query"]},
        expected_output={"answer": it["answer"]},
        metadata={"source": "synthetic", "model": "gpt-4o"},
    )

跑一次 run

run 的含义是"把 dataset 里所有 item 过一遍你的系统,把每条输出挂回 dataset item"。Langfuse 的 dataset.run context 帮你自动关联:

from langfuse import Langfuse

lf = Langfuse()
dataset = lf.get_dataset("customer-bot-regression-v1")

# 这次跑的名字, 一般带版本号 + 时间, UI 里列表按 run 分组
run_name = "run-2026-05-07-gpt4o-mini-prompt-v8"

for item in dataset.items:
    # 关键: 用 item.run() 包裹, span 会挂到这个 run 下
    with item.run(run_name=run_name) as handler:
        output = my_app(item.input["query"])  # 你的应用入口
        handler.update_trace(output=output)

        # 顺手打个基础评分(字符串匹配之类), 下章再讲 LLM-as-Judge
        handler.score(
            name="contains_keyword",
            value=1.0 if "退款" in output else 0.0,
        )

lf.flush()

跑完后 UI 里 Datasets → customer-bot-regression-v1 → Runs 能看到这次 run 的:

每条 item 的 input / expected / actual output 对比
每条 item 挂的 scores
聚合指标:平均分、pass 率、latency 中位数、总 cost
和上次 run 的 diff(UI 一键切换"与上次对比"视图)

多 run 对比:A/B 跑 prompt 与模型

把"prompt 版本 + 模型"当变量,跑几次 run,直接在 UI 对比。典型操作:

for prompt_label in ["v7", "v8"]:
    for model in ["gpt-4o-mini", "claude-haiku-4-5"]:
        run_name = f"prompt-{prompt_label}-{model}"
        for item in dataset.items:
            with item.run(run_name=run_name) as h:
                output = my_app(item.input, prompt_label=prompt_label, model=model)
                h.update_trace(output=output)

跑完在 UI Datasets → Runs 能并排看四个 run 的聚合指标。胜出组合就是下一版生产候选。

CI 集成:回归闸门

把上面那段逻辑塞进 GitHub Actions,每次 prompt 改动都自动跑一次 dataset,低于阈值就挂。

脚本 scripts/run_regression.py

import os, sys
from langfuse import Langfuse
from my_app import answer  # 被测应用

PASS_THRESHOLD = 0.80  # 平均分低于此分数 CI 挂

lf = Langfuse()
dataset = lf.get_dataset("customer-bot-regression-v1")
run_name = f"ci-{os.environ['GITHUB_SHA'][:7]}"

scores = []
for item in dataset.items:
    with item.run(run_name=run_name) as h:
        try:
            out = answer(item.input["query"])
            h.update_trace(output=out)
            score = simple_match_score(out, item.expected_output["answer"])
            h.score(name="match", value=score)
            scores.append(score)
        except Exception as e:
            h.score(name="match", value=0.0, comment=str(e))
            scores.append(0.0)

lf.flush()

avg = sum(scores) / len(scores)
print(f"regression run {run_name}: avg={avg:.3f} (threshold={PASS_THRESHOLD})")
if avg < PASS_THRESHOLD:
    sys.exit(1)  # 挂 CI

.github/workflows/regression.yml

name: llm-regression
on:
  pull_request:
    paths:
      - "prompts/**"
      - "src/llm/**"

jobs:
  regression:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install -r requirements.txt
      - run: python scripts/run_regression.py
        env:
          LANGFUSE_HOST:       ${{ secrets.LANGFUSE_HOST }}
          LANGFUSE_PUBLIC_KEY: ${{ secrets.LANGFUSE_PUBLIC_KEY }}
          LANGFUSE_SECRET_KEY: ${{ secrets.LANGFUSE_SECRET_KEY }}
          OPENAI_API_KEY:      ${{ secrets.OPENAI_API_KEY }}

这个 workflow 只在 prompt 和 LLM 相关代码改动时才触发,避免纯前端 PR 浪费 LLM 额度。

基线(baseline)进 Git

CI 除了"有没有跌破绝对阈值",还需要"比上次降没降"。做法:把每个 PR 的 run 结果写到 benchmarks/latest.json,commit 进主干。新 PR 的 run 跟这个文件 diff:

import json

baseline = json.load(open("benchmarks/latest.json"))
current = {"avg_match": avg, "run_name": run_name}

# 允许 2% 容忍, 超出就挂
if current["avg_match"] < baseline["avg_match"] - 0.02:
    print(f"REGRESSION: {baseline['avg_match']:.3f} -> {current['avg_match']:.3f}")
    sys.exit(1)

# 通过则更新基线(仅在 main 分支 push 时)
if os.environ.get("GITHUB_REF") == "refs/heads/main":
    json.dump(current, open("benchmarks/latest.json", "w"))

别把基线当绝对真理
模型升级、prompt 变更都会带来合理的"不同",不一定是回归。基线降了应该触发人工 review,而不是自动挡人。把它当成 notification,不是 absolute block。

运营侧:QA 修正 expected_output

生产圈进来的 item 经常 expected_output 是错的。UI 里有 Annotation Queue,推荐流程:

自动化任务:每天把前一天低分 trace 批量加入 dataset,metadata 打 needs_review=true
UI 里 QA 点进 item,能看到关联的原 trace、模型原输出、当前 expected
QA 修改 expected_output 后,把 metadata 改成 reviewed=true
CI 脚本只对 reviewed=true 的 item 跑,保证回归集质量

Dataset 的版本治理

dataset 本身没有版本号,只有"你自己怎么命名"这一条约定。建议用后缀标注:

customer-bot-regression-v1:主回归集,稳定
customer-bot-edge-cases-v1:边界 case,专攻怪异输入
customer-bot-synthetic-v1:合成数据,量大但质量一般
customer-bot-holdout-v1:holdout 集,永远不给 LLM-as-Judge 看,防止污染

重大结构变更(比如 schema 改了)就开 v2,旧 v1 保留用于长期趋势对比。

本章小结

    Dataset = LLM 应用的"测试夹具",item 结构 {input, expected_output, metadata}
三种来源:手工冷启动、生产 trace 圈、LLM 合成,混着用最稳
with item.run(run_name) 自动把本次调用挂到 dataset run 上,UI 可对比
CI 集成:脚本跑 regression,低于阈值或跌破基线就挂
基线文件进 Git,允许 2-5% 容忍带,避免模型正常波动被误判
生产 trace 圈进的 item 要走 QA annotation 流程,保证金标可靠

  

Datasets:生产 trace → 回归集 → CI 闸门

Dataset 是什么

创建 dataset 的三个来源

① 手工挑几十条(冷启动)

② 从生产 trace 圈(最常用)

③ 合成(LLM 生成)

跑一次 run

多 run 对比:A/B 跑 prompt 与模型

CI 集成:回归闸门

脚本 scripts/run_regression.py

.github/workflows/regression.yml

基线(baseline)进 Git

运营侧:QA 修正 expected_output

Dataset 的版本治理

本章小结