实验

什么是实验？

实验是对您的应用程序进行的有意更改，以测试一个假设或想法。例如，在检索增强生成（RAG）系统中，您可能会替换检索器模型，以评估新的嵌入模型如何影响聊天机器人的响应。

良好实验的原则

定义可衡量的指标：使用准确率、精确率或召回率等指标来量化更改的影响。
系统化的结果存储：确保结果以有组织的方式存储，以便于比较和跟踪。
隔离变更：一次只进行一项更改，以确定其具体影响。避免同时进行多项更改，因为这会使结果变得模糊不清。
迭代过程：遵循结构化的方法：*进行更改 → 运行评估 → 观察结果 →

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

Ragas 中的实验

实验的组成部分

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

执行过程

设置：定义实验参数并加载测试数据集。
运行：在数据集中的每个样本上执行应用程序。
评估：应用指标来衡量性能。
存储：保存结果以供分析和比较。

使用 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. 一致的命名

使用描述性的名称，包括： - 更改了什么（模型、提示、参数） - 版本号 - 如果相关，还包括日期/时间

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 测试

同时测试两种不同的方法

@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"
    }

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