這篇教學將指引你如何 從零開始構建數據攝取（Data Ingestion）流程，將文檔處理為向量並存入向量數據庫。我們不使用 LlamaIndex 的高階封裝，而是直接構建數據處理管道，幫助你了解內部細節。 完整教學的 官方文件：👉 LlamaIndex 官方文件 🚀 1.從零開始構建數據攝取管道（Data Ingestion） 📌 目標 在這篇教學中，我們將學習如何構建 數據攝取管道，將文本轉換為向量嵌入，並存入 Pinecone 向量數據庫。 主要步驟： 載入文檔 拆分文本 構建節點（Nodes） 提取元數據（Metadata, 選擇性） 生成嵌入向量 插入向量數據庫 查詢向量數據庫 1️⃣ 環境準備 我們需要安裝以下 Python 套件： pip install llama-index-embeddings-openai pip install llama-index-vector-stores-pinecone pip install llama-index-llms-openai pip install llama-index pip install python-dotenv pinecone-client pymupdf 🔑 設定 API Keys 你需要： Pinecone API Key（申請免費帳號） OpenAI API Key（取得 API 金鑰） 建立一個 .env 檔案來儲存 API Key： import os from dotenv import load_dotenv dotenv_path = "env" with open(dotenv_path, "w") as f: f.write('PINECONE_API_KEY=""\n') f.write('OPENAI_API_KEY=""\n') load_dotenv(dotenv_path=dotenv_path) 2️⃣ 建立 Pinecone 向量數據庫 我們將使用 Pinecone 作為向量儲存庫。 from pinecone import Pinecone, Index, ServerlessSpec api_key = os.environ["PINECONE_API_KEY"] pc = Pinecone(api_key=api_key) index_name = "llamaindex-rag-fs" # 創建索引（如不存在） if index_name not in pc.list_indexes().names(): pc.create_index( index_name, dimension=1536, # text-embedding-ada-002 的維度 metric="euclidean", spec=ServerlessSpec(cloud="aws", region="us-east-1"), ) pinecone_index = pc.Index(index_name) # 清除索引（可選） pinecone_index.delete(deleteAll=True) 🌟 建立 LlamaIndex 的 VectorStore from llama_index.vector_stores.pinecone import PineconeVectorStore vector_store = PineconeVectorStore(pinecone_index=pinecone_index) 3️⃣ 載入文檔 下載 LLaMA2 論文 PDF，並解析文本： import fitz file_path = "./data/llama2.pdf" doc = fitz.open(file_path) 4️⃣ 使用文本拆分器 我們使用 LlamaIndex 內建的 SentenceSplitter，將長文本拆分為小段落： from llama_index.core.node_parser import SentenceSplitter text_parser = SentenceSplitter(chunk_size=1024) text_chunks = [] doc_idxs = [] for doc_idx, page in enumerate(doc): page_text = page.get_text("text") cur_text_chunks = text_parser.split_text(page_text) text_chunks.extend(cur_text_chunks) doc_idxs.extend([doc_idx] * len(cur_text_chunks)) 5️⃣ 手動構建節點 LlamaIndex 的 TextNode 是一種低階數據抽象，我們手動創建節點並添加元數據： from llama_index.core.schema import TextNode nodes = [] for idx, text_chunk in enumerate(text_chunks): node = TextNode(text=text_chunk) nodes.append(node) # 測試：印出第一個節點內容 print(nodes[0].get_content(metadata_mode="all")) 6️⃣ （可選）提取元數據 如果想要提取 標題與關鍵問題，我們可以使用 LlamaIndex 的 MetadataExtractor： from llama_index.core.extractors import ( QuestionsAnsweredExtractor, TitleExtractor, ) from llama_index.core.ingestion import IngestionPipeline from llama_index.llms.openai import OpenAI llm = OpenAI(model="gpt-3.5-turbo") extractors = [ TitleExtractor(nodes=5, llm=llm), QuestionsAnsweredExtractor(questions=3, llm=llm), ] pipeline = IngestionPipeline(transformations=extractors) nodes = await pipeline.arun(nodes=nodes, in_place=False) # 測試：印出第一個節點的元數據 print(nodes[0].metadata) 7️⃣ 生成嵌入向量 我們使用 OpenAI 的 text-embedding-ada-002 模型來生成向量： from llama_index.embeddings.openai import OpenAIEmbedding embed_model = OpenAIEmbedding() for node in nodes: node_embedding = embed_model.get_text_embedding( node.get_content(metadata_mode="all") ) node.embedding = node_embedding 8️⃣ 將節點插入向量數據庫 vector_store.add(nodes) 9️⃣ 查詢向量數據庫 數據攝取完成後，我們可以開始查詢： from llama_index.core import VectorStoreIndex, StorageContext index = VectorStoreIndex.from_vector_store(vector_store) query_engine = index.as_query_engine() query_str = "Can you tell me about the key concepts for safety finetuning?" response = query_engine.query(query_str) print(str(response)) 🎯 總結 這篇教學展示了 從零開始構建數據攝取管道 的完整流程，幫助你： 載入文檔（PDF） 拆分文本（Sentence Splitter） 建立節點（TextNode） 提取元數據（TitleExtractor, QuestionsExtractor） 生成向量嵌入（OpenAI Embeddings） 插入向量數據庫（Pinecone） 執行查詢（VectorStoreIndex） 🚀 現在，你已經可以獨立構建 RAG 系統的數據攝取模組了！ 🚀 這篇教學將指引你 從零開始構建檢索（Retrieval）系統，我們將不依賴高階封裝的檢索方法，而是逐步實作： 建立向量數據庫（Pinecone） 載入文檔並將其轉換為向量 手動生成查詢嵌入 查詢向量數據庫 解析結果 封裝成自訂檢索器（Retriever） 整合檢索器與查詢引擎 完整教學的 官方文件：👉 LlamaIndex 官方文件 🚀 2.從零開始構建檢索系統（Retrieval） 📌 目標 本教程將展示如何手動構建一個檢索系統，並將其整合到 LlamaIndex 的查詢引擎中。我們將使用 Pinecone 作為向量存儲庫，並學習如何： 生成查詢嵌入（Query Embeddings） 執行不同檢索模式（密集、稀疏、混合） 解析檢索結果 封裝為自訂檢索器（Retriever） 1️⃣ 環境準備 安裝所需的 Python 套件： pip install llama-index-readers-file pymupdf pip install llama-index-vector-stores-pinecone pip install llama-index-embeddings-openai pip install llama-index 🔑 設定 API Keys 你需要： Pinecone API Key（申請免費帳號） OpenAI API Key（取得 API 金鑰） import os from dotenv import load_dotenv dotenv_path = "env" with open(dotenv_path, "w") as f: f.write('PINECONE_API_KEY=""\n') f.write('OPENAI_API_KEY=""\n') load_dotenv(dotenv_path=dotenv_path) 2️⃣ 建立 Pinecone 向量數據庫 from pinecone import Pinecone, Index, ServerlessSpec api_key = os.environ["PINECONE_API_KEY"] pc = Pinecone(api_key=api_key) dataset_name = "quickstart" # 創建索引（如不存在） if dataset_name not in pc.list_indexes().names(): pc.create_index( dataset_name, dimension=1536, # text-embedding-ada-002 的維度 metric="euclidean", spec=ServerlessSpec(cloud="aws", region="us-east-1"), ) pinecone_index = pc.Index(dataset_name) # 清除索引（可選） pinecone_index.delete(deleteAll=True) 🌟 建立 LlamaIndex 的 VectorStore from llama_index.vector_stores.pinecone import PineconeVectorStore vector_store = PineconeVectorStore(pinecone_index=pinecone_index) 3️⃣ 載入文檔並轉換為向量 下載 LLaMA2 論文 PDF，並解析文本： !mkdir data !wget --user-agent "Mozilla" "https://arxiv.org/pdf/2307.09288.pdf" -O "data/llama2.pdf" from pathlib import Path from llama_index.readers.file import PyMuPDFReader loader = PyMuPDFReader() documents = loader.load(file_path="./data/llama2.pdf") 📌 載入文檔至向量存儲 我們使用 高階方法 VectorStoreIndex.from_documents 來加速載入數據： from llama_index.core import VectorStoreIndex from llama_index.core.node_parser import SentenceSplitter from llama_index.core import StorageContext splitter = SentenceSplitter(chunk_size=1024) storage_context = StorageContext.from_defaults(vector_store=vector_store) index = VectorStoreIndex.from_documents( documents, transformations=[splitter], storage_context=storage_context ) 4️⃣ 生成查詢嵌入 from llama_index.embeddings.openai import OpenAIEmbedding embed_model = OpenAIEmbedding() query_str = "Can you tell me about the key concepts for safety finetuning?" query_embedding = embed_model.get_query_embedding(query_str) 5️⃣ 查詢向量數據庫 我們支援 不同檢索模式： default（密集檢索 Dense） sparse（稀疏檢索 Sparse） hybrid（混合檢索 Hybrid） from llama_index.core.vector_stores import VectorStoreQuery query_mode = "default" # 可切換成 "sparse" 或 "hybrid" vector_store_query = VectorStoreQuery( query_embedding=query_embedding, similarity_top_k=2, mode=query_mode ) query_result = vector_store.query(vector_store_query) print(query_result) 6️⃣ 解析檢索結果 我們將結果封裝成 **NodeWithScore**，包含檢索到的內容與分數： from llama_index.core.schema import NodeWithScore from typing import Optional nodes_with_scores = [] for index, node in enumerate(query_result.nodes): score: Optional[float] = None if query_result.similarities is not None: score = query_result.similarities[index] nodes_with_scores.append(NodeWithScore(node=node, score=score)) 輸出檢索到的內容： from llama_index.core.response.notebook_utils import display_source_node for node in nodes_with_scores: display_source_node(node, source_length=1000) 7️⃣ 封裝成自訂檢索器（Retriever） from llama_index.core import QueryBundle from llama_index.core.retrievers import BaseRetriever from typing import Any, List class PineconeRetriever(BaseRetriever): """自訂檢索器，對 Pinecone 向量數據庫進行查詢""" def __init__( self, vector_store: PineconeVectorStore, embed_model: Any, query_mode: str = "default", similarity_top_k: int = 2, ) -> None: """初始化參數""" self._vector_store = vector_store self._embed_model = embed_model self._query_mode = query_mode self._similarity_top_k = similarity_top_k super().__init__() def _retrieve(self, query_bundle: QueryBundle) -> List[NodeWithScore]: """檢索處理""" if query_bundle.embedding is None: query_embedding = self._embed_model.get_query_embedding(query_bundle.query_str) else: query_embedding = query_bundle.embedding vector_store_query = VectorStoreQuery( query_embedding=query_embedding, similarity_top_k=self._similarity_top_k, mode=self._query_mode, ) query_result = self._vector_store.query(vector_store_query) nodes_with_scores = [] for index, node in enumerate(query_result.nodes): score: Optional[float] = None if query_result.similarities is not None: score = query_result.similarities[index] nodes_with_scores.append(NodeWithScore(node=node, score=score)) return nodes_with_scores 初始化檢索器： retriever = PineconeRetriever(vector_store, embed_model, query_mode="default", similarity_top_k=2) retrieved_nodes = retriever.retrieve(query_str) for node in retrieved_nodes: display_source_node(node, source_length=1000) 8️⃣ 整合查詢引擎 from llama_index.core.query_engine import RetrieverQueryEngine query_engine = RetrieverQueryEngine.from_args(retriever) response = query_engine.query(query_str) print(str(response)) 🎯 總結 這篇指南展示了 如何從零開始構建檢索系統，幫助你掌握： 生成查詢嵌入 執行不同模式的向量檢索 解析檢索結果 封裝為自訂檢索器 整合查詢引擎 🚀 現在，你已經可以獨立打造 RAG 系統的檢索模組！ 🚀 好的，我會根據你的需求撰寫一篇完整的 「完全開源的 RAG 系統」 部落格文章，並在適當位置加入 Optional 章節，幫助讀者更深入理解 RAG 的運作機制，並提供額外的優化技巧。 這篇文章不僅涵蓋 LlamaIndex 官方指南的內容，還會補充： 可選的進階技巧（Optional Sections） 最佳實踐建議 額外的擴展選項 🚀 3.從零開始構建 完全開源 RAG 系統 📌 介紹 大多數 RAG（Retrieval-Augmented Generation）系統依賴 OpenAI API 或 雲端向量數據庫，這雖然方便，但在某些應用場景下，我們可能需要 完全開源的替代方案，例如： 隱私要求高的應用（如醫療或法律） 離線運行（On-Premise） 降低 API 成本 對 LLM、向量存儲和檢索技術有更深理解的需求 在這篇文章中，我們將使用 完全開源的技術堆疊 來構建 RAG 系統： 嵌入模型：Sentence Transformers 向量存儲：PostgreSQL + pgvector LLM（大型語言模型）：Llama 2（透過 llama.cpp） 這篇教學將涵蓋： 數據攝取（Data Ingestion） 向量存儲（Vector Storage） 檢索管道（Retrieval Pipeline） 查詢處理（Query Processing） 1️⃣ 環境準備 🛠️ 安裝必要套件： pip install llama-index-readers-file pymupdf pip install llama-index-vector-stores-postgres pip install llama-index-embeddings-huggingface pip install llama-index-llms-llama-cpp pip install psycopg2-binary pgvector asyncpg "sqlalchemy[asyncio]" greenlet 🔑 設定 PostgreSQL 安裝 PostgreSQL & pgvector Linux / Mac：brew install postgresql brew services start postgresql **安裝 pgvector**：psql -d postgres -c "CREATE EXTENSION IF NOT EXISTS vector;" 建立用戶與資料庫：CREATE ROLE my_user WITH LOGIN PASSWORD 'my_password'; ALTER ROLE my_user SUPERUSER; CREATE DATABASE vector_db; 2️⃣ 嵌入模型（Embedding Model） 我們將使用 Hugging Face 的 BAAI/bge-small-en 模型來計算文本嵌入： from llama_index.embeddings.huggingface import HuggingFaceEmbedding embed_model = HuggingFaceEmbedding(model_name="BAAI/bge-small-en") ✅ Optional: 為何選擇 bge-small-en？ 訓練良好，適合向量檢索 相較於 text-embedding-ada-002，它是 完全開源 的 若需更高性能，可以嘗試 BAAI/bge-large-en 3️⃣ 載入 Llama 2 作為 LLM 我們使用 llama.cpp 來載入 Llama 2-13B： from llama_index.llms.llama_cpp import LlamaCPP model_url = "https://huggingface.co/TheBloke/Llama-2-13B-chat-GGUF/resolve/main/llama-2-13b-chat.Q4_0.gguf" llm = LlamaCPP( model_url=model_url, model_path=None, temperature=0.1, max_new_tokens=256, context_window=3900, model_kwargs={"n_gpu_layers": 1}, verbose=True, ) ✅ Optional: 可以使用哪些開源 LLM？ Llama 3（即將推出） Mistral 7B Falcon 40B ChatGLM 3 4️⃣ 建立 PostgreSQL 向量存儲 import psycopg2 from llama_index.vector_stores.postgres import PGVectorStore db_name = "vector_db" host = "localhost" password = "my_password" port = "5432" user = "my_user" conn = psycopg2.connect( dbname="postgres", host=host, password=password, port=port, user=user, ) conn.autocommit = True with conn.cursor() as c: c.execute(f"DROP DATABASE IF EXISTS {db_name}") c.execute(f"CREATE DATABASE {db_name}") vector_store = PGVectorStore.from_params( database=db_name, host=host, password=password, port=port, user=user, table_name="llama2_paper", embed_dim=384, ) ✅ Optional: 其他開源向量存儲 Chroma Weaviate Qdrant 5️⃣ 數據攝取管道 1. 載入文檔 from llama_index.readers.file import PyMuPDFReader loader = PyMuPDFReader() documents = loader.load(file_path="./data/llama2.pdf") 2. 拆分文本 from llama_index.core.node_parser import SentenceSplitter text_parser = SentenceSplitter(chunk_size=1024) text_chunks = [] doc_idxs = [] for doc_idx, doc in enumerate(documents): cur_text_chunks = text_parser.split_text(doc.text) text_chunks.extend(cur_text_chunks) doc_idxs.extend([doc_idx] * len(cur_text_chunks)) 3. 轉換為節點 from llama_index.core.schema import TextNode nodes = [] for idx, text_chunk in enumerate(text_chunks): node = TextNode(text=text_chunk) nodes.append(node) 4. 生成嵌入 for node in nodes: node.embedding = embed_model.get_text_embedding(node.get_content(metadata_mode="all")) 5. 插入向量存儲 vector_store.add(nodes) 6️⃣ 檢索管道 1. 生成查詢嵌入 query_str = "How does Llama 2 compare to other models?" query_embedding = embed_model.get_query_embedding(query_str) 2. 執行向量檢索 from llama_index.core.vector_stores import VectorStoreQuery vector_store_query = VectorStoreQuery( query_embedding=query_embedding, similarity_top_k=2, mode="default" ) query_result = vector_store.query(vector_store_query) 3. 解析結果 from llama_index.core.schema import NodeWithScore nodes_with_scores = [ NodeWithScore(node=node, score=query_result.similarities[i]) for i, node in enumerate(query_result.nodes) ] 7️⃣ 整合查詢引擎 from llama_index.core.query_engine import RetrieverQueryEngine retriever = VectorDBRetriever(vector_store, embed_model) query_engine = RetrieverQueryEngine.from_args(retriever, llm=llm) response = query_engine.query(query_str) print(str(response)) ✅ Optional: 如何提升檢索性能？ 使用 Hybrid Search（混合檢索） 結合 BM25 作關鍵字匹配 嘗試 Query Expansion（查詢擴展） 🎯 結論 我們構建了一個 完全開源 的 RAG 系統，無需依賴 OpenAI 或商業 API，適合：✅ 高隱私場景✅ 離線應用✅ 客製化需求 🚀 現在，你可以打造自己的開源 RAG 系統了！ 🚀 這是一個 關鍵的 Optional 章節，我會將它納入前面「完全開源的 RAG 系統」部落格中的 Optional 部分，作為「進階檢索技術 - 簡易 Hybrid Search」。這樣讀者在學完基本的向量檢索後，可以更進一步學習如何將 向量檢索（Semantic Search）與關鍵字檢索（Keyword Lookup） 結合，提高檢索的準確度與靈活性。 ✅ Optional: 進階檢索技術 - 簡易 Hybrid Search 在標準的 RAG 流程中，我們通常使用 向量檢索（Vector Retrieval），即： 對 Query 進行嵌入（Embedding） 從向量存儲中檢索 Top-K 相似結果 傳遞給 LLM 進行回答生成 這種方法的優勢是： 能夠理解語意相似性 可以檢索到未必完全匹配關鍵字的內容 但它也有一些問題： 無法處理 精確匹配（Exact Match） 可能忽略關鍵字的重要性 在數據量少時，語意相似度可能不準確 🚀 解決方案：Hybrid Search Hybrid Search（混合檢索） 結合了： 向量檢索（Semantic Search） 關鍵字檢索（Keyword Lookup） 這樣我們可以在查詢時： 若關鍵字命中（Keyword Match），則優先返回完全匹配的內容 若關鍵字沒有命中，則回退至語意檢索（Vector Retrieval） 可使用「AND」與「OR」條件 來調整查詢模式 這種方式特別適合：✅ 企業內部知識庫（Enterprise KB）✅ FAQ 檢索系統✅ 結構化數據的 RAG 應用 🛠 1️⃣ Hybrid Search 的基本架構 這裡我們使用 LlamaIndex 來建立 兩種不同的索引： 向量索引（VectorStoreIndex） - 用來進行語意檢索 關鍵字索引（SimpleKeywordTableIndex） - 用來進行關鍵字匹配 from llama_index.core import SimpleDirectoryReader, StorageContext from llama_index.core import SimpleKeywordTableIndex, VectorStoreIndex # 讀取文檔 documents = SimpleDirectoryReader("./data/paul_graham").load_data() # 轉換為節點 nodes = Settings.node_parser.get_nodes_from_documents(documents) # 建立存儲上下文 storage_context = StorageContext.from_defaults() storage_context.docstore.add_documents(nodes) # 建立向量索引 & 關鍵字索引 vector_index = VectorStoreIndex(nodes, storage_context=storage_context) keyword_index = SimpleKeywordTableIndex(nodes, storage_context=storage_context) 🔍 2️⃣ 自訂 Hybrid Search 檢索器 我們自訂一個 CustomRetriever： 先使用向量檢索 再使用關鍵字檢索 使用 AND 或 OR 模式來決定如何合併結果 from llama_index.core import QueryBundle from llama_index.core.schema import NodeWithScore from llama_index.core.retrievers import ( BaseRetriever, VectorIndexRetriever, KeywordTableSimpleRetriever, ) from typing import List class CustomRetriever(BaseRetriever): """自訂混合檢索 Retriever""" def __init__( self, vector_retriever: VectorIndexRetriever, keyword_retriever: KeywordTableSimpleRetriever, mode: str = "AND", # 可選 "AND" 或 "OR" ) -> None: """初始化參數""" self._vector_retriever = vector_retriever self._keyword_retriever = keyword_retriever if mode not in ("AND", "OR"): raise ValueError("Invalid mode.") self._mode = mode super().__init__() def _retrieve(self, query_bundle: QueryBundle) -> List[NodeWithScore]: """執行檢索""" vector_nodes = self._vector_retriever.retrieve(query_bundle) keyword_nodes = self._keyword_retriever.retrieve(query_bundle) vector_ids = {n.node.node_id for n in vector_nodes} keyword_ids = {n.node.node_id for n in keyword_nodes} combined_dict = {n.node.node_id: n for n in vector_nodes} combined_dict.update({n.node.node_id: n for n in keyword_nodes}) if self._mode == "AND": retrieve_ids = vector_ids.intersection(keyword_ids) else: retrieve_ids = vector_ids.union(keyword_ids) retrieve_nodes = [combined_dict[rid] for rid in retrieve_ids] return retrieve_nodes ⚙️ 3️⃣ 建立 Hybrid Query Engine 我們將檢索器嵌入 RetrieverQueryEngine，並測試查詢： from llama_index.core import get_response_synthesizer from llama_index.core.query_engine import RetrieverQueryEngine # 建立向量檢索器 vector_retriever = VectorIndexRetriever(index=vector_index, similarity_top_k=2) # 建立關鍵字檢索器 keyword_retriever = KeywordTableSimpleRetriever(index=keyword_index) # 設定自訂混合檢索 custom_retriever = CustomRetriever(vector_retriever, keyword_retriever, mode="AND") # 定義 LLM 回應處理 response_synthesizer = get_response_synthesizer() # 建立查詢引擎 custom_query_engine = RetrieverQueryEngine( retriever=custom_retriever, response_synthesizer=response_synthesizer, ) 📝 4️⃣ 測試 Hybrid Search ✅ 測試 1：檢索與 YC 相關的內容 response = custom_query_engine.query("What did the author do during his time at YC?") print(response) 📌 結果 During his time at YC, the author worked on various projects, including writing essays and working on YC itself... ✅ 測試 2：關鍵字不匹配時的效果 response = custom_query_engine.query("What did the author do during his time at Yale?") print(response) 📌 結果 Empty Response 📢 Hybrid Search 能夠避免不相關的檢索結果！ ✅ 測試 3：純向量檢索 vector_response = vector_query_engine.query("What did the author do during his time at Yale?") print(vector_response) 📌 結果 The context information does not provide any information about the author's time at Yale. 📢 向量檢索可能會產生與內容無關的回應，但 Hybrid Search 可避免這種情況！ 🎯 結論 在這個 Optional 章節中，我們學習了： Hybrid Search 的必要性（解決語意檢索的精確匹配問題） 如何結合關鍵字索引與向量索引 如何自訂 Retriever 如何測試 Hybrid Search 📌 適用場景 內部文件檢索 FAQ 知識庫 法規與政策文件查詢 完整教學文件：👉 LlamaIndex 官方文件 🚀 現在，你的 RAG 系統已經支援 Hybrid Search 了！ 🚀