实验

什么是实验？

实验是对应用做出的有意改动，用来验证某个假设或想法。例如，在检索增强生成（RAG）系统中，你可能会替换检索器模型，以评估新的嵌入模型对聊天机器人回复的影响。

好实验的原则

定义可度量的指标：使用准确率、精确率或召回率等指标，量化改动的影响。
系统化存储结果：确保结果以有组织的方式存储，便于比较和追踪。
隔离改动：一次只做一处改动，以识别其具体影响。避免同时做多处改动，否则会掩盖结果。
迭代过程：遵循结构化方法：*做出修改 → 运行评测 → 观察结果 →

graph LR
    A[Make a change] --> B[Run evaluations]
    B --> C[Observe results]
    C --> D[Hypothesize next change]
    D --> A

Ragas 中的实验

实验的组成

测试数据集：用于评测系统的数据。
应用端点：被测试的应用、组件或模型。
指标：用于评估性能的定量度量。

执行过程

Setup（准备）：定义实验参数并加载测试数据集。
Run（运行）：在数据集的每个样本上执行应用。
Evaluate（评测）：应用指标来衡量性能。
Store（存储）：保存结果以供分析和比较。

用 Ragas 创建实验

Ragas 提供 @experiment 装饰器，用于简化实验创建流程。若你更想先动手入门，请参见快速开始指南。

基本实验结构

from ragas import experiment
import asyncio

@experiment()
async def my_experiment(row):
    # Process the input through your system
    response = await asyncio.to_thread(my_system_function, row["input"])

    # Return results for evaluation
    return {
        **row,  # Include original data
        "response": response,
        "experiment_name": "baseline_v1",
        # Add any additional metadata
        "model_version": "gpt-4o",
        "timestamp": datetime.now().isoformat()
    }

运行实验

from ragas import Dataset

# Load your test dataset
dataset = Dataset.load(name="test_data", backend="local/csv", root_dir="./data")

# Run the experiment
results = await my_experiment.arun(dataset)

参数化实验

你可以创建参数化实验，以测试不同配置：

@experiment()
async def model_comparison_experiment(row, model_name: str, temperature: float):
    # Configure your system with the parameters
    response = await my_system_function(
        row["input"], 
        model=model_name, 
        temperature=temperature
    )

    return {
        **row,
        "response": response,
        "experiment_name": f"{model_name}_temp_{temperature}",
        "model_name": model_name,
        "temperature": temperature
    }

# Run with different parameters
results_gpt4 = await model_comparison_experiment.arun(
    dataset, 
    model_name="gpt-4o", 
    temperature=0.1
)

results_gpt35 = await model_comparison_experiment.arun(
    dataset, 
    model_name="gpt-3.5-turbo", 
    temperature=0.1
)

实验管理最佳实践

1. 一致的命名

使用描述性名称，包含：

改了什么（模型、prompt、参数）
版本号
如有需要，加上日期/时间

experiment_name = "gpt4o_v2_prompt_temperature_0.1_20241201"

2. 结果存储

实验会自动把结果保存到 experiments/ 目录下带时间戳的 CSV 文件中：

experiments/
├── 20241201-143022-baseline_v1.csv
├── 20241201-143515-gpt4o_improved_prompt.csv
└── 20241201-144001-comparison.csv

3. 元数据追踪

在实验结果中包含相关元数据：

return {
    **row,
    "response": response,
    "experiment_name": "baseline_v1",
    "git_commit": "a1b2c3d",
    "environment": "staging",
    "model_version": "gpt-4o-2024-08-06",
    "total_tokens": response.usage.total_tokens,
    "response_time_ms": response_time
}

高级实验模式

A/B Testing（A/B 测试）

同时测试两种不同方案：

@experiment()
async def ab_test_experiment(row, variant: str):
    if variant == "A":
        response = await system_variant_a(row["input"])
    else:
        response = await system_variant_b(row["input"])

    return {
        **row,
        "response": response,
        "variant": variant,
        "experiment_name": f"ab_test_variant_{variant}"
    }

# Run both variants
results_a = await ab_test_experiment.arun(dataset, variant="A")
results_b = await ab_test_experiment.arun(dataset, variant="B")

多阶段实验

适用于包含多个组件的复杂系统：

@experiment()
async def multi_stage_experiment(row):
    # Stage 1: Retrieval
    retrieved_docs = await retriever(row["query"])

    # Stage 2: Generation
    response = await generator(row["query"], retrieved_docs)

    return {
        **row,
        "retrieved_docs": retrieved_docs,
        "response": response,
        "num_docs_retrieved": len(retrieved_docs),
        "experiment_name": "multi_stage_v1"
    }

实验中的错误处理

优雅地处理错误，以免丢失部分结果：

@experiment()
async def robust_experiment(row):
    try:
        response = await my_system_function(row["input"])
        error = None
    except Exception as e:
        response = None
        error = str(e)

    return {
        **row,
        "response": response,
        "error": error,
        "success": error is None,
        "experiment_name": "robust_v1"
    }

与指标集成

实验可以与 Ragas 指标无缝配合：

from ragas.metrics import FactualCorrectness

@experiment()
async def evaluated_experiment(row):
    response = await my_system_function(row["input"])

    # Calculate metrics inline
    factual_score = FactualCorrectness().score(
        response=response,
        reference=row["expected_output"]
    )

    return {
        **row,
        "response": response,
        "factual_correctness": factual_score.value,
        "factual_reason": factual_score.reason,
        "experiment_name": "evaluated_v1"
    }

这种集成让你可以自动计算指标分数，并与实验结果一并存储，从而轻松跟踪随时间推移的性能改进。