[Reranking] Reranking: 검색 결과를 진짜로 바꾸는 정밀 정제 단계

당니·2026년 2월 3일

Cohere CrossEncoder LostInTheMiddle bge rag reranking 검색정밀도

LLM

목록 보기

19/19

Embedding 글에서도 언급했듯이, Bi-Encoder는 빠르지만 정밀도에는 한계가 있습니다.
"한국어 자연어 처리"를 검색했을 때 "영어 NLP 튜토리얼"이 높은 스코어로 올라오는 일이 빈번합니다.
이 글에서는 그 간격을 메우는 Reranking의 원리, 세 가지 패러디그램(Pointwise · Pairwise · Listwise),
"Lost in the Middle" 문제와의 관계, 그리고 2026년 실무 스택까지 깊이 다루겠습니다.

Reranking이란? — 왜 필요한가
2단계 검색의 구조
Cross-Encoder의 원리 — Bi-Encoder와 무엇이 다른가
세 가지 Reranking 패러디그램
Lost in the Middle — Reranking이 해결해야 할 진짜 문제
2026년 Reranker 모델 비교
LLM을 Reranker로 사용할 때와 안 할 때
실무 구현: Full Reranking Pipeline
비용과 지연도의 최적화
모델 선택 가이드

1. Reranking이란? — 왜 필요한가

1.1 한줄 정의

Reranking은 1차 검색에서 올라온 후보 문서들을 더 정밀한 모델로 재평가하여 순위를 재정렬하는 단계입니다.

1.2 왜 1차 검색만으로는 안 되는가

Bi-Encoder는 Query와 Document를 독립적으로 임베딩한 후 코사인 유사도로 비교합니다.
이 구조의 본질적 한계는 교차 정보(cross-information)를 활용할 수 없다는 것입니다.

Bi-Encoder의 한계 예시:

Query:  "한국어로 작성된 NLP 튜토리얼"
Doc A:  "영어 NLP 튜토리얼 — Python과 NLTK 활용"        ← 벡터 유사도 높음
Doc B:  "한국어 자연어 처리: 형태소 분석부터 시작"      ← 벡터 유사도 낮음

→ Bi-Encoder 결과: Doc A > Doc B  (틀림!)
→ Reranker 결과:  Doc B > Doc A  (맞음!)

이유: "한국어로 작성된"이라는 조건과 "NLP 튜토리얼"이라는 조건을
     동시에 교차 비교할 수 있는 것은 Cross-Encoder뿐

Databricks 연구에서 Reranking은 검색 품질을 최대 48%까지 향상시킬 수 있다고 보고되었습니다.
실무에서도 Reranking 추가 시 RAG 정확도가 일반적으로 20-35%의 향상을 보이며,
추가 지연도는 200-500ms 수준입니다.

2. 2단계 검색의 구조

전체 파이프라인

사용자 쿼리
    │
    ▼
┌─────────────────────────────┐
│  1단계: Retrieval (회상)     │  ← 빠르게, 넓게 (Top 50-100)
│  Bi-Encoder + Vector Search │     목표: Recall 최대화
└─────────────────────────────┘
    │
    ▼  후보 문서 50-100개
┌─────────────────────────────┐
│  2단계: Reranking (정밀)     │  ← 느리지만 정확하게 (Top 5-10)
│  Cross-Encoder              │     목표: Precision 최대화
└─────────────────────────────┘
    │
    ▼  최종 문서 5-10개
┌─────────────────────────────┐
│  3단계: Generation           │
│  LLM + Context              │
└─────────────────────────────┘

각 단계의 역할

단계	모델	속도	정밀도	목표
Retrieval	Bi-Encoder	⚡⚡⚡	⭐⭐	관련 후보를 놓지 않음 (Recall)
Reranking	Cross-Encoder	⚡	⭐⭐⭐⭐⭐	진짜 정답을 위로 올림 (Precision)
Generation	LLM	⚡⭐	—	답변 생성

핵심 포인트: Reranking은 1차 검색의 후보를 재평가하는 것입니다.
1차 검색에서 정답이 빠지면 Reranking으로도 복구할 수 없습니다.
그러므로 1차 검색의 Recall이 충분히 높아야 합니다.

sweet spot은 후보 문서 50-75개로 검색한 후 Reranking하는 것으로,
품질과 비용의 균형이 가장 좋은 지점입니다.

3. Cross-Encoder의 원리 — Bi-Encoder와 무엇이 다른가

3.1 아키텍처 비교

┌──────── Bi-Encoder ──────────────────────┐
│                                          │
│   Query ──→ [BERT] ──→ Q_vec            │
│   Doc   ──→ [BERT] ──→ D_vec            │
│                                          │
│   Score = cosine_sim(Q_vec, D_vec)       │
│                                          │
│   ✅ 벡터를 미리 저장 가능 (offline)     │
│   ✅ 대량 검색에 최적 (O(1) per doc)     │
│   ❌ Query-Doc 교차 정보 없음            │
└──────────────────────────────────────────┘

┌──────── Cross-Encoder ───────────────────┐
│                                          │
│   [Query + SEP + Doc]                    │
│         ↓                                │
│      [BERT]                              │
│         ↓                                │
│      Score (단일 숫자)                   │
│                                          │
│   ✅ 교차 어텐션으로 깊은 이해           │
│   ✅ 정밀도가 Bi-Encoder >> Cross-Enc    │
│   ❌ 벡터 저장 불가 (매번 실행 필요)     │
│   ❌ O(N) — 후보마다 한 번씩 실행        │
└──────────────────────────────────────────┘

3.2 교차 어텐션이 만드는 차이

Cross-Encoder의 핵심은 Full Cross-Attention입니다.

입력: [CLS] 한국어 NLP 튜토리얼 [SEP] 영어 NLP 튜토리얼 내용... [SEP]

Transformer 내부:
  "한국어" 토큰  ──어텐션──→  "영어" 토큰
      ↓                          ↓
  "이 쿼리는 한국어를 원하는데,    
   이 문서는 영어야. 관련도 낮음."

→ Score: 0.21 (낮음)

이것이 Bi-Encoder에서는 불가능한 논리입니다.
Bi-Encoder는 각각 독립적으로 벡터를 만들기 때문에
"쿼리가 원하는 것"과 "문서가 제공하는 것"을 직접 비교할 수 없습니다.

실제 연구에서 Cross-Encoder 처리 속도는 100개 문서(256 토큰)당 약 150밀리초이며,
4096 토큰 문서의 경우 약 7초로 증가합니다.

4. 세 가지 Reranking 패러디그램

Reranking 모델들은 문서를 몇 개씩 보면서 스코어를 매기는지로 분류됩니다.

4.1 Pointwise — "각각 독립적으로 평가"

┌─────────────────────────────────────────┐
│            Pointwise Reranking           │
│                                         │
│  Query + Doc1  →  [Model]  →  Score1    │
│  Query + Doc2  →  [Model]  →  Score2    │
│  Query + Doc3  →  [Model]  →  Score3    │
│                                         │
│  최종 순위: Score 내림차순 정렬          │
│                                         │
│  복잡도: O(n) — 문서 수에 선형           │
│  대표 모델: BGE-Reranker, ms-marco-MiniLM│
└─────────────────────────────────────────┘

장점: 단순, 빠름, 구현 용이
단점: 문서 간 상대적 비교 불가 — "이 문서가 다른 문서보다 낫다"는 판단 어려움

from sentence_transformers import CrossEncoder

model = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

query = "한국어 자연어 처리 튜토리얼"
candidates = [
    "한국어 NLP 가이드...",
    "영어 NLP 튜토리얼...",
    "Python 프로그래밍..."
]

# Pointwise: 각 (query, doc) 쌍을 독립적으로 스코어링
pairs = [[query, doc] for doc in candidates]
scores = model.predict(pairs)

# 스코어 내림차순 정렬
ranked = sorted(zip(scores, candidates), reverse=True)
for score, doc in ranked:
    print(f"Score: {score:.3f} | {doc[:40]}...")

4.2 Pairwise — "둘씩 비교하여 승부"

┌─────────────────────────────────────────┐
│            Pairwise Reranking            │
│                                         │
│  Query + (Doc1 vs Doc2)  →  Doc1 승     │
│  Query + (Doc1 vs Doc3)  →  Doc1 승     │
│  Query + (Doc2 vs Doc3)  →  Doc2 승     │
│                                         │
│  최종 순위: 승수(win count) 기반 정렬   │
│                                         │
│  복잡도: O(n²) — 조합 폭발              │
│  대표 모델: DuoT5, PRP                  │
└─────────────────────────────────────────┘

장점: 상대적 관계 학습 가능 — "A가 B보다 더 관련"
단점: O(n²) 복잡도로 대량 문서 불가 → 실무에서는 거의 안 사용

# Pairwise 개념적 구현
def pairwise_rerank(query, candidates, model):
    """
    모든 쌍을 비교하여 승수로 랭킹
    n=50이면 조합 수 = 50*49/2 = 1,225회 실행 필요
    → 실무에서는 Top 10-20개만 적용
    """
    from itertools import combinations

    wins = {i: 0 for i in range(len(candidates))}

    for i, j in combinations(range(len(candidates)), 2):
        prompt = f"""
Query: {query}
Doc A: {candidates[i]}
Doc B: {candidates[j]}

어떤 문서가 Query에 더 관련이 있나? A 또는 B로만 답하세요.
"""
        winner = llm.predict(prompt)  # "A" 또는 "B"
        if winner == "A":
            wins[i] += 1
        else:
            wins[j] += 1

    # 승수 내림차순
    ranked_indices = sorted(wins, key=wins.get, reverse=True)
    return [candidates[i] for i in ranked_indices]

4.3 Listwise — "전체 리스트를 한 번에 재정렬"

┌─────────────────────────────────────────┐
│            Listwise Reranking            │
│                                         │
│  Query                                  │
│  + [Doc1, Doc2, Doc3, ..., Doc20]       │
│         ↓                               │
│      [LLM]                              │
│         ↓                               │
│  Output: [Doc3, Doc1, Doc5, ...]        │
│          (재정렬된 순서)                 │
│                                         │
│  복잡도: O(n) (sliding window 기본)     │
│  대표 모델: RankGPT, RankVicuna         │
└─────────────────────────────────────────┘

장점: 문서 간 상호 관계를 holistic하게 파악
단점: LLM 기반이므로 비용과 지연도가 높음

def listwise_rerank(query, candidates, llm):
    """
    RankGPT 스타일 Listwise Reranking

    LLM에게 전체 리스트를 보여주고,
    관련도 순으로 재정렬한 번호 리스트를 출력받음
    """
    # 문서에 번호 부여
    numbered_docs = "\n".join(
        [f"[{i+1}] {doc}" for i, doc in enumerate(candidates)]
    )

    prompt = f"""다음 문서들을 Query에 대한 관련도 높은 순으로 재정렬하세요.
관련도가 가장 높은 문서 번호부터 나열하세요.

Query: {query}

문서 리스트:
{numbered_docs}

관련도 순 번호 리스트 (예: [3, 1, 5, 2, 4]):
"""

    result = llm.predict(prompt)
    # "[3, 1, 5, 2, 4]" 파싱
    reordered_indices = parse_list(result)

    return [candidates[i-1] for i in reordered_indices]

Listwise의 Sliding Window 전략

LLM의 입력 제한으로 100개 문서를 한 번에 넣을 수 없습니다.
Sliding Window로 분할하여 처리합니다.

Top-100 후보 리스트:
[Doc1, Doc2, ..., Doc100]

Sliding Window (size=20, stride=10):

Pass 1: [Doc81 ~ Doc100]  → 재정렬 → 상위 10개 보존
Pass 2: [Doc71 ~ Doc90]   → 재정렬 → 상위 10개 보존
  ...
Pass N: [Doc1  ~ Doc20]   → 최종 정렬

→ 끝(덜 관련)부터 앞(가장 관련)으로 이동하며 정제

4.4 세 패러디그램 비교

특성	Pointwise	Pairwise	Listwise
단위	문서 1개	문서 2개	문서 N개
복잡도	O(n)	O(n²)	O(n) * window
상대 비교	❌	✅	✅
속도	⚡⚡⚡	⚡	⚡
정밀도	⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐⭐⭐
비용	$	$$$
실무 사용	가장 많음	거의 안 함	고품질 필요 시

연구 결과에서 Pointwise와 Listwise 모두 Open-domain QA 작업에서 Pairwise를 초과한 경우가 많었으며,
실무에서는 Pointwise Cross-Encoder가 비용과 정밀도의 최적 균형점이라고 평가됩니다.

5. Lost in the Middle — Reranking이 해결해야 할 진짜 문제

5.1 문제 정의

Stanford 연구팀(Liu et al., 2024)의 논문 "Lost in the Middle"에서 발견된 현상입니다.

LLM은 긴 컨텍스트에서 앞부분과 끝부분에만 주의를 기울이고, 중간부분을 무시하는 경향이 있습니다.

LLM의 주의(Attention) 분포:

주의도
  ▲
高 │ ████                          ████
  │ ████                          ████
  │ ████░░░░░░░░░░░░░░░░░░░░░░░░░████
低 │ ████░░░░░░░░░░░░░░░░░░░░░░░░░████
  └──────────────────────────────────→
     앞부분    중간부분(무시!)    끝부분

→ "U자형 주의 패턴" (Primacy Bias + Recency Bias)

이 연구에서 관련 정보의 위치에 따라 성능이 유의미하게 변화한다는 것이
확인되었으며, 이는 긴 컨텍스트 모델에서도 동일하게 관찰됩니다.

5.2 실제 영향

상황: 20개 문서를 LLM에 전달하고, 정답 문서가 15번째 위치

❌ Reranking 없이 전달:
  → 정답이 중간에 위치 → LLM이 무시 → 환각 답변

✅ Reranking 후 전달:
  → 정답이 1~3번째 위치로 이동 → LLM이 주의 기울임 → 정확한 답변

5.3 해결 전략: Reranking + 문서 배치

Reranking만으로는 부족합니다. 배치(ordering) 전략도 중요합니다.

def lost_in_the_middle_ordering(reranked_docs: list) -> list:
    """
    'Lost in the Middle' 논문의 권장 배치 전략

    가장 관련 있는 문서를 앞과 끝에 배치
    덜 관련 있는 문서를 중간에 배치

    Input (관련도 순):  [A, B, C, D, E]
    Output (배치 순):   [A, C, E, D, B]
                         ↑        ↑
                        앞       끝
    """
    if len(reranked_docs) <= 2:
        return reranked_docs

    result = [None] * len(reranked_docs)
    left, right = 0, len(reranked_docs) - 1

    for i, doc in enumerate(reranked_docs):
        if i % 2 == 0:
            result[left] = doc
            left += 1
        else:
            result[right] = doc
            right -= 1

    return result

# 예시
docs_by_relevance = ["정답", "관련A", "관련B", "관련C", "관련D"]
ordered = lost_in_the_middle_ordering(docs_by_relevance)
# → ["정답", "관련B", "관련D", "관련C", "관련A"]
#     (앞)                              (끝)

5.4 구조화된 프롬프트로 LLM 안내

def build_context_prompt(query, reranked_docs, top_k=5):
    """
    Primary / Secondary 구분으로 LLM에게 명확히 신호 전달
    """
    primary = reranked_docs[:3]       # 최상위 관련 문서
    secondary = reranked_docs[3:top_k] # 보조 문서

    primary_text = "\n".join(
        [f"[Primary {i+1}] {doc}" for i, doc in enumerate(primary)]
    )
    secondary_text = "\n".join(
        [f"[Secondary {i+1}] {doc}" for i, doc in enumerate(secondary)]
    )

    prompt = f"""다음 컨텍스트를 참고하여 질문에 답하세요.

=== Primary 컨텍스트 (가장 관련 있는 정보) ===
{primary_text}

=== Secondary 컨텍스트 (참고용) ===
{secondary_text}

위의 컨텍스트에 답이 없으면 "정보가 부족합니다"로 답하세요.
정보를 날捏하지 마세요.

질문: {query}
답변:"""

    return prompt

6. 2026년 Reranker 모델 비교

6.1 종합 비교표

모델	타입	언어	속도	정밀도	비용	주요 특징
Cohere Rerank v3.5	Cross-Encoder (API)	100+ 언어	⚡⚡	⭐⭐⭐⭐⭐	$$	Nimble 옵션, 실시간 최적
BGE-Reranker-v2	Cross-Encoder (오픈소스)	다국어	⚡⚡⚡	⭐⭐⭐⭐⭐	Free	자체 호스팅, Fine-Tuning 가능
Voyage Rerank-2.5	Cross-Encoder (API)	다국어	⚡⚡	⭐⭐⭐⭐⭐	$$	Agent · 대화형 최적화
ms-marco-MiniLM-L-6	Cross-Encoder (오픈소스)	영어 중심	⚡⚡⚡	⭐⭐⭐⭐	Free	경량(33M params), 빠름
Jina Reranker v2	Cross-Encoder (API)	다국어	⚡⚡	⭐⭐⭐⭐	$$	Multimodal, 긴 컨텍스트
FlashRank	Distilled (오픈소스)	영어 중심	⚡⚡⚡⚡	⭐⭐⭐	Free	초경량(~4MB), CPU 가능
RankGPT (LLM)	Listwise (API)	다국어	🐌	⭐⭐⭐⭐⭐		설명 가능, 비용 높음

6.2 각 모델 깊은 분석

Cohere Rerank — API 기반 최고 신뢰도

Cohere Rerank는 Transformer 아키텍처 기반의 Cross-Encoder로, Query와 Document를 함께 처리하여 정밀한 관련도를 판단합니다. 100개 이상의 언어를 지원하며, "Rerank 3 Nimble" 바전은 생산 환경에서 더 빠른 응답 속도를 제공하면서도 높은 정밀도를 유지합니다.

import cohere

co = cohere.Client("API_KEY")

def cohere_rerank(query: str, documents: list[str], top_k: int = 5):
    """
    Cohere Rerank API 사용

    ⚠️ documents는 텍스트 리스트 (dict 아님)
    """
    response = co.rerank(
        model="rerank-v3.5",          # 최신 모델
        query=query,
        documents=documents,
        top_k=top_k,
        return_documents=True         # 원본 문서 포함 반환
    )

    results = []
    for item in response.results:
        results.append({
            "document": item.document["text"],
            "relevance_score": item.relevance_score,  # 0~1 정규화
            "index": item.index                        # 원본 인덱스
        })

    return results

# 사용 예시
docs = [
    "한국어 자연어 처리 튜토리얼...",
    "영어 NLP 가이드...",
    "Python 프로그래밍..."
]

results = cohere_rerank("한국어 NLP 튜토리얼", docs, top_k=2)
for r in results:
    print(f"Score: {r['relevance_score']:.3f} | {r['document'][:40]}...")

BGE-Reranker-v2 — 오픈소스의 최강

BGE Reranker는 Embedding 모델과 다르게 Query와 Document를 직접 입력받아 유사도를 스코어로 출력합니다. Cross-entropy loss로 학습되었으므로 관련도 스코어가 특정 범위로 제한되지 않습니다.

BGE-Reranker-v2는 M3과 LLM(GEMMA, MiniCPM) 백본 위에서 학습되어 다국어 처리와 큰 입력 길이를 지원하며,
BEIR, C-MTEB, MIRACL 벤치마크에서 대폭적인 랭킹 성능 향상을 보였습니다.

from FlagEmbedding import FlagReranker

# BGE Reranker 로드
reranker = FlagReranker(
    "BAAI/bge-reranker-v2-m3",  # 다국어 v2
    use_fp16=True               # GPU 메모리 절감
)

def bge_rerank(query: str, passages: list[str], top_k: int = 5):
    """
    BGE Reranker — 오픈소스, 자체 호스팅 가능
    """
    # (query, passage) 쌍 생성
    pairs = [[query, passage] for passage in passages]

    # 스코어 계산
    scores = reranker.compute_score(pairs, normalize=True)  # 0~1 정규화

    # Top-k 정렬
    scored = list(zip(scores, passages))
    scored.sort(key=lambda x: x[0], reverse=True)

    return scored[:top_k]

# 사용 예시
results = bge_rerank("한국어 NLP 튜토리얼", docs, top_k=3)
for score, doc in results:
    print(f"Score: {score:.3f} | {doc[:40]}...")

FlashRank — "아무것도 불필요한" 경량 옵션

FlashRank는 초경량(~4MB) Reranking 라이브러리로, PyTorch와 Transformers 의존성 없이도 CPU에서 실행 가능합니다.

from flashrank import Ranker, RerankRequest, RerankItem

# 초경량 — GPU 불필요, CPU로 실행 가능
ranker = Ranker(model_name="ms-marco-MiniLM-L-6-v2")

def flashrank_rerank(query: str, passages: list[str], top_k: int = 5):
    """
    FlashRank — 프로토타입과 Cost-sensitive 환경에 최적
    """
    items = [RerankItem(text=p) for p in passages]

    request = RerankRequest(
        query=query,
        passages=items
    )

    results = ranker.rerank(request, k=top_k)
    return results

ms-marco-MiniLM-L-6-v2 — 경험증명된 baseline

ms-marco-MiniLM-L-6-v2는 일반 사용 케이스의 시작점으로 권장됩니다 — 빠르고, 정확하고, 잘 검증된 모델입니다.

연구에서 33.4M 파라미터의 ms-marco-MiniLM-L12가 278M(jina-reranker-v2)과 560M(bge-reranker-large)보다
큰 모델들을 초과하는 경우가 관찰되었으며, 이는 모델 크기보다 학습 데이터와 아키텍처 설계가 더 중요함을 의미합니다.

6.3 모델 크기 ≠ 정밀도

경험의 역설 (BioASQ 2025 연구 결과):

ms-marco-MiniLM-L12  (33.4M params)  →  최고 Reranking 성능
jina-reranker-v2     (278M params)   →  낮은 성능
bge-reranker-large   (560M params)   →  낮은 성능

→ Pre-training 데이터와 학습 방법이
   모델 크기보다 훨씬 더 중요

7. LLM을 Reranker로 사용할 때와 안 할 때

7.1 "LLM Reranker 아직도 쓰는가?" — Voyage AI의 연구

Voyage AI의 연구(2025년 10월)에서 목적 전용 Reranker와 LLM Reranker를 직접 비교한 결과,
목적 전용 Reranker(rerank-2.5)는 최신 LLM보다 최대 60배 저렴, 48배 빠르고, NDCG@10에서 최대 15% 더 높은 정밀도를 달성했습니다.

7.2 비교 실수치

기준	목적 전용 Reranker	LLM as Reranker
비용	$0.001/call	$0.05-0.10/call
지연도	50-200ms	2,000-5,000ms
정밀도 (NDCG@10)	높음	높음 (단, 일관성 부족)
설명 가능성	❌	✅
도메인 일반화	학습 데이터에 의존	강함 (zero-shot)

7.3 실제 연구 결과 (2025년 August, 22모델 · 40변형)

FlashRank-MiniLM은 속도와 정밀도의 균형이 좋을 때 강점이며, RankGPT는 속도와 정밀도 모두에서 뒤처져 효율성이 낮다고 평가됩니다.

구체적으로, FutureQueryEval(학습 데이터 이후의 새로운 쿼리)에서는
LLM 기반 Reranker가 익숙한 쿼리에서는 잘 작동하지만, 새로운 쿼리에 대한 일반화는 불일관적임을 확인했습니다.

7.4 결론: 언제 LLM Reranker를 쓰는가

LLM Reranker 사용 OK:
  ✅ "왜 이 문서가 관련이 높은가?" 설명이 필요한 경우
  ✅ 극도로 정밀한 단일 검색 (批量이 아닌 소량)
  ✅ 도메인이 매우 특수하고 학습 데이터가 없는 경우

LLM Reranker 사용 안함:
  ❌ 실시간 검색 (지연도 제약)
  ❌ 비용 민감한 환경
  ❌ 대량 검색 파이프라인
  ❌ 일반적인 RAG 시스템

8. 실무 구현: Full Reranking Pipeline

8.1 프로덕션 파이프라인

import time
import logging
import numpy as np
from dataclasses import dataclass, field
from typing import List, Optional
from enum import Enum

logger = logging.getLogger(__name__)

# ─── 데이터 모델 ─────────────────────────────────────

class RerankerType(Enum):
    BGE = "bge"
    COHERE = "cohere"
    FLASHRANK = "flashrank"
    CROSS_ENCODER = "cross_encoder"

@dataclass
class Document:
    """검색 결과 문서"""
    content: str
    metadata: dict = field(default_factory=dict)
    retrieval_score: float = 0.0       # 1차 검색 스코어
    rerank_score: Optional[float] = None  # Reranking 스코어
    source: str = ""

@dataclass
class RerankedResult:
    """최종 Reranking 결과"""
    documents: List[Document]
    query: str
    retrieval_time_ms: float
    reranking_time_ms: float
    model_used: str

# ─── Reranking Pipeline ─────────────────────────────

class RerankingPipeline:
    """
    프로덕션 Reranking 파이프라인

    기능:
    - 여러 Reranker 모델 지원
    - 자동 폴백 (primary 실패 시 fallback)
    - Lost in the Middle 방지 (배치 최적화)
    - 성능 모니터링
    """

    def __init__(
        self,
        primary_reranker: RerankerType = RerankerType.BGE,
        fallback_reranker: RerankerType = RerankerType.FLASHRANK,
        top_k: int = 5,
        candidate_k: int = 50
    ):
        self.primary = self._load_reranker(primary_reranker)
        self.fallback = self._load_reranker(fallback_reranker)
        self.top_k = top_k
        self.candidate_k = candidate_k

    def _load_reranker(self, reranker_type: RerankerType):
        """모델 로드"""
        if reranker_type == RerankerType.BGE:
            from FlagEmbedding import FlagReranker
            return FlagReranker("BAAI/bge-reranker-v2-m3", use_fp16=True)

        elif reranker_type == RerankerType.FLASHRANK:
            from flashrank import Ranker
            return Ranker(model_name="ms-marco-MiniLM-L-6-v2")

        elif reranker_type == RerankerType.COHERE:
            import cohere
            return cohere.Client("API_KEY")

        else:
            raise ValueError(f"지원하지 않는 모델: {reranker_type}")

    def rerank(
        self,
        query: str,
        documents: List[Document]
    ) -> RerankedResult:
        """
        메인 Reranking 실행

        1. 후보 제한
        2. Primary Reranker로 스코어링
        3. 실패 시 Fallback
        4. Lost in the Middle 배치 최적화
        """
        start_time = time.time()

        try:
            # 1. 후보 제한 (candidate_k개까지만)
            candidates = documents[:self.candidate_k]

            # 2. Primary Reranker로 스코어링
            scored_docs = self._score_documents(
                query, candidates, self.primary
            )

            # 3. Top-k 선정
            scored_docs.sort(key=lambda d: d.rerank_score, reverse=True)
            top_docs = scored_docs[:self.top_k]

            # 4. Lost in the Middle 배치 최적화
            ordered_docs = self._apply_litm_ordering(top_docs)

            reranking_time = (time.time() - start_time) * 1000

            logger.info(
                f"Reranking 완료: {len(candidates)}개 → "
                f"{len(top_docs)}개 | {reranking_time:.0f}ms"
            )

            return RerankedResult(
                documents=ordered_docs,
                query=query,
                retrieval_time_ms=0,  # 외부에서 주입
                reranking_time_ms=reranking_time,
                model_used=str(type(self.primary))
            )

        except Exception as e:
            logger.warning(f"Primary 실패: {e} → Fallback 사용")
            return self._fallback_rerank(query, documents)

    def _score_documents(
        self,
        query: str,
        documents: List[Document],
        reranker
    ) -> List[Document]:
        """BGE Reranker로 스코어링"""
        pairs = [[query, doc.content] for doc in documents]
        scores = reranker.compute_score(pairs, normalize=True)

        for doc, score in zip(documents, scores):
            doc.rerank_score = float(score)

        return documents

    def _apply_litm_ordering(
        self,
        docs: List[Document]
    ) -> List[Document]:
        """
        Lost in the Middle 방지 배치

        관련도: [1, 2, 3, 4, 5]  (1이 가장 높음)
        배치:   [1, 3, 5, 4, 2]  (앞과 끝에 중요한 것)
        """
        if len(docs) <= 2:
            return docs

        result = [None] * len(docs)
        left, right = 0, len(docs) - 1

        for i, doc in enumerate(docs):
            if i % 2 == 0:
                result[left] = doc
                left += 1
            else:
                result[right] = doc
                right -= 1

        return result

    def _fallback_rerank(
        self,
        query: str,
        documents: List[Document]
    ) -> RerankedResult:
        """Fallback Reranker 사용"""
        # FlashRank 등 경량 모델로 폴백
        candidates = documents[:self.candidate_k]
        # ... (fallback 로직)
        return RerankedResult(
            documents=candidates[:self.top_k],
            query=query,
            retrieval_time_ms=0,
            reranking_time_ms=0,
            model_used="fallback"
        )


# ─── 사용 예시 ───────────────────────────────────────

pipeline = RerankingPipeline(
    primary_reranker=RerankerType.BGE,
    fallback_reranker=RerankerType.FLASHRANK,
    top_k=5,
    candidate_k=50
)

# 1차 검색 결과 (예시)
retrieved_docs = [
    Document(content="한국어 NLP 튜토리얼...", retrieval_score=0.85),
    Document(content="영어 NLP 가이드...",     retrieval_score=0.82),
    Document(content="Python 프로그래밍...",   retrieval_score=0.78),
    # ... 최대 50개
]

# Reranking 실행
result = pipeline.rerank("한국어 자연어 처리 튜토리얼", retrieved_docs)

# 최종 문서 출력
for i, doc in enumerate(result.documents):
    print(f"#{i+1} | Rerank: {doc.rerank_score:.3f} | "
          f"Retrieval: {doc.retrieval_score:.3f} | "
          f"{doc.content[:50]}...")

8.2 LangChain과의 통합

from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank
from langchain_community.vectorstores import Qdrant
from langchain_community.embeddings import SentenceTransformerEmbeddings

# 1차 검색: Vector Store
embeddings = SentenceTransformerEmbeddings(model_name="BAAI/bge-m3")
vectorstore = Qdrant(...)  # 기존 벡터 DB
base_retriever = vectorstore.as_retriever(
    search_kwargs={"k": 50}  # 후보 50개
)

# 2차 정제: Cohere Rerank
reranker = CohereRerank(
    model="rerank-v3.5",
    top_n=5                    # 최종 5개
)

# 결합: Contextual Compression Retriever
compression_retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=base_retriever
)

# 사용
results = compression_retriever.invoke("한국어 NLP 튜토리얼")

8.3 BGE Reranker의 Fine-Tuning

# BGE Reranker Fine-Tuning (도메인 특화)
from FlagEmbedding import FlagReranker
from datasets import Dataset

# 학습 데이터 형식: (query, positive_doc, negative_doc)
train_data = [
    {
        "query": "환자의 혈당 수치 확인",
        "pos": "당뇨 환자 혈당 관리 가이드라인...",   # 정답
        "neg": "일반 건강 검진 안내..."               # Hard Negative
    },
    # ... 최소 500개 이상
]

# Fine-Tuning (FlagEmbedding 공식 스크립트 사용)
# → Hard Negative가 학습 품질의 핵심

Hard Negative를 학습 데이터로 사용하는 것이 Reranker Fine-Tuning의 핵심이며,
이를 통해 모델이 "비슷하지만 틀린" 문서와 "실제 정답" 문서를 구분하는 능력을 강화합니다.

9. 비용과 지연도의 최적화

9.1 후보 크기(candidate_k) 최적화

후보 수     │  정밀도 (NDCG@10)  │  비용  │  권장
────────────┼────────────────────┼────────┼────────
  20개      │  ██████░░░░        │  $     │  경량 앱
  50개      │  █████████░        │  $$    │  ★ Sweet Spot
  75개      │  ██████████        │  $$$   │  정밀도 우선
 100개      │  ██████████        │  $$$$  │ Diminishing Return

후보가 50-75개 사이에서 품질과 비용의 균형이 가장 좋고,
100개 이상으로 늘리면 비용은 두 배가 되지만 관련성 향상은 줄어듭니다.

9.2 비용 절감 전략

class CostOptimizedReranking:
    """
    비용 최적화 Reranking
    """

    def __init__(self):
        self.light_reranker = FlashRankReranker()   # 빠르고 저렴
        self.heavy_reranker = BGEReranker()          # 정밀하고 비싼 모델

    def rerank(self, query, documents):
        """
        2단계 Reranking으로 비용 절감

        1단계: FlashRank로 빠르게 후보 축소 (100 → 20)
        2단계: BGE로 정밀하게 정제 (20 → 5)
        """
        # 1단계: Light Reranker (100개 → 20개)
        light_results = self.light_reranker.rerank(
            query, documents[:100], top_k=20
        )

        # 2단계: Heavy Reranker (20개 → 5개)
        final_results = self.heavy_reranker.rerank(
            query,
            [doc for _, doc in light_results],
            top_k=5
        )

        return final_results

9.3 캐싱과 비동기 처리

import asyncio
from functools import lru_cache
import hashlib

class AsyncRerankingService:
    """
    비동기 + 캐싱 Reranking 서비스
    """

    def __init__(self):
        self.reranker = BGEReranker()
        self.cache = {}                    # 쿼리별 캐싱
        self.cache_ttl = 300              # 5분 TTL

    def _get_cache_key(self, query: str, doc_hashes: list) -> str:
        """캐시 키 생성"""
        content = query + str(sorted(doc_hashes))
        return hashlib.md5(content.encode()).hexdigest()

    async def rerank_async(
        self,
        query: str,
        documents: List[Document]
    ) -> List[Document]:
        """비동기 Reranking"""
        cache_key = self._get_cache_key(
            query,
            [hashlib.md5(d.content.encode()).hexdigest() for d in documents]
        )

        # 캐시 확인
        if cache_key in self.cache:
            logger.info("Cache hit!")
            return self.cache[cache_key]

        # 비동기로 실행
        loop = asyncio.get_event_loop()
        result = await loop.run_in_executor(
            None, self.reranker.rerank, query, documents
        )

        # 캐시 저장
        self.cache[cache_key] = result
        return result

9.4 ROI 계산

Reranking의 ROI는 LLM에 전달되는 컨텍스트의 품질을 높여서 불필요한 토큰 사용을 줄이는 것에서 옵니다. 큰 LLM을 관련 없는 정보로 돌리는 것보다, 정제된 적은 문서를 전달하는 것이 훨씬 저렴합니다.

Reranking 없이:
  1차 검색 → 50개 문서 → LLM (50개 컨텍스트)
  → LLM 비용: 50 * avg_tokens * price = 높음
  → 정밀도: 낮음

Reranking 있이:
  1차 검색 → 50개 → Reranking → 5개 → LLM (5개 컨텍스트)
  → Reranking 비용: 소규모
  → LLM 비용: 5 * avg_tokens * price = 낮음 (10x 절감)
  → 정밀도: 높음

10. 모델 선택 가이드

10.1 빠른 결정 트리

시작
  │
  ├─ 프라이버시 중요 (온프리미스)?
  │   ├─ YES → BGE-Reranker-v2-m3 (자체 호스팅)
  │   └─ NO  → 다음
  │
  ├─ 최고 정밀도 + 유지보수 최소화?
  │   ├─ YES → Cohere Rerank v3.5 (API)
  │   └─ NO  → 다음
  │
  ├─ 초경량 / CPU 운영 가능?
  │   ├─ YES → FlashRank (MiniLM)
  │   └─ NO  → 다음
  │
  ├─ 다국어 (특히 한국어) 지원?
  │   ├─ YES → BGE-Reranker-v2-m3 / Cohere
  │   └─ NO  → ms-marco-MiniLM-L-6-v2
  │
  └─ "왜?"가 중요한가? (설명 가능성)
      ├─ YES → LLM Reranker (비용 감수)
      └─ NO  → Cross-Encoder (일반 권장)

10.2 상황별 권장 스택

상황	Embedding	Reranker	후보→최종
스타트업 (빠르게)	OpenAI-3-small	FlashRank	30→5
중규모 (균형)	BGE-M3	BGE-Reranker-v2	50→5
엔터프라이즈 (정밀)	Cohere embed-v4	Cohere Rerank v3.5	75→10
프라이버시 우선	BGE-M3	BGE-Reranker-v2 (온프리미스)	50→5
한국어 특화	BGE-M3	BGE-Reranker-v2-m3	50→5

10.3 최종 체크리스트

Reranking 구축 시 반드시 확인:

□ 1차 검색 Recall 확인 (정답이 후보에 포함되는가?)
□ 후보 크기 결정 (sweet spot: 50-75개)
□ Reranker 모델 선택 (다국어? 온프리미스? 속도?)
□ Lost in the Middle 배치 전략 적용
□ Primary → Secondary 구조화된 프롬프트 사용
□ 폴백(Fallback) 모델 준비
□ 캐싱 전략 설정 (반복 쿼리 최적화)
□ 비용 모니터링 설정
□ A/B 테스트로 정밀도 검증
□ Reranking 전후 정밀도 비교 (반드시!)

마치며

Reranking은 RAG 파이프라인에서 "좋은 시스템"과 "사용자가 신뢰하는 시스템"의 차이를 만드는 단계입니다.

이번 글의 핵심 메시지

Reranking은 선택이 아닌 필수 — Bi-Encoder만으로는 교차 정보를 파악할 수 없습니다.
Pointwise Cross-Encoder가 실무의 표준 — 비용과 정밀도의 최적 균형점입니다.
LLM Reranker는 아직 아니다 — 60x 비용, 48x 지연도로 일반 사용 불가합니다.
Lost in the Middle은 반드시 대응 — Reranking + 배치 최적화 + 구조화 프롬프트.
모델 크기 ≠ 정밀도 — 33M의 MiniLM이 560M의 BGE를 초과하는 경우가 존재합니다.

시리즈 흐름

Chunking → Embedding → Reranking (현재)
                            ↓
                     다음: Agent 시스템
                     (ReAct · ToT · Reflexion)

검색 파이프라인 3단계(Chunking · Embedding · Reranking)가 완성되었습니다.
다음 글부터는 Agent 패턴으로 넘어가겠습니다.

참고 자료

Lost in the Middle (Liu et al., 2024): https://aclanthology.org/2024.tacl-1.9/
Voyage AI — Case Against LLMs as Rerankers: https://blog.voyageai.com/2025/10/22/the-case-against-llms-as-rerankers/
BGE Reranker: https://github.com/FlagOpen/FlagEmbedding
FlashRank: https://github.com/MoritzLaurer/FlashRank
RankLLM: https://rankllm.ai
awesome-rerankers: https://github.com/agentset-ai/awesome-rerankers

당니

👩🏻‍💻

이전 포스트