체인 골격 (LCEL)

chain = (
  {"context": retriever,
   "question": RunnablePassthrough()}
  | prompt | llm | StrOutputParser()
)

오늘 더하는 신뢰 장치

• 출처 인용 (Citation)
• 환각 검증 (Hallucination)
• 정량 평가 (RAGAS)

① format_docs 로 context 결합

def format_docs(docs):
    return "\n\n".join(
        d.page_content for d in docs)

② 환각 가드 프롬프트

prompt = ChatPromptTemplate.from_messages([
    ("system",
     "아래 문서만 참고해서 답하세요. "
     "문서에 없으면 '모르겠습니다'라고 "
     "답하세요.\n\n문서:\n{context}"),
    ("user", "{question}"),
])

③ LCEL 체인 (표준형)

chain = (
    RunnablePassthrough.assign(
        context=lambda x: format_docs(
            retriever.invoke(x["question"])))
    | prompt
    | llm                 # temperature=0
    | StrOutputParser()
)

print(chain.invoke(
    {"question": "NVMe 와 SATA 차이?"}))

• assign 이 question 보존 + context 추가
• 문서에 없는 질문 → "모르겠습니다"

프롬프트 + 도큐먼트 라벨링

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

CITATION_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """다음 근거 문서들로만 답변하세요.
각 문장 끝에 사용한 문서의 인용 마커 `[doc N]`
을 반드시 붙이세요. 근거에 없으면 "모릅니다".

근거 문서:
{context}
"""),
    ("user", "{question}"),
])

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def format_with_ids(docs):
    """[doc 1] [doc 2] 형식으로 라벨링."""
    return "\n\n".join(
        f"[doc {i+1}] (출처: "
        f"{d.metadata.get('source','?')} "
        f"p.{d.metadata.get('page','?')})\n"
        f"{d.page_content}"
        for i, d in enumerate(docs)
    )

RAG 체인 + 후처리

def rag_with_citation(question: str):
    docs = retriever.invoke(question)
    context = format_with_ids(docs)
    answer = (CITATION_PROMPT | llm).invoke(
        {"context": context, "question": question})

    # 후처리 — [doc N] 마커를 실제 source 로 치환
    import re
    def expand_citation(m):
        idx = int(m.group(1)) - 1
        d = docs[idx]
        return (f"[{d.metadata.get('source','?')}"
                f" p.{d.metadata.get('page','?')}]")

    final = re.sub(r"\[doc (\d+)\]",
                   expand_citation, answer.content)
    return final, docs

answer, sources = rag_with_citation(
    "RAG 의 핵심 단계는?")
print(answer)
# → "RAG 는 색인 → 검색 → 답변 3단계
#    로 진행됩니다 [rag.md p.3]."

Recall@K

$$ \text{Recall@K} = \frac{|{\text{정답}} \cap {\text{top-K}}|}{|{\text{정답}}|} $$

• 질의 한 개당 0 또는 1 (정답이 K개 중 있으면 1)
• 여러 정답이 있으면 비율
• K=5 가 95% 이상 이면 retrieval 양호

MRR (Mean Reciprocal Rank)

$$ \text{MRR} = \frac{1}{|Q|} \sum_{q \in Q} \frac{1}{\text{rank}_q} $$

• 정답 chunk 의 순위의 역수 평균
• 1위면 1, 2위면 0.5, 10위면 0.1
• 0.7 이상이면 좋음

직접 측정 코드

golden = [
    {"query": "RAG 의 본질은?",
     "answer_chunk_ids": ["c_001", "c_042"]},
    # ... 100개
]

def evaluate_retrieval(retriever, golden, k=5):
    recalls, rrs = [], []
    for item in golden:
        results = retriever.invoke(item["query"])
        retrieved_ids = [d.metadata["chunk_id"]
                         for d in results[:k]]
        golden_set = set(item["answer_chunk_ids"])

        # Recall@K
        hit = len(set(retrieved_ids) & golden_set) \
              / len(golden_set)
        recalls.append(hit)

        # MRR — 첫 정답의 역수
        for rank, rid in enumerate(retrieved_ids, 1):
            if rid in golden_set:
                rrs.append(1 / rank); break
        else:
            rrs.append(0)
    return (sum(recalls)/len(recalls),
            sum(rrs)/len(rrs))

r, m = evaluate_retrieval(retriever, golden)
print(f"Recall@5 = {r:.2f}, MRR = {m:.2f}")

Faithfulness — 답변이 출처에 충실한가 (0~1)

주어진 답변 = 5개 문장
각 문장이 context 에 entailed 되는가?
→ entailed 비율 = faithfulness 점수

def faithfulness(answer, context, llm) -> float:
    """LLM 으로 답변 문장 각각 검증."""
    prompt = f"""답변의 각 문장이 context 에서
도출 가능한지 yes/no:

Context: {context}
답변: {answer}

각 문장마다 줄 단위로 "yes" 또는 "no" 만 출력.
"""
    result = llm.invoke(prompt).content
    decisions = [l.strip().lower()
                 for l in result.split("\n") if l.strip()]
    if not decisions:
        return 0.0
    return decisions.count("yes") / len(decisions)

Answer Relevancy — 질문 ↔ 답변 적합도

def answer_relevancy(question, answer,
                     llm, embeddings) -> float:
    """답변에서 LLM 이 역추론한 질문 N개를
    원 질문과 임베딩 유사도로 비교."""
    reverse_prompt = f"""다음 답변이 주어졌을 때,
가능한 사용자 질문 3개를 만드세요:

답변: {answer}

질문 3개를 한 줄씩 출력.
"""
    generated_questions = llm.invoke(
        reverse_prompt).content.split("\n")
    q_emb = embeddings.embed_query(question)
    sims = [
        cosine_sim(q_emb, embeddings.embed_query(gq))
        for gq in generated_questions if gq.strip()
    ]
    return sum(sims) / len(sims) if sims else 0.0

🎯 두 메트릭 모두 LLM-as-Judge — 자동화 가능하지만 LLM 호출 비용 발생.

평가 데이터 + 실행

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

# 평가 데이터 — (질의, RAG 답변, context, 정답)
data = {
    "question":     [i["query"]      for i in golden],
    "answer":       [i["rag_answer"] for i in golden],
    "contexts":     [i["retrieved"]  for i in golden],
    "ground_truth": [i["true_answer"] for i in golden],
}
ds = Dataset.from_dict(data)

result = evaluate(
    dataset=ds,
    metrics=[
        context_precision,
        context_recall,
        faithfulness,
        answer_relevancy,
        answer_correctness,
    ],
)
print(result)

결과 + 메트릭 진단

# {
#   'context_precision':  0.82,
#   'context_recall':     0.78,
#   'faithfulness':       0.91,
#   'answer_relevancy':   0.85,
#   'answer_correctness': 0.74,
# }

메트릭이 알려주는 진단

1단계 — Golden set 구축 (10분)

• 색인해 둔 문서로 질문 10개 작성
• 각 질문마다 답이 들어있는 chunk_id 1~3개 표시
• LLM 에게 "이 chunk 로 만들 수 있는 질문 3개" 자동 생성도 OK

2단계 — 3가지 retrieval 비교 (10분)

# A. Similarity (baseline)
retriever_sim = vectorstore.as_retriever(
    search_kwargs={"k": 5})

# B. MMR
retriever_mmr = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 5, "fetch_k": 20,
                   "lambda_mult": 0.5},
)

# C. BM25 + Vector hybrid
bm25 = BM25Retriever.from_documents(all_chunks)
bm25.k = 5
retriever_hybrid = EnsembleRetriever(
    retrievers=[bm25, retriever_sim],
    weights=[0.4, 0.6])

# 각각 Recall@5 측정
for name, r in [("sim", retriever_sim),
                ("mmr", retriever_mmr),
                ("hybrid", retriever_hybrid)]:
    recall, mrr = evaluate_retrieval(r, golden)
    print(f"{name}: Recall@5={recall:.2f}"
          f"  MRR={mrr:.2f}")

3단계 — Citation 답변 생성 (5분)

• 위 슬라이드의 rag_with_citation() 적용
• 질문 3개에 출처 인용 답변 확인

4단계 — Faithfulness 측정 (5분)

• RAGAS 또는 직접 self-check 프롬프트
• 점수 0.7 미만 답변 1개 찾아 어디가 hallucination 인지 분석

📤 결과 보고

• Sim / MMR / Hybrid 중 대상 도메인에 가장 적합한 것 + 한 줄 이유
• Faithfulness 가 낮았던 답변 사례 1건

평가 기준

• Recall@5 0.7 이상 = 양호
• Faithfulness 0.85 이상 = 운영 가능
• 두 메트릭이 모두 낮으면 → chunk_size 또는 system prompt 부터 재검토