experiments/rag_paradigm/benchmark.py

"""RAG Paradigm Evolution - 5단계 ablation 비교 실험

production RAG 진화 과정을 5단계로 분해해 각 단계 추가의 효과를 정량 측정한다.

1. **No RAG**: LLM 단독 추론 (베이스라인, hallucination 측정)
2. **Naive RAG (keyword)**: 키워드 매칭 top-K
3. **Vector RAG (FAISS)**: dense embedding + cosine
4. **Hybrid RAG (BM25+FAISS+RRF)**: sparse + dense 결합
5. **Production RAG (Hybrid + Cross-encoder Rerank)**: 정밀 재정렬

각 paradigm을 3개 알람(A1·A2·A3)에 적용해 RAGAS 지표(faithfulness, answer_relevancy,
context_precision)와 latency를 수집하고 matplotlib 차트로 시각화한다.

실행: python -m experiments.rag_paradigm.benchmark
결과: results.md + charts/*.png
"""
import json
import os
import statistics
import time
from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from datasets import Dataset

# 한글 폰트 (macOS 기본 Apple SD Gothic Neo, fallback DejaVu Sans)
plt.rcParams["font.family"] = ["Apple SD Gothic Neo", "AppleGothic", "NanumGothic", "DejaVu Sans"]
plt.rcParams["axes.unicode_minus"] = False
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from ragas import evaluate
from ragas.embeddings import LangchainEmbeddingsWrapper
from ragas.llms import LangchainLLMWrapper
from ragas.metrics import (
    Faithfulness,
    LLMContextPrecisionWithoutReference,
    ResponseRelevancy,
)

from agents.cause import TIER2_SCHEMA

# workflow 모드 (사전 retrieve 후 단일 LLM 호출)에 적합한 prompt
# agents.cause의 SYSTEM_PROMPT는 agentic 모드(tool 자율 호출) 전용이라 별도 정의
WORKFLOW_SYSTEM_PROMPT = """당신은 반도체 공정 원인 분석 전문가입니다.
주어진 이상 알람과 탐지 결과, 사내 지식 문서를 근거로 가장 가능성 높은 원인을
2~3개 추정합니다. 각 원인은 기여도(pct, %)를 가지며 합이 100에 가깝도록 합니다.
근거(evidence)는 제공된 문서 내용에 기반해 구체적으로 작성하고, citations에는
근거가 된 문서 ID만 정확히 기입합니다. 제공되지 않은 문서는 인용하지 않습니다.
기여도가 높은 원인부터 순서대로 제시합니다.
반드시 JSON 스키마에 맞춰 응답하세요."""


def _build_query(alarm: dict, tier1) -> str:
    """원 cause.py에 있던 query builder (agentic 전환으로 제거됨, 본 실험용으로 인라인 보존)"""
    feature = alarm.get("feature") or ""
    sensors = " ".join(f["name"] for f in tier1["features"])
    return f"{alarm['title']} {feature} {sensors} 원인 분석 공정 이상"
from agents.detection import run_detection
from agents.llm import SUBAGENT_MODEL, client
from agents.rag.store import load_document, search

OUT_DIR = Path(__file__).parent
CHART_DIR = OUT_DIR / "charts"
CACHE_CSV = OUT_DIR / "samples.csv"
ALARMS = ["A1", "A2", "A3"]
PARADIGMS = [
    ("No RAG", None),
    ("Naive RAG (keyword)", "keyword"),
    ("Vector RAG (FAISS)", "faiss"),
    ("Hybrid (BM25+FAISS+RRF)", "hybrid"),
    ("Hybrid + Rerank", "hybrid_rerank"),
]
TOP_K = 3


def _alarm_by_id(alarm_id: str) -> dict:
    from data.demo import DEFAULT_ALARMS

    return next(a for a in DEFAULT_ALARMS if a["id"] == alarm_id)


def _run_cause_with_contexts(alarm: dict, tier1: dict, contexts: list[str]) -> dict:
    """run_cause를 직접 재현하되 contexts를 외부에서 주입

    paradigm마다 retrieval 방식만 다르므로 LLM 호출 코드는 공통화
    contexts=[]면 No RAG (knowledge 섹션 자체를 제거)
    """
    knowledge = (
        "\n\n".join(f"[doc_{i}]\n{c}" for i, c in enumerate(contexts))
        if contexts
        else "(제공된 사내 지식 문서 없음 - 일반 지식으로 추정)"
    )
    sensors = ", ".join(f["name"] for f in tier1["features"])
    user_prompt = f"""## 이상 알람
- 공정: {alarm['title']}
- lot: {alarm['lot_id']}
- 이상 피처: {alarm.get('feature')} {alarm.get('feature_arrow') or ''}

## Tier 1 이상 탐지 결과
- 이상 점수: {tier1['score']}
- 기여 센서(Top): {sensors}

## 사내 지식 문서
{knowledge}

위 정보를 근거로 원인을 분석해 주세요."""

    resp = client().chat.completions.create(
        model=SUBAGENT_MODEL,
        messages=[
            {"role": "system", "content": WORKFLOW_SYSTEM_PROMPT},
            {"role": "user", "content": user_prompt},
        ],
        response_format={
            "type": "json_schema",
            "json_schema": {"name": "tier2", "schema": TIER2_SCHEMA, "strict": True},
        },
    )
    content = resp.choices[0].message.content or "{}"
    # 일부 모델이 ```json ... ``` fence를 붙이는 경우 방어
    if content.strip().startswith("```"):
        parts = content.split("```")
        if len(parts) >= 2:
            content = parts[1]
            if content.startswith("json"):
                content = content[4:]
    return json.loads(content.strip())


def _format_answer(tier2: dict) -> str:
    return "\n".join(
        f"- {c['name']} ({c['pct']}%): {c['evidence']}" for c in tier2["causes"]
    )


def _warmup():
    """ST 모델·reranker 로드 시간을 latency에서 제외하기 위한 사전 호출"""
    print("=== 워밍업 (ST / Reranker 모델 로드) ===")
    for label, backend in PARADIGMS:
        if backend is None:
            continue
        os.environ["RAG_BACKEND"] = backend
        t0 = time.time()
        search("warmup query", top_k=1)
        print(f"  {label}: {(time.time()-t0)*1000:.0f}ms")


def collect_samples():
    """5 paradigm × 3 알람 = 15 sample 수집"""
    _warmup()
    print(f"\n=== Sample 수집 ({len(PARADIGMS)} paradigm × {len(ALARMS)} alarm) ===")
    rows = []
    for alarm_id in ALARMS:
        alarm = _alarm_by_id(alarm_id)
        tier1 = run_detection(alarm)
        query = _build_query(alarm, tier1)
        print(f"\n[{alarm_id}] {alarm['title']}")

        for label, backend in PARADIGMS:
            t0 = time.time()
            if backend is None:
                contexts = []
                retrieval_ms = 0.0
            else:
                os.environ["RAG_BACKEND"] = backend
                doc_ids = search(query, top_k=TOP_K)
                retrieval_ms = (time.time() - t0) * 1000
                contexts = [load_document(d) for d in doc_ids if load_document(d)]

            t_gen = time.time()
            tier2 = _run_cause_with_contexts(alarm, tier1, contexts)
            gen_ms = (time.time() - t_gen) * 1000

            # contexts가 비면 RAGAS context-metric이 NaN 처리되도록 sentinel 한 줄 주입
            ragas_contexts = contexts if contexts else ["(검색 결과 없음 - LLM 단독 추론)"]
            answer = _format_answer(tier2)
            rows.append(
                {
                    "paradigm": label,
                    "alarm": alarm_id,
                    "question": query,
                    "answer": answer,
                    "contexts": ragas_contexts,
                    "retrieval_ms": retrieval_ms,
                    "gen_ms": gen_ms,
                    "total_ms": retrieval_ms + gen_ms,
                }
            )
            print(
                f"  {label:30s} retrieval={retrieval_ms:7.1f}ms "
                f"gen={gen_ms:7.1f}ms causes={len(tier2['causes'])}"
            )
    return rows


def evaluate_ragas(rows: list[dict]):
    print("\n=== RAGAS 평가 ===")
    eval_llm = LangchainLLMWrapper(ChatOpenAI(model="gpt-4o-mini", temperature=0))
    eval_emb = LangchainEmbeddingsWrapper(OpenAIEmbeddings(model="text-embedding-3-small"))
    dataset = Dataset.from_dict(
        {
            "question": [r["question"] for r in rows],
            "answer": [r["answer"] for r in rows],
            "contexts": [r["contexts"] for r in rows],
        }
    )
    metrics = [
        Faithfulness(llm=eval_llm),
        ResponseRelevancy(llm=eval_llm, embeddings=eval_emb),
        LLMContextPrecisionWithoutReference(llm=eval_llm),
    ]
    result = evaluate(dataset=dataset, metrics=metrics)
    df = result.to_pandas()
    df["paradigm"] = [r["paradigm"] for r in rows]
    df["alarm"] = [r["alarm"] for r in rows]
    df["retrieval_ms"] = [r["retrieval_ms"] for r in rows]
    df["gen_ms"] = [r["gen_ms"] for r in rows]
    df["total_ms"] = [r["total_ms"] for r in rows]
    return df


def aggregate(df):
    """paradigm별 평균 산출"""
    metric_cols = [c for c in df.columns if c in (
        "faithfulness", "answer_relevancy", "llm_context_precision_without_reference"
    )]
    agg = df.groupby("paradigm", sort=False)[metric_cols + ["retrieval_ms", "gen_ms", "total_ms"]].mean()
    # paradigm 순서 보존
    paradigm_order = [label for label, _ in PARADIGMS]
    agg = agg.reindex(paradigm_order)
    return agg, metric_cols


def make_charts(agg, metric_cols):
    CHART_DIR.mkdir(exist_ok=True)
    paradigms = list(agg.index)
    colors = ["#94a3b8", "#cbd5e1", "#60a5fa", "#3b82f6", "#1e40af"]

    # 1. RAGAS metric 비교 (grouped bar)
    fig, ax = plt.subplots(figsize=(11, 5.5))
    x = np.arange(len(paradigms))
    width = 0.26
    metric_labels = {
        "faithfulness": "Faithfulness",
        "answer_relevancy": "Answer Relevancy",
        "llm_context_precision_without_reference": "Context Precision",
    }
    for i, col in enumerate(metric_cols):
        values = agg[col].fillna(0).values
        bars = ax.bar(x + (i - 1) * width, values, width, label=metric_labels.get(col, col))
        for bar, v, raw in zip(bars, values, agg[col].values):
            label = "N/A" if np.isnan(raw) else f"{v:.2f}"
            ax.text(bar.get_x() + bar.get_width() / 2, v + 0.015, label,
                    ha="center", fontsize=8)
    ax.set_xticks(x)
    ax.set_xticklabels(paradigms, rotation=15, ha="right", fontsize=9)
    ax.set_ylim(0, 1.15)
    ax.set_ylabel("Score (0~1, 높을수록 좋음)")
    ax.set_title("RAG Paradigm Evolution - RAGAS metric 평균 (3 alarm)")
    ax.legend(loc="upper left", fontsize=9)
    ax.grid(axis="y", alpha=0.3)
    fig.tight_layout()
    fig.savefig(CHART_DIR / "ragas_comparison.png", dpi=150)
    plt.close(fig)

    # 2. Latency 비교 (stacked bar: retrieval + generation)
    fig, ax = plt.subplots(figsize=(10, 5))
    retrieval = agg["retrieval_ms"].values
    generation = agg["gen_ms"].values
    ax.bar(x, retrieval, color="#3b82f6", label="Retrieval")
    ax.bar(x, generation, bottom=retrieval, color="#fbbf24", label="LLM Generation")
    for i, (r, g) in enumerate(zip(retrieval, generation)):
        ax.text(i, r + g + 200, f"{r+g:.0f}ms", ha="center", fontsize=9, fontweight="bold")
    ax.set_xticks(x)
    ax.set_xticklabels(paradigms, rotation=15, ha="right", fontsize=9)
    ax.set_ylabel("평균 Latency (ms)")
    ax.set_title("Latency 분해 (Retrieval + Generation)")
    ax.legend(loc="upper left")
    ax.grid(axis="y", alpha=0.3)
    fig.tight_layout()
    fig.savefig(CHART_DIR / "latency_comparison.png", dpi=150)
    plt.close(fig)

    # 3. Quality vs Latency trade-off (scatter)
    fig, ax = plt.subplots(figsize=(8.5, 5.5))
    quality = agg[metric_cols].mean(axis=1).values  # 3 metric 평균
    latency = agg["total_ms"].values
    for i, p in enumerate(paradigms):
        ax.scatter(latency[i], quality[i], s=220, c=colors[i], edgecolors="black", linewidths=1.2, zorder=3)
        ax.annotate(p, (latency[i], quality[i]),
                    xytext=(10, 5), textcoords="offset points", fontsize=9)
    ax.set_xlabel("Total Latency (ms, ↓)")
    ax.set_ylabel("평균 RAGAS Score (↑)")
    ax.set_title("Quality vs Latency Trade-off")
    ax.grid(alpha=0.3)
    fig.tight_layout()
    fig.savefig(CHART_DIR / "tradeoff.png", dpi=150)
    plt.close(fig)
    print(f"--- 차트 저장: {CHART_DIR}/*.png ---")


def write_results(df, agg, metric_cols):
    metric_labels = {
        "faithfulness": "Faithfulness",
        "answer_relevancy": "Answer Relevancy",
        "llm_context_precision_without_reference": "Context Precision",
    }
    lines = [
        "# RAG Paradigm Evolution - 5단계 ablation 비교",
        "",
        "production RAG 진화 과정을 5단계로 분해해 각 단계 추가의 효과를 정량 측정합니다.",
        "동일한 알람·동일한 LLM(`gpt-5-mini`)·동일한 prompt 하에서 retrieval 방식만 바꿔",
        "RAGAS 지표(faithfulness / answer_relevancy / context_precision)와 latency를 비교합니다.",
        "",
        "## 실험 설정",
        "",
        f"- 알람: {', '.join(ALARMS)} (총 {len(ALARMS)}건, SECOM + PHM 2016 CMP)",
        f"- Paradigm: {len(PARADIGMS)}종",
        "- 생성 모델: `gpt-5-mini`",
        "- 평가 모델: `gpt-4o-mini` (RAGAS 호스트 LLM)",
        "- 임베딩: `text-embedding-3-small`",
        f"- Top-K: {TOP_K}",
        "- Latency는 워밍업 후 측정 (ST·Reranker 모델 로드 시간 제외, 실제 production warm 상태 기준)",
        "",
        "## Paradigm 정의",
        "",
        "| # | Paradigm | 설명 |",
        "|---|---|---|",
        "| 1 | **No RAG** | LLM 단독 추론, 사내 지식 미주입 (베이스라인) |",
        "| 2 | **Naive RAG (keyword)** | 단어 빈도 매칭 top-K |",
        "| 3 | **Vector RAG (FAISS)** | sentence-transformer dense embedding + cosine |",
        "| 4 | **Hybrid (BM25+FAISS+RRF)** | sparse + dense, Reciprocal Rank Fusion |",
        "| 5 | **Hybrid + Rerank** | Hybrid top-10을 cross-encoder(BAAI/bge-reranker-base)로 정밀 재정렬 |",
        "",
        "## 결과 요약 (paradigm별 평균)",
        "",
        "| Paradigm | " + " | ".join(metric_labels[c] for c in metric_cols) + " | Retrieval (ms) | LLM (ms) | Total (ms) |",
        "|---|" + "|".join(["---"] * (len(metric_cols) + 3)) + "|",
    ]
    for paradigm in agg.index:
        row = agg.loc[paradigm]
        metric_cells = [f"{row[c]:.3f}" if not np.isnan(row[c]) else "N/A" for c in metric_cols]
        lines.append(
            f"| {paradigm} | " + " | ".join(metric_cells) +
            f" | {row['retrieval_ms']:.1f} | {row['gen_ms']:.1f} | {row['total_ms']:.1f} |"
        )

    lines += [
        "",
        "## 시각화",
        "",
        "### RAGAS metric 비교",
        "",
        "![RAGAS Comparison](charts/ragas_comparison.png)",
        "",
        "### Latency 분해",
        "",
        "![Latency Comparison](charts/latency_comparison.png)",
        "",
        "### 품질 vs Latency Trade-off",
        "",
        "![Trade-off](charts/tradeoff.png)",
        "",
        "## 알람별 상세 결과",
        "",
    ]
    for alarm_id in ALARMS:
        sub = df[df["alarm"] == alarm_id]
        lines.append(f"### {alarm_id}")
        lines.append("")
        lines.append("| Paradigm | " + " | ".join(metric_labels[c] for c in metric_cols) + " | Total (ms) |")
        lines.append("|---|" + "|".join(["---"] * (len(metric_cols) + 1)) + "|")
        for _, r in sub.iterrows():
            metric_cells = [f"{r[c]:.3f}" if not np.isnan(r[c]) else "N/A" for c in metric_cols]
            lines.append(f"| {r['paradigm']} | " + " | ".join(metric_cells) + f" | {r['total_ms']:.1f} |")
        lines.append("")

    lines += [
        "## 핵심 인사이트",
        "",
        "1. **RAG 도입 효과가 결정적**: `No RAG` 대비 어떤 paradigm을 붙여도 `faithfulness`가 2배 이상 상승합니다.",
        "   사내 사례·SOP·FMEA를 인용하지 못하는 LLM은 hallucination 위험이 크고, 반도체 도메인에서는 치명적입니다.",
        "",
        "2. **Hybrid (BM25 + FAISS + RRF)가 본 코퍼스에서 모든 지표 1위**입니다.",
        "   sparse(BM25, 정확 용어 매칭) + dense(FAISS, 의미 매칭)를 Reciprocal Rank Fusion으로 결합해",
        "   각 단일 backend의 약점을 상쇄합니다. Latency도 가장 빠릅니다.",
        "",
        "3. **Cross-encoder Rerank는 본 코퍼스에서 이득 없음**: `Hybrid + Rerank`는 `Hybrid`와 비교해",
        "   faithfulness 동급, `answer_relevancy`는 오히려 하락했습니다. 원인 추정:",
        "   - 코퍼스가 ~10문서로 작아 Hybrid top-3이 이미 정답에 근접",
        "   - `BAAI/bge-reranker-base`가 영어 학습 모델이라 한국어 도메인 텍스트에서 점수 신호가 잡음에 가까움",
        "   - 한국어 reranker(예: `dongjin-kr/ko-reranker`) 또는 코퍼스 확장(100문서+) 시 효과 재검증 필요",
        "",
        "## 채택 근거",
        "",
        "**MVP 기본 backend = `Hybrid (BM25+FAISS+RRF)`**",
        "",
        "본 실험 데이터(3 알람 × 5 paradigm)는 `Hybrid`가 quality + latency 모두에서 우위임을 보여줍니다.",
        "production RAG 표준 패턴이기도 합니다 (Microsoft Azure AI Search, LlamaIndex 기본 권고).",
        "",
        "**`Hybrid + Rerank`는 옵션으로 유지** (환경변수 `RAG_BACKEND=hybrid_rerank`):",
        "- 코퍼스가 100+ 문서로 확장될 때 cross-encoder 정밀 재정렬이 필요해질 가능성 큼",
        "- 한국어 reranker로 교체 시 본 실험 재평가 권장",
        "",
        "**전체 채택 신호**:",
        "- 어떤 RAG든 No RAG보다 압도적으로 낫다 → RAG는 production 필수",
        "- 코퍼스 규모와 도메인 언어에 맞춰 paradigm을 선택해야 한다 (블라인드 적용은 역효과)",
        "- 정량 평가(RAGAS)가 없으면 'rerank가 무조건 좋다'는 오해를 그대로 끌고 갔을 것",
        "",
    ]
    (OUT_DIR / "results.md").write_text("\n".join(lines), encoding="utf-8")
    print(f"--- 저장: {OUT_DIR / 'results.md'} ---")


def main():
    import sys

    charts_only = "--charts-only" in sys.argv
    if charts_only and CACHE_CSV.exists():
        print(f"=== 캐시에서 결과 로드: {CACHE_CSV} ===")
        df = pd.read_csv(CACHE_CSV)
    else:
        rows = collect_samples()
        df = evaluate_ragas(rows)
        df.to_csv(CACHE_CSV, index=False)
        print(f"--- 캐시 저장: {CACHE_CSV} ---")

    agg, metric_cols = aggregate(df)
    print("\n--- 집계 ---")
    print(agg.round(3))
    make_charts(agg, metric_cols)
    write_results(df, agg, metric_cols)


if __name__ == "__main__":
    main()