檢索品質:hybrid search 與 rerank

先建黃金測試集,有數字才能比

在動任何程式碼之前先量「現在的系統有多差」。你需要一份「問題 → 期待命中的 chunk_id」對照表:

[
  {
    "query": "k8s pod crashloopbackoff 怎麼排查",
    "expected_chunk_id": "doc-42-chunk-3"
  },
  {
    "query": "invoice API /v2/invoices 的 rate limit 是多少",
    "expected_chunk_id": "doc-17-chunk-1"
  },
  {
    "query": "2025 Q4 revenue 成長率",
    "expected_chunk_id": "doc-88-chunk-0"
  }
]

測試集最快的來源:從你的文件中隨機抽 30~50 個 chunk,請 Claude 根據每個 chunk 的內容生成多樣化的問題。注意 prompt 要求生成口語版、縮寫版,不然測試集會太理想化:

import anthropic, json

client = anthropic.Anthropic()

def generate_test_questions(chunk_text: str, chunk_id: str, n: int = 2) -> list:
    prompt = f"""根據以下文件段落,生成 {n} 個真實使用者可能詢問的問題。
問題必須能從這個段落直接回答。生成多樣化的問法,包含:
- 正式完整問句
- 口語簡短版(像在 Slack 裡問同事)
- 只有關鍵字的搜尋式查詢

只輸出 JSON 陣列,格式如下:

段落內容:
{chunk_text}

[{{"query": "問題", "expected_chunk_id": "{chunk_id}"}}]"""

    msg = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=512,
        messages=[{"role": "user", "content": prompt}]
    )
    return json.loads(msg.content[0].text)

# 對所有 chunk 跑完後合成測試集
all_cases = []
for chunk in your_chunks:  # 你的 Document 物件清單
    cases = generate_test_questions(
        chunk.page_content,
        chunk.metadata["chunk_id"]
    )
    all_cases.extend(cases)

json.dump(all_cases, open("test_set.json", "w"), ensure_ascii=False, indent=2)
print(f"測試集共 {len(all_cases)} 筆")

30 個 chunk × 每個 2 題 = 60 筆,已經夠量出有意義的命中率了。

量基準:純向量的命中率

先定義兩個指標函數,之後每個版本的 retriever 都用同一套量:

from langchain_voyageai import VoyageAIEmbeddings
from langchain_chroma import Chroma
import json

embedding = VoyageAIEmbeddings(
    model="voyage-4-lite",  # $0.02/MTok,前 200M tokens 免費
    voyage_api_key="va-..."
)
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embedding
)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

test_set = json.load(open("test_set.json"))

def hit_rate(retriever, test_cases: list, k: int = 5) -> float:
    """k 筆結果中命中期待 chunk 的比例"""
    hits = 0
    for case in test_cases:
        results = retriever.invoke(case["query"])
        retrieved_ids = [r.metadata.get("chunk_id") for r in results[:k]]
        if case["expected_chunk_id"] in retrieved_ids:
            hits += 1
    return hits / len(test_cases)

def mrr(retriever, test_cases: list, k: int = 10) -> float:
    """Mean Reciprocal Rank:命中排在越前面分數越高"""
    rr_sum = 0.0
    for case in test_cases:
        results = retriever.invoke(case["query"])
        for rank, doc in enumerate(results[:k], start=1):
            if doc.metadata.get("chunk_id") == case["expected_chunk_id"]:
                rr_sum += 1 / rank
                break
    return rr_sum / len(test_cases)

baseline_hr = hit_rate(vector_retriever, test_set)
baseline_mrr = mrr(vector_retriever, test_set)
print(f"純向量 Hit Rate@5: {baseline_hr:.2%}")
print(f"純向量 MRR@10:    {baseline_mrr:.3f}")

把這兩個數字截圖存下來,這是後續所有改善的對照基準。

加 BM25,建 Hybrid Retriever

pip install langchain-community rank_bm25 jieba

from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
import jieba

# 中文必須自訂分詞函數,英文可以跳過這段
def chinese_tokenizer(text: str) -> list[str]:
    return list(jieba.cut(text))

# BM25 需要原始的 Document 物件清單(就是你 Chroma 裡的 docs)
bm25_retriever = BM25Retriever.from_documents(
    docs,
    preprocess_func=chinese_tokenizer  # 英文文件可省略
)
bm25_retriever.k = 10  # 召回先取多一點

# 向量那邊也放寬,k=10
vector_retriever_wide = vectorstore.as_retriever(search_kwargs={"k": 10})

# EnsembleRetriever 內部使用 RRF 融合
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever_wide],
    weights=[0.5, 0.5]  # 兩邊各半;若你的文件專有名詞多,可試 [0.6, 0.4]
)

print(f"Hybrid Hit Rate@5: {hit_rate(ensemble_retriever, test_set):.2%}")
print(f"Hybrid MRR@10:    {mrr(ensemble_retriever, test_set):.3f}")

正常情況下,hybrid 比純向量的 Hit Rate 提升 5~20 個百分點,尤其在你的文件有大量技術術語、型號、縮寫的時候效果最明顯。

加 Reranker:二階段精排

用 Voyage rerank-2 接法(前 200M tokens 完全免費,推薦開發測試期間使用):

pip install voyageai

import voyageai

vc = voyageai.Client(api_key="pa-...")

def rerank_with_voyage(
    query: str,
    candidates: list,
    model: str = "rerank-2",
    top_k: int = 5
) -> list:
    doc_texts = [d.page_content for d in candidates]
    result = vc.rerank(
        query=query,
        documents=doc_texts,
        model=model,
        top_k=top_k
    )
    # result.results 是 RerankingObject list,含 index 和 relevance_score
    return [candidates[r.index] for r in result.results]


def hybrid_rerank_retriever(query: str, k_final: int = 5) -> list:
    # 第一階段:hybrid 召回 top-20
    candidates = ensemble_retriever.invoke(query)[:20]
    # 第二階段:rerank 精排
    return rerank_with_voyage(query, candidates, top_k=k_final)


# 量測
hits = sum(
    1 for case in test_set
    if case["expected_chunk_id"] in [
        d.metadata.get("chunk_id")
        for d in hybrid_rerank_retriever(case["query"])
    ]
)
print(f"Hybrid+Rerank Hit Rate@5: {hits/len(test_set):.2%}")

如果你已經有 Cohere API key,可以用 LangChain 整合,接法更簡潔:

pip install langchain-cohere

from langchain_cohere import CohereRerank
from langchain.retrievers.contextual_compression import ContextualCompressionRetriever

compressor = CohereRerank(
    model="rerank-v3.5",
    top_n=5,
    cohere_api_key="..."
)
final_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=ensemble_retriever
)

# 這個 retriever 就可以直接 .invoke(query),跟其他 retriever 一樣用
print(f"Hybrid+Cohere Hit Rate@5: {hit_rate(final_retriever, test_set):.2%}")

Cohere Rerank v3.5 是按「搜尋次數」計費($2.00 / 1,000 次搜尋),Voyage rerank-2 則按 token 計費($0.05/MTok);兩者計費單位不同,實際成本取決於每次 rerank 的文件數與長度,但多數 RAG 場景 Cohere 會貴一截。開發測試先用 Voyage,上線後再評估是否需要 Cohere 的更強多語言能力。

找 top-k 的最佳值

reranker 的 top_n 決定最終送進 LLM 的文件數量。量 Hit Rate 在不同 k 下的分布,找邊際效益拐點:

for k in [3, 5, 8, 10]:
    hr = sum(
        1 for case in test_set
        if case["expected_chunk_id"] in [
            d.metadata.get("chunk_id")
            for d in hybrid_rerank_retriever(case["query"], k_final=k)
        ]
    ) / len(test_set)
    print(f"k={k:2d}: Hit Rate = {hr:.2%}")

k= 3: Hit Rate = 72.0%
k= 5: Hit Rate = 84.0%
k= 8: Hit Rate = 87.0%
k=10: Hit Rate = 88.0%

k=5 到 k=8 之間通常有一個明顯的「邊際效益拐點」。超過這個點再加文件,命中率提升不多,但每次生成都要讓 LLM 讀更多 context、增加 token 費用和延遲。找到拐點之後固定 top_k 就好。