RAG 中的檢索器 (Retrievers) 在 RAG (Retrieval-Augmented Generation) 流程中，檢索器 (Retriever) 是一個核心組件，其主要職責是根據使用者的查詢 (Query)，從一個大規模的資料來源（通常是向量儲存）中「取回 (Retrieve)」最相關的文件區塊 (Chunks)。 與直接查詢向量儲存相比，檢索器提供了一個更抽象、更靈活的介面，並支援多種先進的檢索策略，以提升檢索結果的品質和多樣性。本文件將根據 10-Retriever 資料夾中的筆記本內容，介紹幾種關鍵的檢索器類型。 檢索器 vs 向量儲存對比 特性 向量儲存 檢索器 抽象層級 底層儲存 高層介面 功能範圍 基礎 CRUD 操作 複雜檢索策略 可組合性 有限 高度可組合 擴展性 依賴具體實現 統一介面，易擴展 適用場景 簡單搜尋需求 複雜檢索邏輯 1. 基礎：VectorStoreRetriever 這是最基礎的檢索器，它直接將一個向量儲存 (Vector Store) 封裝成檢索器介面。任何向量儲存都可以透過 .as_retriever() 方法輕鬆轉換。 1.1 核心搜尋類型 相似度搜尋 (Similarity Search) 預設的檢索方式，回傳與查詢向量最相似的 k 個文件。 from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings # 建立基礎檢索器 db = FAISS.from_documents(docs, OpenAIEmbeddings()) basic_retriever = db.as_retriever( search_type="similarity", search_kwargs={"k": 5} ) # 執行檢索 results = basic_retriever.invoke("什麼是機器學習？") for doc in results: print(f"Source: {doc.metadata.get('source', 'unknown')}") print(f"Content: {doc.page_content[:100]}...") 最大邊際相關性 (Maximal Marginal Relevance, MMR) 在回傳結果時，不僅考慮文件與查詢的相關性，還會考慮文件之間的多樣性，避免回傳過於相似、冗餘的內容。 # MMR 參數詳解 mmr_retriever = db.as_retriever( search_type="mmr", search_kwargs={ "k": 5, # 最終回傳的文件數量 "fetch_k": 20, # 初始候選文件數量（越大越多樣，但計算成本越高） "lambda_mult": 0.7 # 平衡因子：1.0=純相關性，0.0=純多樣性 } ) # MMR 演算法流程示例 def explain_mmr_process(): """ MMR 演算法步驟： 1. 取得 fetch_k=20 個最相似的候選文件 2. 選擇與查詢最相似的文件作為第一個結果 3. 對於剩餘的 k-1 個位置： - 計算每個候選文件的 MMR 分數 - MMR = λ × 相關性分數 - (1-λ) × max(與已選文件的相似度) - 選擇 MMR 分數最高的文件 """ pass relevant_diverse_docs = mmr_retriever.invoke("深度學習的應用領域") 相似度分數過濾 (Similarity Score Threshold) 只回傳相似度分數高於某個閾值的結果，用於過濾掉低相關性的雜訊。 # 基於分數閾值的檢索 threshold_retriever = db.as_retriever( search_type="similarity_score_threshold", search_kwargs={ "score_threshold": 0.8, # 只回傳相似度 > 0.8 的文件 "k": 10 # 最大檢索數量 } ) # 只有高相關性的文件會被回傳 high_quality_results = threshold_retriever.invoke("PyTorch 訓練技巧") # 檢查結果品質 for doc, score in threshold_retriever.get_relevant_documents_with_score("PyTorch 訓練技巧"): print(f"Score: {score:.3f} - {doc.page_content[:50]}...") 1.2 進階配置選項 # 自訂檢索器配置 class CustomVectorStoreRetriever: def __init__(self, vectorstore, config): self.vectorstore = vectorstore self.config = config def retrieve(self, query): # 根據查詢長度調整檢索參數 if len(query.split()) <= 3: # 短查詢 k = self.config["short_query_k"] search_type = "similarity" else: # 長查詢 k = self.config["long_query_k"] search_type = "mmr" retriever = self.vectorstore.as_retriever( search_type=search_type, search_kwargs={"k": k} ) return retriever.invoke(query) # 使用自訂配置 config = { "short_query_k": 3, "long_query_k": 7 } custom_retriever = CustomVectorStoreRetriever(db, config) 2. 提升檢索品質的進階檢索器 2.1 MultiQueryRetriever 單一的使用者查詢可能因為用詞或觀點的限制，無法涵蓋所有相關的文件。MultiQueryRetriever 透過 LLM 從一個原始查詢自動生成多個不同角度的相似查詢，然後對每個查詢都進行檢索，最後將所有結果合併並去重。 運作流程詳解 from langchain.retrievers.multi_query import MultiQueryRetriever from langchain_openai import ChatOpenAI # 建立多查詢檢索器 llm = ChatOpenAI(model="gpt-4", temperature=0.3) multiquery_retriever = MultiQueryRetriever.from_llm( retriever=db.as_retriever(), llm=llm, parser_key="lines" # 解析生成查詢的方式 ) # 自訂查詢生成提示 from langchain.retrievers.multi_query import LineListOutputParser from langchain.prompts import PromptTemplate QUERY_PROMPT = PromptTemplate( input_variables=["question"], template="""你是一個 AI 語言模型助手。你的任務是生成5個不同版本的使用者問題， 以便從向量資料庫中檢索相關文件。透過對使用者問題產生多種視角， 你的目標是幫助使用者克服基於距離的相似度搜尋的一些限制。 原始問題: {question} 請提供這些替代問題，每行一個：""" ) # 使用自訂提示的多查詢檢索器 multiquery_retriever = MultiQueryRetriever( retriever=db.as_retriever(), llm_chain=llm, prompt=QUERY_PROMPT, parser=LineListOutputParser() ) # 示例：生成的查詢變體 original_query = "LangChain 的優點是什麼？" # 可能生成的查詢： # 1. "LangChain 有哪些主要功能和特色？" # 2. "使用 LangChain 開發有什麼好處？" # 3. "LangChain 框架的核心優勢是什麼？" # 4. "為什麼選擇 LangChain 而不是其他框架？" # 5. "LangChain 在 AI 應用開發中的價值？" results = multiquery_retriever.invoke(original_query) print(f"找到 {len(results)} 個相關文件") 自訂查詢生成策略 class DomainSpecificMultiQueryRetriever(MultiQueryRetriever): """針對特定領域的多查詢檢索器""" def __init__(self, retriever, llm, domain="general"): super().__init__(retriever, llm) self.domain = domain self.domain_prompts = { "medical": "生成醫學相關的替代查詢...", "legal": "生成法律相關的替代查詢...", "technical": "生成技術相關的替代查詢..." } def generate_queries(self, question): domain_prompt = self.domain_prompts.get(self.domain, "") enhanced_question = f"{domain_prompt}\n原始問題: {question}" return super().generate_queries(enhanced_question) # 技術領域專用檢索器 tech_multiquery = DomainSpecificMultiQueryRetriever( retriever=db.as_retriever(), llm=llm, domain="technical" ) 2.2 ContextualCompressionRetriever 有時候，即使是相關的文件區塊，也可能只有一小部分內容與查詢直接相關。ContextualCompressionRetriever 旨在解決這個問題，它會在檢索後對文件進行壓縮，只保留與查詢最相關的部分。 文件壓縮器類型 LLMChainExtractor 使用 LLM 判斷每個文件中的哪些句子與查詢相關，並只保留這些句子。 from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor from langchain_openai import ChatOpenAI # 建立 LLM 提取器 llm = ChatOpenAI(model="gpt-4", temperature=0) compressor = LLMChainExtractor.from_llm(llm) # 建立壓縮檢索器 compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=db.as_retriever() ) # 執行壓縮檢索 compressed_docs = compression_retriever.invoke("深度學習在醫療診斷中的應用") # 比較原始和壓縮後的內容長度 for doc in compressed_docs: print(f"壓縮後長度: {len(doc.page_content)} 字符") print(f"內容: {doc.page_content}") print("-" * 50) EmbeddingsFilter 計算每個文件與查詢的嵌入相似度，只保留高於某個閾值的文件。 from langchain.retrievers.document_compressors import EmbeddingsFilter # 建立嵌入過濾器 embeddings = OpenAIEmbeddings() embeddings_filter = EmbeddingsFilter( embeddings=embeddings, similarity_threshold=0.76, # 相似度閾值 k=10 # 最多保留的文件數 ) # 組合多個壓縮器 from langchain.retrievers.document_compressors import DocumentCompressorPipeline # 建立壓縮管道：先用嵌入過濾，再用 LLM 提取 pipeline_compressor = DocumentCompressorPipeline( transformers=[embeddings_filter, compressor] ) pipeline_retriever = ContextualCompressionRetriever( base_compressor=pipeline_compressor, base_retriever=db.as_retriever() ) # 雙重壓縮的結果 highly_relevant_docs = pipeline_retriever.invoke("機器學習模型評估指標") 自訂壓縮器 from langchain.retrievers.document_compressors.base import BaseDocumentCompressor from typing import List, Sequence import re class KeywordHighlightCompressor(BaseDocumentCompressor): """突出顯示關鍵字並保留相關段落的壓縮器""" def __init__(self, keywords_per_query=5): self.keywords_per_query = keywords_per_query def compress_documents( self, documents: Sequence[Document], query: str ) -> List[Document]: # 提取查詢中的關鍵字 keywords = self._extract_keywords(query) compressed_docs = [] for doc in documents: # 找到包含關鍵字的句子 relevant_sentences = self._find_relevant_sentences( doc.page_content, keywords ) if relevant_sentences: compressed_content = "\n".join(relevant_sentences) compressed_doc = Document( page_content=compressed_content, metadata={**doc.metadata, "compression_ratio": len(compressed_content) / len(doc.page_content)} ) compressed_docs.append(compressed_doc) return compressed_docs def _extract_keywords(self, query: str) -> List[str]: # 簡單的關鍵字提取（實際應用中可使用更複雜的 NLP 技術） words = re.findall(r'\b\w+\b', query.lower()) return words[:self.keywords_per_query] def _find_relevant_sentences(self, text: str, keywords: List[str]) -> List[str]: sentences = re.split(r'[.!?]+', text) relevant = [] for sentence in sentences: if any(keyword in sentence.lower() for keyword in keywords): relevant.append(sentence.strip()) return relevant # 使用自訂壓縮器 custom_compressor = KeywordHighlightCompressor(keywords_per_query=3) custom_compression_retriever = ContextualCompressionRetriever( base_compressor=custom_compressor, base_retriever=db.as_retriever() ) 2.3 LongContextReorder 研究顯示，LLM 在處理長上下文時，對於開頭和結尾的資訊注意力最高，而中間的資訊容易被「遺忘」。LongContextReorder 是一個後處理步驟，它會將檢索到的文件列表進行重排序，將最相關的文件放在列表的開頭和結尾，而將次要的文件放在中間，以最大化 LLM 的注意力。 from langchain.document_transformers import LongContextReorder # 建立重排序檢索器 def create_reordered_retriever(base_retriever): reordering = LongContextReorder() def reordered_retrieve(query): # 先檢索文件 docs = base_retriever.invoke(query) # 重新排序以最佳化長上下文處理 reordered_docs = reordering.transform_documents(docs) return reordered_docs return reordered_retrieve # 使用重排序檢索器 base_retriever = db.as_retriever(search_kwargs={"k": 10}) reordered_retriever = create_reordered_retriever(base_retriever) # 注意力最佳化的結果 optimized_docs = reordered_retriever("複雜的多步驟機器學習流程") # 檢查重排序效果 print("重排序後的文件順序：") for i, doc in enumerate(optimized_docs): relevance = "高" if i < 2 or i >= len(optimized_docs) - 2 else "中" print(f"位置 {i+1}: 重要性={relevance} - {doc.page_content[:50]}...") 自訂重排序策略 class SmartReorder: """智能重排序策略""" def __init__(self, strategy="lost_in_middle"): self.strategy = strategy def reorder_documents(self, docs, query=None): if self.strategy == "lost_in_middle": return self._lost_in_middle_reorder(docs) elif self.strategy == "similarity_decay": return self._similarity_decay_reorder(docs) elif self.strategy == "interleaved": return self._interleaved_reorder(docs) else: return docs def _lost_in_middle_reorder(self, docs): """將最相關的文件放在開頭和結尾""" if len(docs) <= 2: return docs reordered = [] high_relevance = docs[:len(docs)//2] low_relevance = docs[len(docs)//2:] # 交替放置高相關性文件在開頭和結尾 for i, doc in enumerate(high_relevance): if i % 2 == 0: reordered.insert(0, doc) # 放在開頭 else: reordered.append(doc) # 放在結尾 # 低相關性文件放在中間 mid_point = len(reordered) // 2 for doc in low_relevance: reordered.insert(mid_point, doc) mid_point += 1 return reordered def _similarity_decay_reorder(self, docs): """基於相似度衰減的重排序""" # 實現基於相似度分數的動態重排序 return docs # 簡化實現 def _interleaved_reorder(self, docs): """交錯排列不同類型的文件""" # 根據文件類型或來源進行交錯排列 return docs # 簡化實現 # 使用智能重排序 smart_reorder = SmartReorder(strategy="lost_in_middle") def smart_reordered_retriever(query): docs = db.similarity_search(query, k=10) return smart_reorder.reorder_documents(docs, query) 3. 處理複雜文件結構的檢索器 3.1 ParentDocumentRetriever 在處理文件時，我們常常面臨一個兩難：小的文件區塊有利於精準的語意搜尋，但大的區塊能提供更完整的上下文。ParentDocumentRetriever 巧妙地解決了這個問題。 詳細運作流程 from langchain.retrievers import ParentDocumentRetriever from langchain.storage import InMemoryStore from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_community.vectorstores import Chroma # 建立不同粒度的分割器 parent_splitter = RecursiveCharacterTextSplitter( chunk_size=2000, # 父區塊：2000字符 chunk_overlap=200, # 重疊部分 separators=["\n\n", "\n", " ", ""] ) child_splitter = RecursiveCharacterTextSplitter( chunk_size=400, # 子區塊：400字符 chunk_overlap=50, # 重疊部分 separators=["\n\n", "\n", " ", ""] ) # 建立儲存 vectorstore = Chroma( collection_name="child_chunks", embedding_function=OpenAIEmbeddings() ) store = InMemoryStore() # 儲存父區塊 # 建立父文件檢索器 parent_retriever = ParentDocumentRetriever( vectorstore=vectorstore, docstore=store, child_splitter=child_splitter, parent_splitter=parent_splitter, k=3 # 檢索的子區塊數量 ) # 新增文件 parent_retriever.add_documents(docs, ids=None) # 檢索流程示例 def demonstrate_parent_retrieval(query): """展示父文件檢索的詳細流程""" # 1. 在子區塊中搜尋 child_docs = vectorstore.similarity_search(query, k=3) print("找到的相關子區塊：") for i, doc in enumerate(child_docs): print(f"子區塊 {i+1}: {doc.page_content[:100]}...") # 2. 透過父文件檢索器取得完整上下文 parent_docs = parent_retriever.invoke(query) print("\n對應的父區塊：") for i, doc in enumerate(parent_docs): print(f"父區塊 {i+1} (長度: {len(doc.page_content)}): {doc.page_content[:200]}...") return parent_docs # 執行演示 results = demonstrate_parent_retrieval("深度學習的反向傳播演算法") 進階配置選項 class AdaptiveParentDocumentRetriever(ParentDocumentRetriever): """自適應的父文件檢索器""" def __init__(self, vectorstore, docstore, child_splitter, parent_splitter, **kwargs): super().__init__(vectorstore, docstore, child_splitter, parent_splitter, **kwargs) self.query_complexity_threshold = 10 # 查詢複雜度閾值 def invoke(self, query: str): # 根據查詢複雜度調整檢索策略 query_words = len(query.split()) if query_words > self.query_complexity_threshold: # 複雜查詢：檢索更多子區塊，回傳更多父區塊 self.k = 5 self.fetch_k = 20 else: # 簡單查詢：較少的檢索數量 self.k = 3 self.fetch_k = 10 return super().invoke(query) def get_relevant_documents_with_metadata(self, query: str): """回傳文件及其檢索中繼資料""" docs = self.invoke(query) enhanced_docs = [] for doc in docs: # 計算父區塊包含的子區塊數量 child_count = len([d for d in self.vectorstore.similarity_search(query, k=20) if d.metadata.get('parent_id') == doc.metadata.get('doc_id')]) doc.metadata['child_chunk_count'] = child_count doc.metadata['retrieval_method'] = 'parent_document' enhanced_docs.append(doc) return enhanced_docs # 使用自適應檢索器 adaptive_retriever = AdaptiveParentDocumentRetriever( vectorstore=vectorstore, docstore=store, child_splitter=child_splitter, parent_splitter=parent_splitter ) 3.2 MultiVectorRetriever 這個檢索器更進一步，它允許你為一份文件儲存多個不同的向量，這些向量可以代表文件的不同方面。 策略一：儲存摘要 (Summary) from langchain.retrievers.multi_vector import MultiVectorRetriever from langchain.storage import InMemoryStore from langchain_openai import ChatOpenAI import uuid # 建立多向量檢索器 vectorstore = Chroma(collection_name="summaries", embedding_function=OpenAIEmbeddings()) store = InMemoryStore() id_key = "doc_id" # 建立多向量檢索器 mv_retriever = MultiVectorRetriever( vectorstore=vectorstore, docstore=store, id_key=id_key ) # 生成文件摘要 llm = ChatOpenAI(model="gpt-4", temperature=0) def generate_summaries(docs): """為文件生成摘要""" summaries = [] doc_ids = [str(uuid.uuid4()) for _ in docs] summary_prompt = """ 請為以下文件提供一個簡潔的摘要，重點關注主要概念和關鍵資訊： 文件內容： {content} 摘要： """ for doc, doc_id in zip(docs, doc_ids): # 生成摘要 summary = llm.invoke(summary_prompt.format(content=doc.page_content)) # 建立摘要文件 summary_doc = Document( page_content=summary.content, metadata={**doc.metadata, id_key: doc_id} ) summaries.append(summary_doc) # 將原始文件存入 docstore store.mset([(doc_id, doc)]) return summaries, doc_ids # 生成並新增摘要 summaries, doc_ids = generate_summaries(docs) mv_retriever.vectorstore.add_documents(summaries) # 檢索：透過摘要找到完整文件 summary_results = mv_retriever.invoke("機器學習的監督學習方法") print(f"透過摘要檢索到 {len(summary_results)} 個完整文件") 策略二：假設性問題 (Hypothetical Questions) def generate_hypothetical_questions(docs, questions_per_doc=3): """為每個文件生成假設性問題""" question_docs = [] doc_ids = [str(uuid.uuid4()) for _ in docs] question_prompt = """ 基於以下文件內容，生成 {num_questions} 個這個文件可以回答的問題。 問題應該具體、明確，並涵蓋文件的主要內容。 文件內容： {content} 請用以下格式回應： 問題1: [問題內容] 問題2: [問題內容] 問題3: [問題內容] """ for doc, doc_id in zip(docs, doc_ids): # 生成問題 questions_response = llm.invoke( question_prompt.format( content=doc.page_content, num_questions=questions_per_doc ) ) # 解析生成的問題 questions = [ line.split(": ", 1)[1] for line in questions_response.content.split("\n") if line.strip() and ": " in line ] # 為每個問題建立文件 for question in questions: question_doc = Document( page_content=question, metadata={ **doc.metadata, id_key: doc_id, "type": "hypothetical_question" } ) question_docs.append(question_doc) # 將原始文件存入 docstore store.mset([(doc_id, doc)]) return question_docs # 建立基於問題的檢索器 question_vectorstore = Chroma( collection_name="questions", embedding_function=OpenAIEmbeddings() ) question_store = InMemoryStore() question_retriever = MultiVectorRetriever( vectorstore=question_vectorstore, docstore=question_store, id_key=id_key ) # 生成並新增假設性問題 hypothetical_questions = generate_hypothetical_questions(docs) question_retriever.vectorstore.add_documents(hypothetical_questions) # 檢索：透過問題匹配找到相關文件 question_results = question_retriever.invoke("如何調整學習率？") print(f"透過問題匹配找到 {len(question_results)} 個相關文件") 策略三：多視角向量化 def generate_multi_perspective_vectors(docs): """從多個視角對文件進行向量化""" perspectives = { "technical": "從技術實現的角度重新描述這個文件的內容", "conceptual": "從概念理解的角度總結這個文件的核心思想", "practical": "從實際應用的角度說明這個文件的實用價值", "educational": "從教學的角度解釋這個文件適合什麼程度的學習者" } multi_docs = [] doc_ids = [str(uuid.uuid4()) for _ in docs] for doc, doc_id in zip(docs, doc_ids): store.mset([(doc_id, doc)]) # 儲存原始文件 for perspective, prompt in perspectives.items(): # 從不同視角重新描述文件 perspective_content = llm.invoke( f"{prompt}:\n\n{doc.page_content}" ) perspective_doc = Document( page_content=perspective_content.content, metadata={ **doc.metadata, id_key: doc_id, "perspective": perspective } ) multi_docs.append(perspective_doc) return multi_docs # 建立多視角檢索器 multi_perspective_retriever = MultiVectorRetriever( vectorstore=Chroma(collection_name="multi_perspective", embedding_function=OpenAIEmbeddings()), docstore=store, id_key=id_key ) # 新增多視角文件 multi_perspective_docs = generate_multi_perspective_vectors(docs) multi_perspective_retriever.vectorstore.add_documents(multi_perspective_docs) # 檢索時會考慮所有視角 perspective_results = multi_perspective_retriever.invoke("初學者如何學習機器學習？") 組合策略的綜合檢索器 class ComprehensiveMultiVectorRetriever: """組合多種策略的綜合多向量檢索器""" def __init__(self, base_docs, llm): self.base_docs = base_docs self.llm = llm self.retrievers = {} self._setup_retrievers() def _setup_retrievers(self): # 建立不同策略的檢索器 strategies = ["summary", "questions", "perspectives"] for strategy in strategies: vectorstore = Chroma( collection_name=f"{strategy}_store", embedding_function=OpenAIEmbeddings() ) docstore = InMemoryStore() retriever = MultiVectorRetriever( vectorstore=vectorstore, docstore=docstore, id_key="doc_id" ) self.retrievers[strategy] = retriever # 根據策略生成對應的文件 if strategy == "summary": generated_docs = self._generate_summaries() elif strategy == "questions": generated_docs = self._generate_questions() elif strategy == "perspectives": generated_docs = self._generate_perspectives() retriever.vectorstore.add_documents(generated_docs) def retrieve_with_strategy(self, query, strategy="summary", k=3): """使用特定策略檢索""" return self.retrievers[strategy].invoke(query) def retrieve_ensemble(self, query, weights=None, k=3): """組合多個策略的結果""" if weights is None: weights = {"summary": 0.4, "questions": 0.4, "perspectives": 0.2} all_results = [] for strategy, weight in weights.items(): strategy_results = self.retrievers[strategy].invoke(query) # 為每個結果添加權重 for doc in strategy_results: doc.metadata["strategy"] = strategy doc.metadata["weight"] = weight all_results.append(doc) # 去重並排序 unique_results = self._deduplicate_and_rank(all_results) return unique_results[:k] def _deduplicate_and_rank(self, docs): """去重並根據綜合分數排序""" seen_ids = set() unique_docs = [] for doc in docs: doc_id = doc.metadata.get("doc_id") if doc_id not in seen_ids: seen_ids.add(doc_id) unique_docs.append(doc) return unique_docs # 使用綜合檢索器 comprehensive_retriever = ComprehensiveMultiVectorRetriever(docs, llm) # 使用不同策略檢索 summary_results = comprehensive_retriever.retrieve_with_strategy("深度學習", "summary") question_results = comprehensive_retriever.retrieve_with_strategy("深度學習", "questions") # 組合策略檢索 ensemble_results = comprehensive_retriever.retrieve_ensemble( "深度學習的實際應用", weights={"summary": 0.3, "questions": 0.5, "perspectives": 0.2} ) 4. 結合不同檢索策略：EnsembleRetriever 不同的檢索方法各有優劣。例如，傳統的關鍵字搜尋（如 BM25）擅長匹配精確的術語，而向量搜尋擅長捕捉語意相似度。EnsembleRetriever 可以將多個不同的檢索器（如一個 BM25Retriever 和一個 VectorStoreRetriever）組合起來。 4.1 基本 Ensemble 檢索 from langchain.retrievers import BM25Retriever, EnsembleRetriever from langchain_community.vectorstores import FAISS # 準備不同類型的檢索器 # 1. BM25 檢索器（關鍵字匹配） bm25_retriever = BM25Retriever.from_documents(docs) bm25_retriever.k = 5 # 2. 向量檢索器（語意相似度） faiss_vectorstore = FAISS.from_documents(docs, OpenAIEmbeddings()) faiss_retriever = faiss_vectorstore.as_retriever(search_kwargs={"k": 5}) # 3. 建立 Ensemble 檢索器 ensemble_retriever = EnsembleRetriever( retrievers=[bm25_retriever, faiss_retriever], weights=[0.5, 0.5] # 等權重組合 ) # 執行組合檢索 hybrid_results = ensemble_retriever.invoke("機器學習模型評估指標") # 比較不同檢索器的結果 print("BM25 檢索結果：") bm25_results = bm25_retriever.invoke("機器學習模型評估指標") for doc in bm25_results[:3]: print(f"- {doc.page_content[:100]}...") print("\n向量檢索結果：") vector_results = faiss_retriever.invoke("機器學習模型評估指標") for doc in vector_results[:3]: print(f"- {doc.page_content[:100]}...") print("\nEnsemble 檢索結果：") for doc in hybrid_results[:3]: print(f"- {doc.page_content[:100]}...") 4.2 Reciprocal Rank Fusion (RRF) 演算法詳解 def explain_rrf_algorithm(): """ Reciprocal Rank Fusion 演算法： 1. 對每個檢索器的結果進行排序 2. 為每個文件計算 RRF 分數： RRF(d) = Σ(1 / (k + rank_i(d))) 其中 k 是常數（通常為60），rank_i(d) 是文件 d 在檢索器 i 中的排名 3. 根據 RRF 分數重新排序所有文件 4. 回傳 top-k 結果 """ pass class CustomEnsembleRetriever: """自訂的 Ensemble 檢索器，實現更靈活的融合策略""" def __init__(self, retrievers, weights=None, fusion_method="rrf"): self.retrievers = retrievers self.weights = weights or [1.0] * len(retrievers) self.fusion_method = fusion_method def invoke(self, query, k=5): # 收集所有檢索器的結果 all_results = [] for i, retriever in enumerate(self.retrievers): results = retriever.invoke(query) for rank, doc in enumerate(results): all_results.append({ "doc": doc, "retriever_id": i, "rank": rank, "weight": self.weights[i] }) # 根據融合方法計算最終分數 if self.fusion_method == "rrf": return self._rrf_fusion(all_results, k) elif self.fusion_method == "weighted_sum": return self._weighted_sum_fusion(all_results, k) elif self.fusion_method == "borda_count": return self._borda_count_fusion(all_results, k) def _rrf_fusion(self, results, k, rrf_k=60): """Reciprocal Rank Fusion""" doc_scores = {} for result in results: doc_id = id(result["doc"]) # 使用物件 ID 作為唯一標識 if doc_id not in doc_scores: doc_scores[doc_id] = {"doc": result["doc"], "score": 0} # RRF 分數計算 rrf_score = result["weight"] / (rrf_k + result["rank"] + 1) doc_scores[doc_id]["score"] += rrf_score # 排序並回傳 top-k sorted_docs = sorted( doc_scores.values(), key=lambda x: x["score"], reverse=True ) return [item["doc"] for item in sorted_docs[:k]] def _weighted_sum_fusion(self, results, k): """加權求和融合""" doc_scores = {} max_rank = max(result["rank"] for result in results) for result in results: doc_id = id(result["doc"]) if doc_id not in doc_scores: doc_scores[doc_id] = {"doc": result["doc"], "score": 0} # 將排名轉換為分數（排名越高分數越高） rank_score = (max_rank - result["rank"]) / max_rank weighted_score = result["weight"] * rank_score doc_scores[doc_id]["score"] += weighted_score sorted_docs = sorted( doc_scores.values(), key=lambda x: x["score"], reverse=True ) return [item["doc"] for item in sorted_docs[:k]] def _borda_count_fusion(self, results, k): """Borda Count 融合""" doc_scores = {} # 計算每個檢索器的最大排名 retriever_max_ranks = {} for result in results: retriever_id = result["retriever_id"] if retriever_id not in retriever_max_ranks: retriever_max_ranks[retriever_id] = 0 retriever_max_ranks[retriever_id] = max( retriever_max_ranks[retriever_id], result["rank"] ) for result in results: doc_id = id(result["doc"]) if doc_id not in doc_scores: doc_scores[doc_id] = {"doc": result["doc"], "score": 0} # Borda Count 分數 max_rank = retriever_max_ranks[result["retriever_id"]] borda_score = result["weight"] * (max_rank - result["rank"]) doc_scores[doc_id]["score"] += borda_score sorted_docs = sorted( doc_scores.values(), key=lambda x: x["score"], reverse=True ) return [item["doc"] for item in sorted_docs[:k]] # 使用自訂 Ensemble 檢索器 custom_ensemble = CustomEnsembleRetriever( retrievers=[bm25_retriever, faiss_retriever], weights=[0.6, 0.4], # 更重視 BM25 結果 fusion_method="rrf" ) rrf_results = custom_ensemble.invoke("深度學習優化技術") 4.3 多檢索器的進階組合 # 建立更多樣化的檢索器組合 def create_diverse_ensemble(): # 1. 不同參數的向量檢索器 similarity_retriever = faiss_vectorstore.as_retriever( search_type="similarity", search_kwargs={"k": 5} ) mmr_retriever = faiss_vectorstore.as_retriever( search_type="mmr", search_kwargs={"k": 5, "fetch_k": 20, "lambda_mult": 0.7} ) # 2. 壓縮檢索器 compression_retriever = ContextualCompressionRetriever( base_compressor=EmbeddingsFilter( embeddings=OpenAIEmbeddings(), similarity_threshold=0.76 ), base_retriever=similarity_retriever ) # 3. 多查詢檢索器 multi_query_retriever = MultiQueryRetriever.from_llm( retriever=similarity_retriever, llm=ChatOpenAI(model="gpt-4", temperature=0.3) ) # 4. 建立多層 Ensemble diverse_ensemble = EnsembleRetriever( retrievers=[ bm25_retriever, # 關鍵字匹配 mmr_retriever, # 多樣性語意搜尋 compression_retriever, # 壓縮檢索 multi_query_retriever # 多查詢檢索 ], weights=[0.3, 0.3, 0.2, 0.2] ) return diverse_ensemble # 建立多樣化的檢索器 diverse_retriever = create_diverse_ensemble() # 測試不同複雜度的查詢 test_queries = [ "什麼是機器學習？", # 簡單查詢 "深度學習中的梯度消失問題及其解決方案", # 複雜查詢 "如何選擇合適的優化器和學習率？" # 實用查詢 ] for query in test_queries: print(f"\n查詢: {query}") results = diverse_retriever.invoke(query) print(f"找到 {len(results)} 個相關文件") for i, doc in enumerate(results[:2]): print(f" {i+1}. {doc.page_content[:80]}...") 5. 自訂檢索器開發 5.1 基礎自訂檢索器框架 from langchain.schema import BaseRetriever, Document from typing import List from abc import ABC, abstractmethod class CustomRetriever(BaseRetriever, ABC): """自訂檢索器的基礎類別""" def __init__(self, **kwargs): super().__init__(**kwargs) @abstractmethod def _get_relevant_documents(self, query: str) -> List[Document]: """實現具體的檢索邏輯""" pass def invoke(self, query: str) -> List[Document]: """統一的檢索介面""" return self._get_relevant_documents(query) class KeywordBoostRetriever(CustomRetriever): """基於關鍵字加權的檢索器""" def __init__(self, base_retriever, keyword_boost=2.0, **kwargs): super().__init__(**kwargs) self.base_retriever = base_retriever self.keyword_boost = keyword_boost self.important_keywords = [ "機器學習", "深度學習", "神經網路", "人工智慧", "演算法", "模型", "訓練", "優化" ] def _get_relevant_documents(self, query: str) -> List[Document]: # 基礎檢索 base_results = self.base_retriever.invoke(query) # 關鍵字加權 boosted_results = [] for doc in base_results: boost_score = 1.0 # 檢查重要關鍵字 for keyword in self.important_keywords: if keyword in doc.page_content.lower(): boost_score *= self.keyword_boost doc.metadata["boost_score"] = boost_score boosted_results.append(doc) # 根據加權分數重新排序 boosted_results.sort( key=lambda x: x.metadata.get("boost_score", 1.0), reverse=True ) return boosted_results # 使用關鍵字加權檢索器 keyword_retriever = KeywordBoostRetriever( base_retriever=faiss_retriever, keyword_boost=1.5 ) boosted_results = keyword_retriever.invoke("機器學習演算法比較") 5.2 時間感知檢索器 from datetime import datetime, timedelta import dateutil.parser class TimeAwareRetriever(CustomRetriever): """時間感知的檢索器，優先顯示最新內容""" def __init__(self, base_retriever, time_decay_factor=0.1, **kwargs): super().__init__(**kwargs) self.base_retriever = base_retriever self.time_decay_factor = time_decay_factor def _get_relevant_documents(self, query: str) -> List[Document]: base_results = self.base_retriever.invoke(query) current_time = datetime.now() time_weighted_results = [] for doc in base_results: # 嘗試從中繼資料獲取時間資訊 doc_time = self._extract_document_time(doc) if doc_time: # 計算時間衰減分數 time_diff = (current_time - doc_time).days time_score = max(0.1, 1.0 - (time_diff * self.time_decay_factor / 365)) else: time_score = 0.5 # 沒有時間資訊的預設分數 doc.metadata["time_score"] = time_score doc.metadata["document_age_days"] = time_diff if doc_time else None time_weighted_results.append(doc) # 根據時間分數排序 time_weighted_results.sort( key=lambda x: x.metadata.get("time_score", 0), reverse=True ) return time_weighted_results def _extract_document_time(self, doc: Document) -> datetime: """從文件中提取時間資訊""" time_fields = ["created_at", "published_date", "last_modified", "date"] for field in time_fields: if field in doc.metadata: try: return dateutil.parser.parse(doc.metadata[field]) except: continue return None # 使用時間感知檢索器 time_retriever = TimeAwareRetriever( base_retriever=faiss_retriever, time_decay_factor=0.05 # 較慢的時間衰減 ) recent_results = time_retriever.invoke("最新的深度學習技術") 5.3 使用者偏好檢索器 class UserPreferenceRetriever(CustomRetriever): """基於使用者偏好的個人化檢索器""" def __init__(self, base_retriever, user_profile=None, **kwargs): super().__init__(**kwargs) self.base_retriever = base_retriever self.user_profile = user_profile or { "preferred_topics": [], "difficulty_level": "intermediate", "language": "zh-TW", "content_type_preference": {"tutorial": 1.2, "research": 0.8, "news": 1.0} } def _get_relevant_documents(self, query: str) -> List[Document]: base_results = self.base_retriever.invoke(query) personalized_results = [] for doc in base_results: preference_score = self._calculate_preference_score(doc) doc.metadata["preference_score"] = preference_score personalized_results.append(doc) # 根據偏好分數排序 personalized_results.sort( key=lambda x: x.metadata.get("preference_score", 0), reverse=True ) return personalized_results def _calculate_preference_score(self, doc: Document) -> float: score = 1.0 # 主題偏好 doc_topics = doc.metadata.get("topics", []) for topic in self.user_profile["preferred_topics"]: if topic in doc_topics: score *= 1.3 # 難度等級匹配 doc_difficulty = doc.metadata.get("difficulty", "intermediate") if doc_difficulty == self.user_profile["difficulty_level"]: score *= 1.2 # 內容類型偏好 content_type = doc.metadata.get("content_type", "unknown") if content_type in self.user_profile["content_type_preference"]: score *= self.user_profile["content_type_preference"][content_type] return score def update_user_profile(self, **updates): """更新使用者偏好""" self.user_profile.update(updates) # 建立個人化檢索器 user_profile = { "preferred_topics": ["深度學習", "電腦視覺"], "difficulty_level": "advanced", "language": "zh-TW", "content_type_preference": {"research": 1.5, "tutorial": 1.0, "news": 0.7} } personalized_retriever = UserPreferenceRetriever( base_retriever=faiss_retriever, user_profile=user_profile ) personalized_results = personalized_retriever.invoke("卷積神經網路") 6. 檢索器效能評估和最佳化 6.1 檢索品質評估 import numpy as np from sklearn.metrics.pairwise import cosine_similarity class RetrievalEvaluator: """檢索器效能評估工具""" def __init__(self, embedding_model): self.embedding_model = embedding_model def evaluate_relevance(self, query, retrieved_docs, ground_truth_docs=None): """評估檢索相關性""" query_embedding = self.embedding_model.embed_query(query) # 計算與查詢的相似度 doc_similarities = [] for doc in retrieved_docs: doc_embedding = self.embedding_model.embed_query(doc.page_content) similarity = cosine_similarity( [query_embedding], [doc_embedding] )[0][0] doc_similarities.append(similarity) metrics = { "mean_similarity": np.mean(doc_similarities), "std_similarity": np.std(doc_similarities), "min_similarity": np.min(doc_similarities), "max_similarity": np.max(doc_similarities) } return metrics def evaluate_diversity(self, retrieved_docs): """評估檢索結果的多樣性""" if len(retrieved_docs) < 2: return {"diversity_score": 1.0} # 計算文件間的平均相似度 embeddings = [] for doc in retrieved_docs: embedding = self.embedding_model.embed_query(doc.page_content) embeddings.append(embedding) similarities = [] for i in range(len(embeddings)): for j in range(i+1, len(embeddings)): sim = cosine_similarity([embeddings[i]], [embeddings[j]])[0][0] similarities.append(sim) avg_similarity = np.mean(similarities) diversity_score = 1.0 - avg_similarity # 相似度越低，多樣性越高 return { "diversity_score": diversity_score, "avg_inter_document_similarity": avg_similarity } def comprehensive_evaluation(self, retriever, test_queries): """綜合評估檢索器效能""" results = {} for query in test_queries: retrieved_docs = retriever.invoke(query) relevance_metrics = self.evaluate_relevance(query, retrieved_docs) diversity_metrics = self.evaluate_diversity(retrieved_docs) results[query] = { **relevance_metrics, **diversity_metrics, "num_docs": len(retrieved_docs) } return results # 評估不同檢索器 evaluator = RetrievalEvaluator(OpenAIEmbeddings()) test_queries = [ "深度學習的基本概念", "機器學習模型的評估指標", "神經網路的優化技術" ] # 評估基礎檢索器 basic_results = evaluator.comprehensive_evaluation(faiss_retriever, test_queries) # 評估 Ensemble 檢索器 ensemble_results = evaluator.comprehensive_evaluation(ensemble_retriever, test_queries) # 比較結果 for query in test_queries: print(f"\n查詢: {query}") print(f"基礎檢索器 - 平均相似度: {basic_results[query]['mean_similarity']:.3f}") print(f"Ensemble 檢索器 - 平均相似度: {ensemble_results[query]['mean_similarity']:.3f}") print(f"基礎檢索器 - 多樣性: {basic_results[query]['diversity_score']:.3f}") print(f"Ensemble 檢索器 - 多樣性: {ensemble_results[query]['diversity_score']:.3f}") 6.2 A/B 測試框架 ```pythonimport randomfrom typing import Dict, List, Callable class RetrieverABTester: """檢索器 A/B 測試框架""" def __init__(self, evaluator): self.evaluator = evaluator self.test_results = {} def run_ab_test( self, retriever_a, retriever_b, test_queries: List[str], test_name: str = "ab_test" ): """執行 A/B 測試""" results_a = self.evaluator.comprehensive_evaluation(retriever_a, test_queries) results_b = self.evaluator.comprehensive_evaluation(retriever_b, test_queries) comparison = self._compare_results(results_a, results_b) self.test_results[test_name] = { "retriever_a": results_a, "retriever_b": results_b, "comparison": comparison } return comparison def _compare_results(self, results_a, results_b): """比較兩個檢索器的結果""" metrics = ["mean_similarity", "diversity_score"] comparison = {} for metric in metrics: a_scores = [results_a[q][metric] for q in results_a.keys()] b_scores = [results_b[q][metric] for q in results_b.keys()] comparison[metric] = { "a_avg": np.mean(a_scores), "b_avg": np.mean(b_scores), "improvement": (np.mean(b_scores) - np.mean(a_scores)) / np.mean(a_scores) * 100, "winner": "B" if np.mean(b_scores) > np.mean(a_scores) else "A" } return comparison def statistical_significance_test(self, test_name: str, metric: str = "mean_similarity"): """統計顯著性檢驗""" from scipy import stats if test_name not in self.test_results: raise ValueError(f"Test {test_name} not found") results_a = self.test_results[test_name]["retriever_a"] results_b = self.test_results[test_name]["retriever_b"] a_scores = [results_a[q][metric] for q in results_a.keys()] b_scores = [results_b[q][metric] for q in results_b.keys()]