RAG Evaluation: Precision/Recall을 넘어서

def compute_faithfulness(answer: str, contexts: List[str]) -> float:
    """
    1. 답변에서 개별 주장(claim) 추출
    2. 각 주장이 context에서 지지되는지 확인
    3. 지지되는 주장 비율 반환
    """
    claims = extract_claims(answer)
    supported = 0

    for claim in claims:
        if is_supported_by_context(claim, contexts):
            supported += 1

    return supported / len(claims) if claims else 0

def compute_answer_relevance(question: str, answer: str) -> float:
    """
    1. 답변에서 질문을 역생성
    2. 원본 질문과 역생성 질문의 유사도 측정
    """
    # 답변만 보고 질문을 추측
    generated_questions = generate_questions_from_answer(answer, n=3)

    # 원본 질문과의 유사도
    similarities = [
        semantic_similarity(question, gen_q)
        for gen_q in generated_questions
    ]

    return np.mean(similarities)

def compute_context_recall(
    ground_truth: str,
    contexts: List[str]
) -> float:
    """
    1. Ground truth에서 핵심 문장 추출
    2. 각 문장이 context에서 지지되는지 확인
    """
    gt_statements = extract_statements(ground_truth)
    attributed = 0

    for statement in gt_statements:
        if any(supports(ctx, statement) for ctx in contexts):
            attributed += 1

    return attributed / len(gt_statements) if gt_statements else 0

def compute_context_precision(
    question: str,
    answer: str,
    contexts: List[str]
) -> float:
    """
    검색된 각 context가 답변에 기여했는지 확인
    """
    useful = 0

    for ctx in contexts:
        if contributes_to_answer(ctx, question, answer):
            useful += 1

    return useful / len(contexts) if contexts else 0

# pip install ragas

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_recall,
    context_precision,
)
from datasets import Dataset

# 평가 데이터 준비
eval_data = {
    "question": ["Tesla CEO는 누구야?"],
    "answer": ["Elon Musk가 Tesla의 CEO입니다."],
    "contexts": [["Elon Musk는 Tesla의 CEO이자 창업자이다."]],
    "ground_truth": ["Elon Musk"]  # Context Recall 계산에 필요
}

dataset = Dataset.from_dict(eval_data)

# 평가 실행
results = evaluate(
    dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_recall,
        context_precision,
    ]
)

print(results)

def evaluate_rag_batch(
    questions: List[str],
    rag_system,
    ground_truths: List[str] = None
) -> pd.DataFrame:
    """여러 질문에 대해 RAG 시스템 평가"""

    results = []
    for i, question in enumerate(questions):
        # RAG 실행
        answer, contexts = rag_system.query(question)

        # 개별 평가
        result = {
            "question": question,
            "answer": answer,
            "faithfulness": compute_faithfulness(answer, contexts),
            "relevance": compute_answer_relevance(question, answer),
            "context_precision": compute_context_precision(
                question, answer, contexts
            ),
        }

        if ground_truths:
            result["context_recall"] = compute_context_recall(
                ground_truths[i], contexts
            )

        results.append(result)

    return pd.DataFrame(results)

FAITHFULNESS_PROMPT = """
Given the context and answer, determine if each claim in the answer
is supported by the context.

Context:
{context}

Answer:
{answer}

For each claim in the answer, respond with:
- Claim: [the claim]
- Verdict: [Supported/Not Supported]
- Evidence: [quote from context if supported]

Finally, provide the overall faithfulness score (0-1).
"""

def llm_faithfulness(answer: str, context: str, llm) -> float:
    prompt = FAITHFULNESS_PROMPT.format(context=context, answer=answer)
    response = llm.generate(prompt)
    return parse_faithfulness_score(response)

RELEVANCE_PROMPT = """
Given the question and answer, rate how relevant the answer is
to the question on a scale of 0-1.

Question: {question}
Answer: {answer}

Consider:
- Does the answer address the question directly?
- Is the answer complete?
- Is there irrelevant information?

Score (0-1):
Reasoning:
"""

def llm_relevance(question: str, answer: str, llm) -> float:
    prompt = RELEVANCE_PROMPT.format(question=question, answer=answer)
    response = llm.generate(prompt)
    return parse_relevance_score(response)

# 다양한 유형의 질문 포함
eval_set = {
    "simple": [  # 단일 문서로 답 가능
        "Tesla CEO는 누구야?",
        "OpenAI는 언제 설립됐어?",
    ],
    "multi_hop": [  # 여러 문서 필요
        "OpenAI CEO가 해고됐을 때 Microsoft CEO는 뭐라고 했어?",
    ],
    "temporal": [  # 시간 추론 필요
        "Sam Altman 복귀 전 CEO는 누구였어?",
    ],
    "comparison": [  # 비교 질문
        "Tesla와 BYD 중 2023년 판매량이 높은 곳은?",
    ],
    "unanswerable": [  # 답할 수 없는 질문
        "2025년 Tesla 판매량은?",
    ]
}

class RAGEvaluator:
    def __init__(self, rag_system, llm_judge):
        self.rag = rag_system
        self.judge = llm_judge
        self.metrics_history = []

    def evaluate(self, eval_set: Dict[str, List[str]]) -> Dict:
        results = {}

        for category, questions in eval_set.items():
            category_results = []

            for question in questions:
                answer, contexts = self.rag.query(question)

                metrics = {
                    "faithfulness": self.compute_faithfulness(answer, contexts),
                    "relevance": self.compute_relevance(question, answer),
                    "context_precision": self.compute_precision(
                        question, answer, contexts
                    ),
                }
                category_results.append(metrics)

            results[category] = {
                "avg_faithfulness": np.mean([r["faithfulness"] for r in category_results]),
                "avg_relevance": np.mean([r["relevance"] for r in category_results]),
                "avg_precision": np.mean([r["context_precision"] for r in category_results]),
            }

        self.metrics_history.append({
            "timestamp": datetime.now(),
            "results": results
        })

        return results

    def compare_versions(self, v1_results: Dict, v2_results: Dict) -> Dict:
        """두 버전의 RAG 시스템 비교"""
        comparison = {}
        for category in v1_results:
            comparison[category] = {
                metric: v2_results[category][metric] - v1_results[category][metric]
                for metric in v1_results[category]
            }
        return comparison