基本 RAG 實作:逐步教學

這份適合初學者的教學會說明基本 RAG 實作的運作方式、介紹最佳做法,並逐步帶你使用 Python、ChromaDB 以及 Perplexity 的嵌入模型和語言模型,建置可實際運作的命令列 RAG 系統。

Basic RAG implementation process diagram: the interface sends input to the orchestrator, which retrieves data from a vector database, builds the input context for the LLM, and returns the output to the interface.

RAG 是現代 LLM 基礎架構中不可或缺的一環。RAG 讓 AI 模型能夠存取外部資料,有助於解決幻覺與訓練資料不完整等問題。這也代表企業能運用內含敏感或專有資訊的內部知識庫。

RAG 的實作可能非常複雜,但基本的 RAG 系統其實出乎意料地容易建置。對初學者來說,建立一個簡單的架構有助於了解 RAG 在實際環境中的運作方式,也能作為進一步實驗的跳板。

本文將說明基本 RAG 實作的架構、介紹最佳做法,接著使用 Python、ChromaDB 以及兩個分別用於嵌入與答案生成的 Perplexity 模型,建置一個可實際運作的系統。

基本 RAG 實作流程:概覽

基本的 RAG 系統包含三大元件:協調器、資訊來源和 LLM。在典型的工作流程中,使用者輸入提示詞後,協調器會修改提示詞,以搜尋所有已連接的資訊來源。接著,擷取到的資訊會加入原始輸入上下文,再傳送給 LLM 以生成回覆。

建置 RAG 實作時,你需要考量下列步驟:

  1. 索引與嵌入:在這個階段,你的文件會轉換為協調器可搜尋的資訊來源。最常見的做法之一是建立可搜尋的向量索引。我們將使用 Perplexity 的專用模型,將文件區塊轉換為嵌入向量,再透過熱門的開源向量資料庫 ChromaDB 儲存這些嵌入向量。
  2. 協調:這是位於外部資訊來源(在本例中為向量資料庫)與 LLM 之間的應用程式。我們將建立一個簡單的 Python 指令碼來擔任協調器。
  3. 擷取與生成:當使用者透過聊天介面提出問題時,協調器會接收問題、執行向量搜尋,並將擴充後的輸入傳送給 LLM。我們不會另外建置專用前端,而是透過命令列輸入提示詞,與 Python 協調器溝通。

最佳做法:怎樣才是有效的 RAG 實作?

RAG 架構有多種類型,不同的使用情境需要不同的做法。儘管如此,仍有許多最佳做法適用於所有情況,能確保系統準確、具成本效益、可擴充且易於維護。

下列最佳做法適用於所有類型的 RAG 實作:

  • 確保資料匯入管線乾淨:建立索引前,來源資料應盡可能乾淨。你也應該使用適合內容的嵌入模型。
  • 使用混合搜尋:最佳的擷取方法會因使用情境而異,無論是向量資料庫搜尋、透過 API 進行的網路搜尋、知識圖譜遍歷、關鍵字搜尋或其他方法皆然。選擇合適的組合,對於降低成本十分重要。
  • 評估你的系統:實作完成後,你應從三個核心指標進行評估:準確度、相關性和延遲。準確度是指事實是否正確;相關性是指擷取的資訊與使用者查詢的契合程度;延遲則是指擷取資訊的速度。
  • 安全處理資訊:如果你的 RAG 系統會使用內部私人資訊,請確保資訊安全儲存,並妥善管理系統介面的存取權限。切勿分享 API 金鑰。

基本 RAG 實作:7 步驟教學

在本教學中,我們將建置一個命令列應用程式,針對單一文件回答問題。我們會使用 Python、ChromaDB 和 Perplexity API。

本教學的目的並非建置可用於正式環境的 RAG 系統,而是提供適合初學者的教學,說明基礎 RAG 架構實際上如何運作。

步驟 1:安裝軟體

請確認電腦上已安裝下列項目,並擁有 Perplexity API 帳戶:

  • Python:這是我們用來建置協調器的程式語言,協調器負責協調嵌入模型、ChromaDB 與 Perplexity 之間的資料流。
  • Visual Studio:我們將使用 Visual Studio 作為程式碼編輯器。安裝完成後,開啟延伸模組面板並下載 Python 延伸模組。
  • Perplexity API 帳戶:Perplexity 將同時提供嵌入模型(將資料轉換為向量嵌入)以及處理擴充提示詞的 LLM。你不需要付費帳戶,但會消耗少量點數。

步驟 2:建立專案資料夾

首先,建立名為 rag-project 的資料夾,並選取檔案 > 開啟資料夾,在 VS Code 中開啟此資料夾。接著透過終端機 > 新增終端機開啟整合式終端機。

輸入下列內容以建立 Python 虛擬環境:

bash
py -3.12 -m venv .venv

現在你需要安裝四個 Python 程式庫:perplexityai、chromadb、python-dotenv 和 numpy。Perplexity SDK 可讓你透過 API 連線至 Perplexity 的模型。ChromaDB 會建置向量資料庫並在本機執行。Python-dotenv 會從本機檔案安全地載入你的 API 金鑰,而資料匯入過程則需要用到 NumPy。

使用下列指令安裝這些程式庫:

bash
.\.venv\Scripts\python.exe -m pip install perplexityai chromadb python-dotenv numpy

步驟 3:設定檔案

將下列檔案新增至 rag-project(專案根目錄),方法是在檔案總管側邊欄中選取新增檔案…:

  • index.py:這是用來為文件建立索引的應用程式。你可以重新執行它來加入更多文字。
  • rag.py:這是擔任協調器的 Python 應用程式。
  • document.txt:這個檔案將包含你要轉換為向量資料庫的文件內容。
  • .env:這是存放 Perplexity API 金鑰的地方。

在 .env 中輸入下列內容:

plaintext
PERPLEXITY_API_KEY=your_api_key_here

將 your_api_key_here 替換為你實際的金鑰。為此,你需要前往 Perplexity API Console 首頁儀表板,並選取建立 API 金鑰。請妥善保管此 API 金鑰,切勿外流。

我們將使用兩個模型:Perplexity 的 pplx-embed-v1-0.6b,用於在上傳至 ChromaDB 之前嵌入文件內容;以及 GPT-5.6-Luna(透過 Perplexity 的 API 存取),用於處理擷取到的資訊。

在 document.txt 中,加入 RAG 管線要擷取的文字。如果手邊沒有現成的文件,可以在本教學中使用古騰堡計畫(Project Gutenberg)上 William Blake 的《天真與經驗之歌》(Songs of Innocence and of Experience)。這是一部篇幅短、易於理解且屬於公共領域的作品。

步驟 4:撰寫索引程式

將下方程式碼複製並貼到 index.py 中。每個小標題都會簡要說明接下來那段程式碼的作用。這個程式會使用 Perplexity 的嵌入模型 pplx-embed-v1-0.6b 與 NumPy 程式庫,將你的來源資料轉換為向量嵌入,並儲存至 ChromaDB 資料庫。

這個程式的作用如下:

  1. 開啟 document.txt。
  2. 將文字分割成彼此重疊的區塊。
  3. 將這些區塊傳送至 Perplexity 的 pplx-embed-v1-0.6b 嵌入模型。
  4. 使用 NumPy(一個提供運算函式的 Python 程式庫)解碼並正規化傳回的向量。
  5. 將向量與原始區塊儲存至 ChromaDB 資料庫(你會看到它出現在專案目錄中)。

程式經過設定,不會在每次執行時重新為文件建立索引。如果你在 document.txt 中新增資訊,程式只會嵌入新的文字。

python
from pathlib import Path from collections import defaultdict, deque from hashlib import sha256 import base64 import numpy as np import chromadb from dotenv import load_dotenv from perplexity import Perplexity # 1. Connect to Perplexity and ChromaDB load_dotenv() ai = Perplexity() db = chromadb.PersistentClient(path="./rag_database_perplexity") collection = db.get_or_create_collection( name="documents", metadata={"hnsw:space": "cosine"} ) # Convert Perplexity's encoded embeddings into numerical vectors def decode_embedding(encoded): vector = np.frombuffer(base64.b64decode(encoded), dtype=np.int8).astype(np.float32) norm = np.linalg.norm(vector) if norm == 0: raise ValueError("Received a zero-length embedding vector.") return (vector / norm).tolist() # 2. Read the document and split it into chunks text = Path("document.txt").read_text(encoding="utf-8") words = text.split() chunk_size = 180 overlap = 30 step = chunk_size - overlap chunks = [" ".join(words[i:i + chunk_size]) for i in range(0, len(words), step)] if not chunks: raise ValueError("document.txt is empty.") # 3. Retrieve the chunks already stored in ChromaDB existing = collection.get(include=["documents"]) stored_chunks = defaultdict(deque) existing_ids = set(existing["ids"]) for chunk_id, document in zip(existing["ids"], existing["documents"]): stored_chunks[document].append(chunk_id) # 4. Identify unchanged and new chunks new_chunks = [] new_ids = [] kept_ids = set() unchanged = 0 for chunk in chunks: if stored_chunks[chunk]: chunk_id = stored_chunks[chunk].popleft() kept_ids.add(chunk_id) unchanged += 1 else: chunk_hash = sha256(chunk.encode("utf-8")).hexdigest() chunk_id = f"chunk-{chunk_hash}" suffix = 1 while chunk_id in existing_ids or chunk_id in kept_ids or chunk_id in new_ids: chunk_id = f"chunk-{chunk_hash}-{suffix}" suffix += 1 new_chunks.append(chunk) new_ids.append(chunk_id) kept_ids.add(chunk_id) # 5. Generate embeddings in batches of up to 100 chunks batch_size = 100 for i in range(0, len(new_chunks), batch_size): batch_chunks = new_chunks[i:i + batch_size] batch_ids = new_ids[i:i + batch_size] embeddings = ai.embeddings.create( model="pplx-embed-v1-0.6b", input=batch_chunks, encoding_format="base64_int8" ) vectors = [decode_embedding(item.embedding) for item in embeddings.data] # 6. Store each batch in ChromaDB collection.upsert( ids=batch_ids, documents=batch_chunks, embeddings=vectors ) print(f"Indexed batch {(i // batch_size) + 1}") # 7. Remove outdated chunks old_ids = [chunk_id for chunk_id in existing["ids"] if chunk_id not in kept_ids] if old_ids: collection.delete(ids=old_ids) # 8. Display indexing results print("Indexing complete!") print("Unchanged chunks:", unchanged) print("New or modified chunks:", len(new_chunks)) print("Removed chunks:", len(old_ids)) print("Total stored chunks:", collection.count())

步驟 5:撰寫 RAG 系統的協調器

將下方程式碼複製並貼到 rag.py 中。這是一個完整但精簡的 RAG 實作,使用 Perplexity SDK 和 ChromaDB 的本機向量資料庫功能。

執行程式時會發生以下情況:

  1. 連線至 Perplexity 和你新建的 ChromaDB 資料庫。
  2. 檢查資料庫是否包含已建立索引的文件。
  3. 接收使用者提出的問題。
  4. 將問題傳送至 Perplexity 的嵌入模型。
  5. 使用 NumPy 解碼並正規化產生的嵌入向量。
  6. 在 ChromaDB 中搜尋最相似的三個文件區塊。
  7. 建構包含擷取區塊與使用者問題的提示詞。
  8. 透過 Perplexity 的 Agent API 將提示詞傳送至 GPT-5.6 Luna。
  9. 顯示答案並等待下一個問題。
python
import base64 import numpy as np import chromadb from dotenv import load_dotenv from perplexity import Perplexity # 1. Connect to Perplexity and the existing ChromaDB database load_dotenv() ai = Perplexity() db = chromadb.PersistentClient(path="./rag_database_perplexity") collection = db.get_collection("documents") # Convert Perplexity's encoded embeddings into numerical vectors def decode_embedding(encoded): vector = np.frombuffer(base64.b64decode(encoded), dtype=np.int8).astype(np.float32) norm = np.linalg.norm(vector) if norm == 0: raise ValueError("Received a zero-length embedding vector.") return (vector / norm).tolist() # 2. Check that the database contains indexed documents count = collection.count() if count == 0: raise ValueError("The database is empty. Run index.py first.") print("Connected to existing ChromaDB database.") print("Stored chunks:", count) # 3. Accept questions from the user while True: question = input("\nAsk a question (or type exit): ") if question.lower().strip() == "exit": break if not question.strip(): continue # 4. Convert the question into a Perplexity embedding question_response = ai.embeddings.create( model="pplx-embed-v1-0.6b", input=[question], encoding_format="base64_int8" ) question_vector = decode_embedding(question_response.data[0].embedding) # 5. Retrieve the most similar document chunks from ChromaDB results = collection.query( query_embeddings=[question_vector], n_results=min(3, count) ) context = "\n\n".join(results["documents"][0]) # 6. Construct a prompt and send it to the fixed LLM response = ai.responses.create( model="openai/gpt-5.6-luna", instructions=( "Answer using only the supplied document excerpts. " "Do not use external information. " "If the answer is not present, say you don't know." ), input=f"Document excerpts:\n{context}\n\nQuestion: {question}" ) # 7. Display the answer print("\nAnswer:\n", response.output_text)

步驟 6:執行應用程式

在 VS Code 的 PowerShell 終端機中輸入下列指令,嵌入 document.txt 中的內容:

bash
.\.venv\Scripts\python.exe index.py

如果文件已成功建立索引,你應該會看到類似以下的輸出:

plaintext
Indexing complete! Unchanged chunks: 0 New or modified chunks: 58 Removed chunks: 0 Total stored chunks: 58

現在可以執行主要的協調器應用程式了。輸入此指令:

bash
.\.venv\Scripts\python.exe rag.py

如果執行成功,你會看到以下內容:

plaintext
Connected to existing ChromaDB database. Stored chunks: 58 Ask a question (or type exit):

現在你可以提出問題,系統會從你建立的向量資料庫中擷取資訊來回答,例如:

plaintext
Ask a question (or type exit): What animals does William Blake talk about? Answer: The excerpts mention sheep, lambs, birds, beasts, wolves, tigers, and a lion.

步驟 7:了解你建置的系統

你建置的系統包含每個 RAG 架構都具備的兩條核心管線:索引(或稱「資料匯入」)管線,以及擷取與生成管線。

你的基本 RAG 實作現在能做到以下所有事項:

  • 向量儲存:嵌入向量會根據你的知識檔案,儲存在本機環境的 ChromaDB 資料庫中。
  • 增量索引:未變更的文件區塊不會重新嵌入,而且你可以隨時更新文件。
  • 語意擷取:ChromaDB 會根據向量相似度擷取區塊(由協調器發起)。
  • 以資料為依據的答案生成:LLM 會接收擷取到的段落作為上下文,並將其融入答案中。
  • 索引與查詢各自獨立:你可以直接提問,不必重新處理知識檔案中的所有文字。
  • 透過 API 連接模型:所有答案皆透過 Perplexity 使用 GPT-5.6 Luna 生成,而非使用在你電腦上執行的模型。

後續步驟:如何進一步擴充這個實作

本教學的目的是建置一個簡單易懂的基本實作,涵蓋現代系統中部分核心 RAG 架構元件。如果你想更進一步,有多種方式可以在此基礎上擴充。

以下是一些擴充實作範圍的構想:

  • 試用 LangChain:LangChain 是最熱門的開源協調框架之一,內含預先建置的元件,可用於開發能執行複雜 RAG 管線的應用程式。
  • 擴充搜尋功能:將密集向量搜尋與 BM25 等詞彙(關鍵字)搜尋結合。透過 Perplexity Search API 加入網路搜尋,也能讓答案以最新資料為依據。
  • 加入重新排序模型:整合重新排序模型,依據初始提示詞為擷取到的段落排序,並選出最相關的段落,納入提供給 LLM 的輸入上下文中。
  • 建置自適應 RAG 系統:自適應 RAG 會使用 LLM 或分類器(協調器中的元件),在模組化 RAG 系統中決定要使用哪種擷取方法。建置自適應 RAG 系統是探索更進階 RAG 實作的好方法。

雖然這些聽起來可能很複雜,但請記得,Perplexity 等主流 AI 平台能提供進階的程式設計協助。這些平台非常適合用於個人專案和實驗,而且成本可以控制在相對較低的水準。

注意:本教學使用 GPT-5.6 Luna。模型的可用性可能會變動,因此你可能需要更新 API 呼叫。