RAG 앱 평가하고 개선하기

RAG 앱 평가하고 개선하기 (How to Evaluate and Improve a RAG App)

이 가이드에서는 Ragas를 사용해서 RAG(Retrieval-Augmented Generation) 앱을 평가하고 반복적으로 개선하는 방법을 배워요. 평가 데이터셋을 구축하고, RAG 성능을 측정할 메트릭을 세우고, 재사용 가능한 평가 파이프라인을 만든 뒤 오류를 분석해 체계적으로 개선할 수 있어요.

출처: 문서

본문

이 가이드에서는 Ragas를 사용해서 RAG 앱을 평가하고 반복적으로 개선하는 방법을 배워요.

달성할 것 (What you'll accomplish)

평가 데이터셋 구축
RAG 성능을 측정할 메트릭 확립
재사용 가능한 평가 파이프라인 구축
오류 분석 및 RAG 앱 체계적 개선
RAG 평가에 Ragas를 활용하는 방법 배우기

RAG 시스템 설정 및 실행

Hugging Face 문서 데이터셋에서 관련 문서를 검색하고 LLM으로 답변을 생성하는 간단한 RAG 시스템을 만들었어요. 이 데이터셋은 markdown으로 저장된 많은 Hugging Face 패키지의 문서 페이지를 포함해서 RAG 역량을 테스트하기 위한 풍부한 지식 베이스를 제공해요.

전체 구현은 여기: ragas_examples/improve_rag/

flowchart LR
    A[User Query] --> B[Retrieve Documents<br/>BM25]
    B --> C[Generate Response<br/>OpenAI]
    C --> D[Return Answer]

이를 실행하려면 의존성을 설치해요.

uv pip install "ragas-examples[improverag]"

그런 다음 RAG 앱을 실행해요.

import os
import asyncio
from openai import AsyncOpenAI
from ragas_examples.improve_rag.rag import RAG, BM25Retriever

# OpenAI 클라이언트 설정
os.environ["OPENAI_API_KEY"] = "<your_key>"
openai_client = AsyncOpenAI()

# 리트리버와 RAG 시스템 생성
retriever = BM25Retriever()
rag = RAG(openai_client, retriever)

# 시스템에 질의
question = "What architecture is the `tokenizers-linux-x64-musl` binary designed for?"
result = asyncio.run(rag.query(question))
print(f"Answer: {result['answer']}")

출력

Answer: It's built for the x86_64 architecture (specifically the x86_64-unknown-linux-musl target — 64-bit Linux with musl libc).

RAG 구현 이해하기

위 코드는 핵심 RAG 패턴을 보여주는 간단한 RAG 클래스를 사용해요. 작동 방식은 다음과 같아요.

# examples/ragas_examples/improve_rag/rag.py
from typing import Any, Dict, Optional
from openai import AsyncOpenAI

class RAG:
    """문서 검색과 답변 생성을 위한 간단한 RAG 시스템."""

    def __init__(self, llm_client: AsyncOpenAI, retriever: BM25Retriever, system_prompt=None, model="gpt-4o-mini", default_k=3):
        self.llm_client = llm_client
        self.retriever = retriever
        self.model = model
        self.default_k = default_k
        self.system_prompt = system_prompt or "Answer only based on documents. Be concise.\n\nQuestion: {query}\nDocuments:\n{context}\nAnswer:"

    async def query(self, question: str, top_k: Optional[int] = None) -> Dict[str, Any]:
        """RAG 시스템에 질의."""
        if top_k is None:
            top_k = self.default_k

        return await self._naive_query(question, top_k)

    async def _naive_query(self, question: str, top_k: int) -> Dict[str, Any]:
        """naive RAG 처리: 한 번 검색 후 생성."""
        # 1. BM25로 문서 검색
        docs = self.retriever.retrieve(question, top_k)

        if not docs:
            return {"answer": "No relevant documents found.", "retrieved_documents": [], "num_retrieved": 0}

        # 2. 검색된 문서로 컨텍스트 구성
        context = "\n\n".join([f"Document {i}:\n{doc.page_content}" for i, doc in enumerate(docs, 1)])
        prompt = self.system_prompt.format(query=question, context=context)

        # 3. 검색된 컨텍스트로 OpenAI를 사용해 응답 생성
        response = await self.llm_client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}]
        )

        return {
            "answer": response.choices[0].message.content.strip(),
            "retrieved_documents": [{"content": doc.page_content, "metadata": doc.metadata, "document_id": i} for i, doc in enumerate(docs)],
            "num_retrieved": len(docs)
        }

이는 핵심 RAG 패턴을 보여줘요: 관련 문서 검색 → 프롬프트에 주입 → 답변 생성.

평가 데이터셋 만들기

Hugging Face 문서에 대한 질문과 답변이 담긴 데이터셋인 huggingface_doc_qa_eval을 사용할게요.

데이터셋의 몇 가지 샘플 행:

Question	Expected Answer
What architecture is the `tokenizers-linux-x64-musl` binary designed for?	x86_64-unknown-linux-musl
What is the purpose of the BLIP-Diffusion model?	The BLIP-Diffusion model is designed for controllable text-to-image generation and editing.
What is the purpose of the /healthcheck endpoint in the Datasets server API?	Ensure the app is running

평가 스크립트는 여기에서 데이터셋을 다운로드해서 Ragas Dataset 형식으로 변환해요.

# examples/ragas_examples/improve_rag/evals.py
import urllib.request
from pathlib import Path
from ragas import Dataset
import pandas as pd

def download_and_save_dataset() -> Path:
    dataset_path = Path("datasets/hf_doc_qa_eval.csv")
    dataset_path.parent.mkdir(exist_ok=True)

    if not dataset_path.exists():
        github_url = "https://raw.githubusercontent.com/vibrantlabsai/ragas/main/examples/ragas_examples/improve_rag/datasets/hf_doc_qa_eval.csv"
        urllib.request.urlretrieve(github_url, dataset_path)

    return dataset_path

def create_ragas_dataset(dataset_path: Path) -> Dataset:
    dataset = Dataset(name="hf_doc_qa_eval", backend="local/csv", root_dir=".")
    df = pd.read_csv(dataset_path)

    for _, row in df.iterrows():
        dataset.append({"question": row["question"], "expected_answer": row["expected_answer"]})

    dataset.save()
    return dataset

데이터셋 작업에 대해 더 알아보려면 Core Concepts - Datasets를 참고해요.

RAG 평가용 메트릭 설정

이제 평가 데이터셋이 준비됐으니 RAG 성능을 측정할 메트릭이 필요해요. 핵심 사용 사례를 직접 측정하는 간단하고 집중된 메트릭으로 시작해요. 메트릭에 대한 자세한 내용은 Core Concepts - Metrics에서 찾을 수 있어요.

여기서는 RAG 응답이 기대 답변의 핵심 정보를 포함하고 제공된 컨텍스트를 바탕으로 사실적으로 정확한지 평가하는 correctness 이산(discrete) 메트릭을 사용해요.

# examples/ragas_examples/improve_rag/evals.py
from ragas.metrics import DiscreteMetric

# correctness 메트릭 정의
correctness_metric = DiscreteMetric(
    name="correctness",
    prompt="""Compare the model response to the expected answer and determine if it's correct.

Consider the response correct if it:
1. Contains the key information from the expected answer
2. Is factually accurate based on the provided context
3. Adequately addresses the question asked

Return 'pass' if the response is correct, 'fail' if it's incorrect.

Question: {question}
Expected Answer: {expected_answer}
Model Response: {response}

Evaluation:""",
    allowed_values=["pass", "fail"],
)

이제 평가 메트릭이 있으니 이를 데이터셋 전체에 체계적으로 실행해야 해요. 여기서 Ragas 실험(experiment)이 등장해요.

평가 실험 만들기

실험 함수는 각 데이터 샘플에서 RAG 시스템을 실행하고 correctness 메트릭으로 응답을 평가해요. 실험에 대한 자세한 내용은 Core Concepts - Experimentation에서 찾을 수 있어요.

실험 함수는 질문, 기대 컨텍스트, 기대 답변을 포함하는 데이터셋 행을 받아서 다음을 수행해요.

질문으로 RAG 시스템에 질의
correctness 메트릭으로 응답 평가
점수와 이유를 포함한 상세 결과 반환

# examples/ragas_examples/improve_rag/evals.py
import asyncio
from typing import Dict, Any
from ragas import experiment

@experiment()
async def evaluate_rag(row: Dict[str, Any], rag: RAG, llm) -> Dict[str, Any]:
    """
    단일 행에서 RAG 평가 실행.

    Args:
        row: question과 expected_answer를 포함하는 딕셔너리
        rag: 사전 초기화된 RAG 인스턴스
        llm: 평가용 사전 초기화된 LLM 클라이언트

    Returns:
        평가 결과를 담은 딕셔너리
    """
    question = row["question"]

    # RAG 시스템에 질의
    rag_response = await rag.query(question, top_k=4)
    model_response = rag_response.get("answer", "")

    # 비동기로 correctness 평가
    score = await correctness_metric.ascore(
        question=question,
        expected_answer=row["expected_answer"],
        response=model_response,
        llm=llm
    )

    # 평가 결과 반환
    result = {
        **row,
        "model_response": model_response,
        "correctness_score": score.value,
        "correctness_reason": score.reason,
        "mlflow_trace_id": rag_response.get("mlflow_trace_id", "N/A"),  # 디버깅용 MLflow trace ID (나중에 설명)
        "retrieved_documents": [
            doc.get("content", "")[:200] + "..." if len(doc.get("content", "")) > 200 else doc.get("content", "")
            for doc in rag_response.get("retrieved_documents", [])
        ]
    }

    return result

데이터셋, 메트릭, 실험 함수가 준비됐으니 이제 RAG 시스템 성능을 평가할 수 있어요.

초기 RAG 실험 실행

MLflow 서버 시작

평가를 실행하기 전에 MLflow 서버를 시작해야 해요. RAG 시스템은 디버깅과 분석을 위해 MLFlow에 자동으로 트레이스를 기록해요.

# MLflow 서버 시작 (필수 - 별도 터미널에서)
uv run mlflow ui --backend-store-uri sqlite:///mlflow.db --port 5000

MLflow UI는 http://127.0.0.1:5000에서 사용할 수 있어요.

초기 RAG 실험 실행

이제 RAG 시스템의 베이스라인 성능 메트릭을 얻기 위해 완전한 평가 파이프라인을 실행해요.

# 필요한 컴포넌트 임포트
import asyncio
from datetime import datetime
from ragas_examples.improve_rag.evals import (
    evaluate_rag,
    download_and_save_dataset,
    create_ragas_dataset,
    get_openai_client,
    get_llm_client
)
from ragas_examples.improve_rag.rag import RAG, BM25Retriever

async def run_evaluation():
    # 데이터셋 다운로드 및 준비
    dataset_path = download_and_save_dataset()
    dataset = create_ragas_dataset(dataset_path)

    # RAG 컴포넌트 초기화
    openai_client = get_openai_client()
    retriever = BM25Retriever()
    rag = RAG(llm_client=openai_client, retriever=retriever, model="gpt-5-mini", mode="naive")
    llm = get_llm_client()

    # 평가 실험 실행
    exp_name = f"{datetime.now().strftime('%Y%m%d-%H%M%S')}_naiverag"
    results = await evaluate_rag.arun(
        dataset, 
        name=exp_name,
        rag=rag,
        llm=llm
    )

    # 결과 출력
    if results:
        pass_count = sum(1 for result in results if result.get("correctness_score") == "pass")
        total_count = len(results)
        pass_rate = (pass_count / total_count) * 100 if total_count > 0 else 0
        print(f"Results: {pass_count}/{total_count} passed ({pass_rate:.1f}%)")

    return results

# 평가 실행
results = await run_evaluation()
print(results)

이는 데이터셋을 다운로드하고, BM25 리트리버를 초기화하고, 각 샘플에서 평가 실험을 실행하며, 분석용 상세 결과를 experiments/ 디렉토리에 CSV 파일로 저장해요.

출력

Results: 43/66 passed (65.2%)
Evaluation completed successfully!

Detailed results:
Experiment(name=20250924-212541_naiverag,  len=66)

65.2%의 통과율로 이제 베이스라인이 생겼어요. experiments/의 상세 결과 CSV에는 오류 분석과 체계적 개선에 필요한 모든 데이터가 들어 있어요.

MLflow에서 트레이스 보기

실험 결과 CSV에는 각 평가의 mlflow_trace_id와 mlflow_trace_url이 모두 포함되어 있어 상세 실행 트레이스를 분석할 수 있어요. 트레이스는 검색, 생성, 평가 단계 중 정확히 어디에서 실패가 발생하는지 이해하는 데 도움을 줘요.

RAG 시스템은 MLflow 서버(앞서 시작한)에 트레이스를 자동으로 기록하며, http://127.0.0.1:5000에서 볼 수 있어요.

이를 통해:

CSV에서 결과 분석: 응답, 메트릭 점수, 이유 보기
트레이스로 딥다이브: 결과에서 mlflow_trace_url을 클릭하면 MLflow UI에서 해당 평가의 상세 실행 트레이스로 바로 이동

프로 팁: 디버깅에 Trace URL 클릭

각 평가 결과에는 mlflow_trace_url이 포함되어 있어요. MLflow UI의 트레이스로 바로 연결되는 클릭 가능한 링크죠. 수동으로 탐색하거나 trace ID를 복사할 필요 없이 클릭만 하면 상세 실행 트레이스로 바로 이동해요!

오류 및 실패 모드 분석

평가를 실행한 후 experiments/ 디렉토리의 결과 CSV 파일을 검사해서 실패한 케이스의 패턴을 파악해요. 각 행에는 mlflow_trace_id/mlflow_trace_url이 포함되어 있어 MLflow UI에서 상세 실행 트레이스를 볼 수 있어요. 앱을 개선할 수 있도록 각 실패 케이스에 주석을 달아 패턴을 이해해요.

우리 평가의 실제 실패 패턴 분석

이 예제에서 핵심 문제는 검색 실패예요. BM25 리트리버가 답변을 포함하는 문서를 찾지 못하고 있어요. 모델은 문서에 정보가 없을 때 말하라는 지침을 올바르게 따르지만, 잘못된 문서가 검색되고 있는 것이에요.

문서 검색 불량 예제

BM25 리트리버가 답변을 포함하는 관련 문서를 검색하지 못해요.

Question	Expected Answer	Model Response	Root Cause
"What is the default repository type for create_repo?"	`model`	"The provided documents do not state the default repository type..."	BM25 missed docs with create_repo details
"What is the purpose of the BLIP-Diffusion model?"	"controllable text-to-image generation and editing"	"The provided documents do not mention BLIP‑Diffusion..."	BM25 didn't retrieve BLIP-Diffusion docs
"What is the name of the new Hugging Face library for hosting scikit-learn models?"	`Skops`	"The provided documents do not mention or name any new Hugging Face library..."	BM25 missed Skops documentation

이 분석을 바탕으로 검색이 주요 병목이라는 것을 알 수 있어요. 표적 개선을 적용해볼게요.

RAG 앱 개선

검색이 주요 병목으로 확인됐으니 시스템을 두 가지 방법으로 개선할 수 있어요.

전통적인 접근 방식은 더 나은 청킹, 하이브리드 검색, 벡터 임베딩에 초점을 맞춰요. 하지만 우리 BM25 검색이 단일 쿼리로 관련 문서를 일관되게 놓치고 있으므로, 대신 agentic 접근 방식을 탐구할게요.

Agentic RAG는 AI가 검색 전략을 반복적으로 다듬게 해요. 정적 쿼리 하나에 의존하는 대신 여러 검색어를 시도하고 충분한 컨텍스트를 찾았을 때를 스스로 결정하게 하죠.

Agentic RAG 구현

flowchart LR
    A[User Query] --> B[AI Agent<br/>OpenAI]
    B --> C[BM25 Tool]
    C --> B
    B --> D[Final Answer]

샘플 질의로 Agentic RAG 앱을 실행해요.

# agentic 모드로 전환
rag_agentic = RAG(openai_client, retriever, mode="agentic")

question = "What architecture is the `tokenizers-linux-x64-musl` binary designed for?"
result = await rag_agentic.query(question)
print(f"Answer: {result['answer']}")

출력

Answer: It targets x86_64 — i.e. the x86_64-unknown-linux-musl target triple.

Agentic RAG 구현 이해하기

Agentic RAG 모드는 OpenAI Agents SDK를 사용해서 BM25 검색 도구를 가진 AI 에이전트를 만들어요.

# mode="agentic"일 때 RAG 클래스의 핵심 컴포넌트
from agents import Agent, Runner, function_tool

def _setup_agent(self):
    """agentic 모드용 에이전트 설정."""
    @function_tool
    def retrieve(query: str) -> str:
        """주어진 쿼리에 대해 BM25 리트리버로 문서 검색."""
        docs = self.retriever.retrieve(query, self.default_k)
        if not docs:
            return "No documents found."
        return "\n\n".join([f"Doc {i}: {doc.page_content}" for i, doc in enumerate(docs, 1)])

    self._agent = Agent(
        name="RAG Assistant",
        model=self.model,
        instructions="Use short keywords to search. Try 2-3 different searches. Only answer based on documents. Be concise.",
        tools=[retrieve]
    )

async def _agentic_query(self, question: str, top_k: int) -> Dict[str, Any]:
    """agentic 모드 처리: 에이전트가 검색 전략을 제어."""
    result = await Runner.run(self._agent, input=question)
    print(result.answer)

naive 모드의 단일 검색 호출과 달리 에이전트는 언제 어떻게 검색할지 자율적으로 결정해요. 충분한 컨텍스트를 찾을 때까지 여러 키워드 조합을 시도하죠.

실험 다시 실행하고 결과 비교

이제 agentic RAG 접근 방식을 평가해요.

# 필요한 컴포넌트 임포트
import asyncio
from datetime import datetime
from dotenv import load_dotenv

# 환경 변수 로드
load_dotenv()

from ragas_examples.improve_rag.evals import (
    evaluate_rag,
    download_and_save_dataset, 
    create_ragas_dataset,
    get_openai_client,
    get_llm_client
)
from ragas_examples.improve_rag.rag import RAG, BM25Retriever

async def run_agentic_evaluation():
    # 데이터셋 다운로드 및 준비
    dataset_path = download_and_save_dataset()
    dataset = create_ragas_dataset(dataset_path)

    # agentic 모드로 RAG 컴포넌트 초기화
    openai_client = get_openai_client()
    retriever = BM25Retriever()
    rag = RAG(llm_client=openai_client, retriever=retriever, model="gpt-5-mini", mode="agentic")
    llm = get_llm_client()

    # 평가 실험 실행
    exp_name = f"{datetime.now().strftime('%Y%m%d-%H%M%S')}_agenticrag"
    results = await evaluate_rag.arun(
        dataset, 
        name=exp_name,
        rag=rag,
        llm=llm
    )

    # 결과 출력
    if results:
        pass_count = sum(1 for result in results if result.get("correctness_score") == "pass")
        total_count = len(results)
        pass_rate = (pass_count / total_count) * 100 if total_count > 0 else 0
        print(f"Results: {pass_count}/{total_count} passed ({pass_rate:.1f}%)")

    return results

# agentic 평가 실행
results = await run_agentic_evaluation()
print("\nDetailed results:")
print(results)

Agentic RAG 평가 출력

Results: 58/66 passed (87.9%)

훌륭해요! 65.2%(naive)에서 87.9%(agentic)로 큰 개선을 이뤘어요. agentic RAG 접근 방식으로 22.7퍼센트포인트 개선된 거예요!

성능 비교

agentic RAG 접근 방식은 naive RAG 베이스라인 대비 큰 개선을 보여줘요.

Approach	Correctness	Improvement
Naive RAG	65.2%	-
Agentic RAG	87.9%	+22.7%

이 루프를 RAG 시스템에 적용하기

어떤 RAG 시스템이든 이 체계적인 접근 방식으로 개선할 수 있어요.

평가 데이터셋 만들기: 시스템의 실제 쿼리를 사용하거나 LLM으로 합성 데이터 생성
메트릭 정의: 사용 사례에 맞는 간단한 메트릭 선택. 집중적으로 유지
베이스라인 평가 실행: 현재 성능을 측정하고 오류 패턴을 분석해서 체계적인 실패 식별
표적 개선 구현: 오류 분석을 바탕으로 검색(청킹, 하이브리드 검색), 생성(프롬프트, 모델) 개선 또는 agentic 접근 시도
비교 및 반복: 베이스라인에 대한 개선 테스트. 한 번에 하나씩 바꾸면서 정확도가 비즈니스 요구를 충족할 때까지

Ragas 프레임워크는 오케스트레이션과 결과 집계를 자동으로 처리해서 평가 인프라 구축이 아닌 분석과 개선에 집중할 수 있게 해줘요.

RAG 앱 평가하고 개선하기