RAG 中的向量儲存 (Vector Stores) 在 RAG (Retrieval-Augmented Generation) 流程中，向量儲存 (Vector Store) 扮演著至關重要的角色。在文件經過「載入 (Load)」、「分割 (Split)」和「嵌入 (Embed)」之後，所產生的向量化區塊 (Vectorized Chunks) 需要一個專門的資料庫來進行高效的儲存、管理和檢索。這就是向量儲存的核心功能。 本文件將根據 09-VectorStore 資料夾中的筆記本內容，介紹向量儲存的基本概念以及幾種主流的向量資料庫。 1. 為什麼需要向量儲存？ 傳統的關聯式資料庫（如 SQL）是為結構化資料設計的，不適合儲存和查詢高維度的向量資料。向量儲存則專為此而生，它具備以下關鍵優勢： 高效的相似度搜尋：能夠在數百萬甚至數十億個向量中，快速找到與給定查詢向量最相似的向量。透過索引演算法（如 HNSW、IVF）可以將搜尋時間從線性降至對數級別。 可擴展性：能夠應對不斷增長的資料量，而不會顯著降低查詢效能。支援水平擴展和叢集部署，可處理 TB 級別的向量資料。 語意檢索：支援基於語意相似度的檢索，而不僅僅是關鍵字匹配，這對於 RAG 應用至關重要。能夠理解同義詞、上下文和概念關聯。 **中繼資料過濾 (Metadata Filtering)**：允許在進行向量搜尋的同時，根據文件的中繼資料（如標題、日期、來源、標籤）進行過濾，實現更精準的檢索。 多模態支援：能夠同時處理文字、圖像、音訊等不同類型的向量資料，實現跨模態搜尋。 向量儲存 vs 傳統資料庫對比 特性 傳統資料庫 向量資料庫 資料類型 結構化資料 高維向量 + 中繼資料 查詢方式 精確匹配、關鍵字 相似度搜尋、語意理解 索引方法 B-tree、Hash HNSW、IVF、LSH 查詢延遲 毫秒級 毫秒到秒級（取決於資料量） 適用場景 交易處理、報表 AI 搜尋、推薦系統 2. 核心操作 (CRUD & Search) LangChain 為各種向量儲存提供了一個統一的介面，主要包含以下操作： 2.1 資料寫入操作 add_documents(documents): 批次新增文件到向量儲存中。系統會自動對文件進行向量化並儲存相關中繼資料。 # 批次新增文件 db.add_documents([doc1, doc2, doc3]) upsert(vectors, ids, metadata): 更新插入操作，如果文件已存在則更新，否則新增。適用於需要動態更新知識庫的場景。 # 更新或插入特定 ID 的文件 db.upsert( vectors=[vector1, vector2], ids=["doc_1", "doc_2"], metadata=[{"source": "updated"}, {"source": "new"}] ) from_documents(documents, embedding, **kwargs): 類別方法，從文件列表直接建立向量儲存實例。 # 一次性建立向量儲存 db = VectorStore.from_documents( documents=docs, embedding=embeddings, persist_directory="./db" # 持久化路徑 ) 2.2 資料讀取與搜尋 similarity_search(query, k=4, filter=None): 核心搜尋功能，回傳最相似的 K 個文件。 # 基本相似度搜尋 results = db.similarity_search( query="人工智慧的應用", k=5, filter={"category": "technology"} # 中繼資料過濾 ) similarity_search_with_score(query, k=4): 除了回傳文件外，還包含相似度分數，有助於評估結果品質。 # 帶分數的搜尋 results_with_scores = db.similarity_search_with_score( query="機器學習演算法", k=3 ) for doc, score in results_with_scores: print(f"Score: {score:.3f} - {doc.page_content[:100]}") max_marginal_relevance_search(query, k=4, fetch_k=20): 最大邊際相關性搜尋，平衡相關性和多樣性。 # MMR 搜尋避免重複內容 diverse_results = db.max_marginal_relevance_search( query="深度學習", k=5, fetch_k=50, # 先取得更多候選 lambda_mult=0.7 # 平衡相關性(1.0)和多樣性(0.0) ) 2.3 資料管理操作 delete(ids): 根據文件 ID 刪除特定文件。對於需要動態維護知識庫的應用非常重要。 # 刪除過時文件 db.delete(ids=["outdated_doc_1", "outdated_doc_2"]) get(ids): 根據 ID 取得特定文件，用於驗證或偵錯。 # 取得特定文件 specific_docs = db.get(ids=["doc_1", "doc_2"]) 3. 常用的向量儲存方案 LangChain 整合了多種向量儲存方案，開發者可以根據應用需求、部署環境（本地 vs. 雲端）和成本來選擇。 本地端方案 (Local / Self-Hosted) 適合快速原型開發、小型應用或對資料隱私有嚴格要求的場景。 3.1 FAISS (Facebook AI Similarity Search) 由 Facebook 開發的高效能程式庫，非常適合在記憶體中進行快速的相似度搜尋。它輕量且易於整合，但本身不提供持久化儲存（需要額外實現）。 優勢： 極快的搜尋速度（純記憶體運算） 支援多種索引演算法（Flat、IVF、HNSW等） 豐富的量化選項以節省記憶體 支援 GPU 加速 劣勢： 需要手動實現持久化 不支援線上更新（需要重建索引） 單機限制，不支援分散式 適用場景： 快速原型開發 靜態資料集 對延遲要求極高的應用 資料量中等（百萬到千萬級別） from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings import pickle # 建立或載入 FAISS 索引 embedding = OpenAIEmbeddings() # 方法一：從文件建立 db = FAISS.from_documents(docs, embedding) # 方法二：從已有向量建立 # db = FAISS.from_texts(texts, embedding, metadatas=metadatas) # 搜尋 results = db.similarity_search(query="小王子來自哪個星球？", k=3) # 持久化儲存 db.save_local("faiss_index") # 載入已儲存的索引 db_loaded = FAISS.load_local("faiss_index", embedding, allow_dangerous_deserialization=True) # 進階配置：使用不同的索引類型 # IndexFlatL2: 精確搜尋，適合小資料集 # IndexIVFFlat: 近似搜尋，適合大資料集 # IndexHNSWFlat: 圖形索引，平衡速度和精度 3.2 Chroma 一個開源的、專為 AI 應用設計的嵌入式資料庫。它簡單易用，支援本地持久化儲存和中繼資料過濾。 優勢： 原生支援持久化 優秀的中繼資料過濾功能 內建的資料版本控制 支援多種距離度量 輕量級，易於部署 劣勢： 相對較新，生態系統較小 大規模資料效能有限 單機部署限制 適用場景： 中小型應用（百萬級文件） 需要豐富中繼資料過濾 快速原型和開發 本地部署需求 from langchain_chroma import Chroma from langchain_openai import OpenAIEmbeddings import chromadb # 初始化持久化客戶端 client = chromadb.PersistentClient(path="./chroma_db") # 建立向量儲存 db = Chroma( client=client, collection_name="my_collection", embedding_function=OpenAIEmbeddings() ) # 新增文件 db.add_documents(docs) # 進階搜尋：結合中繼資料過濾 results = db.similarity_search( query="機器學習演算法", k=5, filter={"source": "academic_paper", "year": {"$gte": 2020}} ) # 使用 where 和 where_document 進行複雜過濾 results = db.similarity_search( query="深度學習", k=3, filter={ "$and": [ {"category": {"$eq": "AI"}}, {"difficulty": {"$in": ["intermediate", "advanced"]}} ] } ) # 資料統計 collection = client.get_collection("my_collection") print(f"Total documents: {collection.count()}") 3.3 PGVector PostgreSQL 的一個擴充套件，讓傳統的關聯式資料庫具備了儲存和查詢向量的能力。適合希望在現有 PostgreSQL 基礎設施上擴充向量搜尋功能的團隊。 優勢： 結合關聯式資料庫的 ACID 特性 強大的 SQL 查詢能力 成熟的生態系統和工具 支援複雜的混合查詢 優秀的資料一致性保證 劣勢： 向量搜尋效能不如專用向量資料庫 需要 PostgreSQL 運維知識 記憶體使用較高 適用場景： 已有 PostgreSQL 基礎設施 需要複雜的關聯查詢 強一致性要求 企業級應用 from langchain_postgres import PGVector from langchain_openai import OpenAIEmbeddings import os # 資料庫連線設定 CONNECTION_STRING = "postgresql+psycopg://user:password@localhost:5432/vectordb" # 建立向量儲存 db = PGVector( embeddings=OpenAIEmbeddings(), collection_name="documents", connection=CONNECTION_STRING, use_jsonb=True # 使用 JSONB 儲存中繼資料 ) # 從文件建立 db = PGVector.from_documents( documents=docs, embedding=OpenAIEmbeddings(), collection_name="my_docs", connection=CONNECTION_STRING ) # 複雜的 SQL 查詢結合向量搜尋 # 可以直接使用 SQL 進行複雜查詢 query_sql = """ SELECT document, cmetadata, embedding <-> %s as distance FROM langchain_pg_embedding WHERE cmetadata ->> 'category' = 'research' AND cmetadata ->> 'year' > '2020' ORDER BY embedding <-> %s LIMIT 5; """ # 使用 hybrid search 結合全文檢索 results = db.similarity_search( query="人工智慧倫理", k=5, filter={"category": "ethics", "language": "zh-TW"} ) 3.4 Elasticsearch 一個強大的分散式搜尋和分析引擎，除了傳統的全文檢索外，也支援向量相似度搜尋，適合需要混合搜尋（關鍵字 + 語意）的複雜場景。 優勢： 優秀的全文檢索能力 支援混合搜尋（BM25 + 向量） 強大的聚合和分析功能 成熟的叢集和分散式能力 豐富的生態系統 劣勢： 配置複雜，學習曲線陡峭 資源消耗較高 向量搜尋效能不如專用方案 適用場景： 需要混合搜尋能力 已有 Elasticsearch 基礎設施 大規模資料處理 複雜的搜尋和分析需求 from langchain_elasticsearch import ElasticsearchStore from langchain_openai import OpenAIEmbeddings # Elasticsearch 配置 es_config = { "hosts": ["http://localhost:9200"], "http_auth": ("username", "password"), # 如果需要認證 "verify_certs": False } # 建立向量儲存 db = ElasticsearchStore( embedding=OpenAIEmbeddings(), index_name="documents", es_connection=es_config ) # 混合搜尋：結合關鍵字和語意搜尋 hybrid_results = db.similarity_search( query="機器學習在醫療領域的應用", k=5, search_kwargs={ "query": { "bool": { "should": [ # 語意搜尋 { "knn": { "embedding": { "vector": query_vector, "k": 10 } } }, # 關鍵字搜尋 { "match": { "content": "機器學習 醫療" } } ] } } } ) # 使用聚合進行資料分析 aggregation_query = { "aggs": { "categories": { "terms": {"field": "metadata.category.keyword"} }, "avg_score": { "avg": {"script": "_score"} } } } 雲端託管方案 (Cloud-Hosted / Managed) 適合需要高可用性、可擴展性和無需自行維護基礎設施的生產環境應用。 3.5 Pinecone 一個完全託管的向量資料庫服務，專為高效能、低延遲的向量搜尋而設計，提供簡單的 API 和可擴展的基礎設施。 優勢： 完全託管，無需維護 極佳的搜尋效能和擴展性 簡單的 API 設計 內建的資料備份和災難恢復 支援實時更新 劣勢： 成本較高 供應商鎖定風險 對中繼資料查詢支援有限 適用場景： 生產環境應用 大規模向量搜尋（億級別） 需要高可用性 不想維護基礎設施 import os from langchain_pinecone import PineconeVectorStore from langchain_openai import OpenAIEmbeddings from pinecone import Pinecone # 初始化 Pinecone pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY")) index_name = "my-langchain-index" # 建立索引（如果不存在） if index_name not in pc.list_indexes().names(): pc.create_index( name=index_name, dimension=1536, # OpenAI embeddings 維度 metric="cosine", spec=ServerlessSpec( cloud="aws", region="us-east-1" ) ) # 建立向量儲存 vectorstore = PineconeVectorStore( index=pc.Index(index_name), embedding=OpenAIEmbeddings() ) # 批次新增文件 vectorstore.add_documents(docs) # 進階搜尋配置 results = vectorstore.similarity_search( query="小王子和狐狸說了什麼？", k=3, filter={"source": {"$in": ["book1", "book2"]}} # 中繼資料過濾 ) # 使用命名空間進行資料隔離 vectorstore_ns = PineconeVectorStore( index=pc.Index(index_name), embedding=OpenAIEmbeddings(), namespace="project_a" # 資料隔離 ) # 統計資訊 index_stats = pc.Index(index_name).describe_index_stats() print(f"Total vectors: {index_stats['total_vector_count']}") 3.6 Qdrant 提供開源和雲端兩種版本，以其豐富的過濾功能和高效的查詢效能而聞名。 優勢： 豐富的過濾和搜尋選項 優秀的效能表現 支援多租戶架構 強大的 API 和 SDK 開源版本可自部署 劣勢： 相對較新的產品 文件和社群相對較小 某些高級功能需要付費版本 適用場景： 需要複雜過濾邏輯 多租戶應用 高效能要求 混合雲部署 from langchain_qdrant import QdrantVectorStore from langchain_openai import OpenAIEmbeddings from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams # 雲端版本 client = QdrantClient( url="https://your-cluster-url.qdrant.io", api_key="your-api-key" ) # 本地版本 # client = QdrantClient(host="localhost", port=6333) collection_name = "my_documents" # 建立集合（如果不存在） if not client.collection_exists(collection_name): client.create_collection( collection_name=collection_name, vectors_config=VectorParams( size=1536, # OpenAI embeddings 維度 distance=Distance.COSINE ) ) # 建立向量儲存 qdrant = QdrantVectorStore( client=client, collection_name=collection_name, embedding=OpenAIEmbeddings() ) # 複雜的過濾查詢 results = qdrant.similarity_search( query="人工智慧的未來發展", k=5, filter={ "must": [ {"key": "category", "match": {"value": "technology"}}, {"key": "year", "range": {"gte": 2020}} ], "should": [ {"key": "language", "match": {"value": "zh"}}, {"key": "language", "match": {"value": "en"}} ] } ) # 使用 payload 進行進階操作 qdrant.add_documents( documents=docs, ids=["doc1", "doc2"], payloads=[ {"category": "AI", "priority": "high", "tags": ["ML", "DL"]}, {"category": "Tech", "priority": "medium", "tags": ["Software"]} ] ) 3.7 Weaviate 同樣提供開源和雲端版本，支援圖形化資料模型和 GraphQL API，適合需要複雜資料關聯的應用。 優勢： 圖形資料模型支援 GraphQL API 內建的機器學習模組 多模態資料支援 強大的語意搜尋能力 劣勢： 學習曲線較陡峭 配置相對複雜 資源消耗較高 適用場景： 複雜的資料關聯需求 多模態搜尋 知識圖譜應用 GraphQL 生態系統 from langchain_weaviate import WeaviateVectorStore from langchain_openai import OpenAIEmbeddings import weaviate # 連線到 Weaviate client = weaviate.Client( url="https://your-weaviate-instance.weaviate.network", auth_client_secret=weaviate.AuthApiKey(api_key="your-api-key") ) # 本地 Weaviate # client = weaviate.Client("http://localhost:8080") # 建立向量儲存 db = WeaviateVectorStore( client=client, index_name="Document", text_key="content", embedding=OpenAIEmbeddings() ) # GraphQL 查詢 graphql_query = """ { Get { Document( nearText: { concepts: ["人工智慧"] } limit: 5 where: { path: ["category"] operator: Equal valueString: "technology" } ) { content category _additional { distance } } } } """ # 執行 GraphQL 查詢 result = client.query.raw(graphql_query) # 使用 Weaviate 的內建模組 # 支援自動分類、問答等 classification_result = client.query.get("Document", ["content"]) \ .with_near_text({"concepts": ["機器學習"]}) \ .with_additional(["classification"]) \ .with_limit(5) \ .do() 3.8 MongoDB Atlas Vector Search MongoDB 的雲端服務，將向量搜尋功能與其強大的 NoSQL 文件資料庫無縫整合，方便在現有資料上擴充 AI 功能。 優勢： 與 MongoDB 生態系統完美整合 強大的文件資料庫功能 支援複雜的聚合管道 成熟的分片和副本集 豐富的中繼資料查詢能力 劣勢： 向量搜尋功能相對較新 需要 Atlas 雲端服務 成本可能較高 適用場景： 已使用 MongoDB 的應用 需要複雜文件查詢 混合資料類型應用 需要強大的聚合能力 from langchain_mongodb import MongoDBAtlasVectorSearch from langchain_openai import OpenAIEmbeddings from pymongo import MongoClient # 連線到 MongoDB Atlas client = MongoClient("mongodb+srv://username:password@cluster.mongodb.net/") db = client["vector_database"] collection = db["documents"] # 建立向量搜尋索引（在 Atlas UI 中配置） # { # "fields": [ # { # "type": "vector", # "path": "embedding", # "numDimensions": 1536, # "similarity": "cosine" # }, # { # "type": "filter", # "path": "metadata.category" # } # ] # } # 建立向量儲存 vectorstore = MongoDBAtlasVectorSearch( collection=collection, embedding=OpenAIEmbeddings(), index_name="vector_index" ) # 複雜的聚合查詢 pipeline = [ { "$vectorSearch": { "index": "vector_index", "path": "embedding", "queryVector": query_vector, "numCandidates": 100, "limit": 10, "filter": {"metadata.category": "technology"} } }, { "$group": { "_id": "$metadata.author", "docs": {"$push": "$$ROOT"}, "avg_score": {"$avg": "$score"} } }, {"$sort": {"avg_score": -1}} ] results = list(collection.aggregate(pipeline)) 3.9 Neo4j 一個圖資料庫，除了儲存節點和關係外，也支援向量索引。這使得它能夠進行結合了圖遍歷和向量相似度搜尋的複雜查詢。 優勢： 強大的圖查詢能力 結合向量搜尋和圖遍歷 Cypher 查詢語言 優秀的關係資料處理 支援知識圖譜應用 劣勢： 學習曲線陡峭 不適合純向量搜尋場景 資源消耗較高 適用場景： 知識圖譜應用 複雜關係查詢 推薦系統 欺詐檢測 from langchain_community.vectorstores import Neo4jVector from langchain_openai import OpenAIEmbeddings from neo4j import GraphDatabase # 連線到 Neo4j uri = "bolt://localhost:7687" username = "neo4j" password = "password" # 建立向量儲存 vectorstore = Neo4jVector.from_documents( documents=docs, embedding=OpenAIEmbeddings(), url=uri, username=username, password=password, index_name="document_embeddings" ) # 複雜的 Cypher 查詢結合向量搜尋 cypher_query = """ CALL db.index.vector.queryNodes('document_embeddings', 5, $query_vector) YIELD node, score MATCH (node)-[:AUTHORED_BY]->(author:Author) MATCH (node)-[:BELONGS_TO]->(category:Category) WHERE category.name IN ['AI', 'Technology'] RETURN node.content, author.name, score ORDER BY score DESC """ # 使用圖遍歷增強檢索 graph_enhanced_query = """ // 找到與查詢最相似的文件 CALL db.index.vector.queryNodes('document_embeddings', 10, $query_vector) YIELD node as doc, score // 擴展到相關文件 MATCH (doc)-[:RELATED_TO]-(related_doc:Document) // 找到共同的主題 MATCH (doc)-[:HAS_TOPIC]->(topic:Topic)<-[:HAS_TOPIC]-(related_doc) RETURN doc, related_doc, topic, score ORDER BY score DESC LIMIT 5 """ # 執行圖增強查詢 with GraphDatabase.driver(uri, auth=(username, password)) as driver: with driver.session() as session: result = session.run( graph_enhanced_query, query_vector=query_embedding ) enhanced_results = [record for record in result] 4. 向量儲存作為檢索器 (Retriever) 在 LangChain 中，任何向量儲存都可以透過 .as_retriever() 方法轉換為一個**檢索器 (Retriever)**。檢索器是一個更通用的介面，它封裝了從資料來源（如向量儲存）取回文件的邏輯。 4.1 基本檢索器配置 # 基本檢索器 retriever = db.as_retriever( search_type="similarity", # 搜尋類型 search_kwargs={'k': 5} # 搜尋參數 ) # 使用檢索器 relevant_docs = retriever.invoke("關於馴養的意義") 4.2 進階檢索策略 Maximal Marginal Relevance (MMR) 在回傳結果時，不僅考慮與查詢的相似度，還考慮結果之間的多樣性，避免回傳過於相似的內容。 # MMR 檢索器 mmr_retriever = db.as_retriever( search_type="mmr", search_kwargs={ 'k': 5, # 最終回傳數量 'fetch_k': 20, # 候選文件數量 'lambda_mult': 0.7 # 多樣性參數 (0=最大多樣性, 1=最大相關性) } ) # MMR 演算法原理： # 1. 先取得 fetch_k 個最相似的候選文件 # 2. 從候選文件中選擇既相關又多樣的 k 個文件 # 3. 每次選擇時平衡相關性和與已選文件的差異性 Similarity Score Threshold 根據相似度分數過濾結果，只回傳超過特定閾值的文件。 # 基於分數閾值的檢索器 threshold_retriever = db.as_retriever( search_type="similarity_score_threshold", search_kwargs={ 'score_threshold': 0.8, # 分數閾值 'k': 10 # 最大回傳數量 } ) # 只有相似度分數 > 0.8 的文件才會被回傳 relevant_docs = threshold_retriever.invoke("機器學習演算法") Self-Querying Retriever 讓 LLM 根據使用者的自然語言問題，自動生成結構化的中繼資料過濾條件。 from langchain.retrievers.self_query.base import SelfQueryRetriever from langchain.chains.query_constructor.base import AttributeInfo from langchain_openai import ChatOpenAI # 定義中繼資料屬性 metadata_field_info = [ AttributeInfo( name="category", description="文件的類別，如 'technology', 'science', 'business'", type="string" ), AttributeInfo( name="year", description="文件發布的年份", type="integer" ), AttributeInfo( name="author", description="文件作者", type="string" ), AttributeInfo( name="difficulty", description="內容難度等級，如 'beginner', 'intermediate', 'advanced'", type="string" ) ] # 文件內容描述 document_content_description = "技術文章和研究論文的集合" # 建立自查詢檢索器 llm = ChatOpenAI(model="gpt-4", temperature=0) self_query_retriever = SelfQueryRetriever.from_llm( llm=llm, vectorstore=db, document_contents=document_content_description, metadata_field_info=metadata_field_info, verbose=True ) # 使用自然語言查詢 # LLM 會自動解析出過濾條件 results = self_query_retriever.invoke( "找出 2023 年後發布的高級機器學習文章" ) # 自動轉換為: query="機器學習", filter={"year": {"$gte": 2023}, "difficulty": "advanced"} Multi-Query Retriever 為單一使用者問題產生多個不同角度的查詢，提高檢索覆蓋率。 from langchain.retrievers.multi_query import MultiQueryRetriever # 建立多查詢檢索器 multiquery_retriever = MultiQueryRetriever.from_llm( retriever=db.as_retriever(), llm=ChatOpenAI(model="gpt-4", temperature=0) ) # 對於使用者問題 "AI 的倫理問題"，可能會產生： # 1. "人工智慧倫理考量" # 2. "機器學習公平性問題" # 3. "AI 決策透明度" # 4. "演算法偏見處理" results = multiquery_retriever.invoke("AI 的倫理問題") Contextual Compression Retriever 對檢索到的文件進行壓縮和過濾，只保留與查詢最相關的部分。 from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor # 建立文件壓縮器 compressor = LLMChainExtractor.from_llm(ChatOpenAI(model="gpt-4")) # 建立壓縮檢索器 compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=db.as_retriever() ) # 檢索並壓縮結果 compressed_docs = compression_retriever.invoke("深度學習在醫療診斷中的應用") # 每個文件只保留與查詢相關的段落 4.3 組合檢索策略 Ensemble Retriever 結合多個不同的檢索器，利用各自的優勢。 from langchain.retrievers import EnsembleRetriever from langchain_community.retrievers import BM25Retriever # 建立 BM25 檢索器（關鍵字搜尋） bm25_retriever = BM25Retriever.from_documents(docs) bm25_retriever.k = 3 # 建立向量檢索器（語意搜尋） vector_retriever = db.as_retriever(search_kwargs={'k': 3}) # 組合檢索器 ensemble_retriever = EnsembleRetriever( retrievers=[bm25_retriever, vector_retriever], weights=[0.4, 0.6] # BM25: 40%, Vector: 60% ) # 結合關鍵字和語意搜尋的優勢 results = ensemble_retriever.invoke("機器學習模型評估指標") Parent Document Retriever 先檢索小區塊，然後回傳包含該區塊的較大文件段落。 from langchain.retrievers import ParentDocumentRetriever from langchain.storage import InMemoryStore from langchain.text_splitter import RecursiveCharacterTextSplitter # 建立兩個分割器 parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000) child_splitter = RecursiveCharacterTextSplitter(chunk_size=400) # 建立儲存 store = InMemoryStore() # 建立父文件檢索器 parent_retriever = ParentDocumentRetriever( vectorstore=db, docstore=store, child_splitter=child_splitter, parent_splitter=parent_splitter ) # 新增文件 parent_retriever.add_documents(docs) # 檢索時返回包含相關小區塊的大段落 results = parent_retriever.invoke("神經網路訓練技巧") 5. 向量儲存選擇指南 5.1 決策矩陣 考量因素 本地方案 雲端方案 推薦選擇 資料隱私 ✅ 完全控制 ⚠️ 依賴供應商政策 Chroma, PGVector 維運成本 💰 需要人力維護 💰💰 服務費用較高 FAISS (開發), Pinecone (生產) 擴展性 ⚠️ 硬體限制 ✅ 彈性擴展 Pinecone, Qdrant 整合難度 📚 需要學習 🚀 開箱即用 Chroma (簡單), Weaviate (複雜) 資料量 < 1M 文件 > 10M 文件 小: FAISS, 大: Pinecone 5.2 使用場景建議 快速原型開發 # 推薦：FAISS 或 Chroma # 理由：設置簡單，無需外部依賴 from langchain_community.vectorstores import FAISS from langchain_openai import OpenAIEmbeddings # 幾行程式碼即可開始 db = FAISS.from_documents(docs, OpenAIEmbeddings()) results = db.similarity_search("查詢內容") 生產環境應用 # 推薦：Pinecone 或 Qdrant # 理由：高可用性，專業維護 from langchain_pinecone import PineconeVectorStore import pinecone # 企業級向量搜尋 vectorstore = PineconeVectorStore( index=pinecone.Index("production-index"), embedding=OpenAIEmbeddings() ) 混合搜尋需求 # 推薦：Elasticsearch 或 MongoDB Atlas # 理由：同時支援關鍵字和語意搜尋 from langchain_elasticsearch import ElasticsearchStore # 結合 BM25 和向量搜尋 db = ElasticsearchStore( embedding=OpenAIEmbeddings(), index_name="hybrid_search" ) 複雜關係查詢 # 推薦：Neo4j 或 Weaviate # 理由：支援圖查詢和向量搜尋 from langchain_community.vectorstores import Neo4jVector # 知識圖譜 + 向量搜尋 vectorstore = Neo4jVector.from_documents( docs, OpenAIEmbeddings(), url="bolt://localhost:7687" ) 5.3 效能優化建議 索引優化 # FAISS 索引優化 import faiss # 對於大型資料集，使用 IVF 索引 nlist = 100 # 聚類數量 quantizer = faiss.IndexFlatL2(dimension) index = faiss.IndexIVFFlat(quantizer, dimension, nlist) # 訓練索引 index.train(training_vectors) index.add(vectors) # 搜尋參數調整 index.nprobe = 10 # 搜尋的聚類數量 快取策略 from functools import lru_cache import hashlib class CachedVectorStore: def __init__(self, vectorstore): self.vectorstore = vectorstore @lru_cache(maxsize=1000) def _cached_search(self, query_hash, k): return self.vectorstore.similarity_search(query_hash, k=k) def similarity_search(self, query, k=4): query_hash = hashlib.md5(query.encode()).hexdigest() return self._cached_search(query_hash, k) 批次處理 # 批次新增文件以提高效率 def batch_add_documents(vectorstore, documents, batch_size=100): for i in range(0, len(documents), batch_size): batch = documents[i:i + batch_size] vectorstore.add_documents(batch) print(f"已處理 {min(i + batch_size, len(documents))}/{len(documents)} 文件") 6. 最佳實務和常見陷阱 6.1 向量維度一致性 # 確保所有向量維度一致 embedding_model = OpenAIEmbeddings() dimension = len(embedding_model.embed_query("test")) print(f"向量維度: {dimension}") # 驗證向量維度 def validate_vectors(vectors): if not vectors: return True expected_dim = len(vectors[0]) return all(len(v) == expected_dim for v in vectors) 6.2 中繼資料設計 # 良好的中繼資料結構 metadata_template = { "source": "document_source.pdf", "page": 1, "section": "introduction", "category": "technology", "tags": ["AI", "machine learning"], "created_at": "2024-01-01", "updated_at": "2024-01-15", "language": "zh-TW", "difficulty": "intermediate", "author": "張三", "version": "1.0" } # 確保中繼資料可搜尋 def normalize_metadata(metadata): """標準化中繼資料格式""" normalized = {} for key, value in metadata.items(): if isinstance(value, str): normalized[key] = value.lower().strip() elif isinstance(value, list): normalized[key] = [str(v).lower().strip() for v in value] else: normalized[key] = value return normalized 6.3 向量品質監控 import numpy as np from sklearn.metrics.pairwise import cosine_similarity def monitor_vector_quality(vectorstore, test_queries): """監控向量搜尋品質""" results = {} for query in test_queries: # 檢索結果 docs = vectorstore.similarity_search_with_score(query, k=5) # 計算分數分佈 scores = [score for _, score in docs] results[query] = { "avg_score": np.mean(scores), "min_score": np.min(scores), "max_score": np.max(scores), "score_std": np.std(scores) } return results # 定期執行品質監控 test_queries = ["人工智慧", "機器學習", "深度學習"] quality_report = monitor_vector_quality(db, test_queries) 6.4 資料版本管理 class VersionedVectorStore: def __init__(self, base_vectorstore): self.base_store = base_vectorstore self.versions = {} self.current_version = "v1.0" def create_version(self, version_name, documents): """建立新版本""" # 為文件添加版本標籤 versioned_docs = [] for doc in documents: doc.metadata["version"] = version_name versioned_docs.append(doc) self.base_store.add_documents(versioned_docs) self.versions[version_name] = len(versioned_docs) self.current_version = version_name def search_version(self, query, version=None, k=4): """在特定版本中搜尋""" if version is None: version = self.current_version filter_condition = {"version": version} return self.base_store.similarity_search( query, k=k, filter=filter_condition ) 選擇合適的向量儲存是建構 RAG 應用的重要決策。你需要根據你的資料規模、查詢延遲要求、預算、部署環境以及是否需要進階過濾或混合搜尋功能來做出選擇。 記住，沒有「最好」的向量儲存，只有「最適合」你特定需求的解決方案。建議在小規模資料上測試多個選項，評估其效能、易用性和成本，然後再做出最終決定。