建構客製化增強生成式AI知識函式庫

– 26 分鐘閱讀

當今生成式 AI 的應用越來越廣泛，但如何有效控制和管理 AI 流程仍是一大挑戰。檢索增強生成（RAG）技術提供了一個有效的解決方案，它允許開發者透過追蹤輸出來源，提升 AI 系統的可控性和可靠性。本文將探討 RAG 的核心概念，並逐步指導讀者如何從零開始建構一個根據 RAG 的知識函式庫。我們將會涵蓋向量儲存、資料分塊、索引和排序等關鍵技術，並結合 LlamaIndex、Pinecone、Deep Lake 等框架以及 OpenAI 和 Hugging Face 等生成式 AI 平台，提供理論與實踐的結合。此外，我們也會探討如何透過整合人類回饋持續改進模型準確性，並在微調成本和效益之間取得平衡。

掌握 RAG 的精髓：邁向增強生成式 AI 之路

在當今 AI 領域，設計和管理可控、可靠的多模態生成式 AI 流程是一項複雜的挑戰。本文將引領讀者探索檢索增強生成（Retrieval Augmented Generation, RAG）的精髓，並提供建構高效能、低成本的 LLM、電腦視覺和生成式 AI 系統的藍圖。

從基礎概念到複雜的實作，本文將探討 RAG 如何透過追蹤每個輸出到其來源檔案來控制和增強 AI 系統。RAG 的可追蹤性讓人類回饋得以有效整合，持續改進系統，並最大限度地減少錯誤、幻覺和偏差。本文將逐步指導讀者從零開始建構 RAG 框架，提供關於向量儲存、分塊、索引和排序的實用知識。

讀者將學習如何最佳化效能和成本、透過整合人類回饋提高模型準確性、平衡微調成本與效益，以及利用嵌入式索引知識圖譜提高準確性和檢索速度。透過 LlamaIndex、Pinecone 和 Deep Lake 等框架以及 OpenAI 和 Hugging Face 等生成式 AI 平台，讀者將獲得理論與實踐的完美結合。

讀完本文後，讀者將掌握實作智慧解決方案的技能，無論是生產、客戶服務還是任何其他專案，都能保持競爭力。

為何選擇 RAG？

本章將介紹 RAG 的基本概念，概述其對不同資料型別的適應性，並探討將 RAG 框架整合到現有 AI 平台的複雜性。透過 Python 的實務操作，讀者將學習建構多樣化的 RAG 組態，包括簡易型、進階型和模組化 RAG，為後續章節的進階應用奠定堅實基礎。

RAG 的核心優勢

RAG 的核心優勢在於其能夠將外部知識函式庫整合到 LLM 中，從而提升 LLM 的能力。相較於傳統 LLM，RAG 具有以下優勢：

  • 提升準確性：RAG 可以存取最新的資訊，避免 LLM 產生過時或不準確的內容。
  • 降低幻覺：RAG 可以追蹤答案的來源，減少 LLM 產生幻覺的可能性。
  • 增強可解釋性：RAG 可以提供答案的來源依據，增加 LLM 的可解釋性。
  • 提高效率：RAG 可以減少微調 LLM 的需求，降低訓練成本和時間。

RAG 的應用場景

RAG 的應用場景非常廣泛，包括：

  • 問答系統：建構根據特定領域知識函式庫的問答系統。
  • 聊天機器人：開發更具知識性和資訊量的聊天機器人。
  • 文字摘要：產生更準確和全面的文字摘要。
  • 程式碼生成：根據特定程式碼函式庫生成程式碼。

簡易型 RAG 的 Python 實作

以下是一個使用 LlamaIndex 和 OpenAI 建構簡易型 RAG 的 Python 範例：

from llama_index import VectorStoreIndex, SimpleDirectoryReader, LLMPredictor, ServiceContext
from langchain.chat_models import ChatOpenAI

# 載入資料
documents = SimpleDirectoryReader('./data').load_data()

# 初始化 LLM
llm_predictor = LLMPredictor(llm=ChatOpenAI(temperature=0, model_name="gpt-3.5-turbo"))

# 建立服務上下文
service_context = ServiceContext.from_defaults(llm_predictor=llm_predictor)

# 建立索引
index = VectorStoreIndex.from_documents(documents, service_context=service_context)

# 執行查詢
query_engine = index.as_query_engine()
response = query_engine.query("What is RAG?")

# 輸出結果
print(response)

內容解密：

這段程式碼展示瞭如何使用 LlamaIndex 建立一個簡易的 RAG 系統。首先，它載入指定目錄下的資料，然後初始化 OpenAI 的 LLM 模型。接著，它使用 LlamaIndex 建立向量儲存索引，並使用該索引建立查詢引擎。最後，它執行查詢並輸出結果。

RAG 嵌入向量儲存：Deep Lake 與 OpenAI 的完美結合

本章探討 RAG 驅動的生成式 AI 的複雜性，重點關注嵌入向量及其儲存解決方案。我們將探討如何使用 Activeloop Deep Lake 和 OpenAI 模型將原始資料轉換為有組織的向量儲存，並詳細介紹建立和管理嵌入的過程，這些嵌入能夠捕捉深層語義。讀者將學習如何透過將 RAG 生態系統分解成獨立的元件，從零開始建構可擴充套件的多團隊 RAG 流程。最後，讀者將具備使用複雜檢索功能處理大型資料集的能力。

深入淺出索引式 RAG：LlamaIndex、Deep Lake 與 OpenAI 的完美結合

在 AI 領域，精準、快速與透明的資訊檢索至關重要。索引式 RAG（Retrieval-Augmented Generation）技術正是為此而生。本文將探討如何利用 LlamaIndex、Deep Lake 和 OpenAI 構建一個高效與可追蹤的 RAG 流程。

多模態模組化 RAG：無人機技術的革新

生成式 AI 的應用日新月異，而多模態模組化 RAG 框架更是將其提升到新的高度。本文將開發一個不僅能處理文字資訊，還能整合先進影像識別功能的生成式 AI 系統。

圖表翻譯： 上圖展示了多模態模組化 RAG 的流程。使用者查詢首先經過多模態分析，將文字和影像資訊分別送入對應的索引。接著，融合來自兩個索引的資訊，送入生成式 AI 模型，最終輸出結果。

利用檢索增強生成技術克服生成式 AI 的限制

即使最先進的生成式 AI 模型也只能根據其訓練資料生成回應。它們無法針對訓練資料以外的資訊提供準確答案。檢索增強生成（RAG）是一種透過結合根據檢索的方法與生成模型來解決此限制的框架。它會即時從外部來源檢索相關資料，並使用此資料生成更準確與上下文相關的回應。

簡易型 RAG 的 Python 實作

from llama_index import VectorStoreIndex, SimpleDirectoryReader, LLMPredictor, ServiceContext
from langchain.chat_models import ChatOpenAI

# 載入資料
documents = SimpleDirectoryReader('./data').load_data()

# 初始化 LLM
llm_predictor = LLMPredictor(llm=ChatOpenAI(temperature=0, model_name="gpt-3.5-turbo"))

# 建立服務上下文
service_context = ServiceContext.from_defaults(llm_predictor=llm_predictor)

# 建立索引
index = VectorStoreIndex.from_documents(documents, service_context=service_context)

# 執行查詢
query_engine = index.as_query_engine()
response = query_engine.query("What is RAG?")

# 輸出結果
print(response)

內容解密：

這段程式碼展示瞭如何使用 LlamaIndex 建立一個簡易的 RAG 系統。首先，它載入指定目錄下的資料，然後初始化 OpenAI 的 LLM 模型。接著，它使用 LlamaIndex 建立向量儲存索引，並使用該索引建立查詢引擎。最後，它執行查詢並輸出結果。

RAG 與微調的巧妙平衡

RAG（Retrieval-Augmented Generation）和微調並非對立的技術，它們各有千秋。當 RAG 資料集過於龐大時，系統可能會變得笨重難以管理。反之，對於每日天氣預報、股市行情、公司新聞等動態資料，微調模型就顯得力不從心。

選擇 RAG 或微調的關鍵在於引數化和非引數化訊息的比例。經過從頭訓練或微調的模型與 RAG 的根本區別就在於此：

  • 引數化知識： 在 RAG 系統中，引數化部分指的是生成式 AI 模型透過訓練資料學習到的權重。模型的知識儲存在這些權重和偏差中。原始訓練資料被轉換成數學形式，即引數化表示。模型“記住”了從資料中學到的內容，但資料本身並未被明確儲存。
  • 非引數化知識： 與之相對，RAG 的非引數化部分涉及儲存可直接存取的資料。這意味著資料始終可用，可以隨時查詢。與知識間接嵌入權重的引數化模型不同，RAG 中的非引數化資料允許我們檢視和使用每個輸出的實際資料。

圖表翻譯：

此圖示展示了評估模型潛力後，根據 RAG 或微調的需求比例，選擇增強 RAG 或微調模型的流程。評估模型潛力是第一步，然後根據具體需求決定是增強 RAG 還是微調模型。

RAG 和微調並非互斥的。RAG 可以與微調結合使用，以提高模型的整體效率，並增強 RAG 框架內檢索和生成元件的效能。

深入 RAG 生態系統

RAG 系統的架構彈性很高，但無論如何組態，都離不開以下四個核心領域：

  • 資料： 資料來源為何？可靠性如何？資料量是否充足？是否存在版權、隱私和安全問題？
  • 儲存： 資料在處理前後將如何儲存？需要儲存多少資料？
  • 檢索： 如何檢索正確的資料來增強使用者輸入，使其滿足生成式模型的需求？哪種 RAG 框架適合專案？
  • 生成： 哪種生成式 AI 模型適合所選的 RAG 框架？

圖表翻譯：

此圖表呈現了 RAG 生態系統的核心組成部分，包括檢索器、生成器、評估器和訓練器，以及它們之間的互動流程。每個元件都有其特定的功能和子流程，共同構成了完整的 RAG 生成式 AI 流程。

檢索器 (D) 深度解析

檢索器元件負責收集、處理、儲存和檢索資料。RAG 生態系統的起點是資料攝取過程，第一步是收集資料。

  • 收集 (D1)： 現今的 AI 資料型別豐富多樣，從部落格文章中的文字片段到迷因圖，甚至是最新的熱門歌曲。資料格式也各不相同，例如 PDF、網頁、純文字檔案、JSON 檔案、MP3 音樂、MP4 影片以及 PNG 和 JPG 圖片。許多資料是非結構化的，以複雜的方式存在。許多平台（如 Pinecone、OpenAI、Chroma 和 Activeloop）提供了現成的工具來處理和儲存這些資料。
  • 處理 (D2)： 在多模態資料處理的資料收集階段 (D1)，可以使用網路爬蟲技術或其他訊息來源從網站中提取各種資料，例如文字、影像和影片。這些資料物件隨後會被轉換，以建立統一的特徵表示。例如，資料可以被分塊、嵌入（轉換成向量）和索引，以提高搜尋和檢索效率。
  • 儲存 (D3)： 在流程的這個階段，我們已經從網路上收集並開始處理大量的資料，包括影片、圖片、文字等等。向量儲存函式庫（如 Deep Lake、Pinecone 和 Chroma）可以將資料轉換成向量，並應用各種索引方法和其他技術以便快速存取。
  • 檢索查詢 (D4)： 檢索過程由使用者輸入或自動輸入 (G1) 觸發。為了快速檢索資料，我們將資料轉換成合適的格式後載入到向量儲存函式庫和資料集中。然後，使用關鍵字搜尋、嵌入和索引的組合，我們可以有效地檢索資料。例如，餘弦相似度可以找到密切相關的專案，確保搜尋結果不僅快速而與高度相關。檢索到資料後，我們會增強輸入。

理解 RAG 系統的輸入與輸出

構建高效的 RAG 系統，首先要理解其輸入、增強輸入、提示工程以及輸出之間的關係。

輸入 (G1)

RAG 系統的輸入可以是自動化任務批次（例如處理電子郵件）或透過使用者介面 (UI) 輸入的人工提示。這種靈活性讓你可以將 AI 無縫整合到各種專業環境中，提升跨產業的生產力。

結合人類回饋的增強輸入 (G2)

正如評估器章節中「人類回饋 (E2)」所述，人類回饋 (HF) 可以增加到輸入中。HF 能顯著提升 RAG 生態系統的適應性，並提供對資料檢索和生成式 AI 輸入的全面控制。在本章的「使用 Python 建構混合自適應 RAG」部分，我們將示範如何構建結合人類回饋的增強輸入。

提示工程 (G3)

檢索器 (D) 和生成器 (G) 都高度依賴提示工程來準備生成式 AI 模型需要處理的標準和增強訊息。提示工程將檢索器的輸出和使用者輸入結合在一起。

生成與輸出 (G4)

生成式 AI 模型的選擇取決於專案目標。Llama、Gemini、GPT 和其他模型可以滿足各種需求。然而，提示必須符合每個模型的規範。我們將在本文中使用的 LangChain 等框架，透過提供可適應的介面和工具，幫助簡化各種 AI 模型與應用程式的整合。

評估器：衡量 RAG 系統的效能

評估生成式 AI 模型的效能，不能僅僅依賴數學指標，還需要考量人類的實際體驗。

指標 (E1)

與任何 AI 系統一樣，模型評估離不開數學指標，例如餘弦相似度。這些指標確保檢索到的資料具有相關性和準確性。透過量化資料點之間的關係和相關性，它們為評估模型的效能和可靠性提供了堅實的基礎。

人類回饋 (E2)

無論數學指標看起來多麼充分，任何生成式 AI 系統，無論是否由 RAG 驅動，最終都無法迴避人類評估。系統是否被使用者接受、讚揚或批評，最終取決於人類的評估。

訓練器：微調 RAG 系統

標準的生成式 AI 模型使用大量的通用資料進行預訓練。之後，我們可以使用特定領域的資料對模型進行微調 (T2)。

Python 程式碼實作：原生、進階和模組化 RAG

本文將透過基礎教學範例介紹原生、進階和模組化 RAG。程式將建構關鍵字比對、向量搜尋和根據索引的檢索方法。它使用 OpenAI 的 GPT 模型，根據輸入查詢和檢索到的檔案生成回應。

基礎與基本實作

環境設定與生成器函式實作

首先，我們需要安裝必要的函式庫並進行環境設定：

# 安裝必要的函式庫
!pip install openai

# 載入 OpenAI API 金鑰
import os
os.environ["OPENAI_API_KEY"] = "your_openai_api_key"

# 載入 OpenAI 函式庫
import openai

# 設定 OpenAI API 金鑰
openai.api_key = os.getenv("OPENAI_API_KEY")

內容解密：

此段程式碼首先安裝了 openai 函式庫，然後載入 OpenAI API 金鑰並設定環境變數，最後初始化 OpenAI API 金鑰。這樣我們就可以在後續程式碼中使用 OpenAI 的 GPT 模型進行文字生成。

# 定義生成器函式，使用 GPT-4o 模型生成回應
def generate_response(prompt):
    response = openai.ChatCompletion.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=150
    )
    return response.choices[0].message.content.strip()

# 示例用法
prompt = "請介紹 RAG 的基本概念。"
response = generate_response(prompt)
print(response)

內容解密：

此段程式碼定義了一個名為 generate_response 的函式，該函式接受一個提示（prompt）作為輸入，並使用 OpenAI 的 GPT-4o 模型生成回應。函式傳回生成的文字，並在示例用法中展示瞭如何呼叫該函式來取得對特定提示的回應。

接下來，我們將繼續探索更多關於 RAG 的實作細節和進階技術。

環境設定與 OpenAI API 金鑰管理

在開始建構 RAG（Retrieval Augmented Generation）應用之前，首先需要設定適當的開發環境並取得 OpenAI 的 API 金鑰。本章節將詳細介紹如何安裝必要的套件、設定 API 金鑰，以及初始化 OpenAI 客戶端。

安裝 OpenAI 套件

要使用 OpenAI 的 GPT-4o 模型，首先需要安裝 openai 套件。建議使用特定版本以確保相容性：

!pip install openai==1.40.3

安裝完成後，務必凍結已安裝的套件版本，以避免與其他函式庫發生衝突。

取得與設定 OpenAI API 金鑰

  1. 在 OpenAI 平台上建立帳戶並取得 API 金鑰。
  2. 將 API 金鑰儲存在安全的位置，例如 Google Drive。
  3. 使用以下 Python 程式碼讀取 API 金鑰並設定環境變數：
from google.colab import drive
drive.mount('/content/drive')

# 從檔案讀取 API 金鑰
with open("drive/MyDrive/files/api_key.txt", "r") as f:
    API_KEY = f.readline().strip()

import os
import openai
os.environ['OPENAI_API_KEY'] = API_KEY
openai.api_key = os.getenv("OPENAI_API_KEY")

內容解密：

上述程式碼首先掛載 Google Drive，然後從指設定檔案讀取 API 金鑰。這種做法避免了將敏感資訊直接寫入程式碼中，提高了安全性。接著，使用 os.environ 設定環境變數，並將其指定給 openai.api_key，確保 OpenAI 客戶端能夠正確使用該金鑰。

初始化 OpenAI 客戶端

完成 API 金鑰設定後，初始化 OpenAI 客戶端：

import openai
from openai import OpenAI
import time

client = OpenAI()
gpt_model = "gpt-4o"
start_time = time.time()  # 記錄請求開始時間

內容解密：

這段程式碼匯入必要的模組並初始化 OpenAI 客戶端。time.time() 用於記錄請求的開始時間，便於後續計算處理時間。

RAG 系統核心組成部分

下圖展示了 RAG 系統的核心組成部分及其之間的關係：

圖表翻譯：

此圖表展示了 RAG 系統的整體架構，包括輸入處理、提示工程、生成輸出等關鍵步驟，以及指標評估和人類回饋如何影響系統的自適應調整。

向量搜尋與相似性度量在 RAG 中的應用

在建構 RAG 應用時，向量搜尋和相似性度量技術扮演著至關重要的角色。本章節將探討如何利用這些技術來提升 RAG 系統的效能。

資料準備與預處理

假設資料已經過收集、清理和分割，本文將重點放在如何有效地檢索這些資料。以下是一個簡化的資料範例：

db_records = [
    "Retrieval Augmented Generation (RAG) represents a sophisticated...",
    # 其他資料記錄
]

import textwrap

paragraph = ' '.join(db_records)
wrapped_text = textwrap.fill(paragraph, width=80)
print(wrapped_text)

查詢處理與使用者意圖理解

使用者輸入的查詢可能不夠精確，因此需要構建能夠處理模糊輸入的系統。例如，當使用者查詢 “define a rag store” 時，系統需要能夠正確理解其意圖。

query = "define a rag store"
llm_response = call_llm_with_full_text(query)
print_formatted_response(llm_response)

進階技術：向量相似性度量

為了評估檢索檔案的準確性，本文將探討餘弦相似性在評估文字檔案相關性中的作用。

餘弦相似性計算

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

def calculate_cosine_similarity(text1, text2):
    vectorizer = TfidfVectorizer(
        stop_words='english',
        use_idf=True,
        norm='l2',
        ngram_range=(1, 2),
        sublinear_tf=True,
        analyzer='word'
    )
    tfidf = vectorizer.fit_transform([text1, text2])
    similarity = cosine_similarity(tfidf[0:1], tfidf[1:2])
    return similarity[0][0]

內容解密：

這段程式碼使用 TfidfVectorizer 將文字轉換為 TF-IDF 特徵矩陣，並計算兩個向量之間的餘弦相似度。TF-IDF 能夠量化單詞在檔案中的重要性，而餘弦相似度則用於衡量兩個向量之間的相似程度。

向量資料函式庫在 RAG 中的應用

向量資料函式庫能夠將文字資料轉換為數值表示，從而提升搜尋效率。以下是一個簡化的流程圖，展示瞭如何計算使用者查詢與檔案之間的餘弦相似度：

圖表翻譯：

此流程圖展示了向量搜尋的核心步驟：將使用者查詢和檔案向量化，然後計算它們之間的餘弦相似度，最終得到一個量化相似程度的分數。

利用向量資料函式庫提升 RAG 效能

本章節將探討如何透過向量資料函式庫進一步提升 RAG 系統的效能。

增強型相似度計算

透過使用自然語言處理工具（如 spaCy 和 NLTK），可以更準確地捕捉詞彙之間的語義關係。以下是一個增強型相似度計算的範例：

import spacy
import nltk
nltk.download('wordnet')
from nltk.corpus import wordnet
from collections import Counter
import numpy as np

# 定義取得同義詞、預處理文字、擴充同義詞和計算增強相似度的函式
def get_synonyms(word):
    # 從 WordNet 取得同義詞
    synonyms = set()
    for syn in wordnet.synsets(word):
        for lemma in syn.lemmas():
            synonyms.add(lemma.name())
    return list(synonyms)

def preprocess_text(text):
    # 文字預處理：小寫轉換、詞形還原、去除停用詞和標點符號
    nlp = spacy.load('en_core_web_sm')
    doc = nlp(text.lower())
    tokens = [token.lemma_ for token in doc if not token.is_stop and not token.is_punct]
    return tokens

def expand_with_synonyms(words):
    # 使用同義詞擴充詞彙列表
    expanded_words = set(words)
    for word in words:
        synonyms = get_synonyms(word)
        expanded_words.update(synonyms)
    return list(expanded_words)

def calculate_enhanced_similarity(text1, text2):
    # 計算預處理和同義詞擴充後的文字向量之間的餘弦相似度
    tokens1 = preprocess_text(text1)
    tokens2 = preprocess_text(text2)
    expanded_tokens1 = expand_with_synonyms(tokens1)
    expanded_tokens2 = expand_with_synonyms(tokens2)
    
    # 結合擴充後的詞彙並計算向量表示
    all_tokens = list(set(expanded_tokens1 + expanded_tokens2))
    vector1 = np.array([token in expanded_tokens1 for token in all_tokens], dtype=int)
    vector2 = np.array([token in expanded_tokens2 for token in all_tokens], dtype=int)
    
    dot_product = np.dot(vector1, vector2)
    magnitude1 = np.linalg.norm(vector1)
    magnitude2 = np.linalg.norm(vector2)
    similarity = dot_product / (magnitude1 * magnitude2) if magnitude1 * magnitude2 != 0 else 0
    
    return similarity

內容解密：

這段程式碼透過一系列步驟提升了文字相似度計算的準確性。首先，使用 spaCy 對文字進行預處理，包括小寫轉換、詞形還原和去除停用詞。接著，利用 NLTK 的 WordNet 取得詞彙的同義詞，並擴充原始詞彙列表。最後，計算擴充後詞彙列表的向量表示，並使用餘弦相似度公式計算兩個向量之間的相似程度。

透過這些技術，RAG 系統能夠更準確地理解使用者查詢，並提供更相關的資訊。

增強型相似度在指標評估中的進步與挑戰

增強型相似度評估方法在技術指標上取得了顯著進步。然而，將 RAG（檢索增強生成）與生成式 AI 整合時，仍面臨多項挑戰。無論採用何種指標，都會遇到以下限制：

  • 查詢與檔案長度的差異： 使用者查詢通常簡短，而擷取的檔案內容更豐富、更長，這使得直接進行相似度評估變得複雜。
  • 創意擷取機制： 系統可能會創造性地選擇符合使用者預期的較長檔案，但由於意外內容的對齊，可能導致指標分數不理想。
  • 人工回饋的必要性： 通常需要人工判斷來準確評估擷取內容的相關性和有效性，因為自動化指標可能無法完全捕捉使用者的滿意度。

我們需要在數學指標和人工回饋之間找到適當的平衡點，以最佳化系統效能。

基礎 RAG 與關鍵字搜尋的結合

基礎 RAG 搭配關鍵字搜尋和比對，在組織內定義明確的檔案（如法律和醫療檔案）中，可以證明是有效的。這些檔案通常具有清晰的標題或影像標籤。在基礎 RAG 函式中，我們將實作關鍵字搜尋和比對。為此，我們將在程式碼中套用一個簡單的擷取方法：

  1. 將查詢陳述式拆分為個別關鍵字。
  2. 將資料集中的每個記錄拆分為關鍵字。
  3. 計算共同比對的關鍵字數量。
  4. 選擇得分最高的記錄。

生成方法將執行以下步驟：

  1. 使用擷取查詢的結果來擴增使用者輸入。
  2. 請求生成模型（在本例中為 gpt-4o）。
  3. 顯示生成的回應。

讓我們編寫關鍵字搜尋和比對函式。


#### 關鍵字搜尋函式實作
def find_best_match_keyword_search(query, db_records):
    # 初始化最佳分數和最佳記錄
    best_score = 0
    best_record = None
    
    # 將查詢陳述式轉換為小寫並拆分為關鍵字集合
    query_keywords = set(query.lower().split())
    
    # 遍歷資料函式庫中的每個記錄
    for record in db_records:
        # 將記錄轉換為小寫並拆分為關鍵字集合
        record_keywords = set(record.lower().split())
        
        # 計算查詢關鍵字與記錄關鍵字的交集數量
        common_keywords = query_keywords.intersection(record_keywords)
        current_score = len(common_keywords)
        
        # 如果目前分數高於最佳分數，則更新最佳分數和最佳記錄
        if current_score > best_score:
            best_score = current_score
            best_record = record
    
    # 傳回最佳分數和最佳比對記錄
    return best_score, best_record

# 假設 'query' 和 'db_records' 已在先前定義
best_keyword_score, best_matching_record = find_best_match_keyword_search(query, db_records)
print(f"最佳關鍵字比對分數：{best_keyword_score}")
print_formatted_response(best_matching_record)

內容解密：

此函式的主要目的是在資料函式庫記錄中找到與查詢陳述式最比對的記錄。它透過比較查詢關鍵字與每個記錄中的關鍵字，計算共同關鍵字的數量，並選擇具有最多共同關鍵字的記錄。函式傳回最佳比對記錄及其對應的分數。

指標評估與相似度計算

我們在本篇文章的「擷取指標」部分建立了相似度評估方法。首先，我們套用餘弦相似度來評估查詢與最佳比對記錄之間的相似程度：


#### 餘弦相似度計算
# 計算查詢與最佳比對記錄之間的餘弦相似度
score = calculate_cosine_similarity(query, best_matching_record)
print(f"最佳餘弦相似度分數：{score:.3f}")

接著，我們使用增強型相似度評估方法來獲得更好的相似度分數：


#### 增強型相似度計算
# 計算查詢與最佳比對記錄之間的增強型相似度
response = best_matching_record
print(query, ": ", response)
similarity_score = calculate_enhanced_similarity(query, response)
print(f"增強型相似度：{similarity_score:.3f}")

增強型相似度方法能夠提供更精確的相似度評估結果。

擴增輸入與生成式 AI 回應

我們將檢索到的最佳比對記錄與使用者查詢結合，形成擴增輸入，以提高生成模型的效能：


#### 擴增輸入構建
# 將查詢與最佳比對記錄結合，形成擴增輸入
augmented_input = query + ": " + best_matching_record
print_formatted_response(augmented_input)

然後，我們呼叫 GPT-4o 生成模型，並顯示生成的回應：


#### 生成式 AI 回應生成
# 使用擴增輸入呼叫 GPT-4o 生成回應
llm_response = call_llm_with_full_text(augmented_input)
print_formatted_response(llm_response)

基礎 RAG 在許多場景下表現良好，但當檔案數量龐大或內容複雜時，進階 RAG 設定將提供更優異的結果。

RAG 處理流程圖示

圖表翻譯： 上圖展示了 RAG 系統的基本處理流程。使用者查詢和檔案資料首先經過向量化處理，然後進行相似度計算，以擷取相關檔案。最後，相關檔案被送入生成式 AI 模型，生成最終的輸出結果。

圖表翻譯： 此圖展示了基礎 RAG 的處理流程。系統首先執行關鍵字搜尋，找到最佳比對記錄，然後將其與使用者查詢結合，形成擴增輸入。最後，擴增輸入被送入生成式 AI 模型，生成最終輸出結果。

向量搜尋與索引搜尋的實作比較

在本文中，我們將比較向量搜尋和索引搜尋這兩種技術的實作方式，並分析它們各自的優缺點。

向量搜尋技術

向量搜尋的核心概念是將資料集中的每個句子轉換成數值向量。透過計算查詢向量（使用者查詢）和這些檔案向量之間的餘弦相似度，我們可以快速找到最相關的檔案。


#### 向量搜尋實作
def find_best_match(text_input, records):
    # 初始化最佳分數和最佳記錄
    best_score = 0
    best_record = None
    
    # 遍歷所有記錄，計算餘弦相似度
    for record in records:
        current_score = calculate_cosine_similarity(text_input, record)
        if current_score > best_score:
            best_score = current_score
            best_record = record
    
    # 傳回最佳分數和最佳比對記錄
    return best_score, best_record

# 假設 'query' 和 'records' 已在先前定義
best_similarity_score, best_matching_record = find_best_match(query, records)
print_formatted_response(best_matching_record)

內容解密：

此函式的核心功能是遍歷資料集中的每條記錄，並計算查詢向量與每條記錄之間的餘弦相似度。函式會持續追蹤目前為止找到的最高相似度分數和對應的記錄，最終傳回最佳比對的記錄和其相似度分數。該方法在小規模資料集上表現良好，但在處理海量資料時，效率會顯著下降。

索引搜尋技術

索引搜尋則是將所有句子使用 TF-IDF（詞頻-逆檔案頻率）轉換成向量，並建立索引。TF-IDF 是一種統計方法，用於評估一個詞對於檔案集合的重要性。這些向量作為矩陣中的索引，允許快速進行相似度比較，而無需完整解析每個檔案。


#### 索引搜尋實作
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

def find_best_match(query, vectorizer, tfidf_matrix):
    # 將查詢轉換為 TF-IDF 向量
    query_tfidf = vectorizer.transform([query])
    
    # 計算查詢向量與 TF-IDF 矩陣之間的餘弦相似度
    similarities = cosine_similarity(query_tfidf, tfidf_matrix)
    
    # 取得最佳比對索引和分數
    best_index = similarities.argmax()
    best_score = similarities[0, best_index]
    
    # 傳回最佳分數和最佳比對索引
    return best_score, best_index

# 假設 'db_records' 已在先前定義，並已建立 TF-IDF 向量化器和矩陣
vectorizer, tfidf_matrix = setup_vectorizer(db_records)
best_similarity_score, best_index = find_best_match(query, vectorizer, tfidf_matrix)
best_matching_record = db_records[best_index]
print_formatted_response(best_matching_record)

內容解密：

此函式首先使用 TfidfVectorizer 將查詢轉換為 TF-IDF 向量，然後計算查詢向量與 tfidf_matrix 中所有向量之間的餘弦相似度。函式傳回最高相似度分數和對應的索引。相較於向量搜尋，索引搜尋在大規模資料集上效率更高，因為它避免了對每個檔案的完整遍歷。

特徵提取與 TF-IDF 矩陣建立

在使用索引搜尋之前，我們需要先對資料集進行特徵提取。以下程式碼片段展示瞭如何使用 setup_vectorizer 函式建立詞彙表和 TF-IDF 矩陣：


#### TF-IDF 矩陣建立
# 建立 TF-IDF 向量化器並轉換資料集為 TF-IDF 矩陣
vectorizer, tfidf_matrix = setup_vectorizer(records)
# 輸出矩陣格式（示意）
print(tfidf_matrix)

內容解密：

setup_vectorizer 函式會分析輸入的記錄，建立一個詞彙表，並將每個記錄轉換為 TF-IDF 向量。生成的 tfidf_matrix 中，每一行代表一個記錄，每一列代表詞彙表中的一個詞彙。矩陣中的值表示每個詞彙在對應記錄中的 TF-IDF 值。

擴增輸入與生成式 AI 回應

無論使用向量搜尋還是索引搜尋，我們都可以將檢索到的資訊增加到使用者查詢中，以擴增輸入，從而提高生成模型的效能。


#### 擴增輸入構建與生成式 AI 回應生成（同上）
# 將查詢與最佳比對記錄結合，形成擴增輸入，並呼叫 GPT-4o 生成回應（具體實作同前述）
augmented_input = query + ": " + best_matching_record
print_formatted_response(augmented_input)
llm_response = call_llm_with_full_text(augmented_input)
print_formatted_response(llm_response)

從原始資料到向量儲存中的嵌入

嵌入技術將任何形式的資料（文字、影像或音訊）轉換為實數，也就是將檔案轉換為向量。這些檔案的數學表示形式使我們能夠計算檔案之間的距離並檢索相似資料。

資料清理與嵌入

原始資料（書籍、文章、部落格、圖片或歌曲）首先會被收集和清理，以去除雜訊。接著，準備好的資料會被輸入到 OpenAI text-embedding-3-small 等模型中進行嵌入。例如，我們將在本文中使用的 Activeloop Deep Lake 會將文字分解成由特定字元數定義的預定義塊。例如，一個塊的大小可以是 1,000 個字元。

圖表翻譯： 此圖示展示了從原始資料到向量儲存的流程。首先，原始資料經過資料清理，接著透過 OpenAI 嵌入模型轉換為向量，最後儲存在 Activeloop Deep Lake 向量儲存中。

向量儲存的作用

一旦我們有了文字和嵌入，下一步就是有效地儲存它們以便快速檢索。這就是向量儲存的用武之地。向量儲存是一種專門設計用於處理嵌入等高維資料的資料函式庫。我們可以在 Activeloop 等無伺服器平台上建立資料集。我們可以透過 API 以程式碼方式建立和存取它們。

圖表翻譯： 此圖示說明瞭如何透過 API 建立和存取 Activeloop Deep Lake 中的向量儲存和資料集。

向量儲存的另一個特性是它們能夠使用最佳化方法檢索資料。向量儲存使用強大的索引方法構建，這種檢索能力使 RAG 模型能夠在生成階段快速找到並檢索最相關的嵌入，增強使用者輸入，並提高模型產生高品質輸出的能力。

組織 RAG Pipeline

RAG Pipeline 通常會收集資料，並透過清理、分塊檔案、嵌入檔案以及將其儲存在向量儲存資料集中來準備資料。然後，查詢向量資料集以增強生成式 AI 模型的使用者輸入，從而產生輸出。然而，在使用向量儲存時，強烈建議不要在單個程式中執行此 RAG 序列。我們應該至少將流程分成三個元件：

  1. 資料收集和準備
  2. 資料嵌入和載入到向量儲存的資料集中
  3. 查詢向量化資料集以增強生成式 AI 模型的輸入，從而產生回應

元件化方法的優勢

這種根據元件的方法有幾個主要原因：

  • 專業化： 允許團隊中的每個成員做他們最擅長的事情，無論是收集和清理資料、執行嵌入模型、管理向量儲存，還是調整生成式 AI 模型。
  • 可擴充性： 隨著技術的發展，更容易升級單獨的元件，並使用專門的方法擴充不同的元件。
  • 平行開發： 允許每個團隊按照自己的步調前進，而無需等待其他人。
  • 維護： 元件獨立。一個團隊可以在一個元件上工作，而不會影響系統的其他部分。
  • 安全性： 最大限度地減少安全問題和隱私問題，因為每個團隊可以單獨工作，每個元件都有特定的授權、存取許可權和角色。

RAG 驅動的生成式 AI Pipeline

讓我們深入瞭解一下真實的 RAG Pipeline。想像一下，我們是一個必須在幾週內交付整個系統的團隊。首先，我們需要設計一個高效且可擴充套件的架構，以滿足專案需求。

藍圖設計

  1. 資料收集與預處理： 清理和準備原始資料。
  2. 嵌入與向量儲存： 將清理後的資料嵌入並儲存在向量儲存中。
  3. 查詢與生成： 使用查詢向量化資料集，並將結果輸入生成式 AI 模型以產生最終輸出。

這種模組化設計不僅提高了開發效率，也使得後續的維護和升級變得更加容易。

建構可重複使用的檢索元件

以下程式碼示範如何建立一個 RetrievalComponent 類別，可以在專案的每個階段呼叫，執行所需的檢索任務。

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

class RetrievalComponent:
    def __init__(self, method='vector'):
        self.method = method
        if self.method == 'vector' or self.method == 'indexed':
            self.vectorizer = TfidfVectorizer()
            self.tfidf_matrix = None

    def fit(self, records):
        if self.method == 'vector' or self.method == 'indexed':
            self.tfidf_matrix = self.vectorizer.fit_transform(records)

    def retrieve(self, query):
        if self.method == 'keyword':
            return self.keyword_search(query)
        elif self.method == 'vector':
            return self.vector_search(query)
        elif self.method == 'indexed':
            return self.indexed_search(query)

    def keyword_search(self, query):
        # 實作細節
        pass

    def vector_search(self, query):
        # 實作細節
        pass

    def indexed_search(self, query):
        # 實作細節
        pass

# 使用範例
retrieval = RetrievalComponent(method='vector')
retrieval.fit(db_records)
best_matching_record = retrieval.retrieve(query)
print_formatted_response(best_matching_record)

內容解密：

這段程式碼定義了一個 RetrievalComponent 類別，用於根據不同的檢索方法（關鍵字、向量或索引）檢索最比對的記錄。它首先初始化檢索方法，並根據需要建立 TfidfVectorizer 物件。在 fit 方法中，它對輸入記錄進行向量化處理。在 retrieve 方法中，根據所選的檢索方法呼叫相應的搜尋函式（如 keyword_search、vector_search 或 indexed_search），並傳回最比對的記錄。

建構高效能 RAG Pipeline 的實務

在這個技術中，玄貓將帶領各位探討如何建構一個高效能的 Retrieval Augmented Generation (RAG) pipeline。這個流程涉及資料收集、嵌入、儲存以及最終的內容生成，每個環節都至關重要。為了提高效率，我們可以將整個 pipeline 分解成數個平行執行的模組，如同一個高效能團隊般協同合作。

分工合作，提升效率

想像一下，我們將整個 RAG pipeline 的建構過程比喻成一個團隊專案，每個團隊負責不同的任務：

  • 資料收集與準備團隊 (D4): 負責從各種來源收集資料，並進行清洗、轉換等準備工作，確保資料品質符合後續處理的需求。這個團隊的工作類別似於專案的前期調查，需要對資料有深入的理解和處理能力。
  • 資料嵌入與儲存團隊 (G1-G4): 負責將文字資料轉換成向量表示，並儲存到向量資料函式庫中。這一步驟至關重要，因為它決定了後續檢索的效率和準確性。這個團隊需要精通向量嵌入技術和向量資料函式庫的管理。
  • 增強生成團隊 (E1): 根據使用者輸入和檢索結果，生成最終的內容。這個團隊的核心是 GPT-4 等大語言模型，他們的工作是將收集到的資訊和使用者的需求整合，創造出有價值的內容。

圖表翻譯： 以上流程圖展示了 RAG pipeline 的各個階段以及團隊之間的協作關係。資料收集團隊首先對資料進行清洗和轉換，並進行品品檢查。透過檢查後，資料會交給嵌入團隊進行向量化和儲存。最後，生成團隊利用使用者輸入和檢索結果生成最終內容。

建構 RAG Pipeline 的步驟

接下來，玄貓將逐步講解如何建構 RAG pipeline，並以程式碼示例說明每個步驟的具體實作。

  1. 設定環境:

在開始之前，我們需要設定好開發環境，安裝必要的套件和函式庫。以下列出了一些常用的套件：

pip install beautifulsoup4 requests deeplake openai

內容解密： beautifulsoup4 和 requests 用於網頁資料的抓取和解析；deeplake 是一個向量資料函式庫，用於儲存和檢索向量嵌入；openai 則是用於存取 OpenAI 的 API。

  1. 掛載雲端硬碟 (適用於 Google Colab):

如果在 Google Colab 上執行程式碼，需要先掛載 Google Drive，以便存取 API 金鑰和其他檔案。

from google.colab import drive
drive.mount('/content/drive')

內容解密： 這段程式碼將 Google Drive 掛載到 Colab 環境中，方便存取儲存在雲端硬碟中的檔案。

  1. 下載必要檔案:

可以使用 curl 或其他工具從 GitHub 等平台下載所需的程式碼檔案。

  1. 驗證 API 金鑰:

設定 OpenAI 和 Activeloop 的 API 金鑰，確保程式碼可以正常存取相關服務。

資料收集和準備

資料收集和準備是 pipeline 的第一個元件。團隊 #1 將專注於此元件，如圖所示：

我們的任務是擷取和處理 10 篇涵蓋太空探索各個導向的 Wikipedia 文章。首先，我們需要匯入必要的函式庫：

import requests
from bs4 import BeautifulSoup
import re

然後，我們選擇所需的網址：

# Wikipedia 文章的網址
urls = [
    "https://en.wikipedia.org/wiki/Space_exploration",
    "https://en.wikipedia.org/wiki/Apollo_program",
    "https://en.wikipedia.org/wiki/Hubble_Space_Telescope",
    "https://en.wikipedia.org/wiki/Mars_rover",
    "https://en.wikipedia.org/wiki/International_Space_Station",
    "https://en.wikipedia.org/wiki/SpaceX",
    "https://en.wikipedia.org/wiki/Juno_(spacecraft)",
    "https://en.wikipedia.org/wiki/Voyager_program",
    "https://en.wikipedia.org/wiki/Galileo_(spacecraft)",
    "https://en.wikipedia.org/wiki/Kepler_Space_Telescope"
]

內容解密： 這段程式碼定義了一個包含多個 Wikipedia 文章網址的列表 urls。這個列表將用於後續的資料抓取和處理。

接著，我們編寫一個清理函式，用於移除數字參考：

def clean_text(content):
    # 移除通常顯示為 [1]、[2] 等的參考。
    content = re.sub(r'\[\d+\]', '', content)
    return content

內容解密： clean_text 函式使用正規表示式移除文字中的數字參照，使其更適合後續處理。

然後，我們編寫一個典型的抓取和清理函式：

def fetch_and_clean(url):
    response = requests.get(url)
    soup = BeautifulSoup(response.content, 'html.parser')
    content = soup.find('div', {'class': 'mw-parser-output'})
    # 清理提取的內容
    cleaned_content = clean_text(content.get_text())
    return cleaned_content

內容解密： fetch_and_clean 函式首先使用 requests 取得指定 URL 的內容，然後使用 BeautifulSoup 解析 HTML，從中提取主要內容並進行清理。

結語

建構一個高效能的 RAG pipeline 需要仔細的規劃和執行。透過團隊合作和模組化設計，可以有效提升開發效率，並確保系統的穩定性和可維護性。希望這個能幫助你更好地理解 RAG pipeline 的建構過程，並在實務中應用。

資料處理 Pipeline 的第二部分：資料嵌入和儲存

在團隊 #1 完成資料收集和準備工作後，團隊 #2 將接手處理資料嵌入和儲存的任務。這個過程涉及將清理後的資料進行分塊、嵌入，並儲存到向量儲存函式庫中。

資料嵌入和儲存流程

首先，讓我們回顧一下資料處理 pipeline 的流程：

@startuml
skinparam backgroundColor #FEFEFE
skinparam componentStyle rectangle

title 建構客製化增強生成式AI知識函式庫

package "NumPy 陣列操作" {
    package "陣列建立" {
        component [ndarray] as arr
        component [zeros/ones] as init
        component [arange/linspace] as range
    }

    package "陣列操作" {
        component [索引切片] as slice
        component [形狀變換 reshape] as reshape
        component [堆疊 stack/concat] as stack
        component [廣播 broadcasting] as broadcast
    }

    package "數學運算" {
        component [元素運算] as element
        component [矩陣運算] as matrix
        component [統計函數] as stats
        component [線性代數] as linalg
    }
}

arr --> slice : 存取元素
arr --> reshape : 改變形狀
arr --> broadcast : 自動擴展
arr --> element : +, -, *, /
arr --> matrix : dot, matmul
arr --> stats : mean, std, sum
arr --> linalg : inv, eig, svd

note right of broadcast
  不同形狀陣列
  自動對齊運算
end note

@enduml

圖表翻譯： 此圖表展示了資料處理 pipeline 的兩個主要階段：資料收集與準備，以及資料嵌入與儲存。第一階段的輸出將作為第二階段的輸入。

擷取一批準備好的檔案

團隊 #2 首先需要從伺服器上下載由團隊 #1 提供的一批檔案。假設這批檔案與太空探索相關：

from grequests import download

source_text = "llm.txt"
directory = "Chapter02"
filename = "llm.txt"
download(directory, filename)

內容解密：

這段程式碼使用 grequests 函式庫中的 download 函式從指定位置下載 llm.txt 檔案。變數 source_text 儲存了檔案名稱，而 directory 和 filename 變數則指定了下載目錄和檔案名稱。

下載完成後，我們需要驗證檔案內容以確保正確性：

# 開啟檔案並讀取前 20 行
with open('llm.txt', 'r', encoding='utf-8') as file:
    lines = file.readlines()
# 列印前 20 行
for line in lines[:20]:
    print(line.strip())

內容解密：

這段程式碼讀取並列印 llm.txt 檔案的前 20 行，用於快速驗證檔案內容是否正確無誤。

資料分塊

驗證完檔案內容後，下一步是對資料進行分塊。我們根據字元數來決定塊大小，在這個例子中，CHUNK_SIZE 被設定為 1000。不同的策略可以用於選擇合適的塊大小。

CHUNK_SIZE = 1000

def chunk_data(data):
    chunks = []
    for i in range(0, len(data), CHUNK_SIZE):
        chunks.append(data[i:i + CHUNK_SIZE])
    return chunks

with open('llm.txt', 'r', encoding='utf-8') as file:
    data = file.read()
chunks = chunk_data(data)

內容解密：

這段程式碼定義了一個 chunk_data 函式，用於將輸入的資料按照指定的 CHUNK_SIZE 分割成多個塊。然後，它讀取 llm.txt 檔案的內容，並將其分成多個塊。

資料嵌入

接下來，我們需要對這些資料塊進行嵌入處理。這通常涉及使用某種嵌入模型將文字轉換為向量表示。

import numpy as np
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')

def embed_data(chunks):
    embeddings = model.encode(chunks)
    return embeddings

embeddings = embed_data(chunks)

內容解密：

這段程式碼使用 sentence-transformers 函式庫中的預訓練模型 all-MiniLM-L6-v2 將文字塊轉換為向量表示。embed_data 函式接受文字塊列表，並傳回對應的嵌入向量列表。

資料儲存

最後一步是將這些嵌入向量儲存到向量儲存函式庫中，例如 FAISS 或 Annoy。

import faiss

index = faiss.IndexFlatL2(embeddings.shape[1])
index.add(embeddings)

faiss.write_index(index, "embeddings.index")

內容解密：

這段程式碼使用 FAISS 函式庫建立一個平面索引（Flat Index），並將嵌入向量新增到索引中。最後，它將索引儲存到名為 embeddings.index 的檔案中。

透過這些步驟，團隊 #2 成功地完成了資料嵌入和儲存的任務，為後續的查詢和檢索工作奠定了基礎。

玄貓 BlackCat

玄貓 BlackCat

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。