解析 LangSmith 的 ‘Dataset Backtesting’：如何利用历史生产数据对新版 Graph 进行离线回归测试？

各位同仁，欢迎来到今天的讲座。在人工智能，特别是大型语言模型（LLM）驱动的应用开发领域，我们正经历着前所未有的创新浪潮。我们构建的系统不再是简单的函数调用，而是复杂的、多步骤的、有时甚至是高度非确定性的“图”（Graph）结构，它们能够执行检索、推理、规划乃至自主行动。这种复杂性带来了巨大的能力，但也带来了同样巨大的挑战：如何确保这些快速迭代的“图”在每次更新后都能保持甚至超越原有的性能和鲁棒性，尤其是在面对真实世界的生产数据时？

传统的软件测试方法，如单元测试和集成测试，在LLM应用中显得力不从心。LLM的非确定性、对提示词的敏感性以及输出的开放性，使得“预期输出”的定义变得模糊。我们不能简单地断言一个回答是“对”或“错”，而是需要从多个维度评估其质量：相关性、连贯性、准确性、安全性等。

今天，我们将深入探讨LangSmith提供的一个强大功能——“数据集回溯测试”（Dataset Backtesting），以及如何利用它来对我们新开发的或修改的Graph进行离线回归测试，确保在部署到生产环境之前，新版本能够稳健运行，并达到我们预期的效果。我们将重点关注如何利用历史生产数据，这是一种极其宝贵的资源，它代表了用户真实的请求和系统的真实行为。

1. 为什么我们需要离线回归测试？LLM应用的独特挑战

在深入LangSmith的具体机制之前，我们首先要理解为什么离线回归测试对于LLM驱动的Graph至关重要。

1.1 LLM应用的特性与测试困境

• 非确定性： 即使使用相同的输入和提示词，LLM的输出也可能略有不同。这使得基于精确匹配的单元测试变得困难。
• 复杂性： 实际的LLM应用通常是多阶段的，涉及检索增强生成（RAG）、工具使用（Tool Use）、多步推理等。一个微小的改动可能在整个Graph的不同环节产生连锁反应。
• 语义而非语法： 我们关注的是LLM输出的语义质量，而非特定的词语序列。
• 数据依赖性强： 模型的性能很大程度上取决于输入数据的质量和多样性。
• 迭代速度快： 提示词、模型版本、检索策略、工具逻辑等都可能频繁更新。

1.2 传统测试方法的局限性

• 单元测试： 对于独立的函数或模块可能有效，但无法捕捉整个Graph的端到端行为，尤其是在LLM输出作为下一个模块输入的情况下。
• 集成测试： 尝试测试多个模块的协同工作，但面对LLM的非确定性，验证输出仍然是一个挑战。
• 人工评估： 准确但耗时、昂贵且难以扩展，不适合每次代码提交后的回归测试。
• A/B测试（在线测试）： 虽然是最终的真理，但它是在生产环境中进行的，风险高、成本大，且发现问题后回滚需要时间。我们希望在进入A/B测试阶段之前，就尽可能地发现和解决问题。

1.3 离线回归测试的价值

离线回归测试通过在受控环境中，使用代表真实生产场景的历史数据来运行新版本的Graph，并自动化评估其性能，提供了以下关键优势：

• 安全与低成本： 在部署前发现并修复问题，避免影响真实用户。
• 快速反馈： 允许开发者在修改后立即获得关于性能影响的反馈。
• 可重复性： 使用固定数据集，可以多次运行测试，比较不同版本间的性能。
• 聚焦关键场景： 利用历史生产数据，可以直接测试那些真实用户遇到过的问题或高频场景。
• 自动化与可扩展性： 通过LangSmith等工具，实现评估流程的自动化，减少人工干预。

2. LangSmith与数据集回溯测试的核心概念

LangSmith是LangChain生态系统中的一个关键组件，旨在帮助开发者理解、调试、测试和监控他们的LLM应用。它的核心能力包括：

• 追踪（Tracing）： 记录LLM应用中每个步骤的输入、输出、中间结果和延迟。
• 评估（Evaluation）： 提供多种自动评估器，并支持自定义评估逻辑，用于量化LLM输出的质量。
• 数据集（Datasets）： 存储输入/输出对或更复杂的会话数据，作为测试和评估的基准。
• 回溯测试（Backtesting）： 运行一个或多个LLM应用版本（Graph）对抗一个数据集，并使用定义的评估器进行自动评估和比较。

2.1 “Graph”的定义

在本次讲座中，“Graph”指的是一个LLM应用的工作流，例如一个LangChain的Chain或Agent。它可能包括：

• LLM调用
• 检索器（Retriever）调用
• 工具调用（Tool Calls）
• 自定义逻辑
• 提示词模板（Prompt Templates）

当我说“新版Graph”时，我指的是对现有Graph的任何修改，例如：

• 更新了LLM模型（如从GPT-3.5到GPT-4）
• 修改了提示词（Prompt Engineering）
• 调整了RAG的检索策略或块大小
• 添加或移除了工具
• 改变了Agent的规划逻辑

2.2 数据集回溯测试的工作原理

LangSmith的Dataset Backtesting流程大致如下：

1. 准备数据集： 从历史生产数据中提取输入，并可选地包含期望的输出或上下文，构建成一个LangSmith Dataset。
2. 定义评估器： 选择或创建评估函数，它们将接收Graph的输入和输出，并返回一个或多个衡量指标。
3. 运行回溯测试： 将新版Graph（或多个版本）与数据集和评估器关联，LangSmith会为数据集中的每个输入运行Graph，并用评估器评估其输出。
4. 分析结果： 比较新旧Graph版本的评估分数，识别性能下降（回归）或提升的区域。

3. 构建“黄金数据集”：从历史生产数据中提取价值

回溯测试的基石是高质量的“黄金数据集”。这些数据应该代表你的应用程序在生产环境中遇到的真实、多样且有挑战性的场景。LangSmith通过其追踪功能，使得从历史生产数据中构建数据集变得异常方便。

3.1 捕获生产追踪数据

假设我们有一个已经在生产环境中运行的LLM应用。通过集成LangSmith，我们可以自动捕获每次运行的详细追踪信息。

首先，确保你的LangChain应用已经配置了LangSmith。

import os
from langsmith import Client
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 设置LangSmith环境变量
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "YOUR_LANGSMITH_API_KEY"
os.environ["LANGCHAIN_PROJECT"] = "Production-Graph-Monitoring" # 定义项目名称

# 确保你的OpenAI API Key也已设置
os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

# 这是一个简单的生产Graph示例：一个聊天机器人
def create_chat_graph(model_name: str):
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个友好的AI助手，擅长简洁地回答问题。"),
        ("human", "{question}")
    ])
    llm = ChatOpenAI(model_name=model_name, temperature=0.7)
    output_parser = StrOutputParser()
    return prompt | llm | output_parser

# 实例化生产Graph
production_graph = create_chat_graph("gpt-3.5-turbo")

# 模拟一些生产请求
production_queries = [
    "解释一下量子纠缠。",
    "世界上最高的山峰是哪座？",
    "给我推荐一本关于人工智能的入门书籍。",
    "Python中如何实现多线程？",
    "什么是黑洞？",
    "写一句关于秋天的诗句。",
    "谁是特斯拉的创始人？",
    "如何用烤箱制作披萨？"
]

print("--- 模拟生产请求并追踪 ---")
for i, query in enumerate(production_queries):
    print(f"处理请求 {i+1}: {query}")
    try:
        response = production_graph.invoke({"question": query})
        print(f"响应: {response[:50]}...") # 打印前50个字符
        # 实际生产中，这些追踪会自动发送到LangSmith
    except Exception as e:
        print(f"请求处理失败: {e}")
print("--- 生产请求模拟完成 ---")

当上述代码运行时，每次 production_graph.invoke 调用都会生成一个LangSmith追踪，并上传到你指定的LANGCHAIN_PROJECT中。这些追踪包含了完整的输入、输出、中间步骤和元数据，是构建数据集的原始素材。

3.2 从追踪中创建LangSmith数据集

一旦你积累了一定量的生产追踪，你就可以使用LangSmith客户端库将它们转换为一个Dataset。

from langsmith import Client
from langsmith.schemas import Dataset, Example

client = Client()

# 定义数据集名称
dataset_name = "Production Chat History Q3 2023"

# 尝试获取最近的生产追踪（以'Production-Graph-Monitoring'项目为例）
# 实际应用中，你可能需要更复杂的过滤条件，例如时间范围、特定的Graph ID等
# 这里我们假设我们已经有了一些追踪存在于"Production-Graph-Monitoring"项目中
runs = client.list_runs(
    project_name="Production-Graph-Monitoring",
    # run_type="chain" 或 "llm" 取决于你的Graph结构
    # 这里我们假设我们想从顶层chain的追踪中提取
    run_type="chain",
    # 限制获取最近的20个追踪作为示例
    limit=20
)

# 过滤掉失败的或没有完整输入/输出的追踪
valid_runs = []
for run in runs:
    # 检查输入是否是一个字典，并且包含'question'键
    # 检查输出是否存在
    if isinstance(run.inputs, dict) and "question" in run.inputs and run.outputs:
        valid_runs.append(run)

if not valid_runs:
    print("没有找到有效的生产追踪来创建数据集。请确保您的生产应用已运行并生成了追踪。")
else:
    print(f"找到 {len(valid_runs)} 个有效追踪来创建数据集。")

    # 准备数据集的示例（Examples）
    examples = []
    for run in valid_runs:
        # LangSmith Example通常包含inputs和outputs
        # 对于回溯测试，inputs是必需的，outputs是可选的（作为“黄金标准”）
        # 如果你希望评估Graph的输出与历史生产输出的相似度，可以包含outputs
        # 如果你只关心Graph是否能生成“好”的输出（由评估器决定），则outputs可以省略
        examples.append(
            Example(
                inputs={"question": run.inputs["question"]},
                outputs={"answer": run.outputs["output"]}, # 假设顶层chain的输出键是'output'
                dataset_id=None # 稍后会填充
            )
        )

    # 检查数据集是否已存在
    existing_datasets = client.list_datasets(dataset_name=dataset_name)
    if any(ds.name == dataset_name for ds in existing_datasets):
        print(f"数据集 '{dataset_name}' 已存在。我们将创建一个新版本或更新它。")
        # 实际场景中，你可能选择更新现有数据集或创建一个新版本
        # 这里为了演示，我们创建一个新数据集，带上时间戳
        import datetime
        dataset_name = f"{dataset_name} - {datetime.datetime.now().strftime('%Y%m%d%H%M%S')}"

    print(f"正在创建数据集 '{dataset_name}'...")
    dataset = client.create_dataset(
        dataset_name=dataset_name,
        description="从生产环境中捕获的聊天机器人历史问题与回答。",
        examples=examples
    )
    print(f"数据集 '{dataset.name}' (ID: {dataset.id}) 创建成功。包含 {len(examples)} 个示例。")

表格1: LangSmith Dataset Example 的关键组件

组件名称	类型	描述	是否必需（回溯测试）	备注
inputs	dict	Graph的输入，例如用户查询。	是	键值对形式，与Graph的invoke方法接收的参数匹配。
outputs	dict	可选的“黄金标准”输出。	否	如果提供，可用于评估Graph输出与黄金标准的匹配度。
reference_key	str	可选，用于指示outputs中哪个键是主要参考。	否
metadata	dict	可选，任何与该示例相关的额外信息。	否	例如，用户ID、时间戳、原始追踪ID等。

3.3 数据集选择与清洗的考量

• 代表性： 数据集应尽可能代表真实生产流量的分布，包括常见场景和边缘情况。
• 多样性： 避免过拟合于特定类型的查询。
• 问题识别： 优先包含那些在生产中导致失败、低质量响应或高延迟的请求。
• 隐私与敏感信息： 在将生产数据用于测试前，务必进行脱敏处理，移除任何个人身份信息（PII）或其他敏感数据。
• 数据量： 初始阶段可以从几十到几百个示例开始，随着系统成熟，可以扩展到数千甚至更多。

4. 定义评估指标：量化Graph性能

对于LLM应用，没有一个单一的“正确”答案，因此我们需要多维度的评估指标。LangSmith提供了多种内置评估器，并允许我们创建自定义评估器。

4.1 评估器的类型

• LLM-as-a-judge (LLM作为评判者)： 利用另一个强大的LLM来评估被测LLM的输出。这是最灵活也是最强大的方法之一，可以评估相关性、连贯性、有用性、安全性等主观质量。
• 启发式/规则评估器： 基于预定义规则（如关键词存在、输出长度、JSON格式验证）进行评估。
• 语义相似度评估器： 使用嵌入模型计算Graph输出与参考输出之间的语义相似度。
• 自定义评估器： 完全根据业务逻辑或特定需求编写的评估函数。

4.2 创建自定义评估器

假设我们的Graph是一个问答系统。我们可能关心以下几点：

1. 相关性： 回答是否与问题相关？
2. 连贯性： 回答是否流畅、易懂？
3. 简洁性： 回答是否过于冗长？
4. 是否包含特定关键词（业务需求）

我们可以结合使用LangSmith的内置评估器和自定义评估器。

from langsmith.evaluation import evaluate
from langsmith.schemas import Run, Example, EvaluationResult
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
from typing import Optional, Dict, Any, List

client = Client()

# --- 1. 使用LangChain内置的LLM-as-a-judge评估器 ---
# LangChain提供了一些预构建的评估链，例如"labeled_qa"用于QA任务
# 这些评估器通常需要一个LLM来作为评判者
llm_judge = ChatOpenAI(model_name="gpt-4", temperature=0) # 通常使用更强大的模型作为评判者

# 我们可以定义一个辅助函数来创建评估器，或者直接在evaluate函数中传入
# LangChain的评估器通常是一个Chain，接收输入、输出和可选的参考（ground_truth）
from langchain.evaluation import load_evaluator

# 相关性评估器 (LLM-as-a-judge)
qa_relevance_evaluator = load_evaluator(
    "labeled_score_string",
    criteria="relevance",
    llm=llm_judge,
    # 你的Graph的输入键和输出键
    input_key="question",
    prediction_key="answer"
)

# 连贯性评估器 (LLM-as-a-judge)
qa_coherence_evaluator = load_evaluator(
    "labeled_score_string",
    criteria="coherence",
    llm=llm_judge,
    input_key="question",
    prediction_key="answer"
)

# --- 2. 创建一个自定义的启发式评估器 ---
# 这个评估器将检查回答是否包含某个预期的关键词列表中的至少一个
def keyword_presence_evaluator(run: Run, example: Optional[Example] = None) -> EvaluationResult:
    """
    自定义评估器：检查Graph的输出是否包含预期的关键词。
    """
    prediction = run.outputs.get("output") if run.outputs else ""
    # 假设我们期望回答中至少包含“AI”或“人工智能”
    expected_keywords = ["AI", "人工智能", "机器学习"]

    score = 0
    feedback = ""
    for keyword in expected_keywords:
        if keyword.lower() in prediction.lower():
            score = 1
            feedback = f"回答包含关键词: {keyword}"
            break
    if score == 0:
        feedback = f"回答未包含任何预期关键词: {', '.join(expected_keywords)}"

    return EvaluationResult(
        key="keyword_presence",
        score=score,
        comment=feedback,
        correction=None,
        evaluator_info={"expected_keywords": expected_keywords}
    )

# --- 3. 创建一个更通用的自定义评估器，可以被evaluate函数直接调用 ---
# 这种形式的评估器更容易被LangSmith的evaluate函数集成
def check_answer_length_evaluator(run: Run, example: Optional[Example] = None) -> EvaluationResult:
    """
    自定义评估器：检查回答的长度是否在合理范围内。
    """
    prediction = run.outputs.get("output") if run.outputs else ""
    min_length = 20
    max_length = 200

    length = len(prediction)
    score = 0
    feedback = ""

    if min_length <= length <= max_length:
        score = 1
        feedback = f"回答长度 {length} 在预期范围 [{min_length}, {max_length}] 内。"
    elif length < min_length:
        feedback = f"回答过短 ({length} < {min_length})。"
    else: # length > max_length
        feedback = f"回答过长 ({length} > {max_length})。"

    return EvaluationResult(
        key="answer_length_check",
        score=score,
        comment=feedback,
        evaluator_info={"min_length": min_length, "max_length": max_length, "actual_length": length}
    )

# 我们将所有评估器组合成一个列表，稍后传给evaluate函数
evaluators = [
    qa_relevance_evaluator,
    qa_coherence_evaluator,
    keyword_presence_evaluator, # 自定义函数
    check_answer_length_evaluator # 自定义函数
]

print("评估器已定义。")

表格2: 评估器类型及其适用场景

评估器类型	优点	缺点	适用场景
LLM-as-a-judge	灵活，能评估主观质量，适应性强。	成本较高，可能引入LLM的偏见，非确定性。	相关性、连贯性、有用性、安全性、语气等。
启发式/规则	快速，成本低，确定性。	规则难以覆盖所有情况，不灵活，易被绕过。	格式验证（JSON）、关键词存在、长度检查、安全词过滤。
语义相似度	客观衡量语义接近度，比关键词更鲁棒。	需要高质量的参考输出，对同义词处理较好但对反义词或语境变化敏感。	问答系统中的答案相似度、摘要质量。
自定义	完全控制评估逻辑，可实现复杂业务规则。	开发成本，需要人工维护。	特定业务指标、外部系统集成验证、特定领域知识校验。

5. 实现新版Graph：待测试的“候选者”

现在我们有了一个黄金数据集和一套评估器。接下来，我们需要构建或修改我们的Graph，使其成为待测试的“候选者”。

假设我们对生产Graph做了如下改进：

• 将LLM模型从gpt-3.5-turbo升级到gpt-4o-mini，期待更好的推理能力。
• 修改了系统提示词，使其回答更具信息量，而不是仅仅“简洁”。

# 实例化新版Graph
# 假设这是我们想要测试的新版本，它可能在提示词、模型或逻辑上有所不同

def create_new_chat_graph(model_name: str):
    # 修改了系统提示词
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个知识渊博的AI专家，请提供详细、准确且富有洞见的回答。"),
        ("human", "{question}")
    ])
    # 升级了模型
    llm = ChatOpenAI(model_name=model_name, temperature=0.5) # 稍微降低温度以获取更稳定输出
    output_parser = StrOutputParser()
    return prompt | llm | output_parser

# 新版Graph，使用不同的模型和提示词
new_candidate_graph = create_new_chat_graph("gpt-4o-mini")

print("新版Graph已创建，准备进行回溯测试。")

6. 运行回溯测试：LangSmith的evaluate函数

LangSmith的evaluate函数是进行回溯测试的核心。它将数据集、Graph和评估器连接起来，自动化整个评估流程。

from langsmith.evaluation import evaluate

# 确保之前创建的数据集ID可用
# dataset_id = dataset.id # 如果你运行了前面的创建数据集代码

# 如果你没有运行前面的创建数据集代码，或者需要使用一个已有的数据集ID
# 可以手动查找或指定一个数据集名称
dataset_name_to_use = "Production Chat History Q3 2023 - 20240730103000" # 替换为你的实际数据集名称或ID
try:
    existing_datasets = client.list_datasets(dataset_name=dataset_name_to_use)
    if not existing_datasets:
        raise ValueError(f"未找到名为 '{dataset_name_to_use}' 的数据集。请确保数据集名称正确或先创建数据集。")
    dataset_id = existing_datasets[0].id
    print(f"找到并使用数据集 '{dataset_name_to_use}' (ID: {dataset_id})。")
except Exception as e:
    print(f"获取数据集失败: {e}")
    print("请手动设置 dataset_id 为一个有效的LangSmith数据集ID，例如：dataset_id = 'your_dataset_id_here'")
    # 退出或提供一个默认值，以防在实际运行中出错
    exit()

print("--- 启动离线回归测试 ---")

# evaluate函数会为数据集中的每个Example运行我们的candidate_graph
# 并使用所有指定的评估器进行评估
experiment_results = evaluate(
    # The LangChain runnable (your new graph) to test
    llm_or_chain_factory=new_candidate_graph,
    # The dataset to run against
    data=dataset_id,
    # The evaluators to use
    evaluators=evaluators,
    # Configuration for the experiment run
    experiment_prefix="NewGraph-GPT4oMini-DetailedPrompt", # 用于LangSmith UI中区分实验
    # Other optional parameters
    max_concurrency=5, # 限制并发请求，避免API速率限制
    num_repetitions=1, # 每个例子运行的次数 (对于非确定性系统可以多次运行取平均)
    # The keys in the dataset inputs that correspond to the chain's input keys
    # input_keys=["question"], # evaluate函数通常能自动推断
    # The key in the dataset outputs that corresponds to the chain's output key (for labeled evaluators)
    # reference_key="answer", # evaluate函数通常能自动推断
)

print("--- 回溯测试完成 ---")
print(f"实验名称: {experiment_results.experiment_name}")
print(f"实验项目URL: {experiment_results.url}")
print("n请访问上述URL在LangSmith UI中查看详细结果和比较。")

在evaluate函数中：

• llm_or_chain_factory：传入你的新版Graph实例。
• data：传入数据集的ID或名称。
• evaluators：传入我们之前定义的评估器列表。
• experiment_prefix：一个有意义的前缀，用于在LangSmith UI中标识这次实验。
• max_concurrency：控制同时运行的Graph实例数量，以避免API速率限制。
• num_repetitions：对于非确定性LLM，可以运行多次，取平均分数，以减少单次运行的偶然性。

运行此代码后，LangSmith会将每个数据集示例的输入传递给new_candidate_graph，收集其输出，然后将这些输入和输出传递给所有定义的评估器。所有这些结果都将作为一次“实验”存储在LangSmith平台中。

7. 分析结果与迭代优化

回溯测试的结果将上传到LangSmith平台。最直观的分析方式是通过LangSmith UI。

7.1 LangSmith UI中的结果分析

1. 访问实验URL： 运行evaluate函数后，会打印一个实验URL。点击它进入LangSmith UI。
2. 概览页面： 你会看到实验的整体评估分数，例如所有评估器的平均分。
3. 比较视图： LangSmith最强大的功能之一是其比较视图。你可以将当前实验（新版Graph）与基线实验（旧版生产Graph对同一数据集的运行结果）进行并排比较。
   • 指标对比： 直接看到每个评估指标在新旧版本间的变化。例如，“relevance”得分是否提升了？“keyword_presence”是否下降了？
   • 逐个示例对比： 对于每个数据集示例，你可以看到新旧Graph的完整追踪、原始输入、各自的输出以及所有评估器的详细反馈。这对于诊断问题至关重要。如果某个评估指标下降了，你可以深入查看是哪些具体的示例导致了下降，进而分析Graph在该示例上的行为。
   • 筛选与排序： 可以根据评估分数、错误类型等对示例进行筛选和排序，快速定位表现不佳的案例。

7.2 程序化分析结果

除了UI，我们也可以通过LangSmith客户端库程序化地获取和分析结果。

# experiment_results 对象包含了实验的ID和名称，可以用来获取详细结果
# experiment_results.runs 包含所有运行的ID
# experiment_results.feedback 包含了所有评估结果

# 获取实验的所有评估结果
# 注意：这可能会返回大量的评估结果，特别是对于大型数据集和多个评估器
# 实际场景中，你可能需要分页或聚合
all_eval_results = client.list_feedback(
    run_ids=[r.id for r in experiment_results.runs]
)

# 聚合评估结果
aggregated_scores: Dict[str, List[float]] = {}
for feedback in all_eval_results:
    if feedback.key not in aggregated_scores:
        aggregated_scores[feedback.key] = []
    if feedback.score is not None:
        aggregated_scores[feedback.key].append(feedback.score)

print("n--- 聚合评估分数（新版Graph）---")
for key, scores in aggregated_scores.items():
    if scores:
        print(f"评估器 '{key}': 平均分 = {sum(scores) / len(scores):.2f}, 样本数 = {len(scores)}")
    else:
        print(f"评估器 '{key}': 没有找到有效分数。")

# 进一步分析：找出低分案例
print("n--- 低分案例分析 ---")
low_score_threshold = 0.5 # 假设低于0.5分被认为是低分

# 遍历所有运行和它们的评估结果
for run in experiment_results.runs:
    run_feedback = client.list_feedback(run_ids=[run.id])
    for feedback in run_feedback:
        if feedback.score is not None and feedback.score < low_score_threshold:
            print(f"n运行 ID: {run.id}, 评估器: {feedback.key}, 分数: {feedback.score}")
            print(f"  输入: {run.inputs.get('question')}")
            print(f"  输出: {run.outputs.get('output', 'N/A')[:200]}...") # 打印部分输出
            print(f"  评论: {feedback.comment}")
            # 你可以通过 run.reference_examples 找到对应的 Example ID
            # 然后通过 client.read_example(example_id) 获取原始 Example
            if run.reference_examples:
                original_example = client.read_example(run.reference_examples[0].id)
                print(f"  原始 Example ID: {original_example.id}")
                print(f"  原始 Example 输出 (如果存在): {original_example.outputs.get('answer', 'N/A')[:200]}...")

表格3: 示例回溯测试结果对比

评估指标	旧版Graph (gpt-3.5-turbo)	新版Graph (gpt-4o-mini)	变化	备注
相关性	0.85	0.92	↑ 0.07	新模型在理解问题核心上表现更好。
连贯性	0.80	0.88	↑ 0.08	新提示词和模型使得回答更流畅。
关键词存在	0.70	0.65	↓ 0.05	新版回答可能更通用，减少了特定关键词的出现。需要进一步检查。
长度检查	0.95	0.80	↓ 0.15	新版Graph回答更详细，部分回答可能超出了预期长度范围。
平均延迟	1.5s	2.2s	↑ 0.7s	GPT-4o-mini可能响应速度稍慢。这是一个重要的非功能性指标。

7.3 迭代优化

根据分析结果，你可能需要进行以下迭代：

1. 如果出现回归（分数下降）：
   • 检查导致下降的具体示例，分析Graph在该示例上的行为。
   • 是否是提示词的副作用？
   • 是否是模型选择不当？
   • 是否是某个工具调用失败？
   • 修改Graph，例如调整提示词、更换模型、优化检索策略或工具逻辑。
2. 如果出现预期外的提升：
   • 理解为什么会提升，将这些经验应用到其他Graph的开发中。
3. 如果发现评估器本身有问题：
   • 评估器是否准确？LLM-as-a-judge的提示词是否清晰？自定义评估器的逻辑是否有漏洞？
   • 调整或增加评估器。

这个过程是一个循环：修改Graph -> 运行回溯测试 -> 分析结果 -> 优化Graph，直到新版Graph达到预期的性能指标。

8. 高级实践与集成

8.1 版本控制与数据集/评估器管理

• Graph版本： 将你的LangChain Graph代码与LangSmith实验结果关联起来，最好通过Git commit hash或其他版本标识符。
• 数据集版本： 随着时间推移，生产数据会发生变化。定期更新或创建新的数据集版本，以保持其代表性。LangSmith支持数据集版本化。
• 评估器版本： 评估器的逻辑也可能随着业务需求而演进，对其进行版本控制同样重要。

8.2 CI/CD集成

将LangSmith回溯测试集成到你的持续集成/持续部署（CI/CD）流程中：

• 每次代码提交时自动运行： 在Pull Request合并前，自动触发回溯测试。
• 设定通过标准： 如果新版Graph的某个核心评估指标低于预设阈值，或者与基线版本相比下降超过一定百分比，则CI/CD流程失败，阻止合并。
• 报告生成： 自动将LangSmith实验的URL和关键指标发布到CI/CD报告中，便于开发者快速查看。

示例伪代码 (GitHub Actions):

name: LangSmith Backtesting

on:
  pull_request:
    branches:
      - main
  push:
    branches:
      - main

jobs:
  backtest:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v3
    - name: Set up Python
      uses: actions/setup-python@v4
      with:
        python-version: '3.10'
    - name: Install dependencies
      run: |
        pip install -r requirements.txt
    - name: Run LangSmith Backtest
      env:
        LANGCHAIN_API_KEY: ${{ secrets.LANGCHAIN_API_KEY }}
        OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        LANGCHAIN_TRACING_V2: "true"
        LANGCHAIN_PROJECT: "CI-CD-Backtesting"
      run: |
        python run_backtest.py # 这是一个包含上述evaluate调用的脚本

    - name: Check Backtest Results (Example)
      # 这一步需要调用LangSmith API来获取实验结果并进行判断
      # 如果主要指标下降，可以失败Job
      run: |
        python check_results.py # 脚本获取最新的实验结果，并根据预设阈值判断是否通过

8.3 处理非确定性

LLM的非确定性是一个挑战。为了获得更稳定的评估结果：

• 多次运行取平均： 在evaluate函数中设置num_repetitions > 1。
• 降低LLM温度： 在Graph中使用的LLM可以设置较低的temperature参数，使其输出更稳定。

8.4 管理大型数据集

对于非常大的数据集：

• 抽样： 从完整的生产数据集中随机抽样，确保样本的代表性。
• 分布式评估： LangSmith的evaluate函数在底层可以利用并发，但如果数据集巨大，可能需要考虑更复杂的分布式评估策略。

总结

我们今天深入探讨了如何利用LangSmith的“数据集回溯测试”功能，对新版LLM Graph进行离线回归测试。通过捕获和利用历史生产数据构建黄金数据集，结合多维度评估器，并在受控环境中运行新旧Graph版本，我们能够系统、高效地识别性能回归或提升，从而在部署前做出数据驱动的决策。这种方法为LLM应用的快速迭代和稳健发展提供了坚实的基础，是确保AI产品质量和用户体验的关键实践。