Reranking 深度解析：從 Pointwise 到 LLM Zero-Shot 的完整指南

# 問題：計算成本

# 方案 A：用強大模型直接檢索 10,000 個文檔
時間：10,000 × 20ms = 200 秒（太慢！）
成本：極高

# 方案 B：兩階段
階段 1：用快速模型檢索 → 50ms
階段 2：用強大模型 Rerank Top 50 → 200ms
總時間：250ms（可接受！）
成本：合理

結論：Reranking 是效率與效果的最佳平衡

# Pointwise 流程
for doc in candidate_docs:
    score = model.score(query, doc)
    # 每個文檔獨立評分
    
# 按分數排序
ranked_docs = sort_by_score(candidate_docs)

from sentence_transformers import CrossEncoder

class PointwiseReranker:
    def __init__(self):
        # 載入 Cross-Encoder 模型
        self.model = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
    
    def rerank(self, query, documents, top_k=5):
        """
        Pointwise Reranking
        
        Args:
            query: 查詢問題
            documents: 候選文檔列表
            top_k: 返回前 k 個結果
        """
        # 為每個文檔獨立評分
        pairs = [[query, doc] for doc in documents]
        scores = self.model.predict(pairs)
        
        # 按分數排序
        ranked_indices = np.argsort(scores)[::-1][:top_k]
        
        return [
            {
                'document': documents[i],
                'score': scores[i]
            }
            for i in ranked_indices
        ]

# 使用範例
reranker = PointwiseReranker()
results = reranker.rerank(
    query="Django 如何部署",
    documents=candidate_docs,
    top_k=5
)

for i, result in enumerate(results, 1):
    print(f"{i}. {result['document'][:50]}... (分數: {result['score']:.3f})")

def llm_pointwise_rerank(query, documents, top_k=5):
    """用 LLM 進行 Pointwise 評分"""
    scores = []
    
    for doc in documents:
        prompt = f"""
請評估以下文檔與查詢的相關度（0-10 分）。

查詢：{query}

文檔：{doc}

只回答一個數字（0-10）：
"""
        score = float(llm.generate(prompt).strip())
        scores.append(score)
    
    # 排序
    ranked_indices = np.argsort(scores)[::-1][:top_k]
    return [documents[i] for i in ranked_indices]

# Pairwise 流程
for doc_i in candidate_docs:
    for doc_j in candidate_docs:
        if i < j:  # 避免重複比較
            # 問：「Doc i 比 Doc j 更相關嗎？」
            comparison = model.compare(query, doc_i, doc_j)
            update_ranking(comparison)

# 根據比較結果排序
ranked_docs = sort_by_comparisons()

def llm_pairwise_rerank(query, documents, top_k=5):
    """
    LLM Pairwise Reranking
    使用 LLM 進行兩兩比較
    """
    n = len(documents)
    
    # 初始化勝場計數
    win_counts = [0] * n
    
    # 兩兩比較
    for i in range(n):
        for j in range(i + 1, n):
            prompt = f"""
請判斷哪個文檔與查詢更相關。

查詢：{query}

文檔 A：
{documents[i][:200]}...

文檔 B：
{documents[j][:200]}...

請只回答「A」或「B」：
"""
            winner = llm.generate(prompt).strip()
            
            if winner == 'A':
                win_counts[i] += 1
            else:
                win_counts[j] += 1
    
    # 按勝場數排序
    ranked_indices = np.argsort(win_counts)[::-1][:top_k]
    return [documents[i] for i in ranked_indices]

from itertools import combinations
import numpy as np

class PairwiseReranker:
    def __init__(self, model):
        self.model = model
    
    def rerank(self, query, documents, top_k=5):
        """
        Pairwise Reranking with Elo Rating System
        使用 Elo 評分系統（類似棋類排名）
        """
        n = len(documents)
        
        # 初始化 Elo 分數（所有文檔起始分數相同）
        elo_scores = np.ones(n) * 1500
        
        # 兩兩比較
        for i, j in combinations(range(n), 2):
            # LLM 判斷誰更相關
            comparison = self._compare(query, documents[i], documents[j])
            
            # 更新 Elo 分數
            if comparison == 'A':
                elo_scores[i], elo_scores[j] = self._update_elo(
                    elo_scores[i], elo_scores[j], winner='A'
                )
            else:
                elo_scores[i], elo_scores[j] = self._update_elo(
                    elo_scores[i], elo_scores[j], winner='B'
                )
        
        # 按 Elo 分數排序
        ranked_indices = np.argsort(elo_scores)[::-1][:top_k]
        return [documents[i] for i in ranked_indices]
    
    def _compare(self, query, doc_a, doc_b):
        """使用 LLM 比較兩個文檔"""
        prompt = f"""
查詢：{query}
文檔 A：{doc_a[:200]}
文檔 B：{doc_b[:200]}
哪個更相關？回答 A 或 B：
"""
        return self.model.generate(prompt).strip()
    
    def _update_elo(self, rating_a, rating_b, winner, k=32):
        """
        更新 Elo 分數
        K=32 是標準的 K 因子
        """
        expected_a = 1 / (1 + 10 ** ((rating_b - rating_a) / 400))
        expected_b = 1 - expected_a
        
        if winner == 'A':
            new_rating_a = rating_a + k * (1 - expected_a)
            new_rating_b = rating_b + k * (0 - expected_b)
        else:
            new_rating_a = rating_a + k * (0 - expected_a)
            new_rating_b = rating_b + k * (1 - expected_b)
        
        return new_rating_a, new_rating_b

class HybridReranker:
    """混合 Pointwise 和 Pairwise 的優勢"""
    
    def rerank(self, query, documents, top_k=5):
        # 階段 1：Pointwise 快速篩選 Top 20
        pointwise_top = self.pointwise_rerank(query, documents, top_k=20)
        
        # 階段 2：Pairwise 精確排序 Top 5
        final_top = self.pairwise_rerank(query, pointwise_top, top_k=top_k)
        
        return final_top

# 優勢
✅ 速度：只需 20 + 190 = 210 次比較（vs 1225 次）
✅ 準確率：接近純 Pairwise
✅ 成本：可接受

# 傳統 Reranker（需要訓練）
訓練資料：10,000+ 個標註的查詢-文檔對
訓練時間：數天
訓練成本：$$$

# LLM Zero-shot Reranker（無需訓練）
訓練資料：0
訓練時間：0
成本：只有 API 費用
效果：接近甚至超越訓練的模型！

# LLM 已經在預訓練時學會了：
1. 理解查詢的意圖
2. 理解文檔的內容
3. 判斷相關性

# 只需設計好 Prompt，就能直接使用

def llm_zero_shot_rerank(query, documents, top_k=5):
    """最簡單的 LLM Reranking"""
    scores = []
    
    for doc in documents:
        prompt = f"""
任務：評估文檔與查詢的相關度（0-10 分）

查詢：{query}

文檔：
{doc}

請只回答一個 0-10 的數字，表示相關度：
"""
        score = float(llm.generate(prompt, temperature=0).strip())
        scores.append(score)
    
    # 排序
    ranked_indices = np.argsort(scores)[::-1][:top_k]
    return [documents[i] for i in ranked_indices]

def llm_cot_rerank(query, documents, top_k=5):
    """
    使用思維鏈讓 LLM 「思考」
    效果更好但稍慢
    """
    scores = []
    
    for doc in documents:
        prompt = f"""
任務：評估文檔與查詢的相關度

查詢：{query}

文檔：
{doc}

請按以下步驟思考：
1. 查詢的核心意圖是什麼？
2. 文檔主要討論什麼？
3. 文檔是否直接回答查詢？
4. 相關度評分（0-10）：

格式：
思考：[你的分析]
分數：[0-10]
"""
        response = llm.generate(prompt, temperature=0)
        
        # 提取分數
        score_line = [l for l in response.split('\n') if l.startswith('分數')][0]
        score = float(score_line.split('：')[1])
        scores.append(score)
    
    ranked_indices = np.argsort(scores)[::-1][:top_k]
    return [documents[i] for i in ranked_indices]

def llm_batch_rerank(query, documents, top_k=5):
    """
    一次性評估所有文檔
    速度快但可能不夠精確
    """
    # 格式化所有文檔
    docs_text = '\n\n'.join([
        f"[文檔 {i+1}]\n{doc[:200]}..."
        for i, doc in enumerate(documents)
    ])
    
    prompt = f"""
任務：為每個文檔評分（0-10）

查詢：{query}

文檔列表：
{docs_text}

請以 JSON 格式輸出每個文檔的分數：
{{"scores": [8, 5, 9, 3, ...]}}
"""
    response = llm.generate(prompt, temperature=0)
    scores = json.loads(response)['scores']
    
    ranked_indices = np.argsort(scores)[::-1][:top_k]
    return [documents[i] for i in ranked_indices]

class RerankerFeedbackCollector:
    """收集 Reranker 的評分作為訓練資料"""
    
    def __init__(self, retriever, reranker):
        self.retriever = retriever
        self.reranker = reranker
        self.training_data = []
    
    def collect(self, queries):
        """收集訓練資料"""
        for query in queries:
            # 1. Retriever 檢索
            candidates = self.retriever.retrieve(query, k=50)
            
            # 2. Reranker 重新排序
            reranked = self.reranker.rerank(query, candidates)
            
            # 3. 記錄：哪些文檔被認為相關
            positive_docs = reranked[:5]  # Top 5 是正例
            negative_docs = reranked[-10:]  # 最後 10 個是負例
            
            self.training_data.append({
                'query': query,
                'positive': positive_docs,
                'negative': negative_docs
            })
        
        return self.training_data

from sentence_transformers import SentenceTransformer, InputExample, losses
from torch.utils.data import DataLoader

def train_retriever_with_feedback(model, training_data, epochs=3):
    """
    使用 Reranker 的回饋訓練 Retriever
    
    目標：讓 Retriever 學習 Reranker 的偏好
    """
    # 1. 準備訓練樣本
    train_examples = []
    for item in training_data:
        query = item['query']
        
        # 正例：Reranker 認為相關的文檔
        for pos_doc in item['positive']:
            train_examples.append(
                InputExample(texts=[query, pos_doc], label=1.0)
            )
        
        # 負例：Reranker 認為不相關的文檔
        for neg_doc in item['negative']:
            train_examples.append(
                InputExample(texts=[query, neg_doc], label=0.0)
            )
    
    # 2. 建立 DataLoader
    train_dataloader = DataLoader(
        train_examples,
        shuffle=True,
        batch_size=16
    )
    
    # 3. 選擇 Loss Function
    train_loss = losses.CosineSimilarityLoss(model)
    
    # 4. 訓練
    model.fit(
        train_objectives=[(train_dataloader, train_loss)],
        epochs=epochs,
        warmup_steps=100
    )
    
    print("✅ Retriever 訓練完成！")
    return model

# 使用流程
# 1. 收集資料
collector = RerankerFeedbackCollector(retriever, reranker)
training_data = collector.collect(user_queries)

# 2. 訓練 Retriever
improved_retriever = train_retriever_with_feedback(
    model=retriever.encoder,
    training_data=training_data
)

# 3. 更新系統
retriever.encoder = improved_retriever

def distill_reranker_to_retriever(
    student_model,  # Retriever（學生）
    teacher_model,  # Reranker（老師）
    training_queries,
    epochs=5
):
    """
    知識蒸餾：讓 Retriever 模仿 Reranker
    
    不僅學習「對/錯」，還學習「多相關」
    """
    train_examples = []
    
    for query in training_queries:
        # 檢索候選文檔
        candidates = student_model.retrieve(query, k=50)
        
        # 用 Reranker（老師）評分
        teacher_scores = []
        for doc in candidates:
            score = teacher_model.score(query, doc)
            teacher_scores.append(score)
        
        # 標準化分數（0-1）
        teacher_scores = normalize(teacher_scores)
        
        # 建立訓練樣本：讓 Student 學習 Teacher 的分數
        for doc, score in zip(candidates, teacher_scores):
            train_examples.append(
                InputExample(
                    texts=[query, doc],
                    label=score  # Teacher 的評分作為標籤
                )
            )
    
    # 訓練 Student
    train_dataloader = DataLoader(train_examples, batch_size=16)
    train_loss = losses.CosineSimilarityLoss(student_model)
    
    student_model.fit(
        train_objectives=[(train_dataloader, train_loss)],
        epochs=epochs
    )
    
    return student_model

✅ 使用 Cross-Encoder
✅ Pointwise 評分
→ 快速見效，準確率 +15%

✅ 加入快取機制
✅ 批次處理
→ 速度提升 2-3 倍

✅ 嘗試 LLM Zero-shot
✅ 或 Pointwise + Pairwise 混合
→ 準確率再 +5-10%

✅ 收集 Reranker 回饋
✅ 訓練改進 Retriever
→ 系統整體提升

from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_rerank(query, docs_tuple):
    """快取常見查詢的 Reranking 結果"""
    return reranker.rerank(query, list(docs_tuple))

# 命中率 30-50%，節省 30-50% 成本

def smart_rerank(query, candidates):
    """根據候選文檔品質決定是否 Rerank"""
    # 快速評估候選文檔品質
    if all_candidates_look_good(candidates[:5]):
        # 前 5 個都不錯，不需要 Rerank
        return candidates[:5]
    else:
        # 品質不穩定，需要 Rerank
        return reranker.rerank(query, candidates)

# 便宜模型初排 → 昂貴模型精排
candidates_50 = retriever.get(k=50)
candidates_20 = cheap_reranker.rerank(candidates_50, k=20)
final_5 = expensive_reranker.rerank(candidates_20, k=5)

# 成本降低，效果不變