RAG 파이프라인 평가하기

RAG 파이프라인 평가하기 (Evaluation)

만든 RAG 파이프라인이 얼마나 잘 동작하는지 평가(evaluation) 하는 방법을 살펴볼게요. 모델 기반 지표와 통계 기반 지표를 모두 활용해서, 검색 단계와 생성 단계를 각각 그리고 전체를 한 번에 점검하는 방식이에요.

출처: 공식문서

개요

RAG 파이프라인은 기본적으로 검색(Retrieval) 과 생성(Generation) 두 단계로 이뤄져요. 파이프라인 전체를 평가하려면 각 단계를 따로 보면서 동시에 전체로도 봐야 해요. 검색 단계는 정답 라벨이 있으면 통계적 지표로 평가할 수 있지만, 생성 단계는 그렇게 하기 어려워요. 그래서 생성 단계는 LLM을 '평가자'로 쓰는 모델 기반 지표에 주로 의존해요.

이 튜토리얼에서 쓰는 지표는 세 가지예요.

Document MRR (Mean Reciprocal Rank): 정답 문서가 검색 결과 목록의 몇 번째 순위에 나왔는지 확인해요. 정답 라벨이 필요해요.
Semantic Answer Similarity (SAS): 예측 답과 정답 답의 의미적 유사도를 파인튜닝된 언어 모델로 측정해요. 정답 라벨이 필요해요.
Faithfulness: 생성된 답이 주어진 컨텍스트에서 추론 가능한지 LLM으로 판단해요. 정답 라벨이 필요 없어요.

설치

pip install haystack-ai
pip install "datasets>=2.6.1"
pip install sentence-transformers-haystack

평가할 RAG 파이프라인 만들기

평가하려면 평가할 파이프라인이 먼저 있어야 해요. 라벨이 붙은 PubMed 데이터셋(vblagoje/PubMedQA_instruction)을 사용해요. 질문·컨텍스트·답이 있어서, 컨텍스트는 문서로 쓰고 라벨 데이터는 평가 지표에 활용해요. 예시로 첫 1000행만 사용해요.

from datasets import load_dataset
from haystack import Document

dataset = load_dataset("vblagoje/PubMedQA_instruction", split="train")
dataset = dataset.select(range(1000))
all_documents = [Document(content=doc["context"]) for doc in dataset]
all_questions = [doc["instruction"] for doc in dataset]
all_ground_truth_answers = [doc["response"] for doc in dataset]

문서를 색인하는 파이프라인을 만들고 InMemoryDocumentStore에 저장해요.

from typing import List
from haystack import Pipeline
from haystack_integrations.components.embedders.sentence_transformers import SentenceTransformersDocumentEmbedder
from haystack.components.writers import DocumentWriter
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.document_stores.types import DuplicatePolicy

document_store = InMemoryDocumentStore()

document_embedder = SentenceTransformersDocumentEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
document_writer = DocumentWriter(document_store=document_store, policy=DuplicatePolicy.SKIP)

indexing = Pipeline()
indexing.add_component(instance=document_embedder, name="document_embedder")
indexing.add_component(instance=document_writer, name="document_writer")

indexing.connect("document_embedder.documents", "document_writer.documents")

indexing.run({"document_embedder": {"documents": all_documents}})

이제 단순한 RAG 파이프라인을 만들어요. 리트리버는 InMemoryEmbeddingRetriever, 생성기는 OpenAIChatGenerator예요. 다른 ChatGenerator로 바꿔도 돼요.

import os
from getpass import getpass
from haystack.components.builders import AnswerBuilder, ChatPromptBuilder
from haystack.dataclasses import ChatMessage
from haystack_integrations.components.embedders.sentence_transformers import SentenceTransformersTextEmbedder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever

if "OPENAI_API_KEY" not in os.environ:
    os.environ["OPENAI_API_KEY"] = getpass("Enter OpenAI API key:")

template = [
    ChatMessage.from_user(
        """
        You have to answer the following question based on the given context information only.

        Context:
        {% for document in documents %}
            {{ document.content }}
        {% endfor %}

        Question: {{question}}
        Answer:
        """
    )
]

rag_pipeline = Pipeline()
rag_pipeline.add_component(
    "query_embedder", SentenceTransformersTextEmbedder(model="sentence-transformers/all-MiniLM-L6-v2")
)
rag_pipeline.add_component("retriever", InMemoryEmbeddingRetriever(document_store, top_k=3))
rag_pipeline.add_component("prompt_builder", ChatPromptBuilder(template=template))
rag_pipeline.add_component("generator", OpenAIChatGenerator(model="gpt-4o-mini"))
rag_pipeline.add_component("answer_builder", AnswerBuilder())

rag_pipeline.connect("query_embedder", "retriever.query_embedding")
rag_pipeline.connect("retriever", "prompt_builder.documents")
rag_pipeline.connect("prompt_builder.prompt", "generator.messages")
rag_pipeline.connect("generator.replies", "answer_builder.replies")
rag_pipeline.connect("retriever", "answer_builder.documents")

질문을 던질 때는 query_embedder, prompt_builder, answer_builder 세 컴포넌트 모두에 질문을 넘겨요.

question = "Do high levels of procalcitonin in the early phase after pediatric liver transplantation indicate poor postoperative outcome?"

response = rag_pipeline.run(
    {
        "query_embedder": {"text": question},
        "prompt_builder": {"question": question},
        "answer_builder": {"query": question},
    }
)
print(response["answer_builder"]["answers"][0].data)

파이프라인 평가하기

25개의 랜덤 질문과 정답 라벨을 골라서 RAG 파이프라인을 돌리고, 파이프라인이 반환한 답과 검색된 문서를 기록해요.

import random

questions, ground_truth_answers, ground_truth_docs = zip(
    *random.sample(list(zip(all_questions, all_ground_truth_answers, all_documents)), 25)
)

rag_answers = []
retrieved_docs = []

for question in list(questions):
    response = rag_pipeline.run(
        {
            "query_embedder": {"text": question},
            "prompt_builder": {"question": question},
            "answer_builder": {"query": question},
        }
    )
    print(f"Question: {question}")
    print("Answer from pipeline:")
    print(response["answer_builder"]["answers"][0].data)
    print("\n-----------------------------------\n")

    rag_answers.append(response["answer_builder"]["answers"][0].data)
    retrieved_docs.append(response["answer_builder"]["answers"][0].documents)

각 평가자는 단독 컴포넌트로도 돌릴 수 있지만, 파이프라인에 넣어 한 번에 실행할 수도 있어요. 아래는 세 평가자를 하나의 eval_pipeline로 묶은 예시예요. 각 평가자에 필요한 입력(정답 문서, 예측 답, 컨텍스트 등)을 파이프라인 run()에 넘겨요.

from haystack.components.evaluators.document_mrr import DocumentMRREvaluator
from haystack.components.evaluators.faithfulness import FaithfulnessEvaluator
from haystack.components.evaluators.sas_evaluator import SASEvaluator

eval_pipeline = Pipeline()
eval_pipeline.add_component("doc_mrr_evaluator", DocumentMRREvaluator())
eval_pipeline.add_component("faithfulness", FaithfulnessEvaluator())
eval_pipeline.add_component("sas_evaluator", SASEvaluator(model="sentence-transformers/all-MiniLM-L6-v2"))

results = eval_pipeline.run(
    {
        "doc_mrr_evaluator": {
            "ground_truth_documents": list([d] for d in ground_truth_docs),
            "retrieved_documents": retrieved_docs,
        },
        "faithfulness": {
            "questions": list(questions),
            "contexts": list([d.content] for d in ground_truth_docs),
            "predicted_answers": rag_answers,
        },
        "sas_evaluator": {"predicted_answers": rag_answers, "ground_truth_answers": list(ground_truth_answers)},
    }
)

평가 리포트 만들기

EvaluationRunResult를 이용하면 집계 리포트를 만들 수 있어요.

from haystack.evaluation.eval_run_result import EvaluationRunResult

inputs = {
    "question": list(questions),
    "contexts": list([d.content] for d in ground_truth_docs),
    "answer": list(ground_truth_answers),
    "predicted_answer": rag_answers,
}

evaluation_result = EvaluationRunResult(run_name="pubmed_rag_pipeline", inputs=inputs, results=results)
evaluation_result.aggregated_report()

샘플별 상세 점수를 DataFrame으로도 볼 수 있어요.

results_df = evaluation_result.detailed_report(output_format="df")
results_df

DataFrame이라면 이렇게 pandas로 상·하위 샘플을 골라볼 수 있어요.

import pandas as pd

top_3 = results_df.nlargest(3, "sas_evaluator")
bottom_3 = results_df.nsmallest(3, "sas_evaluator")
pd.concat([top_3, bottom_3])

RAG 파이프라인 평가하기