2026-04-10 강의 정리

---

RAG의 후처리 단계를 한 흐름으로 묶어 보면 각 단계의 책임과 중복이 드러난다. 리랭킹·필터링·압축·MMR을 모두 사용하는 것 자체를 목표로 삼지 않고, 어느 단계가 어떤 실패를 줄이는지 살펴본다. 이 글의 축약 예제는 앞선 구현들을 연결하는 구조를 정리하는 용도로 본다.

목차

1. RAG 파이프라인 개요
2. Reranking 기법
3. Score Filtering
4. Context Compression
5. MMR (Maximum Marginal Relevance)
6. RAG 평가 지표 (AP, MAP, ILS, A/B test)
7. LLM-as-a-Judge 평가 전략
8. 자주 나오는 실수 / 주의사항
9. [보충] 평가 프레임워크 설계 원칙
10. 복잡도를 추가할 근거 남기기

---

1. RAG 파이프라인 개요

[사용자 쿼리]
     ↓
[Retriever] → 후보 문서 5~20개
     ↓
[Reranker] → 상위 k개 선별
     ↓
[Generator (LLM)] → 최종 답변 생성

Bi-encoder vs Cross-encoder

방식	구조	특징
Bi-encoder	query → emb / doc → emb / 코사인 유사도	빠름, 사전 인덱싱 가능
Cross-encoder	(query, doc) → 점수	정확, 쌍 단위 처리로 느림

Bi-encoder로 1차 검색 → Cross-encoder/LLM으로 2차 리랭킹하는 게 실무 표준

---

2. Reranking 기법

2-1. Keyword Reranking

가장 단순한 방식. 쿼리 단어가 문서에 몇 개 등장하는지 카운트해 점수 보정.

def keyword_rerank(query, search_results):
    query_terms = set(query.lower().split())
    reranked = []
    for doc, orig_score in search_results:
        doc_terms = doc.page_content.lower().split()
        keyword_hits = sum(1 for t in doc_terms if t in query_terms)
        new_score = orig_score + 0.1 * keyword_hits
        reranked.append((doc, new_score, orig_score))
    reranked.sort(key=lambda x: x[1], reverse=True)
    return reranked

2-2. BM25 Reranker (TF-IDF 기반)

TF-IDF를 개선한 BM25 알고리즘으로 문서를 재정렬.

class BM25Reranker:
    def __init__(self, k1=1.5, b=0.75):
        self.k1 = k1   # 단어 빈도 포화 파라미터
        self.b = b     # 문서 길이 정규화 파라미터

    def rerank(self, query, search_results):
        docs = [doc for doc, _ in search_results]
        tokenized = [doc.page_content.lower().split() for doc in docs]
        avg_dl = np.mean([len(t) for t in tokenized])
        # BM25 점수 계산 후 정렬
        ...

• k1: 단어 빈도가 많아도 점수가 포화되는 정도 조절 (1.2~2.0)
• b: 문서 길이에 따라 정규화 (0~1, 클수록 길이 영향 강함)

2-3. LLM Reranker (Pointwise)

각 문서에 대해 LLM이 0.0~1.0 점수 부여.

def llm_rerank(query, search_results, top_k=3):
    docs_text = '\n'.join(
        f"[{doc.metadata['id']}] {doc.page_content}"
        for doc, _ in search_results
    )
    # LLM에 JSON 형태로 점수 요청
    scoring_chain = ChatPromptTemplate.from_messages([
        ('system', '당신은 scoring 시스템입니다. 항상 json 형태로 출력하세요'),
        ('human', '쿼리: {query}\n\n문서들:\n{docs}'),
    ]) | llm | StrOutputParser()
    ...

2-4. Hybrid Reranker

BM25 점수 + LLM 점수를 가중 합산.

def hybrid_rerank(query, search_results, bm25_weight=0.3):
    # BM25 점수 (정규화)
    bm25_scored = BM25Reranker().rerank(query, search_results)
    # LLM 점수 (정규화)
    llm_scored = llm_rerank(query, search_results)
    # 가중 합산: final = bm25_weight * bm25 + (1-bm25_weight) * llm

2-5. Listwise vs Pointwise 비교

Pointwise (1개씩)
  장점: 정확, 이유 명확, 병렬 처리 가능, 빠름
  단점: 비용, 문서 간 상대 비교 불가

Listwise (N개씩 한꺼번에)
  장점: 문서 간 비교 가능
  단점: 할루시네이션 위험, 컨텍스트 길어짐

---

3. Score Filtering

검색 결과에서 낮은 품질의 문서를 제거하는 전략.

class ScoreFilter:
    @staticmethod
    def fixed_threshold(scored_docs, threshold=0.5):
        # threshold 이하 문서 제거
        return [(doc, s) for doc, s in scored_docs if s >= threshold]

    @staticmethod
    def dynamic_threshold(scored_docs, std_factor=1.0):
        # mean - std_factor * std 이상인 문서만 유지
        scores = [s for _, s in scored_docs]
        cutoff = np.mean(scores) - std_factor * np.std(scores)
        return [(doc, s) for doc, s in scored_docs if s >= cutoff]

    @staticmethod
    def score_gap(scored_docs, min_docs=2):
        # 점수 급락 지점 이전까지만 유지
        ...

AdaptiveFilter: 점수 분포에 따라 전략을 자동 선택.

---

4. Context Compression

이 절에는 앞선 압축 구현을 축약한 함수가 포함되어 있다. 함수 이름과 반환 흐름을 설명하는 코드와 독립적으로 실행할 수 있는 코드를 구분하고, 연결할 때는 문서·점수·메타데이터가 어느 형태로 넘어가는지 확인한다.

LLM 컨텍스트 길이(128K 토큰)는 제한이 있어, 문서를 압축해서 넣어야 효율적.

4-1. Extractive Compress

def extractive_compress(query, document, max_sentences=3):
    # 문장 분리 후 쿼리 단어와 겹치는 문장만 선택
    sentences = re.split(r'[.!?]\s*', document)
    query_terms = set(query.lower().split())
    scored = [(s, sum(1 for w in s.lower().split() if w in query_terms))
              for s in sentences if s.strip()]
    scored.sort(key=lambda x: x[1], reverse=True)
    return '. '.join([s for s, _ in scored[:max_sentences]])

4-2. LLM Compress

LLM에게 직접 문서 요약 요청.

compress_chain = ChatPromptTemplate.from_messages([
    ("system", "당신은 문서 요약 전문가입니다"),
    ("human", "쿼리: {query}\n\n문서: {document}\n\n쿼리에 관련된 내용만 압축 요약:"),
]) | llm | StrOutputParser()

4-3. MMR Compress

관련성은 유지하면서 중복 문장 제거.

def mmr_compress(query, document, max_sentences=3, param=0.5):
    # param: Relevance 가중치 (높을수록 쿼리 관련성 중시)
    # mmr = param * Relevance(s, query) - (1-param) * max(Similarity(s, selected))

4-4. 압축 품질 평가

def evaluate_compression(original, compressed, query):
    # 압축률: len(compressed) / len(original)
    # 키워드 보존율: 쿼리 단어가 얼마나 남았나
    # 의미 보존도: 원문 임베딩 vs 압축 임베딩 코사인 유사도

---

5. MMR (Maximum Marginal Relevance)

쿼리 관련성은 높이면서 선택된 문서들 간의 중복을 최소화.

MMR = P * Relevance(s, query) - (1-P) * max(Similarity(s, selected))

P = 0.1 → 다양성 중시
P = 0.9 → 관련성 중시

# MMR retriever 사용
mmr_results = vectorstore.max_marginal_relevance_search(query, k=5, fetch_k=20)

---

6. RAG 평가 지표

Average Precision (AP)

검색 결과에서 정답 문서가 얼마나 높은 순위에 있는지 측정.

def average_precision(retrieved_ids, relevant_ids):
    relevant_set = set(relevant_ids)
    hits = 0
    precision_sum = 0.0
    for i, doc_id in enumerate(retrieved_ids):
        if doc_id in relevant_set:
            hits += 1
            precision_sum += hits / (i + 1)
    return precision_sum / len(relevant_ids) if relevant_ids else 0.0

MAP (Mean Average Precision)

여러 쿼리에 대한 AP의 평균.

ILS (Intra-List Similarity)

검색 결과의 다양성 측정. 낮을수록 다양한 문서 검색.

def intra_list_similarity(doc_ids, doc_embeddings, top_k=5):
    # 선택된 문서들의 쌍별 코사인 유사도 평균
    ids = doc_ids[:top_k]
    embs = [doc_embeddings[id] for id in ids]
    ...

A/B Test

def ab_test(scores_a, scores_b, alpha=0.05):
    # t-test로 통계적 유의성 검정
    t_stat, p_value = stats.ttest_ind(scores_a, scores_b)
    return {'significant': p_value < alpha, 'p_value': p_value}

---

7. LLM-as-a-Judge 평가 전략

평가 단계까지 연결할 때는 모델 응답을 바로 평균내기 전에 반환 형태를 고정해야 한다. 점수와 판단 이유를 분리하고, 파싱 실패나 평가 불가 사례를 어떤 기준으로 집계에서 처리할지도 정한다.

7-1. Basic Judge (Pointwise)

def basic_judge(question, answer, scale='1~5'):
    prompt = f"""다음 답변을 {scale}점으로 평가하세요.
    질문: {question}
    답변: {answer}"""
    return llm.invoke(prompt)

7-2. ROUGE F1 (참고용 키워드 기반)

def simple_rouge_f1(reference, candidate):
    ref_words = Counter(reference.split())
    cand_words = Counter(candidate.split())
    # precision, recall, F1 계산

7-3. Rubric-based Judge

RUBRIC = {
    5: "정확하고 완전하며, 예시와 설명이 풍부하다",
    4: "정확하고 핵심을 다루지만, 일부 세부사항이 부족하다",
    3: "대체로 정확하지만 불완전하거나 모호한 부분이 있다",
    2: "일부 정보가 맞지만 중요한 오류가 있다",
    1: "잘못된 정보이거나 질문과 무관하다",
}

7-4. Pairwise Judge

두 답변을 동시에 주고 더 나은 쪽 선택.

def pairwise_judge(question, answer_a, answer_b):
    prompt = """두 답변을 비교하여 더 나은 쪽을 선택하세요
    질문: {question}
    A: {answer_a}
    B: {answer_b}
    → {'winner': 'A' or 'B', 'reason': '...'}"""

위치 편향 방지: (A,B)와 (B,A) 순서 바꿔서 2회 평가 후 다수결

def pairwise_judge_with_swap(question, answer_a, answer_b):
    result1 = pairwise_judge(question, answer_a, answer_b)
    result2 = pairwise_judge(question, answer_b, answer_a)
    # 두 결과 종합

7-5. Reference-based Judge

정답 예시(Reference)를 함께 제공해 채점.

def reference_based_judge(question, reference, answer):
    prompt = """정답을 기준으로 답변을 평가하세요.
    정답: {reference}
    답변: {answer}"""

7-6. Chain-of-Thought (CoT) Judge

단계별로 분석 후 점수 부여 → 이유 추적 가능.

def cot_judge(question, answer):
    prompt = """다음 답변을 단계별로 분석하세요.
    1. 정확성 분석
    2. 완전성 분석
    3. 최종 점수"""

7-7. Few-shot Judge

평가 예시를 포함해 일관된 채점 기준 유지.

7-8. Multi-dimensional Judge

EVAL_DIMENSIONS = {
    "accuracy":     "사실적으로 정확한 정보만 포함하는가 (1-5)",
    "relevance":    "질문의 의도와 범위에 적합한 답변인가 (1-5)",
    "completeness": "질문의 모든 측면을 다루는가 (1-5)",
    "coherence":    "논리적 흐름과 구조가 일관적인가 (1-5)",
}

7-9. Ensemble Judge

여러 judge 결과를 가중평균.

def ensemble_judge(question, answer, judge_fns, weight=None):
    scores = []
    for name, fn in judge_fns:
        result = fn(question, answer)
        scores.append(result)
    return np.average(scores, weights=weight)

7-10. 재현성 & 분별력 테스트

# 재현성: 변동계수(std/mean) < 0.1 이면 안정적
def test_reproducibility(question, answer, n_trials=5):
    scores = [pointwise_judge(question, answer) for _ in range(n_trials)]
    cv = np.std(scores) / np.mean(scores)
    return {'mean': np.mean(scores), 'cv': cv}

# 분별력: 좋은 답변 vs 나쁜 답변 점수 차이가 충분한지
def test_discriminability(question, good_answer, bad_answer):
    good_scores = [pointwise_judge(question, good_answer) for _ in range(3)]
    bad_scores  = [pointwise_judge(question, bad_answer)  for _ in range(3)]
    return np.mean(good_scores) - np.mean(bad_scores)

---

자주 나오는 실수 / 주의사항

• 위치 편향 (Position Bias): LLM은 첫 번째로 주어진 문서/답변에 더 높은 점수를 주는 경향 → 반드시 순서 바꿔서 2회 평가
• 길이 편향 (Length Bias): 긴 답변에 점수를 후하게 주는 경향 → 길이가 다른 답변을 의도적으로 테스트
• 자기 강화 편향: LLM이 자신과 유사한 스타일의 문장에 후하게 평가
• BM25 파라미터 착각: k1, b 기본값(1.5, 0.75)이 항상 최적이 아님 → 도메인에 맞게 튜닝
• threshold 고정값 위험: 데이터 분포가 바뀌면 fixed_threshold 성능 급락 → dynamic threshold 권장

---

[보충] 평가 프레임워크 설계 원칙

좋은 평가 기준(Criterion) 조건:

조건	설명
관찰 가능	예시가 포함되어 있는지, 채점자가 확인할 수 있어야 함
독립적	다른 기준과 내용이 겹치지 않아야 함
측정 가능	1~5 등 객관적으로 정량화할 수 있어야 함
평가 목적 부합	실제 서비스 목적(정확성, 안전성 등)에 맞아야 함

class EvaluationCriterion:
    def __init__(self, name, description, scale=(1, 5)):
        self.name = name
        self.description = description
        self.scale = scale

STANDARD_CRITERIA = [
    EvaluationCriterion("accuracy",     "사실적으로 정확한 정보만 포함하는가"),
    EvaluationCriterion("relevance",    "질문의 의도에 적합한가"),
    EvaluationCriterion("completeness", "모든 측면을 다루는가"),
    EvaluationCriterion("coherence",    "논리적 흐름이 일관적인가"),
]

---

10. 복잡도를 추가할 근거 남기기

전체 구성을 비교하는 기준은 검색과 생성만 연결한 단순한 흐름으로 잡을 수 있다. 여기에 한 단계를 추가할 때마다 질문별 점수, 근거 손실, 처리 시간을 기록하면 유지할 단계와 제거할 단계를 구분할 수 있다.

개별 기법이 좋아 보여도 조합한 뒤 같은 효과가 유지된다고 가정하지 않는다. 앞 단계에서 제거된 문서를 뒤 단계가 되살릴 수 있는지처럼 데이터가 사라지는 지점을 확인하고, 최종 답변까지 이어지는 영향을 평가한다.