前言
Naive RAG 的核心問題:搜尋品質決定了答案品質。
一個常見的現象是,明明知識庫裡有答案,但因為使用者的問題措辭跟文件不同,向量搜尋就找不到。或者,找到的 Top-5 結果裡,真正相關的其實排在第 4 位,LLM 因此被無關資訊干擾。
這篇介紹四個能顯著提升搜尋品質的技術。
技術 1:混合搜尋(Hybrid Search)
核心問題
純向量搜尋(Semantic Search)擅長找「語意相近」的內容,但對精確術語、專有名詞、縮寫效果差。
問題:「GPT-4o 的 context window 是多少?」
純向量搜尋找到:「大型語言模型通常有輸入長度限制...」(語意相近但沒答案)
BM25 關鍵字搜尋找到:「GPT-4o 支援 128K token 的 context window」(精確命中)
混合搜尋 = 語意搜尋 + 關鍵字搜尋,兩者結果用 RRF 或加權融合。
BM25 簡介
BM25 是 TF-IDF 的改進版,計算關鍵字與文件的相關度:
Score(D, Q) = Σ IDF(qi) * (tf(qi, D) * (k1 + 1)) / (tf(qi, D) + k1 * (1 - b + b * |D| / avgdl))
不需要理解公式,只需知道:BM25 對精確詞彙匹配非常靈敏。
Reciprocal Rank Fusion(RRF)
RRF 是融合兩個排序結果的標準方法:
1def rrf_fusion(rankings: list[list[str]], k: int = 60) -> dict[str, float]:
2 """
3 rankings: 每個搜尋方法的文件 ID 排序列表
4 k: 平滑常數(通常用 60)
5 回傳:每個文件 ID 的融合分數
6 """
7 scores = {}
8 for ranking in rankings:
9 for rank, doc_id in enumerate(ranking, start=1):
10 scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank)
11 return dict(sorted(scores.items(), key=lambda x: x[1], reverse=True))
完整混合搜尋實作
1from rank_bm25 import BM25Okapi
2import numpy as np
3import openai
4import jieba # 中文分詞
5
6client = openai.OpenAI(api_key="your-api-key")
7
8
9class HybridRetriever:
10 def __init__(self, chunks: list[str]):
11 self.chunks = chunks
12
13 # 建立 BM25 index(中文需要先分詞)
14 tokenized = [list(jieba.cut(c)) for c in chunks]
15 self.bm25 = BM25Okapi(tokenized)
16
17 # 建立向量 index
18 self.embeddings = self._embed_all(chunks)
19
20 def _embed_all(self, texts: list[str]) -> np.ndarray:
21 response = client.embeddings.create(
22 input=texts, model="text-embedding-3-small"
23 )
24 return np.array([r.embedding for r in response.data])
25
26 def _embed(self, text: str) -> np.ndarray:
27 response = client.embeddings.create(
28 input=text, model="text-embedding-3-small"
29 )
30 return np.array(response.data[0].embedding)
31
32 def semantic_search(self, query: str, top_k: int = 10) -> list[int]:
33 """語意搜尋,回傳文件索引排序"""
34 q_emb = self._embed(query)
35 # 餘弦相似度
36 sims = self.embeddings @ q_emb / (
37 np.linalg.norm(self.embeddings, axis=1) * np.linalg.norm(q_emb) + 1e-9
38 )
39 return np.argsort(-sims)[:top_k].tolist()
40
41 def keyword_search(self, query: str, top_k: int = 10) -> list[int]:
42 """BM25 關鍵字搜尋,回傳文件索引排序"""
43 tokens = list(jieba.cut(query))
44 scores = self.bm25.get_scores(tokens)
45 return np.argsort(-scores)[:top_k].tolist()
46
47 def hybrid_search(
48 self, query: str, top_k: int = 5, alpha: float = 0.5
49 ) -> list[str]:
50 """
51 混合搜尋
52 alpha=0.0: 純關鍵字, alpha=1.0: 純語意, alpha=0.5: 各半
53 """
54 sem_ranking = self.semantic_search(query, top_k=20)
55 kw_ranking = self.keyword_search(query, top_k=20)
56
57 # RRF 融合
58 k = 60
59 scores: dict[int, float] = {}
60 for rank, idx in enumerate(sem_ranking, 1):
61 scores[idx] = scores.get(idx, 0) + alpha * (1 / (k + rank))
62 for rank, idx in enumerate(kw_ranking, 1):
63 scores[idx] = scores.get(idx, 0) + (1 - alpha) * (1 / (k + rank))
64
65 top_indices = sorted(scores, key=scores.get, reverse=True)[:top_k]
66 return [self.chunks[i] for i in top_indices]
67
68
69# 使用範例
70chunks = [
71 "GPT-4o 支援 128K token 的 context window,是目前 OpenAI 最強的多模態模型。",
72 "大型語言模型通常有輸入長度的限制,這個限制被稱為 context window。",
73 "Python 在機器學習領域被廣泛使用,常見框架有 TensorFlow 和 PyTorch。",
74]
75
76retriever = HybridRetriever(chunks)
77results = retriever.hybrid_search("GPT-4o context window 大小", top_k=2)
78for r in results:
79 print(r)
優點:結合兩種搜尋的優點,對術語和語意都能命中
缺點:需要維護兩個索引;alpha 值需要根據資料集調整
最佳使用場景:技術文件、法律/醫療知識庫、有大量專有名詞的領域
技術 2:假設性文件嵌入(HyDE)
核心問題
使用者的問題和文件的語言風格不同:
- 問題:「如何減少 API 延遲?」(問句)
- 文件:「使用連線池可以將 API 延遲降低 40%」(陳述句)
問句和陳述句的向量距離比陳述句之間的距離更大,導致搜尋效果差。
HyDE 的解法
讓 LLM 先根據問題生成一個假設性的答案文件,再用這個假設答案去搜尋。假設答案的語言風格跟知識庫文件更接近,搜尋效果更好。
問題(問句)→ LLM → 假設答案(陳述句)→ 向量化 → 搜尋 → 真實文件
即使假設答案的事實不正確也沒關係,它的語言模式已經足夠引導搜尋。
完整實作
1import openai
2
3client = openai.OpenAI(api_key="your-api-key")
4
5
6def generate_hypothetical_document(question: str) -> str:
7 """讓 LLM 生成一個假設性的回答文件"""
8 prompt = f"""請根據以下問題,寫一段簡短的技術說明(3-5 句話),
9就像這是從技術文件中摘錄的一段內容。
10不需要完全正確,重點是語言風格要像技術文件。
11
12問題:{question}
13
14技術文件段落:"""
15
16 response = client.chat.completions.create(
17 model="gpt-4o-mini",
18 messages=[{"role": "user", "content": prompt}],
19 temperature=0.7, # 稍高一點,讓生成的文件更多樣
20 max_tokens=200,
21 )
22 return response.choices[0].message.content
23
24
25def get_embedding(text: str) -> list[float]:
26 return client.embeddings.create(
27 input=text, model="text-embedding-3-small"
28 ).data[0].embedding
29
30
31class HyDERetriever:
32 def __init__(self, chunks: list[str]):
33 self.chunks = chunks
34 self.embeddings = [get_embedding(c) for c in chunks]
35
36 def retrieve(self, question: str, top_k: int = 3) -> list[str]:
37 # Step 1: 生成假設性文件
38 hypothetical_doc = generate_hypothetical_document(question)
39 print(f"🤔 假設性文件:{hypothetical_doc[:100]}...")
40
41 # Step 2: 用假設性文件的向量做搜尋(而非原始問題)
42 hypo_embedding = get_embedding(hypothetical_doc)
43
44 import numpy as np
45 emb_matrix = np.array(self.embeddings)
46 q_emb = np.array(hypo_embedding)
47
48 sims = emb_matrix @ q_emb / (
49 np.linalg.norm(emb_matrix, axis=1) * np.linalg.norm(q_emb) + 1e-9
50 )
51 top_indices = np.argsort(-sims)[:top_k]
52 return [self.chunks[i] for i in top_indices]
53
54
55# 使用範例
56chunks = [
57 "使用連線池(Connection Pooling)可以將 API 的平均延遲降低 30-50%,因為避免了反覆建立 TCP 連線的開銷。",
58 "非同步 I/O(asyncio)可以讓單一 Python 程序同時處理數千個並發請求,大幅提升吞吐量。",
59 "CDN(內容分發網路)可以將靜態資源快取到離使用者最近的節點,減少網路傳輸延遲。",
60 "資料庫查詢優化:使用索引、避免 N+1 問題、適當的快取策略,是降低後端延遲的關鍵。",
61]
62
63retriever = HyDERetriever(chunks)
64results = retriever.retrieve("如何減少 API 的回應時間?")
65for r in results:
66 print(f" 📄 {r}")
優點:有效彌補問句與文件語言風格的差異;實作簡單
缺點:多一次 LLM 呼叫(延遲 + 成本);假設性文件可能引入偏見
最佳使用場景:使用者問題措辭跟文件風格差異大的場景,例如:口語化問題 vs 正式技術文件
技術 3:多查詢檢索(Multi-Query Retrieval)
核心問題
一個問題換一種說法,搜尋結果可能完全不同:
原始問題:「Python 比 Java 快嗎?」
→ 搜尋結果:[Python 效能評測, Java 效能測試]
換一種說法:「Java 的執行速度是否優於 Python?」
→ 搜尋結果:[JVM 效能優化, 動態語言速度分析] ← 可能更相關!
單一查詢的搜尋結果受措辭影響很大,會遺漏相關文件。
解法:自動生成多個查詢版本
1import openai
2from typing import Any
3
4client = openai.OpenAI(api_key="your-api-key")
5
6
7def generate_query_variations(original_query: str, n: int = 3) -> list[str]:
8 """用 LLM 生成同一個問題的多種不同表達方式"""
9 prompt = f"""你是一個搜尋查詢優化助手。
10請將以下問題改寫成 {n} 個不同的表達方式,
11角度可以不同(例如:換用不同詞彙、從反面問、更具體、更抽象)。
12每個改寫版本單獨一行,不要加編號或額外解釋。
13
14原始問題:{original_query}
15
16改寫版本:"""
17
18 response = client.chat.completions.create(
19 model="gpt-4o-mini",
20 messages=[{"role": "user", "content": prompt}],
21 temperature=0.7,
22 )
23 variations = response.choices[0].message.content.strip().split("\n")
24 return [v.strip() for v in variations if v.strip()][:n]
25
26
27def get_embedding(text: str) -> list[float]:
28 return client.embeddings.create(
29 input=text, model="text-embedding-3-small"
30 ).data[0].embedding
31
32
33class MultiQueryRetriever:
34 def __init__(self, chunks: list[str]):
35 self.chunks = chunks
36 import numpy as np
37 self.embeddings = np.array([get_embedding(c) for c in chunks])
38
39 def retrieve(self, question: str, top_k: int = 3) -> list[str]:
40 import numpy as np
41
42 # 生成多個查詢版本
43 variations = generate_query_variations(question, n=3)
44 all_queries = [question] + variations
45
46 print("🔍 查詢版本:")
47 for q in all_queries:
48 print(f" - {q}")
49
50 # 對每個查詢版本做搜尋,收集所有結果
51 seen_indices = set()
52 candidate_scores: dict[int, float] = {}
53
54 for query in all_queries:
55 q_emb = np.array(get_embedding(query))
56 sims = self.embeddings @ q_emb / (
57 np.linalg.norm(self.embeddings, axis=1) * np.linalg.norm(q_emb) + 1e-9
58 )
59 # 取每個查詢的 top_k 結果
60 top_indices = np.argsort(-sims)[:top_k]
61 for rank, idx in enumerate(top_indices):
62 # 用 RRF 融合多個查詢的排名
63 candidate_scores[int(idx)] = (
64 candidate_scores.get(int(idx), 0) + 1 / (60 + rank + 1)
65 )
66
67 # 按融合分數排序,去重後取 top_k
68 top_indices = sorted(candidate_scores, key=candidate_scores.get, reverse=True)
69 return [self.chunks[i] for i in top_indices[:top_k]]
70
71
72# 使用範例
73chunks = [
74 "Python 是一種直譯語言,執行速度比 Java 慢約 5-10 倍,但開發效率更高。",
75 "Java 透過 JIT 編譯器將程式碼編譯成機器碼,執行效能接近 C++。",
76 "對於 CPU 密集型任務,Java 的效能通常優於 Python;但 Python 的 NumPy 等函式庫利用 C 底層可以媲美 Java。",
77 "Python 的 GIL(全域解釋鎖)限制了真正的多執行緒並行,而 Java 沒有這個限制。",
78]
79
80retriever = MultiQueryRetriever(chunks)
81results = retriever.retrieve("Python 和 Java 哪個比較快?")
82print("\n📑 檢索結果:")
83for r in results:
84 print(f" {r}")
優點:覆蓋更多相關文件,減少因措辭導致的遺漏;實作相對簡單
缺點:多次 Embedding 計算(成本增加 N 倍);可能引入不相關結果
最佳使用場景:使用者問題措辭不穩定、知識庫術語多樣的場景
技術 4:Cross-Encoder Reranker
核心問題
向量搜尋的 Top-K 結果,排序不一定最優。向量搜尋用雙塔模型(Bi-Encoder)——問題和文件分別向量化,再計算距離——速度快但精準度有限。
Reranker(Cross-Encoder):把問題和每個候選文件一起送進模型,讓模型直接判斷這對「問題+文件」的相關程度。精準度更高,但速度慢(適合在取得候選後做二次排序)。
向量搜尋(Bi-Encoder): 問題向量 vs 文件向量 → 距離計算
Cross-Encoder Reranker: [問題 + 文件] → 相關分數(0-1)
完整實作
1# 安裝:pip install sentence-transformers
2from sentence_transformers import CrossEncoder
3import openai
4import numpy as np
5
6client = openai.OpenAI(api_key="your-api-key")
7
8# 載入 Cross-Encoder 模型(本地,不需要 API)
9# 中文推薦:cross-encoder/ms-marco-MiniLM-L-6-v2(英文為主)
10# 或 BAAI/bge-reranker-base(中英文)
11reranker = CrossEncoder("BAAI/bge-reranker-base")
12
13
14def get_embedding(text: str) -> list[float]:
15 return client.embeddings.create(
16 input=text, model="text-embedding-3-small"
17 ).data[0].embedding
18
19
20class RerankRetriever:
21 def __init__(self, chunks: list[str]):
22 self.chunks = chunks
23 self.embeddings = np.array([get_embedding(c) for c in chunks])
24
25 def retrieve(
26 self, query: str, initial_top_k: int = 20, final_top_k: int = 5
27 ) -> list[dict]:
28 """
29 兩階段檢索:
30 1. 向量搜尋取 top 20(召回)
31 2. Cross-Encoder rerank 取 top 5(精排)
32 """
33 # Stage 1: 向量搜尋(粗召回)
34 q_emb = np.array(get_embedding(query))
35 sims = self.embeddings @ q_emb / (
36 np.linalg.norm(self.embeddings, axis=1) * np.linalg.norm(q_emb) + 1e-9
37 )
38 initial_indices = np.argsort(-sims)[:initial_top_k].tolist()
39 candidates = [(i, self.chunks[i]) for i in initial_indices]
40
41 # Stage 2: Cross-Encoder Rerank(精排)
42 pairs = [[query, chunk] for _, chunk in candidates]
43 scores = reranker.predict(pairs) # 每對的相關分數
44
45 # 按 reranker 分數重新排序
46 ranked = sorted(
47 zip(candidates, scores),
48 key=lambda x: x[1],
49 reverse=True,
50 )[:final_top_k]
51
52 return [
53 {"chunk": chunk, "rerank_score": float(score)}
54 for (idx, chunk), score in ranked
55 ]
56
57
58# 使用範例
59chunks = [
60 "Python 的 async/await 語法讓非同步程式設計變得更直覺。",
61 "要減少 API 延遲,可以使用 asyncio 搭配 aiohttp 進行並發請求。",
62 "asyncio 是 Python 的標準非同步 I/O 框架,適合 I/O 密集型工作。",
63 "Python GIL 在多執行緒場景下限制了 CPU 密集型任務的並行。",
64 "使用 Redis 作為快取層可以顯著降低資料庫查詢次數,減少延遲。",
65 "CDN 加速靜態資源的分發,降低使用者的載入時間。",
66 "連線池(Connection Pool)避免反覆建立 TCP 連線的開銷。",
67]
68
69retriever = RerankRetriever(chunks)
70results = retriever.retrieve(
71 "如何用 Python 優化 API 的回應速度?",
72 initial_top_k=6,
73 final_top_k=3,
74)
75
76print("📊 Rerank 後的結果:")
77for i, r in enumerate(results, 1):
78 print(f" {i}. [分數: {r['rerank_score']:.4f}] {r['chunk']}")
優點:排序精準度顯著優於純向量搜尋;可以有效過濾掉「語意相近但不相關」的結果
缺點:推理較慢(每個候選都要跑一次 model);不適合作為第一層搜尋(要先用向量搜縮小候選集)
最佳使用場景:高精準度要求的問答系統;候選集大(20-100 個)但需要精確 Top-5 的場景
四種技術的組合策略
實際生產系統通常組合使用多種技術:
使用者問題
│
├── [Multi-Query] 生成 3 個查詢版本
│
├── 每個版本做 [Hybrid Search](語意 + BM25)
│ → 合併去重,取 Top-30 候選
│
├── [Reranker] 對 Top-30 精排
│ → 取 Top-5 最相關 chunks
│
└── 送給 LLM 生成答案
對特定問題類型,加入 HyDE:
使用者問題(措辭非正式/口語化)
│
├── [HyDE] 生成假設性文件
│
└── 用假設性文件向量做 [Hybrid Search] + [Reranker]
小結
| 技術 | 解決的問題 | 延遲增加 | 成本增加 |
|---|---|---|---|
| Hybrid Search | 術語/關鍵字精確匹配 | 低 | 低 |
| HyDE | 問句 vs 文件語言風格差異 | 中(一次 LLM 呼叫) | 中 |
| Multi-Query | 措辭不穩定、遺漏相關文件 | 中(N 次 Embedding) | 中 |
| Reranker | 召回後的排序精準度 | 中高(本地模型推理) | 低(用本地模型) |
下一篇,我們進入查詢轉換與Context 壓縮——當問題本身就很複雜,需要拆解或轉換,以及 context 太多時如何過濾噪音。
系列導覽