Chapter 8: 檢索增強生成 (RAG)

歡迎來到 PocketFlow 教學系列的第八章！在上一章 代理人/智能體 (Agent) 中，我們探索了如何賦予流程動態決策的能力，使其能像智能助手一樣根據情境選擇行動。然而，即使是智能代理人，如果其內部知識有限或過時，其回答的品質也可能受到影響。想像一下，您希望代理人能根據一本非常龐大且持續更新的醫學百科全書來回答專業問題，單純依賴代理人內建的知識可能不夠。

這時，「檢索增強生成 (Retrieval Augmented Generation, RAG)」就派上用場了。RAG 是一種強大的設計模式，它能讓大型語言模型 (LLM) 在回答問題或生成內容前，先從外部知識庫中「檢索」相關資訊，然後利用這些資訊來「增強」其「生成」的結果。這大大提高了答案的準確性、相關性和時效性。

為何需要 RAG？——讓語言模型博覽群書

大型語言模型雖然知識淵博，但它們的知識通常截至其訓練數據的最後日期，且對於非常特定或私有的領域知識可能不夠深入。直接向 LLM 提問時，它可能會：

• 提供過時的資訊。
• 對於它不確定的問題，可能會「捏造」答案（稱為「幻覺」）。
• 無法存取您私有的文件或資料庫中的特定知識。

檢索增強生成 (RAG) 就像是為您的 LLM 配備了一位超級研究助理。當您提出一個複雜問題時，這位助理不會馬上憑空回答，而是會：

1. 去圖書館/資料庫查找資料（檢索階段）：它會先到一個龐大的知識庫（例如您的公司文件、產品手冊、最新的研究論文等，這些通常會被預處理並存儲在一個稱為「向量數據庫」的特殊資料庫中）中，查找與問題最相關的幾份文件或段落。
2. 基於找到的資料組織答案（生成階段）：然後，LLM 會像一位專家一樣，閱讀這些找到的資料，並基於這些具體的、相關的資訊來組織和生成一個準確、詳盡的答案。

這種方式有效地將 LLM 的強大語言生成能力與外部知識庫的廣度和深度結合起來。

在 PocketFlow 中，我們通常透過一系列精心設計的節點 (Node)來實現 RAG 的流程。這個流程主要分為兩個階段：

1. 離線階段 (Offline Stage) - 建立索引：預處理您的知識文件，將它們轉換成可供快速檢索的格式，並存儲起來。這就像是整理圖書館的藏書並編制索引。
2. 線上階段 (Online Stage) - 查詢與回答：當使用者提出問題時，利用先前建立的索引來檢索相關資訊，並結合 LLM 生成答案。這就像是圖書管理員根據您的查詢找到相關書籍，然後您閱讀這些書籍來解答疑問。

讓我們一步步來看看如何在 PocketFlow 中建構一個 RAG 系統。

RAG 流程圖概覽

下面這張圖展示了一個典型的 RAG 系統的兩個主要階段：

階段一：離線索引 (Offline Indexing) - 準備您的知識庫

這個階段的目標是將您的原始文件（例如：文字檔案、PDF、網頁內容）轉換成一個可以被 LLM 高效檢索的知識庫。這通常只需要執行一次，或者在您的文件內容更新時定期執行。

我們會建立三個主要的節點 (Node)：

1. ChunkDocsNode：將長文件分割成較小、易於處理的文本塊 (chunks)。
2. EmbedDocsNode：將每個文本塊轉換成「嵌入向量」(embedding vector)，這是一種能夠捕捉文本語義的數字表示。
3. StoreIndexNode：將這些嵌入向量存儲到一個向量數據庫中，並為它們建立索引，以便快速搜尋。

為了簡化，我們假設有一些輔助函數：

• get_embedding(text)：接收文本，回傳其嵌入向量。
• create_index(embeddings)：接收嵌入向量列表，建立並回傳一個向量索引（例如 FAISS 索引）。
• search_index(index, query_embedding, top_k)：在索引中搜尋與查詢嵌入最相似的 top_k 個向量。

1. ChunkDocsNode (文件分塊節點)

由於 LLM 的上下文長度有限，且檢索時更精確的匹配通常發生在較小的文本單元上，我們需要將長文件切分成小塊。我們使用批次處理 (Batch)中的 BatchNode 來處理多個文件。

from pocketflow import BatchNode # 引入 BatchNode

# 假設我們有多個文件需要處理
# class ChunkDocsNode(BatchNode):
#     def prep(self, shared):
#         # 從 shared store 獲取文件路徑列表
#         return shared.get("files_to_process", []) 

#     def exec(self, filepath): # 處理單個文件
#         # 簡化：假設讀取文件並按固定大小分塊
#         # 在真實應用中，會有更完善的讀取和分塊邏輯
#         with open(filepath, "r", encoding="utf-8") as f: # 讀取文件
#             text = f.read()
#         chunks = [text[i:i+100] for i in range(0, len(text), 100)] # 每100字元分塊
#         print(f"📄 文件 '{filepath}' 已分割成 {len(chunks)} 塊。")
#         return chunks # 回傳此文件的文本塊列表
    
#     def post(self, shared, prep_res, exec_res_list):
#         # exec_res_list 是每個文件對應的文本塊列表所組成的列表
#         all_chunks = [] # 扁平化所有文本塊到一個列表中
#         for chunk_list_for_one_file in exec_res_list:
#             all_chunks.extend(chunk_list_for_one_file)
#         shared["all_document_chunks"] = all_chunks # 存到 shared store
#         print(f"📚 總共得到 {len(all_chunks)} 個文本塊。")

• prep(shared)：從共享儲存 (Shared Store) 中獲取一個包含文件路徑的列表。BatchNode 會為列表中的每個文件路徑調用一次 exec 方法。
• exec(filepath)：接收單個文件路徑，讀取文件內容，並將其分割成小的文本塊。回傳該文件的文本塊列表。
• post(shared, prep_res, exec_res_list)：exec_res_list 包含了所有文件處理後各自的文本塊列表。此方法將這些列表「扁平化」成一個包含所有文本塊的單一列表，並存儲在共享儲存 (Shared Store) 的 "all_document_chunks" 中。

2. EmbedDocsNode (文件嵌入節點)

接下來，我們需要將每個文本塊轉換成嵌入向量。嵌入向量能夠捕捉文本的語義含義，使得語義上相似的文本塊在向量空間中也彼此靠近。同樣，我們使用 BatchNode。

# 假設 get_embedding(text) 函數已定義
# class EmbedDocsNode(BatchNode):
#     def prep(self, shared):
#         # 從 shared store 獲取所有文本塊
#         return shared.get("all_document_chunks", [])

#     def exec(self, chunk_text): # 處理單個文本塊
#         # 簡化：呼叫模型產生嵌入向量
#         embedding = get_embedding(chunk_text) 
#         # print(f"🧬 已為文本塊 '{chunk_text[:15]}...' 產生嵌入。") # 可選的日誌
#         return embedding # 回傳此文本塊的嵌入向量
    
#     def post(self, shared, prep_res, exec_res_list):
#         # exec_res_list 是所有文本塊對應的嵌入向量列表
#         shared["all_document_embeddings"] = exec_res_list # 存到 shared store
#         print(f"💡 總共產生了 {len(exec_res_list)} 個嵌入向量。")

• prep(shared)：從共享儲存 (Shared Store) 中獲取 ChunkDocsNode 產生的所有文本塊。
• exec(chunk_text)：接收單個文本塊，並使用 get_embedding 函數（通常會呼叫一個嵌入模型，如 OpenAI 的 text-embedding-ada-002）為其生成嵌入向量。
• post(shared, prep_res, exec_res_list)：收集所有文本塊的嵌入向量，並將它們存儲在共享儲存 (Shared Store) 的 "all_document_embeddings" 中。

3. StoreIndexNode (儲存索引節點)

最後，我們將所有嵌入向量組織到一個向量數據庫（或一個簡單的索引結構，如 FAISS）中。這個索引允許我們快速地根據一個查詢向量找到最相似的文本塊向量。

from pocketflow import Node # 引入 Node

# 假設 create_index(embeddings) 函數已定義
# class StoreIndexNode(Node): # 這是普通的 Node
#     def prep(self, shared):
#         # 從 shared store 獲取所有嵌入向量
#         return shared.get("all_document_embeddings", [])

#     def exec(self, all_embeddings):
#         # 簡化：建立向量索引
#         if not all_embeddings:
#             print("⚠️ 沒有嵌入向量可以建立索引。")
#             return None # 或拋出錯誤
#         vector_index = create_index(all_embeddings) 
#         return vector_index # 回傳建立好的索引
    
#     def post(self, shared, prep_res, exec_res_index):
#         shared["knowledge_base_index"] = exec_res_index # 將索引存到 shared store
#         # 通常還會將 all_document_chunks 也保存起來，以便後續檢索時能對應回原始文本
#         # shared["knowledge_base_chunks"] = shared.get("all_document_chunks") 
#         print(f"💾 向量索引已建立並儲存。")

• prep(shared)：從共享儲存 (Shared Store) 中獲取所有嵌入向量。
• exec(all_embeddings)：使用 create_index 函數根據這些嵌入向量建立一個向量索引。
• post(shared, prep_res, exec_res_index)：將建立好的向量索引存儲在共享儲存 (Shared Store) 的 "knowledge_base_index" 中。為了能在檢索到向量索引後找到對應的原始文本塊，我們通常也會將 "all_document_chunks" 一併保存或使其在此階段可訪問。

離線索引流程 (OfflineFlow)

現在，我們將這三個節點串聯起來，形成離線索引的流程 (Flow)：

from pocketflow import Flow

# 建立節點實例 (假設上面三個類別已定義)
chunk_node = ChunkDocsNode()
embed_node = EmbedDocsNode()
store_node = StoreIndexNode()

# 連接節點
chunk_node >> embed_node
embed_node >> store_node

# 建立離線索引流程
offline_indexing_flow = Flow(start=chunk_node)

# 如何運行 (假設您已準備好文件列表)
# shared_offline_data = {
#     "files_to_process": ["./my_document1.txt", "./my_document2.txt"] 
# }
# offline_indexing_flow.run(shared_offline_data)
# 執行完畢後, shared_offline_data["knowledge_base_index"] 
# 和 shared_offline_data["all_document_chunks"] (或類似名稱) 就緒了。

運行此 offline_indexing_flow 後，您的共享儲存 (Shared Store) 中就會包含 "knowledge_base_index" (向量索引) 和 "all_document_chunks" (原始文本塊列表)，它們共同構成了您的知識庫，準備好在線上階段被查詢。

階段二：線上查詢與回答 (Online Query & Answer) - 使用知識庫

當使用者提出問題時，線上階段開始運作。它會利用離線階段建立的知識庫來生成答案。

我們也需要三個主要的節點 (Node)：

1. EmbedQueryNode：將使用者的問題轉換成嵌入向量。
2. RetrieveDocsNode：使用問題的嵌入向量，在知識庫索引中搜尋最相關的文本塊。
3. GenerateAnswerNode：將原始問題和檢索到的相關文本塊一起提供給 LLM，讓 LLM 生成最終答案。

1. EmbedQueryNode (查詢嵌入節點)

與嵌入文件塊類似，我們需要將使用者的問題也轉換成相同向量空間中的嵌入向量，這樣才能進行比較。

# class EmbedQueryNode(Node):
#     def prep(self, shared):
#         # 從 shared store 獲取使用者問題
#         return shared.get("user_question", "沒有問題") 

#     def exec(self, question_text):
#         # 簡化：產生問題的嵌入向量
#         query_embedding = get_embedding(question_text) 
#         return query_embedding
    
#     def post(self, shared, prep_res, exec_res_q_embedding):
#         shared["query_embedding"] = exec_res_q_embedding # 存到 shared store
#         print(f"💡 問題 '{prep_res[:20]}...' 的嵌入已產生。")

• prep(shared)：從共享儲存 (Shared Store) 中獲取使用者提出的 "user_question"。
• exec(question_text)：使用 get_embedding 函數為問題文本生成嵌入向量。
• post(shared, prep_res, exec_res_q_embedding)：將產生的查詢嵌入向量存儲在共享儲存 (Shared Store) 的 "query_embedding" 中。

2. RetrieveDocsNode (檢索文件節點)

此節點使用問題的嵌入向量，在離線階段建立的向量索引中搜尋，找出最相關的 k 個文本塊。

# 假設 search_index(index, query_embedding, top_k) 函數已定義
# class RetrieveDocsNode(Node):
#     def prep(self, shared):
#         query_emb = shared.get("query_embedding")
#         kb_index = shared.get("knowledge_base_index")
#         # 確保同時能取回原始文本塊
#         all_chunks = shared.get("all_document_chunks") 
#         return query_emb, kb_index, all_chunks

#     def exec(self, inputs):
#         query_emb, kb_index, all_chunks = inputs
#         if not query_emb or not kb_index or not all_chunks:
#             print("⚠️ 檢索所需資訊不完整。")
#             return ["錯誤：檢索資訊不足"] # 或其他錯誤處理
        
#         # 簡化：搜尋最相關的1個文本塊
#         # search_index 回傳的是索引ID和距離
#         ids, distances = search_index(kb_index, query_emb, top_k=1) 
#         if not ids or not ids[0]: return ["未找到相關文件"]

#         retrieved_chunk_text = all_chunks[ids[0][0]] # 根據ID獲取原始文本塊
#         return [retrieved_chunk_text] # 回傳檢索到的文本塊列表 (即使只有一個)
    
#     def post(self, shared, prep_res, exec_res_retrieved_chunks):
#         shared["retrieved_context_chunks"] = exec_res_retrieved_chunks
#         print(f"🔍 檢索到 {len(exec_res_retrieved_chunks)} 個相關文本塊。")
#         if exec_res_retrieved_chunks:
#             print(f"  最相關的：'{exec_res_retrieved_chunks[0][:30]}...'")

• prep(shared)：從共享儲存 (Shared Store) 中獲取 "query_embedding"（問題嵌入）、"knowledge_base_index"（向量索引）和 "all_document_chunks"（原始文本塊列表）。
• exec(inputs)：使用 search_index 函數，在 kb_index 中根據 query_emb 搜尋最相關的文本塊的索引。然後，根據這些索引從 all_chunks 中取出實際的文本內容。
• post(shared, prep_res, exec_res_retrieved_chunks)：將檢索到的相關文本塊列表存儲在共享儲存 (Shared Store) 的 "retrieved_context_chunks" 中。

3. GenerateAnswerNode (生成答案節點)

最後，這個節點將原始問題和上一步檢索到的相關文本塊（作為上下文）一起傳遞給 LLM，讓 LLM 生成最終的、基於證據的答案。

# 假設 call_llm(prompt) 函數已定義
# class GenerateAnswerNode(Node):
#     def prep(self, shared):
#         question = shared.get("user_question")
#         context_chunks = shared.get("retrieved_context_chunks", [])
#         return question, context_chunks

#     def exec(self, inputs):
#         question, context_chunks = inputs
#         # 簡化：將所有上下文塊合併成一個字串
#         context_str = "\n".join(context_chunks) 
        
#         prompt = f"""
# 請根據以下提供的上下文來回答問題。
# 如果上下文未提供足夠資訊，請直說無法回答。

# 上下文：
# {context_str}

# 問題：{question}

# 答案：
# """
#         answer = call_llm(prompt) # 呼叫 LLM
#         return answer
    
#     def post(self, shared, prep_res, exec_res_answer):
#         shared["final_llm_answer"] = exec_res_answer
#         print(f"💬 LLM 生成的答案：{exec_res_answer}")

• prep(shared)：從共享儲存 (Shared Store) 中獲取原始的 "user_question" 和檢索到的 "retrieved_context_chunks"。
• exec(inputs)：建構一個包含上下文和問題的提示 (prompt)，然後呼叫 call_llm 函數讓 LLM 生成答案。
• post(shared, prep_res, exec_res_answer)：將 LLM 生成的最終答案存儲在共享儲存 (Shared Store) 的 "final_llm_answer" 中。

線上查詢回答流程 (OnlineFlow)

將這三個線上階段的節點串聯起來：

# 建立節點實例 (假設上面三個類別已定義)
embed_q_node = EmbedQueryNode()
retrieve_node = RetrieveDocsNode()
generate_ans_node = GenerateAnswerNode()

# 連接節點
embed_q_node >> retrieve_node
retrieve_node >> generate_ans_node

# 建立線上查詢回答流程
online_query_flow = Flow(start=embed_q_node)

# 如何運行 (假設離線階段已完成，shared_offline_data 包含所需索引和文本塊)
# shared_online_data = {**shared_offline_data} # 複製離線階段的結果
# shared_online_data["user_question"] = "PocketFlow 如何處理批次任務？"

# online_query_flow.run(shared_online_data)
# 執行完畢後, shared_online_data["final_llm_answer"] 就是答案。

當使用者提出問題後，運行此 online_query_flow，它就會執行整個 RAG 的線上查詢和生成過程，最終在共享儲存 (Shared Store) 中提供基於知識庫的答案。

RAG 內部運作機制

RAG 在 PocketFlow 中的實現，是透過我們已經熟悉的節點 (Node)、流程 (Flow) 和共享儲存 (Shared Store) 這些核心組件協同工作來完成的。其內部運作可以概括為數據在這些組件之間的有序流動和處理。

非程式碼逐步解析（以線上流程為例）：

1. 接收問題：使用者提供一個問題，這個問題被存入共享儲存 (Shared Store)（例如，鍵為 "user_question"）。
2. 啟動流程：online_query_flow.run(shared_data) 被呼叫。
3. 查詢嵌入 (EmbedQueryNode)：
   • prep：從共享儲存 (Shared Store) 讀取 "user_question"。
   • exec：將問題文本轉換成嵌入向量。
   • post：將查詢嵌入向量存回共享儲存 (Shared Store)（例如，鍵為 "query_embedding"）。
4. 文件檢索 (RetrieveDocsNode)：
   • prep：從共享儲存 (Shared Store) 讀取 "query_embedding"、先前離線建立的 "knowledge_base_index" 和 "all_document_chunks"。
   • exec：在向量索引中搜尋與查詢嵌入最相似的文本塊的 ID，然後根據 ID 從所有文本塊中獲取實際文本。
   • post：將檢索到的相關文本塊列表存回共享儲存 (Shared Store)（例如，鍵為 "retrieved_context_chunks"）。
5. 答案生成 (GenerateAnswerNode)：
   • prep：從共享儲存 (Shared Store) 讀取原始的 "user_question" 和檢索到的 "retrieved_context_chunks"。
   • exec：建構一個包含問題和上下文的提示，呼叫 LLM 服務。
   • post：將 LLM 生成的答案存回共享儲存 (Shared Store)（例如，鍵為 "final_llm_answer"）。
6. 流程結束：所有節點執行完畢，最終答案已存儲。

序列圖 (線上流程簡化版)：

這個圖清晰地展示了線上查詢階段中，數據如何在不同節點和共享儲存 (Shared Store)之間流動，以及如何與外部的向量索引庫和 LLM 服務互動。

總結

在本章中，我們深入了解了「檢索增強生成 (RAG)」這一強大的設計模式：

• RAG 透過檢索相關資訊並將其作為上下文提供給大型語言模型 (LLM)，來顯著提升 LLM 回答問題的準確性、相關性和時效性。
• 典型的 RAG 系統包含兩個主要階段：
  • 離線索引階段：對知識文件進行分塊、嵌入並建立向量索引。
  • 線上查詢回答階段：對使用者問題進行嵌入、從索引中檢索相關文件、並基於檢索到的內容生成答案。
• PocketFlow 的節點 (Node)和流程 (Flow) 架構非常適合用來模塊化地構建 RAG 的各個步驟，並透過共享儲存 (Shared Store)來管理數據流。
• 我們學習了如何為 RAG 的每個關鍵步驟（如文件分塊、嵌入、索引、查詢嵌入、檢索和答案生成）設計相應的 PocketFlow 節點。

掌握了 RAG，您就擁有了讓您的 LLM 應用程式能夠利用特定領域知識或最新資訊的關鍵能力，使其更加智能和實用。

恭喜您完成了 PocketFlow 的核心概念教學！從基礎的節點 (Node)、共享儲存 (Shared Store) 到流程 (Flow)和工作流 (Workflow)的編排，再到進階的批次處理 (Batch)、異步處理 (Async)、代理人/智能體 (Agent)以及本章的檢索增強生成 (RAG)，您已經對 PocketFlow 的設計理念和核心功能有了全面的了解。

希望這些教學能幫助您開始使用 PocketFlow 建構屬於您自己的、強大而靈活的自動化流程和智能應用。PocketFlow 的設計旨在簡潔和易於擴展，我們鼓勵您動手實踐，探索更多可能性。祝您在 PocketFlow 的世界中旅途愉快！