Datasets：评估数据集的创建与管理

把数据集想成一张表就不会乱。Dataset 是表，Example 是其中一行。每一行（Example）至少有 inputs——一个 dict，结构和你被测函数 / chain 的入参对齐；通常还带 outputs——参照答案（reference / gold），评估时拿它和模型实际输出比对。inputs 和 outputs 都是 自由 schema 的 dict，键名你自己定，但同一数据集内要保持一致，否则 evaluator 取不到字段。此外每行可挂 metadata（来源、难度、标签等任意附加信息）和 split（train / test 等子集标签）。数据集本身有 name（在你的 workspace 内唯一）和 id（UUID）。

环境准备：装 SDK、配 Key

数据集操作走 LangSmith 的 Client，本质是 REST API 的 Python 封装，不依赖 LangChain，纯 SDK 就能跑。需要的只有一个 API Key 和（可选）endpoint 环境变量。

# 安装官方 SDK（数据集 / 评估都在这个包里）
pip install -U langsmith

# 认证：去 https://smith.langchain.com 拿 API Key
export LANGSMITH_API_KEY="lsv2_pt_..."
# 自托管 / 切区时才需要改 endpoint，默认就是 SaaS 美国区
export LANGSMITH_ENDPOINT="https://api.smith.langchain.com"   # EU 区用 https://eu.api.smith.langchain.com

① 建集 + 批量写样例：create_dataset / create_examples

最常见的起手式：先 create_dataset 开一张空表，再 create_examples 一次性把所有样例灌进去。create_examples 接受并行数组（inputs=[...]、outputs=[...]）或对象数组，批量写远快于循环逐条 create_example，因为是一次 HTTP 请求。

from langsmith import Client

client = Client()  # 自动读 LANGSMITH_API_KEY / LANGSMITH_ENDPOINT

DATASET_NAME = "qa-regression-set"

# 1) 建集：name 在 workspace 内唯一。重复建同名会报错，先做存在性判断更稳妥
if client.has_dataset(dataset_name=DATASET_NAME):
    dataset = client.read_dataset(dataset_name=DATASET_NAME)
else:
    dataset = client.create_dataset(
        dataset_name=DATASET_NAME,
        description="问答质量回归测试集：input=问题, output=参照答案",
    )
print("dataset id:", dataset.id)

# 2) 批量写样例：一次 HTTP 请求灌入多条
#    inputs / outputs 都是自由 schema 的 dict，键名自定但全集要一致
client.create_examples(
    dataset_id=dataset.id,
    examples=[
        {
            "inputs": {"question": "地球到月球多远？"},
            "outputs": {"answer": "约 38.4 万公里"},
            "metadata": {"topic": "astronomy", "difficulty": "easy"},
        },
        {
            "inputs": {"question": "水的沸点是多少？"},
            "outputs": {"answer": "标准大气压下 100°C"},
            "metadata": {"topic": "physics", "difficulty": "easy"},
        },
        {
            "inputs": {"question": "光速是多少？"},
            "outputs": {"answer": "约 30 万公里每秒"},
            "metadata": {"topic": "physics", "difficulty": "medium"},
        },
    ],
)
print("样例已写入")

② 从生产 Trace 沉淀真实样本：add_examples_to_dataset

评估数据集最有价值的来源不是拍脑袋编的，而是 线上真实流量。LangSmith 的核心闭环之一就是：在 Tracing 里发现一条翻车 / 边界 case，一键把那条 run 的 inputs（和你修正后的 outputs）沉淀成 Example。UI 上是 run 详情页的 Add to Dataset 按钮；SDK 里则是先用 list_runs 捞出目标 run，再 add_examples_to_dataset 批量写回。

from langsmith import Client

client = Client()
DATASET_NAME = "qa-regression-set"
dataset = client.read_dataset(dataset_name=DATASET_NAME)

# 从某个被追踪的项目里，捞出最近一批真实 run（这里筛根 run）
runs = list(
    client.list_runs(
        project_name="my-prod-app",   # 你 Tracing 时用的项目名
        is_root=True,                 # 只要顶层 trace，不要子 span
        limit=20,
    )
)

# 把这些真实 run 的输入/输出沉淀成数据集样例
# add_examples_to_dataset 接受并行数组：inputs / outputs / source_run_ids
client.add_examples_to_dataset(
    dataset_id=str(dataset.id),
    inputs=[r.inputs for r in runs],
    # 用 run 当时的实际输出当 gold；若要人工修正，这里替换成审核后的答案
    outputs=[(r.outputs or {}) for r in runs],
    # 记录每个样例来自哪条 trace，便于回溯（可选但强烈建议）
    source_run_ids=[str(r.id) for r in runs],
)
print(f"已从生产 trace 沉淀 {len(runs)} 条样例")

③ 按 splits 划分 train / test

同一个数据集常常要切成多份子集：少量 train 用来调 prompt / few-shot，大头 test 留作回归。LangSmith 用 splits 实现——它本质是给 Example 打的标签（一条样例可属于多个 split）。划分用 update_examples 写 splits，读取时 list_examples(splits=[...]) 即可只取某一份。

from langsmith import Client

client = Client()
DATASET_NAME = "qa-regression-set"

# 先把所有样例拉出来
all_examples = list(client.list_examples(dataset_name=DATASET_NAME))
print("总样例数:", len(all_examples))

# 简单按序 8/2 切分：前 80% 进 train，后 20% 进 test
cut = int(len(all_examples) * 0.8)
train_ids = [ex.id for ex in all_examples[:cut]]
test_ids = [ex.id for ex in all_examples[cut:]]

# 用 update_examples 批量打 split 标签
# 关键参数：example_ids 与 splits 并行对齐，每个元素是该样例的 split 列表
client.update_examples(
    example_ids=train_ids,
    splits=[["train"]] * len(train_ids),
)
client.update_examples(
    example_ids=test_ids,
    splits=[["test"]] * len(test_ids),
)
print(f"已划分 train={len(train_ids)}  test={len(test_ids)}")

# 读取：list_examples 用 splits 过滤，只拿你要的子集
from langsmith import Client

client = Client()
DATASET_NAME = "qa-regression-set"

# 只取 test 子集（评估回归时用这一份，保证和调参用的 train 不重叠）
test_set = list(
    client.list_examples(dataset_name=DATASET_NAME, splits=["test"])
)
for ex in test_set:
    # 每个 Example 暴露 inputs / outputs / metadata / id / dataset_id
    print(ex.id)
    print("  inputs :", ex.inputs)    # dict, 喂给被测应用
    print("  outputs:", ex.outputs)   # dict, 参照答案(gold)

# 也能按 metadata 过滤（如只取 difficulty=medium 的难样例）
hard = list(
    client.list_examples(
        dataset_name=DATASET_NAME,
        metadata={"difficulty": "medium"},
    )
)
print("中等难度样例数:", len(hard))

数据集如何成为 evaluate 的输入源

前面所有工作的回报在下一章兑现：建好的数据集（或它的某个 split）直接作为 evaluate(data=...) 的输入源。evaluate 会遍历数据集里每个 Example，把 example.inputs 喂给你的目标函数，再把目标输出和 example.outputs（gold）一起交给 evaluator 打分。下面是承上启下的最小预览。

# 预览：数据集作为 evaluate 的 data 源（完整讲解见下一章 Evaluation）
from langsmith import Client
from langsmith.evaluation import evaluate

client = Client()

# 被测目标：接收 example.inputs（dict），返回输出（dict）
def my_app(inputs: dict) -> dict:
    # ... 你的真实 LLM/chain 调用，这里用占位 ...
    return {"answer": f"对『{inputs['question']}』的回答"}

# 一个最简 evaluator：拿 run 输出与 example 参照答案比对
def exact_match(run, example) -> dict:
    pred = (run.outputs or {}).get("answer", "")
    gold = (example.outputs or {}).get("answer", "")
    return {"key": "exact_match", "score": 1.0 if gold in pred else 0.0}

# data 直接传数据集名；也可传 list_examples(splits=["test"]) 的结果只评 test 子集
results = evaluate(
    my_app,
    data=client.list_examples(dataset_name="qa-regression-set", splits=["test"]),
    evaluators=[exact_match],
    experiment_prefix="baseline",   # 实验前缀，UI 里据此聚合一次运行
)
print("评估完成，去 LangSmith UI 看 experiment 结果")

能被度量的问题才能被完成。数据集就是度量的基线——没有可复现的对照样例，评估只是又一次拍脑袋；建好集，评估才从「感觉变好了」变成「test 集上分数从 0.78 涨到 0.91」。

— Karpathy Lens