第11课：LangSmith 数据集Datasets创建、导入、标注与版本管理实战

文章目录

• • 一、开篇导读
  • 二、知识前置铺垫
  • • 2.1 数据集在LLM应用评估中的核心价值
    • 2.2 Dataset、Example与Input/Output的设计哲学
    • 2.3 数据集全生命周期工作流
    • 2.4 Annotation Queue与数据的“人工标注回路”
  • 三、核心概念精讲
  • • 3.1 Dataset与Example的CRUD操作
    • 3.2 数据集的创建来源
    • 3.3 数据集的版本控制机制
    • 3.4 数据集的标记系统
    • 3.5 Annotation Queue中的批量标注与数据回流
  • 四、原理底层剖析
  • • 4.1 数据集中Example的存储模型
    • 4.2 版本控制的实现机制
    • 4.3 Annotation Queue的分发与存储
    • 4.4 Schema校验的实现路径
  • 五、环境配置手把手实战
  • • 5.1 确认LangSmith追踪配置
    • 5.2 安装必要依赖
    • 5.3 验证环境与权限
    • 5.4 准备示例数据文件（CSV格式）
    • 5.5 UI：创建你的第一个Dataset
  • 六、完整可运行代码案例
  • • 6.1 基础创建：通过Python SDK构建结构化数据集
    • 6.2 批量导入：从CSV文件批量创建数据集
    • 6.3 从现有Runs构建数据集
    • 6.4 单示例精细查询与增量更新
    • 6.5 版本快照与标记
    • 6.6 完整的评测闭环：数据集驱动的自动化评估
    • 6.7 Annotation Queue引导人工标注
  • 七、代码逐行详解
  • • 7.1 `create_dataset_basic.py`核心逻辑
    • 7.2 `upload_csv_dataset.py`列映射
    • 7.3 `evaluation_workflow.py`评估调用
  • 八、常见坑点与避坑指南
  • • 8.1 CSV列命名不当导致映射失败
    • 8.2 误解版本控制的行为
    • 8.3 版本标记时间戳粒度与冲突
    • 8.4 Annotation Queue创建过多导致资源浪费
    • 8.5 缺少输出字段导致自动评估失败
    • 8.6 高并发批量导入时触达速率限制
    • 8.7 数据集跨Project污染全局ID引用
  • 九、企业级落地最佳实践
  • • 9.1 数据集驱动的CI/CD评测流水线
    • 9.2 数据集的生命周期命名规范
    • 9.3 Annotation Queue的人工审核运营
    • 9.4 多租户环境下避免数据冲突
    • 9.5 数据集金丝雀发布策略
    • 9.6 禁用冗余历史版本节省存储
  • 十、本节知识点总结
  • 十一、课后思考练习题
  • • 练习题1：理论理解
    • 练习题2：动手实践
    • 练习题3：场景设计
    • 练习题4：源码分析（选做）
  • 🔗《20节课 LangSmith 从入门到精通》系列课程导航

一、开篇导读

欢迎来到《LangSmith从入门到精通》专栏的第11节课。回顾前10节课的学习旅程，我们从LangSmith的整体认知出发，完成了账号注册、环境搭建、核心概念拆解、快速入门实战，再到环境变量配置、全链路监控、LCEL链路拆解、面板功能精讲，以及标签元数据的精细化管理。至此，你已经掌握了LangSmith追踪能力的全部核心技能——数据能够稳定上报、链路能够清晰展示、Tag和Metadata能够按业务维度精细化分类。

然而，当你的应用从开发阶段走向持续迭代优化时，一个根本性的问题浮现出来：如何系统性地评估和改进模型效果？

想象这样的场景：你花了两周时间优化了一版Prompt，信心满满地部署到生产环境，上线后发现回答质量不仅没有提升，反而出现了之前从未有过的新问题。为什么会发生这种事？因为你从未对模型输出进行过系统的质量度量。缺少标准答案、缺少自动化评估手段的大模型项目，本质上就是盲人摸象——每个人凭感觉说“效果变好了”或“变差了”，没有人能拿出可量化的证据。

这正是LangSmith数据集系统要解决的核心问题。通过将用户问题、期望答案（标准答案）组织成标准化数据集，你可以批量运行评测任务，自动化对比不同模型版本、不同Prompt配置的效果差异。更重要的是，LangSmith的数据集是自动版本化的——每一次增删改操作都会生成一个新版本，你可以精确地知道“v3版本的训练数据到底是哪一天的快照”，在任意版本上进行评测，并与其他版本做差分对比。这使得模型的回归测试和迭代效果验证成为可能。

本节课你将收获：

• 数据集概念体系：深入理解Dataset、Example、Input/Output/Output Reference的层级结构
• 四大创建方式：手动创建、从现有Runs转换、CSV/JSON批量导入、克隆公共数据集
• 版本管理与评测闭环：理解LangSmith自动版本化机制，学会标记里程碑版本并在评估中指定版本
• Python SDK全掌握：使用Client类的create_dataset/create_examples/upload_csv等常用方法进行程序化管理
• 标注队列实战：建立从生产数据筛选→人工QA标注→修正后合并扩充数据集的完整工作流
• 企业级实验流程：构建数据集驱动的评测CI/CD，用数据驱动Prompt迭代

二、知识前置铺垫

2.1 数据集在LLM应用评估中的核心价值

为什么你需要一套系统的数据集管理方案？因为LLM的输出具有非确定性。传统软件测试中，输入1+1，期望输出2，可以用assert expected == actual完成断言。但在LLM应用中，同一个问题“LangSmith是什么”，大模型可以生成至少十几种不同但都正确的表达方式。单纯靠人工一条条验证无法规模化，必须依靠自动化评估体系。

LangSmith的数据集系统正是为此而生。它提供了一套完整的数据管理和评测基础设施：你可以将问题和期望答案组织成结构化的数据集；在每次Prompt调优后，在该数据集上批量运行评测任务；系统自动计算准确率、相似度等多个评估指标；通过对比不同版本的实验结果，你可以数据驱动地判断这次改动到底是“变好了”还是“变差了”。数据集与评估（Evaluation）紧密集成——数据集在LangSmith中不仅是存储测试用例的容器，更是配合评估工具进行自动化回归测试和模型迭代验证的核心工作台。

2.2 Dataset、Example与Input/Output的设计哲学

LangSmith的数据集体系包含两层核心概念：

• Dataset（数据集）：是Example（示例）的逻辑容器，对应现实世界中的一个测试集。例如“客服问答测试集”、“情感分类验证集”，每个Dataset可以包含成百上千个Example。

• Example（示例）：是Dataset中的单条数据记录，通常包含三个关键字段：

这种设计的精妙之处在于灵活性与标准化的最佳平衡。Inputs作为评估时实际喂给大模型的参数；Outputs与Output Reference则为不同评估策略提供支撑——Outputs可作为基准参考，Output Reference则更适合高精度匹配评估。

2.3 数据集全生命周期工作流

在实际开发中，数据集的存在贯穿整个持续迭代的生命周期：

这一闭环工作流正是LangSmith数据集系统设计的核心价值所在。

2.4 Annotation Queue与数据的“人工标注回路”

自动化评估虽然高效，但无法保证100%准确。当自动化评估给出的置信度不够高，或你需要构建更高精度的“黄金测试集”时，人工标注就成为必不可少的环节。

Annotation Queue（标注队列）为此而生。它提供一个专门的视图，供人工标注员对筛选出的Runs进行细化审核和打分。标注完成后，这些带标注反馈的数据可以一键导出为新示例或直接合并回数据集，实现“从生产数据发现问题→人工标注修正→扩充数据集→重新评测”的闭环。

三、核心概念精讲

3.1 Dataset与Example的CRUD操作

与前10节课中使用的Runs和Traces追踪体系不同，数据集模块提供了独立的CRUD接口。每个Dataset由唯一的ID和名称标识，管理方式与Project的组织逻辑类似。通过Python SDK或Web控制台，你可以随时增删改查数据集及其中的示例。

3.2 数据集的创建来源

LangSmith支持四种Dataset创建来源，满足从研究探索到生产监控的各种需求：

其中“从Runs转换”是LangSmith独特的能力：当你部署生产监控后，可以将线上发生的异常Trace一键转换成Dataset中的Example，实现从真实Case到离线回归测试的自动化流转。

3.3 数据集的版本控制机制

数据集版本控制是整个系统中极为精妙的一环。在传统的测试数据管理方式中，你通常需要手动维护“v1.csv”“v2.csv”等散落在各地的文件副本，不方便追溯且极易出错。

LangSmith与之不同：数据集自动版本化。每当你对Dataset执行添加、更新或删除Example的操作时，系统自动创建一个新的数据集版本。版本的标识基于操作发生的时间戳。在UI中，你需要通过点击“示例”选项卡中的具体时间戳来查看某个历史时刻的数据集快照。

这一设计带来三大核心价值：

• 完整的可审计轨迹：你可以回答“上周二执行评估时用的是哪一版数据”这类问题
• 评测的可复现性：在评估任务中通过dataset_version参数指定版本，确保不同实验之间测评基准相同
• 无损的历史探索：查看旧版本数据时，Example是只读的，永远不会误触污染历史基准

此外，针对生产级数据集的稳定迭代，LangSmith还支持数据集模式（Dataset Schema）定义。你可以为数据集预先指定Inputs和Outputs的数据结构，所有新增Example必须遵循该结构，确保数据质量的一致性。

3.4 数据集的标记系统

除了基于时间戳的自动版本化外，LangSmith还支持标记版本（Tag Version） ，为数据集添加更具可读性的语义名称。在生产中，你可能需要标准化地将某个稳定状态的数据集版本标记为“prod”，并确保所有在线评估任务只引用这个标记版本，避免业务方无意中用到了“进行中”的最新数据。

3.5 Annotation Queue中的批量标注与数据回流

Annotation Queue是数据集的“上游工厂”。基于明确的审查规则，标注员对聚集的任务逐个进行质量评分和内容修正。标注完成后，你可以根据需要将这些已评分数据追加为Dataset的新Example，也可以与已存数据进行冲突合并。最终形成一条“生产问题发现→人工修正→格式导入→离线评测”的数据闭环。

四、原理底层剖析

4.1 数据集中Example的存储模型

在LangSmith的后端数据库中，Example可以理解为独立于Project而存在的实体表。examples表中每条记录包含inputs（输入）、outputs（输出）、reference（参考）、dataset_id等多字段。每当调用create_examples()方法时，系统会将这些结构化的JSON字段持久化，并触发生成数据集新版本的程序逻辑。

4.2 版本控制的实现机制

LangSmith的实现并非简单地在数据库层面克隆全部Example副本，而是一种“差异快照+变更日志”的混合实现方式。数据库层仅为每次操作存储变更diff，应用层计算需要时才聚合回滚完整的快照。这种方案既保证了大规模数据集存储成本可控，又维持了任意时间点上数据精准复现的能力。

4.3 Annotation Queue的分发与存储

当开发者使用Annotation Queue时，UI中的数据来源于对特定Run集合的一层逻辑封装。标注员对Run的每一次打分和评语都通过Feedback API写回LangSmith平台。若你决定将所有已标注的Run导出为新的Example，LangSmith会读取Run的inputs和outputs字段，将其转换为符合Dataset结构的数据行，并挂载到指定的Dataset下完成“生产数据”到“测评数据”的转换。

4.4 Schema校验的实现路径

启用Dataset Schema功能时，LangSmith会解析你提供的JSON Schema定义。创建新Example时，后端校验引擎动态验证inputs/outputs的结构是否与Schema定义匹配。Schema版本与数据版本协同演进：若更新Schema，会自动创建数据集新版本，完整保留历史语境。

五、环境配置手把手实战

5.1 确认LangSmith追踪配置

继续使用前几节课配置的.env文件：

# .env
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=lsv2_pt_你的Key
LANGCHAIN_PROJECT=lesson11-dataset-demo
OPENAI_API_KEY=sk-你的Key

5.2 安装必要依赖

确保已安装LangSmith SDK的最新版本：

pip install -U langsmith langchain langchain-openai

5.3 验证环境与权限

运行验证脚本确保API Key有效：

import os
from dotenv import load_dotenv
from langsmith import Client

load_dotenv()
client = Client()

# 测试连接
try:
# 简单操作验证API连通性
print("✅ LangSmith客户端初始化成功")
except Exception as e:
print(f"❌ 连接失败: {e}")

5.4 准备示例数据文件（CSV格式）

创建一个qa_dataset.csv文件，准备用于导入的测试数据：

input.question,input.context,output.answer
什么是LangSmith?,LangSmith是LangChain官方的可观测性平台,LangSmith是用于调试、评估和监控LLM应用的平台
LCEL是什么?,LCEL是LangChain表达式语言,LCEL是一种声明式构建LLM工作流的方法
如何使用LangSmith进行评测?,可以创建数据集并运行评估任务,通过创建测试数据集并配置评估器即可自动评测

注意CSV列的命名规范：LangSmith客户端在上传CSV时，会将列名映射为Example的inputs或outputs字段。

5.5 UI：创建你的第一个Dataset

如果不喜欢命令行工作流，也可以通过UI来完成数据集的初始化：

六、完整可运行代码案例

6.1 基础创建：通过Python SDK构建结构化数据集

本示例展示如何通过Python SDK创建一个包含5个问答对的数据集，并存入LangSmith平台。

# 文件名: create_dataset_basic.py
# 说明: 通过Python SDK创建数据集并批量添加示例

import os
from dotenv import load_dotenv
from langsmith import Client
from typing import Dict, Any

load_dotenv()

def create_qa_dataset():
client = Client()

dataset_name = "basic_qa_test_set"

# 检查数据集是否已存在
try:
existing = client.read_dataset(dataset_name=dataset_name)
print(f"📁 数据集已存在: {existing.name} (ID: {existing.id})")
return existing
except Exception:
pass

# 创建新数据集
dataset = client.create_dataset(
dataset_name=dataset_name,
description="基础问答测试集，用于验证LangSmith数据集功能",
data_type="kv" # kv表示键值对格式
)
print(f"✅ 创建数据集: {dataset.name} (ID: {dataset.id})")

# 准备示例数据
examples = [
{
"inputs": {"question": "LangSmith是什么？"},
"outputs": {"answer": "LangSmith是LangChain官方推出的LLM应用可观测性平台，用于调试、评估和监控大模型应用。"}
},
{
"inputs": {"question": "LCEL是什么有什么优势？"},
"outputs": {"answer": "LCEL是LangChain表达式语言，支持并行执行和流式输出，并能无缝集成LangSmith追踪。"}
},
{
"inputs": {"question": "如何创建LangSmith数据集？"},
"outputs": {"answer": "可以使用Client.create_dataset()方法通过SDK创建，或在UI中手动创建。"}
}
]

for ex in examples:
client.create_example(
inputs=ex["inputs"],
outputs=ex["outputs"],
dataset_id=dataset.id
)
print(f" 📝 添加示例: {ex['inputs']['question']}")

print(f"🎉 数据集创建完成，共 {len(examples)} 个示例")
return dataset

if __name__ == "__main__":
create_qa_dataset()

6.2 批量导入：从CSV文件批量创建数据集

展示如何通过upload_csv方法完成大批量数据的高速导入。

# 文件名: upload_csv_dataset.py
# 说明: 从CSV文件批量创建数据集

import os
import pandas as pd
from io import StringIO
from dotenv import load_dotenv
from langsmith import Client

load_dotenv()

def create_dataset_from_csv(csv_content: str, dataset_name: str):
"""
从CSV字符串创建数据集
"""
client = Client()

# 尝试读取数据集，若不存在则创建
try:
dataset = client.read_dataset(dataset_name=dataset_name)
print(f"📁 数据集已存在: {dataset_name}")
except Exception:
dataset = client.create_dataset(
dataset_name=dataset_name,
description="从CSV文件导入的数据集",
data_type="kv"
)
print(f"✅ 创建新数据集: {dataset_name}")

# 使用upload_csv批量导入
# CSV列名中的input.xxx或output.xxx会自动映射
client.upload_csv(
dataset_id=dataset.id,
csv=csv_content,
input_keys=["input.question", "input.context"], # 指定哪些列作为inputs
output_keys=["output.answer"] # 指定哪些列作为outputs
)
print(f"📊 CSV数据已导入，数据集ID: {dataset.id}")
return dataset

def demonstrate_csv_upload():
# CSV内容示例
csv_data = """input.question,input.context,output.answer,metadata.difficulty
什么是LangSmith?,LangSmith是LangChain官方的平台,LangSmith是LLM应用可观测性平台,简单
LCEL表达式有什么特点?,LCEL支持并行执行和流式输出,LCEL提供声明式语法并可无缝追踪,中等
如何调试RAG应用?,可以使用LangSmith进行全链路追踪,通过LangSmith的Trace树逐层排查问题,困难
"""

dataset = create_dataset_from_csv(csv_data, "csv_imported_dataset")
print("✅ CSV导入完成")

if __name__ == "__main__":
demonstrate_csv_upload()

6.3 从现有Runs构建数据集

该示例在生产环境中尤其有用——你可以将线上出现异常的Trace一键转换为回归测试集的示例。

# 文件名: runs_to_dataset.py
# 说明: 从现有LangSmith Traces快速构建数据集

import os
from dotenv import load_dotenv
from langsmith import Client
from datetime import datetime, timedelta

load_dotenv()

def create_dataset_from_recent_runs(dataset_name: str, hours_back: int = 24, max_runs: int = 50):
"""
从最近指定时间内的一组Runs中抽取数据构建数据集
"""
client = Client()

# 获取近期的Runs（假设Project事先配置好了）
since = datetime.now() – timedelta(hours=hours_back)

runs = list(client.list_runs(
# 可指定project_name筛选特定项目的Runs
start_time=since,
# 只取根Run或包含input的Run
filter='eq(is_root, true)',
limit=max_runs
))

if not runs:
print("⚠️ 未找到符合条件的Runs")
return None

# 创建数据集
try:
dataset = client.create_dataset(
dataset_name=dataset_name,
description=f"从最近{hours_back}小时的生产Runs转换而来，共{len(runs)}个示例"
)
print(f"✅ 创建数据集: {dataset_name}")
except Exception:
dataset = client.read_dataset(dataset_name=dataset_name)
print(f"📁 数据集已存在，将增量添加示例")

converted_count = 0
for run in runs:
# 只保留有有效inputs和outputs的Run
if run.inputs and run.outputs:
client.create_example(
inputs=run.inputs,
outputs=run.outputs,
dataset_id=dataset.id,
# 可选：关联原始Run ID便于追溯
metadata={"source_run_id": run.id}
)
converted_count += 1

print(f"🎉 成功从{len(runs)}个Runs中转换了{converted_count}个示例到数据集")
return dataset

if __name__ == "__main__":
dataset = create_dataset_from_recent_runs("production_converted_dataset", hours_back=72, max_runs=30)

6.4 单示例精细查询与增量更新

当数据集里的某个示例被人工评审为错误时，最好通过SDK精确查询和修正该示例。

# 文件名: query_update_example.py
# 说明: 查询数据集中的具体示例并进行更新

import os
from dotenv import load_dotenv
from langsmith import Client

load_dotenv()

def manage_dataset_examples():
client = Client()
dataset_name = "basic_qa_test_set"

try:
dataset = client.read_dataset(dataset_name=dataset_name)
print(f"📁 数据集: {dataset.name}")
except Exception as e:
print(f"❌ 数据集不存在: {e}")
return

# 获取数据集中的所有示例
examples = list(client.list_examples(dataset_id=dataset.id))
print(f"📋 数据集中共有 {len(examples)} 个示例")

# 查找特定问题的示例
target_question = "LangSmith是什么？"
target_example = None
for ex in examples:
if ex.inputs and ex.inputs.get("question") == target_question:
target_example = ex
break

if target_example:
print(f"🔍 找到示例: {target_example.inputs.get('question')}")
print(f" 原输出: {target_example.outputs.get('answer')}")

# 更新示例
updated_output = "LangSmith是LangChain官方提供的LLM应用可观测性与评估平台。"
client.update_example(
example_id=target_example.id,
outputs={"answer": updated_output}
)
print(f"✅ 已更新示例输出")

# 删除不再需要的示例
for ex in examples:
if "LCEL" in ex.inputs.get("question", ""):
print(f"🗑️ 删除示例: {ex.inputs.get('question')}")
client.delete_example(example_id=ex.id)
break

# 新增一个示例
new_example = client.create_example(
inputs={"question": "LangSmith如何帮助调试Agent？"},
outputs={"answer": "LangSmith提供全链路追踪树，可查看Agent的每一次思考和工具调用过程。"},
dataset_id=dataset.id
)
print(f"✨ 新增示例: {new_example.inputs.get('question')}")

if __name__ == "__main__":
manage_dataset_examples()

6.5 版本快照与标记

数据集的版本管理是LangSmith区别于普通文件存储的关键差异。学会使用版本标记，才能真正发挥数据集在持续迭代中的价值。

# 文件名: version_management.py
# 说明: 数据集版本管理与标记示例

import os
from datetime import datetime, timedelta
from dotenv import load_dotenv
from langsmith import Client

load_dotenv()

def demo_dataset_versioning():
client = Client()
dataset_name = "versioned_qa_dataset"

# 创建数据集
try:
dataset = client.read_dataset(dataset_name=dataset_name)
print(f"📁 找到已存在数据集: {dataset.name}")
except Exception:
dataset = client.create_dataset(
dataset_name=dataset_name,
description="演示版本管理的数据集"
)
print(f"✅ 创建数据集: {dataset_name}")

# 初始一批示例
client.create_examples(dataset_id=dataset.id, examples=[
{"inputs": {"q": "A"}, "outputs": {"a": "Answer A"}},
{"inputs": {"q": "B"}, "outputs": {"a": "Answer B"}},
])
print("📝 初始示例集 = 2个")

# 模拟等待（实际运行时可根据需要留出版本时间差）
import time; time.sleep(1)

# 添加更多数据集示例（触发新版本）
client.create_examples(dataset_id=dataset.id, examples=[
{"inputs": {"q": "C"}, "outputs": {"a": "Answer C"}},
])
print("📝 新增示例后 → 共3个")

# 使用client.list_dataset_versions获取历史版本
versions = list(client.list_dataset_versions(dataset_id=dataset.id))
print(f"📋 历史版本数: {len(versions)}")
for v in versions[:3]:
print(f" – 版本时间: {v.as_of}, 快照大小: {v.count}")

# 标记某个特定版本
if len(versions) >= 2:
target_version = versions[0] # 第一个版本(最早的快照)
client.update_dataset_tag(
dataset_id=dataset.id,
as_of=target_version.as_of,
tag="stable_v1"
)
print(f"✅ 已将数据集快照标记为 stable_v1")

# 获取指定tag的数据集版本用于评估
tagged_version = client.read_dataset_version(dataset_id=dataset.id, tag="stable_v1")
if tagged_version:
print(f"🔖 获取到标记版本: {tagged_version.as_of}")

if __name__ == "__main__":
demo_dataset_versioning()

6.6 完整的评测闭环：数据集驱动的自动化评估

这是一套完整的端到端工作流——创建数据集、运行评估、对比分析结果。

# 文件名: evaluation_workflow.py
# 说明: 使用数据集进行自动化评估的完整示例

import os
from dotenv import load_dotenv
from langsmith import Client
from langsmith.evaluation import evaluate
from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate

load_dotenv()

def create_benchmark_dataset():
"""创建基准测试数据集"""
client = Client()
dataset_name = "benchmark_qa_set"

try:
dataset = client.read_dataset(dataset_name=dataset_name)
print(f"📁 使用现有数据集: {dataset_name}")
except Exception:
dataset = client.create_dataset(
dataset_name=dataset_name,
description="用于自动化评估的基准数据集",
data_type="kv"
)
client.create_examples(dataset_id=dataset.id, examples=[
{
"inputs": {"question": "什么是大语言模型？"},
"outputs": {"answer": "大语言模型是基于海量数据训练的大型神经网络，能理解和生成自然语言。"}
},
{
"inputs": {"question": "LangSmith的主要功能有哪些？"},
"outputs": {"answer": "LangSmith提供全链路追踪、数据集管理、自动化评估等核心功能。"}
}
])
print(f"✅ 创建基准数据集: {dataset_name}")

return dataset

def target_model(inputs: dict) –> dict:
"""被评估的目标模型"""
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
prompt = PromptTemplate.from_template("请回答：{question}")
chain = prompt | llm | (lambda x: {"answer": x.content})
return chain.invoke(inputs)

def evaluate_model_on_dataset():
"""在数据集上运行模型评估"""
dataset = create_benchmark_dataset()

# 运行评估实验
experiment_results = evaluate(
target_model,
data=dataset.name,
evaluators=[
"qa", # 使用内置的问答评估器
"embedding_distance" # 基于Embedding相似度
],
experiment_prefix="benchmark_v1",
metadata={"version": "1.0", "model": "gpt-3.5-turbo"}
)

print(f"\\n🎯 评估完成！实验名称: {experiment_results['experiment_name']}")
print("💡 在LangSmith控制台的Tests选项卡中查看详细评估结果")
return experiment_results

if __name__ == "__main__":
evaluate_model_on_dataset()

6.7 Annotation Queue引导人工标注

该Agent-like脚本展示如何从标注测试队列获取人工反馈的流程。

# 文件名: annotation_queue_demo.py
# 说明: 演示Annotation Queue的使用流程

import os
from dotenv import load_dotenv
from langsmith import Client

load_dotenv()

def demo_annotation_queue():
client = Client()

# 创建一个新的标注队列
try:
queue = client.create_annotation_queue(
name="质量审核队列",
description="用于审核LangSmith自动化评估中的低分样本"
)
print(f"✅ 创建标注队列: {queue.name} (ID: {queue.id})")
except Exception:
# 获取已存在的队列或继续
queues = list(client.list_annotation_queues())
if queues:
queue = queues[0]
print(f"📁 使用现有队列: {queue.name}")
else:
print("❌ 队列创建失败")
return

# 获取队列中待审查的Runs
runs_in_queue = list(client.list_runs(annotation_queue_id=queue.id))
print(f"队列中待审核的Runs数: {len(runs_in_queue)}")

# 模拟添加Runs到队列（真实场景中Runs自动根据过滤规则推入）
# 此处演示手动推入方式
recent_runs = list(client.list_runs(limit=5))
for run in recent_runs[:2]:
client.add_runs_to_annotation_queue(queue_id=queue.id, run_ids=[run.id])
print(f" ➕ 将Run {run.id} 加入标注队列")

# 统计队列信息（新增后重新读取）
updated_runs = list(client.list_runs(annotation_queue_id=queue.id))
print(f"添加后队列中待审核Run总数: {len(updated_runs)}")

if __name__ == "__main__":
demo_annotation_queue()

七、代码逐行详解

7.1 create_dataset_basic.py核心逻辑

dataset = client.create_dataset(
dataset_name=dataset_name,
description="基础问答测试集…",
data_type="kv"
)

data_type="kv"表示数据集的存储格式为键值对。创建完成后返回的dataset对象包含id和name属性，供后续添加示例时引用。

examples = [...]
client.create_examples(dataset_id=dataset.id, examples=examples)

create_examples支持批量插入，比循环调用create_example效率更高。

7.2 upload_csv_dataset.py列映射

client.upload_csv(
dataset_id=dataset.id,
csv=csv_content,
input_keys=["input.question", "input.context"],
output_keys=["output.answer"]
)

upload_csv是LangSmith处理大批量数据的核心入口，支持动态映射CSV列名到Example的inputs/outputs路径，减少手动解析工作量。

7.3 evaluation_workflow.py评估调用

experiment_results = evaluate(
target_model,
data=dataset.name,
evaluators=["qa", "embedding_distance"],
experiment_prefix="benchmark_v1"
)

evaluate函数会自动遍历数据集的每一个Example，调用目标模型生成输出，再调用指定的评估器计算结果。experiment_prefix作为实验命名的前缀，便于在LangSmith的Tests选项卡中区分不同实验。

八、常见坑点与避坑指南

8.1 CSV列命名不当导致映射失败

症状：执行upload_csv()后数据集为空，或Inputs/Outputs字段与预期不符。

原因：LangSmith要求CSV列名采用input.xxx或output.xxx格式。若列名仅用question、answer而非input.question、output.answer，LangSmith将无法进行模式识别。

解决方案：确保CSV第一行列名遵循input.xxx和output.xxx的用户标准化结构。

8.2 误解版本控制的行为

坑点：部分用户以为删除集中的一个示例不会影响已在其他实验中的版本，但对历史实验引用的数据版本具有不可变性；删除当前版本的Example会创建新版本。已引用旧版本的实验仍会保留旧快照，但新评测会默认引用最新版本。

8.3 版本标记时间戳粒度与冲突

坑点：update_dataset_tag中必须精确提供时间戳（as_of），若该时间戳无对应版本会报错。

解决方案：建议先从list_dataset_versions中获取确切的as_of值，再进行标记。

8.4 Annotation Queue创建过多导致资源浪费

坑点：若不设流转规则，Annotation Queue会无限堆积，造成审查效率低下。

解决方案：在队列设置中配置“自动出队”流转逻辑，标注完成的数据自动合并到Dataset后移出队列。

8.5 缺少输出字段导致自动评估失败

坑点：QA评估器要求必须提供output字段作为参考答案，否则评估失败。

解决方案：确保创建数据集时每个Example都包含outputs字段。

8.6 高并发批量导入时触达速率限制

坑点：LangSmith对各租户有API频率限制。一次性大批量写入会触发HTTP 429。

解决方案：实现有退避策略的重试循环，或利用upload_csv等方法内部已经做好了批量缓冲优化。

8.7 数据集跨Project污染全局ID引用

坑点：数据集ID在Project层面是全局唯一的，但部分开发者误以为数据集可以像Project一样由不同用户完全隔离。

解决方案：理解组织内数据集资源的开放性，做好命名空间设计保障数据集的逻辑隔离。

九、企业级落地最佳实践

9.1 数据集驱动的CI/CD评测流水线

在企业的LLM应用迭代中，建议将数据集评估纳入CI流程：

# GitHub Actions伪代码
steps:
– name: 拉取最新模型代码
– name: 在LangSmith基准数据集上执行评估
run: |
python run_evaluation.py \\
–dataset "核心功能测试集" \\
–model-version ${GITHUB_SHA}
– name: 获取评估结果分数
if: 分数低于阈值
run: exit 1 # 阻止低质量模型合并

9.2 数据集的生命周期命名规范

为每个数据集建立清晰的命名体系：{领域}_{用途}_{版本状态}

• 客服系统_意图识别_v1
• 金融领域_Prompt优化_golden_set

9.3 Annotation Queue的人工审核运营

企业可以搭建半自动化的审核流程：每日凌晨定时任务将生产环境低置信度Traces自动推送到标注队列→QA团队次日逐条审查打分→审查通过的Example自动追加到Golden Dataset→触发新一轮离线评测。不仅大幅提升评测数据的质量，也持续驱动模型能力的自我进化。

9.4 多租户环境下避免数据冲突

生产与非生成环境下建议使用不同的Dataset名称前缀（如prod_、staging_）。通过SDK的list_datasets配合租户ID进行动态名称选择，避免重要评测基准被意外写入。

9.5 数据集金丝雀发布策略

对于已锚定为“黄金评测集”的数据，在发布新版本模型时采用金丝雀灰度策略：先在小流量数据集上验证模型效果，通过后再用全量核心测试集。LangSmith的版本标记功能为此提供了精确控制。

9.6 禁用冗余历史版本节省存储

数据集不断积累会持续增加存储成本。利用list_dataset_versions和diff_dataset_versions检查版本变更，定期删除无需保留的中间版本。

十、本节知识点总结

核心技术链条回顾： 创建Dataset → 导入/追加Examples → 标记关键版本 → 配置评估器 → 执行evaluate() → 分析实验结果 → 通过Annotation Queue收集人工反馈 → 修正/扩充Dataset → 继续迭代评估

十一、课后思考练习题

练习题1：理论理解

1.1 简述LangSmith中Dataset与Project在数据维度的重要差别，以及各自适用的业务场景。

1.2 解释自动版本控制对评估复现性的实际价值。如果在生产评测中引用一个经常变动的未标记版本，会产生什么后果？

1.3 标注队列（Annotation Queue）的引入如何打通从生产监控到离线评估的全链路数据闭环？

练习题2：动手实践

2.1 基于本节示例，编写一个脚本从外部JSON文件导入200条问答对，为每个输入/输出字段增加清晰Schema定义。

2.2 使用评估框架（evaluate）对LangChain应用在某一自定义数据集上进行效果测试，输出错误分析日志。

2.3 创建一个人工标注队列，选择性的将某些历史比较可疑的查询推送给人工审核。

练习题3：场景设计

3.1 电商智能客服系统在每日真实对话中发现一些问题，请你设计一套基于annotation queue的数据回流方案，实现从“用户反馈”到“新测试样本”的增量循环。

3.2 某公司AI团队希望针对一个2000条数据的核心评估集进行版本标记，以支撑线上金丝雀验证和离线评估。请给出LangSmith方案。

练习题4：源码分析（选做）

4.1 阅读LangSmith Python SDK中关于版本控制与版本差分的实现伪代码，分析list_dataset_versions与diff_dataset_versions的元数据差异点。

4.2 研究LangSmith Evaluate中evaluators参数内置的qa和embedding_distance的实现策略，思考在实际项目里扩充自定义评估映射的方法。

下节课预告：

第12节课我们将建立LLM应用评测体系，结合LangSmith的数据集功能为我们核心的问答、RAG以及Agent场景设计高级评估器。届时，你不仅能够持续跟踪模型效果的变化，还能对不同实验组进行定量对比，让每一次Prompt优化都有据可循！

下一节课见！

🔗《20节课 LangSmith 从入门到精通》系列课程导航

去订阅

🌟 感谢您耐心阅读到这里！ 💡 如果本文对您有所启发欢迎： 👍 点赞📌 收藏 📤 分享给更多需要的伙伴。 🗣️ 期待在评论区看到您的想法, 共同进步。 🔔 关注我，持续获取更多干货内容～ 🤗 我们下篇文章见～