실험(Experiments)

실험은 애플리케이션에서 가설이나 아이디어를 검증하기 위해 의도적으로 가하는 변경이에요. RAG 시스템을 예로 들면, 새 임베딩 모델이 챗봇 응답에 어떤 영향을 주는지 평가하려고 리트리버 모델을 바꿔보는 거죠. 좋은 실험을 설계하고 Ragas로 실행하는 방법을 정리해볼게요.

출처: 문서

본문

실험이란 무엇인가요?(What is an experiment?)

실험은 애플리케이션에서 가설이나 아이디어를 검증하기 위해 의도적으로 가하는 변경이에요. 예를 들어 RAG(Retrieval-Augmented Generation) 시스템에서 리트리버 모델을 교체해 새 임베딩 모델이 챗봇 응답에 어떤 영향을 주는지 평가할 수 있어요.

좋은 실험의 원칙(Principles of a Good Experiment)

측정 가능한 지표 정의: 정확도(accuracy), 정밀도(precision), 재현율(recall) 같은 지표를 사용해 변경의 영향을 정량화해요.
체계적인 결과 저장: 결과를 비교·추적하기 쉽도록 정리된 방식으로 저장해요.
변경 분리: 한 번에 하나의 변경만 해서 그 특정 영향만 파악해요. 여러 변경을 동시에 하면 결과가 모호해질 수 있어요.
반복 프로세스: 구조화된 접근을 따라서 — 변경 가하기 → 평가 실행 → 결과 관찰 → 다음 변경 가설 세우기.

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

Ragas에서의 실험(Experiments in Ragas)

실험의 구성 요소(Components of an Experiment)

테스트 데이터셋: 시스템을 평가하는 데 사용하는 데이터.
애플리케이션 엔드포인트: 테스트 중인 애플리케이션, 컴포넌트 또는 모델.
지표(Metrics): 성능을 평가하는 양적 측정.

실행 과정(Execution Process)

설정(Setup): 실험 파라미터를 정의하고 테스트 데이터셋을 불러와요.
실행(Run): 데이터셋의 각 샘플에 애플리케이션을 실행해요.
평가(Evaluate): 성능을 측정하기 위해 지표를 적용해요.
저장(Store): 분석과 비교를 위해 결과를 저장해요.

Ragas로 실험 만들기(Creating Experiments with Ragas)

Ragas는 실험 생성 과정을 간소화하는 @experiment 데코레이터를 제공해요. 손으로 먼저 시작해보고 싶다면 퀵 스타트 가이드를 참고하세요.

기본 실험 구조(Basic Experiment Structure)

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()
    }

실험 실행하기(Running Experiments)

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)

파라미터화된 실험(Parameterized Experiments)

서로 다른 구성을 테스트하는 파라미터화된 실험을 만들 수 있어요:

@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
)

실험 관리 모범 사례(Experiment Management Best Practices)

1. 일관된 명명(Consistent Naming)

다음을 포함한 설명적인 이름을 사용하세요:

무엇이 바뀌었는지 (모델, 프롬프트, 파라미터)
버전 번호
필요하면 날짜/시간

experiment_name = "gpt4o_v2_prompt_temperature_0.1_20241201"

2. 결과 저장(Result Storage)

실험은 결과를 타임스탬프와 함께 experiments/ 디렉토리의 CSV 파일에 자동 저장해요:

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

3. 메타데이터 추적(Metadata Tracking)

실험 결과에 관련 메타데이터를 포함하세요:

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
}

고급 실험 패턴(Advanced Experiment Patterns)

A/B 테스트(A/B Testing)

두 접근 방식을 동시에 테스트해요:

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

다단계 실험(Multi-Stage Experiments)

여러 컴포넌트로 구성된 복잡한 시스템의 경우:

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

실험에서의 오류 처리(Error Handling in Experiments)

부분 결과를 잃지 않도록 오류를 우아하게 처리하세요:

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

지표와의 통합(Integrating with Metrics)

실험은 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"
    }

이 통합으로 실험 결과와 함께 지표 점수를 자동으로 계산·저장할 수 있어요. 시간이 지나며 성능 개선을 추적하기가 훨씬 쉬워지죠.

실험(Experiments)