Split View: RAG 챗봇 평가 실전: 오프라인/온라인 품질 측정부터 프로덕션 가드레일까지

"""
RAG 평가용 Golden Dataset.
각 테스트 케이스는 질문, 정답, 정답 근거 문서, 기대 카테고리를 포함한다.
"""
from dataclasses import dataclass, field
from typing import List, Optional

@dataclass
class GoldenTestCase:
    """단일 평가 케이스"""
    question_id: str
    question: str
    expected_answer: str
    relevant_doc_ids: List[str]     # 정답을 포함하는 문서 ID 목록
    category: str                    # faq, policy, product, troubleshooting
    difficulty: str                  # easy, medium, hard
    expected_citations: List[str]    # 답변에 인용되어야 할 문서 ID
    negative_assertions: List[str] = field(default_factory=list)  # 답변에 포함되면 안 되는 내용

# 평가 셋 예시
GOLDEN_DATASET = [
    GoldenTestCase(
        question_id="faq_001",
        question="환불 처리는 며칠이 걸리나요?",
        expected_answer="환불은 요청일로부터 영업일 기준 3-5일 이내에 처리됩니다.",
        relevant_doc_ids=["doc_refund_policy_v3", "doc_faq_payment"],
        category="faq",
        difficulty="easy",
        expected_citations=["doc_refund_policy_v3"],
        negative_assertions=["즉시 환불", "당일 처리"],
    ),
    GoldenTestCase(
        question_id="policy_002",
        question="해외 배송 시 관세는 누가 부담하나요?",
        expected_answer="해외 배송 시 관세 및 부가세는 수령인 부담입니다.",
        relevant_doc_ids=["doc_shipping_international", "doc_customs_guide"],
        category="policy",
        difficulty="medium",
        expected_citations=["doc_shipping_international"],
        negative_assertions=["관세 면제", "무료 배송"],
    ),
    GoldenTestCase(
        question_id="trouble_003",
        question="앱에서 결제 오류 P4021이 발생합니다. 어떻게 해결하나요?",
        expected_answer="P4021 오류는 카드사 인증 실패로 발생합니다. 카드사 앱에서 온라인 결제를 활성화한 후 재시도하세요.",
        relevant_doc_ids=["doc_error_codes", "doc_payment_troubleshoot"],
        category="troubleshooting",
        difficulty="hard",
        expected_citations=["doc_error_codes"],
        negative_assertions=["고객센터에 문의", "알 수 없는 오류"],
    ),
]

"""
검색 단계의 품질을 측정하는 모듈.
Recall@K, MRR(Mean Reciprocal Rank), nDCG를 계산한다.
"""
from typing import List, Dict
import numpy as np

def recall_at_k(
    retrieved_doc_ids: List[str],
    relevant_doc_ids: List[str],
    k: int = 5,
) -> float:
    """
    상위 K개 검색 결과에 관련 문서가 포함된 비율.
    예: relevant_docs = ["A", "B"], retrieved = ["C", "A", "D", "B", "E"]
    recall@5 = 2/2 = 1.0
    """
    retrieved_set = set(retrieved_doc_ids[:k])
    relevant_set = set(relevant_doc_ids)
    if not relevant_set:
        return 1.0  # 관련 문서가 없으면 recall은 무의미
    return len(retrieved_set & relevant_set) / len(relevant_set)

def mean_reciprocal_rank(
    retrieved_doc_ids: List[str],
    relevant_doc_ids: List[str],
) -> float:
    """
    첫 번째 관련 문서의 역수 순위.
    예: relevant = ["B"], retrieved = ["A", "B", "C"] -> MRR = 1/2 = 0.5
    """
    relevant_set = set(relevant_doc_ids)
    for i, doc_id in enumerate(retrieved_doc_ids):
        if doc_id in relevant_set:
            return 1.0 / (i + 1)
    return 0.0

def evaluate_retrieval(
    test_cases: List[dict],
    retriever,
    k_values: List[int] = [3, 5, 10],
) -> Dict[str, float]:
    """
    전체 테스트 셋에 대해 retrieval 품질을 측정한다.
    """
    results = {f"recall@{k}": [] for k in k_values}
    results["mrr"] = []

    for case in test_cases:
        retrieved = retriever.search(
            query=case["question"],
            top_k=max(k_values),
        )
        retrieved_ids = [doc.id for doc in retrieved]

        for k in k_values:
            r = recall_at_k(retrieved_ids, case["relevant_doc_ids"], k)
            results[f"recall@{k}"].append(r)

        mrr = mean_reciprocal_rank(retrieved_ids, case["relevant_doc_ids"])
        results["mrr"].append(mrr)

    return {
        metric: round(float(np.mean(values)), 4)
        for metric, values in results.items()
    }

"""
LLM-as-a-Judge 패턴으로 RAG 답변의 Grounding과 Relevance를 평가한다.
OpenAI GPT-4o 또는 Claude 3.5 Sonnet을 판정자로 사용한다.
"""
from dataclasses import dataclass
from typing import List, Optional
import json

FAITHFULNESS_PROMPT = """당신은 RAG 챗봇 답변의 충실도를 평가하는 전문 판정자입니다.

주어진 정보:
- 사용자 질문: {question}
- 검색된 문서들: {retrieved_contexts}
- 챗봇 답변: {answer}

평가 기준:
1. 답변의 모든 주장(claim)이 검색된 문서에 근거하는가?
2. 문서에 없는 내용을 추가하거나 꾸며내지 않았는가?
3. 문서의 내용을 왜곡하지 않았는가?

반드시 아래 JSON 형식으로 응답하세요:
{{
  "faithfulness_score": <0.0-1.0>,
  "claims": [
    {{
      "claim": "<답변에서 추출한 주장>",
      "supported": <true/false>,
      "evidence": "<근거 문서 내용 또는 'no evidence found'>"
    }}
  ],
  "unsupported_claims_count": <int>,
  "hallucination_detected": <true/false>
}}"""

RELEVANCE_PROMPT = """당신은 RAG 챗봇 답변의 관련성을 평가하는 전문 판정자입니다.

주어진 정보:
- 사용자 질문: {question}
- 챗봇 답변: {answer}

평가 기준:
1. 답변이 질문에 직접적으로 대답하는가?
2. 불필요한 정보가 과도하게 포함되어 있지 않은가?
3. 핵심 정보가 누락되지 않았는가?

반드시 아래 JSON 형식으로 응답하세요:
{{
  "relevance_score": <0.0-1.0>,
  "addresses_question": <true/false>,
  "missing_information": "<누락된 핵심 정보 또는 'none'>",
  "unnecessary_information": "<불필요한 정보 또는 'none'>"
}}"""

@dataclass
class JudgmentResult:
    question_id: str
    faithfulness_score: float
    relevance_score: float
    hallucination_detected: bool
    unsupported_claims: int
    missing_info: str
    raw_judgment: dict

async def evaluate_with_llm_judge(
    question: str,
    answer: str,
    retrieved_contexts: List[str],
    judge_client,  # OpenAI 또는 Anthropic 클라이언트
    question_id: str = "",
) -> JudgmentResult:
    """LLM 판정자로 faithfulness와 relevance를 평가한다."""

    # Faithfulness 평가
    faith_prompt = FAITHFULNESS_PROMPT.format(
        question=question,
        retrieved_contexts="\n---\n".join(retrieved_contexts),
        answer=answer,
    )
    faith_response = await judge_client.chat.completions.create(
        model="gpt-4o-2024-11-20",
        messages=[{"role": "user", "content": faith_prompt}],
        response_format={"type": "json_object"},
        temperature=0.0,  # 판정 일관성을 위해 temperature 0
    )
    faith_result = json.loads(faith_response.choices[0].message.content)

    # Relevance 평가
    rel_prompt = RELEVANCE_PROMPT.format(question=question, answer=answer)
    rel_response = await judge_client.chat.completions.create(
        model="gpt-4o-2024-11-20",
        messages=[{"role": "user", "content": rel_prompt}],
        response_format={"type": "json_object"},
        temperature=0.0,
    )
    rel_result = json.loads(rel_response.choices[0].message.content)

    return JudgmentResult(
        question_id=question_id,
        faithfulness_score=faith_result.get("faithfulness_score", 0.0),
        relevance_score=rel_result.get("relevance_score", 0.0),
        hallucination_detected=faith_result.get("hallucination_detected", False),
        unsupported_claims=faith_result.get("unsupported_claims_count", 0),
        missing_info=rel_result.get("missing_information", "unknown"),
        raw_judgment={"faithfulness": faith_result, "relevance": rel_result},
    )

"""
프로덕션 RAG 챗봇의 실시간 품질 지표를 수집하는 미들웨어.
모든 요청-응답 쌍에 대해 경량 품질 신호를 수집한다.
"""
import time
import hashlib
from dataclasses import dataclass, field
from typing import List, Optional, Dict
from prometheus_client import Histogram, Counter, Gauge

# Prometheus 메트릭 정의
RESPONSE_LATENCY = Histogram(
    "rag_response_latency_seconds",
    "RAG 응답 전체 소요 시간",
    ["pipeline_stage"],  # retrieval, generation, total
    buckets=[0.1, 0.25, 0.5, 1.0, 2.0, 5.0, 10.0],
)
RETRIEVAL_EMPTY = Counter(
    "rag_retrieval_empty_total",
    "검색 결과가 0건인 요청 수",
)
CITATION_RATE = Gauge(
    "rag_citation_rate",
    "최근 100건 중 인용을 포함한 응답 비율",
)
THUMBS_DOWN = Counter(
    "rag_thumbs_down_total",
    "사용자 부정적 피드백 수",
    ["category"],
)

@dataclass
class QualitySignals:
    """한 건의 요청-응답에서 수집하는 경량 품질 신호"""
    request_id: str
    retrieval_count: int              # 검색된 문서 수
    retrieval_latency_ms: float
    top_similarity_score: float       # 최상위 검색 결과의 유사도 점수
    generation_latency_ms: float
    response_length_chars: int
    has_citation: bool                # 응답에 출처 인용이 포함되었는가
    language_detected: str            # 응답 언어
    contains_hedging: bool            # "아마", "~일 수 있습니다" 같은 불확실 표현 포함 여부

def collect_quality_signals(
    request_id: str,
    query: str,
    retrieved_docs: list,
    response: str,
    retrieval_time: float,
    generation_time: float,
) -> QualitySignals:
    """요청-응답 쌍에서 품질 신호를 추출한다."""
    top_score = max((doc.score for doc in retrieved_docs), default=0.0)
    has_citation = any(
        marker in response
        for marker in ["[출처:", "[참고:", "[문서", "출처:"]
    )
    hedging_phrases = ["아마", "~일 수 있습니다", "정확하지 않을 수", "확인이 필요합니다"]
    contains_hedging = any(phrase in response for phrase in hedging_phrases)

    signals = QualitySignals(
        request_id=request_id,
        retrieval_count=len(retrieved_docs),
        retrieval_latency_ms=retrieval_time * 1000,
        top_similarity_score=top_score,
        generation_latency_ms=generation_time * 1000,
        response_length_chars=len(response),
        has_citation=has_citation,
        language_detected="ko" if any('\uac00' <= c <= '\ud7a3' for c in response) else "en",
        contains_hedging=contains_hedging,
    )

    # Prometheus 메트릭 기록
    RESPONSE_LATENCY.labels(pipeline_stage="retrieval").observe(retrieval_time)
    RESPONSE_LATENCY.labels(pipeline_stage="generation").observe(generation_time)
    RESPONSE_LATENCY.labels(pipeline_stage="total").observe(retrieval_time + generation_time)

    if signals.retrieval_count == 0:
        RETRIEVAL_EMPTY.inc()

    return signals

# prometheus-rules.yaml
groups:
  - name: rag_quality_alerts
    interval: 30s
    rules:
      # 검색 빈 결과 비율이 10%를 초과하면 경고
      - alert: RAGRetrievalEmptyRateHigh
        expr: |
          rate(rag_retrieval_empty_total[10m]) /
          rate(rag_response_latency_seconds_count{pipeline_stage="total"}[10m]) > 0.10
        for: 5m
        labels:
          severity: warning
          team: chatbot
        annotations:
          summary: 'RAG 검색 빈 결과 비율이 10%를 초과합니다'
          description: |
            현재 빈 결과 비율: {{ $value | humanizePercentage }}
            인덱스 상태, 임베딩 모델, 쿼리 전처리를 확인하세요.

      # 인용 포함 비율이 70% 미만으로 떨어지면 경고
      - alert: RAGCitationRateLow
        expr: rag_citation_rate < 0.70
        for: 10m
        labels:
          severity: warning
          team: chatbot
        annotations:
          summary: 'RAG 인용 포함 비율이 70% 미만입니다'
          description: |
            현재 인용 비율: {{ $value | humanizePercentage }}
            LLM 프롬프트의 인용 지시 또는 검색 품질을 확인하세요.

      # 응답 지연이 급증하면 경고
      - alert: RAGResponseLatencyHigh
        expr: |
          histogram_quantile(0.95,
            rate(rag_response_latency_seconds_bucket{pipeline_stage="total"}[5m])
          ) > 5.0
        for: 3m
        labels:
          severity: critical
          team: chatbot
        annotations:
          summary: 'RAG 응답 p95 지연이 5초를 초과합니다'

      # 사용자 부정적 피드백 급증
      - alert: RAGNegativeFeedbackSpike
        expr: |
          rate(rag_thumbs_down_total[30m]) > 2 * rate(rag_thumbs_down_total[24h] offset 1d)
        for: 15m
        labels:
          severity: warning
          team: chatbot
        annotations:
          summary: '사용자 부정적 피드백이 전일 대비 2배 이상 증가했습니다'

"""
RAG 품질 회귀 테스트.
Golden dataset에 대해 4개 차원의 품질을 측정하고,
이전 버전 대비 회귀 여부를 판정한다.
"""
import json
import sys
from pathlib import Path
from typing import Dict

# 품질 기준선 (이전 배포 버전의 성적)
BASELINE_SCORES = {
    "recall@5": 0.88,
    "mrr": 0.75,
    "faithfulness": 0.91,
    "relevance": 0.87,
    "safety_pass_rate": 1.00,
    "citation_rate": 0.82,
}

# 허용 가능한 하락 폭 (절대값)
REGRESSION_TOLERANCE = {
    "recall@5": 0.03,       # 3% 하락까지 허용
    "mrr": 0.03,
    "faithfulness": 0.02,   # faithfulness는 엄격하게
    "relevance": 0.03,
    "safety_pass_rate": 0.0,  # safety는 회귀 불허
    "citation_rate": 0.05,
}

def check_regression(current_scores: Dict[str, float]) -> dict:
    """
    현재 점수를 기준선과 비교하여 회귀 여부를 판정한다.
    Returns: {"passed": bool, "regressions": [...], "improvements": [...]}
    """
    regressions = []
    improvements = []

    for metric, baseline in BASELINE_SCORES.items():
        current = current_scores.get(metric, 0.0)
        tolerance = REGRESSION_TOLERANCE.get(metric, 0.0)
        delta = current - baseline

        if delta < -tolerance:
            regressions.append({
                "metric": metric,
                "baseline": baseline,
                "current": current,
                "delta": round(delta, 4),
                "tolerance": tolerance,
            })
        elif delta > 0.01:  # 1% 이상 개선
            improvements.append({
                "metric": metric,
                "baseline": baseline,
                "current": current,
                "delta": round(delta, 4),
            })

    passed = len(regressions) == 0
    return {
        "passed": passed,
        "regressions": regressions,
        "improvements": improvements,
        "summary": (
            f"PASSED: {len(improvements)} improvements, 0 regressions"
            if passed
            else f"FAILED: {len(regressions)} regressions detected"
        ),
    }

if __name__ == "__main__":
    # CI에서 실행: python eval/regression_check.py results.json
    results_path = sys.argv[1] if len(sys.argv) > 1 else "eval_results.json"
    with open(results_path) as f:
        current_scores = json.load(f)

    result = check_regression(current_scores)
    print(json.dumps(result, indent=2, ensure_ascii=False))

    if not result["passed"]:
        print("\n=== REGRESSION DETAILS ===")
        for reg in result["regressions"]:
            print(f"  {reg['metric']}: {reg['baseline']:.4f} -> {reg['current']:.4f} "
                  f"(delta: {reg['delta']:.4f}, tolerance: {reg['tolerance']:.4f})")
        sys.exit(1)

# .github/workflows/rag-eval.yml
name: RAG Quality Evaluation
on:
  push:
    paths:
      - 'src/rag/**'
      - 'prompts/**'
      - 'config/retrieval/**'
  pull_request:
    paths:
      - 'src/rag/**'
      - 'prompts/**'

jobs:
  offline-eval:
    runs-on: ubuntu-latest
    timeout-minutes: 30
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: pip install -r requirements-eval.txt

      - name: Run retrieval evaluation
        run: |
          python eval/retrieval_eval.py \
            --dataset eval/golden_dataset.json \
            --k-values 3,5,10 \
            --output eval_results_retrieval.json

      - name: Run LLM judge evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          python eval/llm_judge_eval.py \
            --dataset eval/golden_dataset.json \
            --judge-model gpt-4o-2024-11-20 \
            --output eval_results_judge.json

      - name: Run safety evaluation
        run: |
          python eval/safety_eval.py \
            --dataset eval/golden_dataset.json \
            --output eval_results_safety.json

      - name: Merge results and check regression
        run: |
          python eval/merge_results.py \
            eval_results_retrieval.json \
            eval_results_judge.json \
            eval_results_safety.json \
            --output eval_results.json

          python eval/regression_check.py eval_results.json

      - name: Upload evaluation report
        uses: actions/upload-artifact@v4
        if: always()
        with:
          name: eval-report
          path: eval_results*.json

"""
RAGAS 프레임워크를 활용한 RAG 평가 예시.
pip install ragas==0.2.6
"""
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset

# 평가 데이터 준비 (RAGAS 형식)
eval_data = {
    "question": [
        "환불 처리는 며칠이 걸리나요?",
        "해외 배송 시 관세는 누가 부담하나요?",
    ],
    "answer": [
        "환불은 영업일 기준 3-5일 이내에 처리됩니다.",
        "해외 배송 시 관세와 부가세는 수령인이 부담합니다.",
    ],
    "contexts": [
        ["환불 요청일로부터 영업일 기준 3-5일 이내에 환불이 처리됩니다. 주말 및 공휴일은 영업일에 포함되지 않습니다."],
        ["해외 배송 시 발생하는 관세 및 부가세는 수령인 부담입니다. 관세 금액은 수입 국가의 세율에 따라 달라집니다."],
    ],
    "ground_truth": [
        "환불은 요청일로부터 영업일 기준 3-5일 이내에 처리됩니다.",
        "해외 배송 시 관세 및 부가세는 수령인 부담입니다.",
    ],
}

dataset = Dataset.from_dict(eval_data)

# RAGAS 평가 실행
results = evaluate(
    dataset=dataset,
    metrics=[
        faithfulness,        # 답변이 컨텍스트에 충실한가
        answer_relevancy,    # 답변이 질문에 관련되는가
        context_precision,   # 검색된 컨텍스트가 정확한가
        context_recall,      # 관련 컨텍스트를 빠짐없이 검색했는가
    ],
)

print(results)
# {'faithfulness': 0.95, 'answer_relevancy': 0.92,
#  'context_precision': 0.88, 'context_recall': 0.90}

증상: 오프라인 평가에서 faithfulness가 0.91에서 0.72로 하락
     사용자 리포트: "근거 없는 답변이 늘었다"

점검 순서:
  1. LLM 모델 버전 변경 여부 확인
     -> gpt-4o-2024-08-06에서 gpt-4o-2024-11-20으로 업데이트됨
  2. 시스템 프롬프트의 grounding 지시 확인
     -> 모델 업데이트 후 프롬프트 튜닝이 안 됨

해결:
  1. 시스템 프롬프트에 "반드시 검색된 문서에 근거하여 답변하세요.
     문서에 없는 내용은 '해당 정보를 찾을 수 없습니다'라고 답변하세요." 강화
  2. 모델 업데이트 전 반드시 오프라인 평가 수행하는 규칙 수립
  3. 모델별 프롬프트 버전 관리 (model_version -> prompt_version 매핑 테이블)

증상: rag_retrieval_empty_total 카운터가 평소 대비 5배 증가
     에러 로그: 없음 (검색 자체는 성공하지만 결과가 0건)

원인: 임베딩 모델 업데이트 후 새 쿼리 임베딩과 기존 문서 임베딩의
     벡터 공간이 달라짐 (cosine similarity가 전반적으로 낮아짐)

해결:
  1. 즉시 이전 임베딩 모델로 롤백
  2. 새 임베딩 모델 적용 시 전체 문서 재임베딩 필수
  3. 임베딩 모델 변경 절차에 "재인덱싱 완료 확인" 단계 추가

증상: 월 LLM Judge 비용이 $3,000에서 $12,000으로 증가
원인: 테스트 셋이 50건에서 500건으로 증가했고, 매 PR마다 실행

해결:
  1. Golden dataset을 core(50건) + extended(450건)으로 분리
  2. PR에서는 core만 실행, main merge 시 extended 포함
  3. LLM Judge 캐싱: 동일 입력에 대한 판정 결과를 30일간 캐시
  4. 저비용 모델(gpt-4o-mini)로 1차 필터링 후 실패 케이스만 gpt-4o로 재판정