在前幾天的實作中,我們已經完成:
但實際測試後會發現:
語意相似,不一定代表內容完全符合使用者的需求
例如,使用者詢問:
Ubuntu 24.04 的安裝方式是什麼
向量搜尋可能找到「Ubuntu 系統安裝說明」,但不一定優先找到包含 Ubuntu 24.04 的精確內容
尤其當文件中出現以下資訊時,單純依賴語意搜尋可能會降低檢索準確度:
因此,今天將引入 Hybrid Search (混合搜尋),結合語意搜尋與關鍵字搜尋,建立更完整的文件檢索流程
目前的 Vector Search 主要根據文字的語意向量,尋找與查詢內容相似的文件。\
這種方式適合處理:
使用者:如何重新啟動系統
文件:安裝程式完成後,必須重新開機
即使使用者沒有使用「重新開機」這個完全相同的詞,系統仍可能透過語意相似度找到相關內容
但是,當使用者查詢特定字串時,關鍵字搜尋可能更有優勢:
使用者:Ubuntu 24.04
文件 A:支援 Ubuntu 系統
文件 B:支援 Ubuntu 22.04 與 Ubuntu 24.04
關鍵字搜尋可以直接確認文件是否包含 Ubuntu 24.04,而不只是判斷整體語意是否相似
因此,我們今天希望解決:
Vector Search 會將文件與使用者問題轉換成向量,再透過向量距離或相似度尋找相關內容
例如:
查詢:如何重新啟動電腦
文件:安裝完成後,需要重新開機
兩段文字雖然用詞不同,但語意相近,因此可能被檢索出來
Keyword Search 主要根據文件中是否出現查詢字詞,計算文件與查詢的相關程度
Vector Search 與 Keyword Search 各有不同的優勢。
Hybrid Search 的核心概念是:
使用 Vector Search 找出語意相關內容,再使用 Keyword Search 補足精確字詞匹配能力
兩者不是互相取代,而是互相補強
這個流程可以拆成五個主要階段:
本次沿用前幾天的資料
pip install chromadb sentence-transformers rank-bm25
本產品支援 Windows 10、Windows 11 以及 Ubuntu 22.04 與 Ubuntu 24.04
使用者可以透過官方網站下載安裝程式
安裝完成後,需要重新啟動系統
若使用 Ubuntu 24.04,請確認系統已安裝必要的相依套件
這份文件包含:
接下來使用不同類型的查詢進行測試
from rank_bm25 import BM25Okapi
class KeywordSearcher:
def __init__(self, documents):
self.documents = documents
tokenized_documents = [
document.split()
for document in documents
]
self.bm25 = BM25Okapi(tokenized_documents)
def search(self, query, top_k=3):
query_tokens = query.split()
scores = self.bm25.get_scores(query_tokens)
ranked_indices = sorted(
range(len(scores)),
key=lambda index: scores[index],
reverse=True
)
results = []
for index in ranked_indices[:top_k]:
results.append({
"document": self.documents[index],
"score": float(scores[index]),
"index": index
})
return results
單純使用 split() 可能無法產生有效的中文關鍵字
實際應用時,可以考慮:
jieba 分詞工具因此,本次程式主要用來理解 BM25 的流程,並非完整的中文搜尋引擎實作
延續 Day 5 的 ChromaDB 設定。
import chromadb
from sentence_transformers import SentenceTransformer
class VectorSearcher:
def __init__(
self,
collection_name="documents",
persist_path="data/chroma"
):
self.client = chromadb.PersistentClient(
path=persist_path
)
self.collection = self.client.get_collection(
name=collection_name
)
self.embedding_model = SentenceTransformer(
"all-MiniLM-L6-v2"
)
def search(self, query, top_k=3):
query_embedding = self.embedding_model.encode(
query
).tolist()
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=top_k,
include=[
"documents",
"metadatas",
"distances"
]
)
documents = results.get("documents", [[]])[0]
metadatas = results.get("metadatas", [[]])[0]
distances = results.get("distances", [[]])[0]
output = []
for document, metadata, distance in zip(
documents,
metadatas,
distances
):
output.append({
"document": document,
"metadata": metadata,
"score": float(distance)
})
return output
這裡需要特別注意:
distance 通常是距離概念,數值越小可能越接近因此,不能直接將兩者的分數相加
在整合之前,必須先處理:
最簡單的方式,是先將兩種搜尋結果合併,再去除重複文件。
class HybridSearcher:
def __init__(self, vector_searcher, keyword_searcher):
self.vector_searcher = vector_searcher
self.keyword_searcher = keyword_searcher
def search(self, query, top_k=5):
vector_results = self.vector_searcher.search(
query,
top_k=top_k
)
keyword_results = self.keyword_searcher.search(
query,
top_k=top_k
)
merged_results = []
seen_documents = set()
for result in vector_results + keyword_results:
document = result["document"]
if document in seen_documents:
continue
seen_documents.add(document)
merged_results.append(result)
return merged_results[:top_k]
上述程式是「合併結果」的簡化版本,並不是真正完整的加權排序
原因是:
因此,這個版本適合用來理解流程,但在正式環境中還需要進一步改善排序策略
除了直接合併結果,也可以使用 Reciprocal Rank Fusion,根據文件在不同搜尋結果中的排名進行整合
RRF 的優點是:
def reciprocal_rank_fusion(
result_lists,
top_k=5,
k=60
):
scores = {}
documents = {}
for results in result_lists:
for rank, result in enumerate(results, start=1):
document = result["document"]
if document not in scores:
scores[document] = 0.0
documents[document] = result
scores[document] += 1 / (k + rank)
ranked_documents = sorted(
scores.items(),
key=lambda item: item[1],
reverse=True
)
output = []
for document, score in ranked_documents[:top_k]:
result = documents[document].copy()
result["fusion_score"] = score
output.append(result)
return output
假設同一份文件:
這份文件同時被兩種方法找到,RRF 會累積它在不同結果列表中的排名分數,使其有機會排在前面
這種方式不代表文件一定正確,只是透過不同檢索方法的共同結果,增加排序穩定性
from src.retrieval.rrf import reciprocal_rank_fusion
class HybridSearchPipeline:
def __init__(
self,
vector_searcher,
keyword_searcher
):
self.vector_searcher = vector_searcher
self.keyword_searcher = keyword_searcher
def search(self, query, top_k=5):
vector_results = self.vector_searcher.search(
query,
top_k=top_k
)
keyword_results = self.keyword_searcher.search(
query,
top_k=top_k
)
fused_results = reciprocal_rank_fusion(
[
vector_results,
keyword_results
],
top_k=top_k
)
return fused_results
Hybrid Search 的目的不是直接產生答案,而是取得更可靠的參考文件
因此,後續仍然需要建立 Context
def build_context(results):
context_parts = []
for index, result in enumerate(results, start=1):
document = result["document"]
metadata = result.get("metadata", {})
source = metadata.get(
"source",
"unknown"
)
chunk_id = metadata.get(
"chunk_id",
"unknown"
)
context_parts.append(
f"[文件 {index}]\n"
f"來源:{source}\n"
f"Chunk ID:{chunk_id}\n"
f"內容:{document}"
)
return "\n\n".join(context_parts)
Context 傳給 LLM
def build_prompt(query, context):
return f"""
你是一個文件問答助理。
請根據提供的文件內容回答問題。
回答規則:
1. 只能根據提供的文件回答。
2. 如果文件沒有足夠資訊,請明確說明。
3. 不要自行捏造文件中沒有的內容。
4. 回答時列出參考來源。
文件內容:
{context}
使用者問題:
{query}
"""
今天從單一 Vector Search,進一步建立 Hybrid Search 的概念與實作流程
重要學習包括:
Vector Search 著重語意相似度,Keyword Search 著重精確文字匹配
兩種方式適合處理不同類型的問題
取得更多文件不代表品質更好
真正需要關注的是:
檢索結果是否包含與使用者問題直接相關、可供 LLM 使用的依據
不同搜尋方法的分數不一定能直接比較,因此可以使用:
即使檢索結果改善,仍然需要持續評估:
目前的系統已經從單純的向量檢索,逐步發展成結合多種搜尋策略的 RAG 架構
今天我們整合了 Vector Search 與 Keyword Search,但搜尋結果仍可能包含不相關文件。
因此,下一步可以進一步研究:
Day 10:Reranking 讓真正相關的文件排在前面
從今天的「結合不同搜尋方式」,進一步走向「重新判斷文件與問題的關聯程度」,讓 RAG 檢索結果更貼近實際應用需求