RAG 管道建構：模組化設計與向量資料函式庫應用

– 15 分鐘閱讀

在現代技術發展中，向量資料函式庫與生成式 AI 模型的整合已成為提升應用效能的關鍵。本文將深入探討如何建構一個模組化的 RAG（Retrieval Augmented Generation）管道，藉由向量資料函式庫有效增強生成式 AI 模型的輸入，進而產生更優質的輸出。此管道設計的核心概念在於將資料收集、準備、嵌入、儲存以及查詢等步驟拆解為獨立模組，提升整體系統的彈性與擴充套件性。透過此架構，不同團隊可專注於各自專業領域，例如資料處理、模型訓練或向量資料函式倉管理，有效提升開發效率。文章將逐步說明每個模組的實作細節，包含程式碼範例與流程圖，並以太空探索相關文章為例，示範如何實際應用 RAG 管道進行資料處理與分析。此外，文章也涵蓋了環境設定、套件安裝、版本控制等實務議題，提供開發者更全面的參考。

RAG 管道的組織

RAG 管道通常會收集資料，準備它（例如，分塊檔案、嵌入它們、將它們儲存在向量儲存的資料集中），然後查詢向量化的資料集以增強使用者輸入，從而產生輸出。然而，強烈建議不要在使用向量儲存時執行這個 RAG 序列。至少應該將過程分為三個元件：

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

這種方法使得 RAG 管道更具模組化和可擴充套件性，從而提高了整個系統的效率和效能。

內容解密：

上述過程涉及多個步驟，包括資料收集、嵌入、儲存和查詢。每個步驟都對整個 RAG 管道的效能和效率至關重要。透過使用向量儲存和 RAG 管道，我們可以實作高品質的輸出和增強使用者經驗。

圖表翻譯：

以下是 RAG 管道的簡化流程圖：

  flowchart TD
    A[資料收集] --> B[資料嵌入]
    B --> C[儲存到向量儲存]
    C --> D[查詢向量化資料集]
    D --> E[增強使用者輸入]
    E --> F[產生輸出]

這個流程圖展示了 RAG 管道的主要步驟，從資料收集到產生輸出。每個步驟都對整個管道的效能和效率至關重要。

建立一個根據RAG的生成式AI管道

在實際生產環境或大型專案中，很少有一個單一的程式或團隊能夠管理端對端的過程。因此，我們需要將專案分解為多個元件，以便不同的團隊可以平行工作。

RAG管道的優點

使用RAG管道的優點包括：

  • 專業化：每個團隊可以專注於自己最擅長的部分，例如資料收集、嵌入模型、向量儲存或生成式AI模型。
  • 可擴充套件性：每個元件可以獨立升級和擴充套件，從而提高整個系統的可擴充套件性。
  • 平行開發：每個團隊可以在不影響其他團隊的情況下獨立開發自己的元件。
  • 維護：每個團隊可以獨立維護自己的元件，而不會影響其他部分的系統。

RAG管道的組成

一個典型的RAG管道由三個主要元件組成：

  1. 資料收集和準備（Data Collection and Prep）：負責收集和清理資料。
  2. 資料嵌入和儲存（Data Embedding and Storage）：負責將資料嵌入到向量空間中並儲存在向量資料函式庫中。
  3. 增強生成（Augmented Generation）：負責根據使用者輸入和檢索查詢生成內容。

建立RAG管道

要建立一個RAG管道，我們需要設定環境，然後實作每個元件。首先，我們需要安裝必要的包和依賴項。然後，我們可以開始實作每個元件。

設定環境

設定環境是建立RAG管道的第一步。這涉及安裝必要的包和依賴項。由於不同的包和依賴項可能有衝突的版本，因此我們需要小心地選擇正確的版本。

實作元件

一旦環境設定完成，我們就可以開始實作每個元件。這涉及實作資料收集和準備、資料嵌入和儲存以及增強生成等功能。

安裝套件和函式庫

為了建立本文的 RAG 管線，我們需要安裝一些套件和凍結套件版本，以防止相依性衝突和函式庫問題，例如： 可能的版本衝突。 當一個函式庫需要更新以便應用程式執行時可能發生的衝突。例如，在 2024 年 8 月，安裝 Deep Lake 需要 Pillow 版本 10.x.x，但 Google Colab 的版本是 9.x.x。因此，需要解除安裝 Pillow 並重新安裝最新版本後才能安裝 Deep Lake。Google Colab 無疑會更新 Pillow。許多此類別情況發生在快速變化的市場中。 如果版本凍結太久，可能發生的淘汰問題。 如果版本凍結太久且錯誤未被 玄貓 糾正，可能發生的問題。 因此，如果我們凍結版本，應用程式可能會保持穩定一段時間，但之後可能會遇到問題。但如果我們升級版本太快，其他一些函式庫可能就不會再工作了。沒有銀彈！這是一個持續的品質控制過程。 對於我們的程式，在本文中，我們將凍結版本。現在讓我們進行安裝步驟，以建立管線的環境。

安裝過程中的元件

讓我們從 玄貓 開始。這些元件不一定安裝在所有筆記本中；本文作為套件的庫存。 在第一個管線節（1. 資料收集和準備），我們只需要安裝 Beautiful Soup 和 Requests：

!pip install beautifulsoup4==4.12.3
!pip install requests==2.31.0

這解釋了為什麼管線的這個元件應該保持分離。對於喜歡建立與網頁互動介面的開發人員來說，這是一項簡單的工作。對於想要參與資料收集和分析的初級開發人員來說，這也是一個很好的匹配。 我們將在本文中建立的管線的另外兩個元件（2. 資料嵌入和儲存以及 3. 增強生成），將需要更多關注以及安裝 requirements01.txt，如前一節所述。現在，讓我們繼續安裝步驟，由 玄貓 進行。

掛載磁碟機

在這個場景中，程式在 Google Colab 中掛載 Google Drive，以安全地讀取 OpenAI API 金鑰以存取 OpenAI 模型和 Activeloop API 權杖以驗證存取 Activeloop Deep Lake 資料集：

# Google Drive 選項以儲存 API 金鑰
# 將您的金鑰儲存在檔案中並讀取（您可以直接輸入）
drive.mount('/content/drive')

您可以選擇在其他地方儲存您的金鑰和權杖。只要確保它們位於安全位置。

建立子程式下載 GitHub 檔案

這裡的目標是編寫一個函式以從 GitHub 下載 grequests.py 檔案。這個程式包含一個使用 curl 下載檔案的函式，並提供新增私人權杖的選項：

使用Python下載檔案並驗證下載過程

在進行檔案下載的過程中，使用Python的subprocess模組可以呼叫系統的命令，例如使用curl命令下載檔案。以下範例展示瞭如何下載檔案並處理下載過程中的異常。

基本下載命令

首先，我們需要準備好要下載的檔案URL和檔案名稱。然後，使用subprocess模組執行curl命令進行下載。

import subprocess

output_file = "grequests.py"
url = "https://example.com/grequests.py"  # 請替換為實際的URL

# 準備curl命令
curl_command = [
    "curl",
    "-o", output_file,
    url
]

try:
    # 執行curl命令
    subprocess.run(curl_command, check=True)
    print("下載成功。")
except subprocess.CalledProcessError:
    print("下載失敗。")

增加私人權杖的下載命令

如果需要使用私人權杖（private token）進行認證，則需要在curl命令中新增適當的標頭。

import subprocess

def download_file(directory, filename, private_token=None):
    base_url = "https://example.com/"  # 請替換為實際的基礎URL
    file_url = f"{base_url}{directory}/{filename}"

    try:
        if private_token:
            # 如果提供了私人權杖，則在curl命令中新增Authorization標頭
            curl_command = f"curl -H 'Authorization: token {private_token}' -o {filename} {file_url}"
        else:
            curl_command = f"curl -o {filename} {file_url}"

        # 執行curl命令
        subprocess.run(curl_command, check=True, shell=True)
        print(f"下載'{filename}'成功。")
    except subprocess.CalledProcessError:
        print(f"下載'{filename}'失敗。請檢查URL或權杖。")

# 使用範例
download_file("path/to/directory", "filename.py", "your_private_token")

圖表翻譯：下載過程流程圖

  flowchart TD
    A[開始] --> B[準備curl命令]
    B --> C[執行curl命令]
    C --> D{下載是否成功?}
    D -->|是| E[印出下載成功訊息]
    D -->|否| F[印出下載失敗訊息]
    E --> G[結束]
    F --> G

內容解密：下載命令的作用與邏輯

上述範例中，使用subprocess模組執行curl命令進行檔案下載。當提供私人權杖時，會在curl命令中新增適當的Authorization標頭，以便進行認證。下載過程中，如果遇到任何異常，會捕捉並處理相關的異常訊息。

在實際應用中，需要根據具體情況替換URL、檔案名稱和私人權杖等引數，以確保下載過程的正確性和安全性。

安裝需求

在開始使用 Activeloop Deep Lake 和 OpenAI 之前，我們需要安裝一些必要的套件。以下是所需的安裝命令：

!pip install deeplake==3.9.18
!pip install openai==1.40.3

請注意，截至 2024 年 8 月，Google Colab 的 Pillow 版本與 deeplake 套件有衝突。但是，deeplake 的安裝套件已經自動處理了這個問題。因此，您只需要重新啟動會話並再次執行安裝命令即可。

啟用 Activeloop 公共 DNS 伺服器

安裝完畢後，我們需要執行一行程式碼來啟用 Activeloop 的公共 DNS 伺服器：

# 對於 Google Colab 和 Activeloop（Deeplake 函式庫）
# 這行程式碼將字串 "nameserver 8.8.8.8" 寫入檔案中。
# 應該使用的是 IP 地址 8.8.8.8，這是 Google 的其中一個 DNS 伺服器。
file.write("nameserver 8.8.8.8")

驗證流程

要使用 OpenAI，您需要註冊並取得 API 金鑰。以下是如何啟用 OpenAI API 金鑰的步驟：

# 取得和設定 OpenAI API 金鑰
f = open("drive/MyDrive/files/api_key.txt", "r")
API_KEY = f.readline().strip()
f.close()

# 設定 OpenAI API 金鑰
import os
import openai

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

接下來，我們需要啟用 Activeloop 的 API 權杖以使用 Deep Lake：

# 取得和設定 Activeloop API 權杖
f = open("drive/MyDrive/files/activeloop.txt", "r")
API_token = f.readline().strip()
f.close()

內容解密：

上述程式碼片段展示瞭如何安裝必要的套件、啟用 Activeloop 公共 DNS 伺服器、以及設定 OpenAI 和 Activeloop 的 API 金鑰和權杖。這些步驟是使用 Activeloop Deep Lake 和 OpenAI 的必要條件。

玄貓的資料收集與準備之旅

在開始資料收集與準備的旅程中，我們需要先設定環境變數 ACTIVELOOP_TOKEN，以便使用 Activeloop 的 API。這個步驟非常重要，因為它確保我們可以順暢地存取所需的資料。

import os

# 設定環境變數
ACTIVELOOP_TOKEN = "您的 API Token"
os.environ['ACTIVELOOP_TOKEN'] = ACTIVELOOP_TOKEN

在取得 API Token 之後，我們就可以開始著手資料收集與準備的工作。這個過程包括從各個來源收集資料，例如維基百科文章，並對這些資料進行處理，以便在後續的分析中使用。

資料收集

我們將從維基百科收集 10 篇文章，涵蓋了太空探索的各個方面，包括：

  1. 太空探索概覽：介紹太空探索的歷史、技術、任務和計劃。
  2. 阿波羅計畫：詳細介紹NASA的阿波羅計畫，包括登入月球的人類首次任務。
  3. 哈伯太空望遠鏡：關於哈伯太空望遠鏡的資訊，包括其重要性和在天文發現中的作用。
  4. 火星探測車：關於已經被送往火星以研究其表面和環境的探測車。
  5. 國際太空站（ISS）：詳細介紹國際太空站，包括其建造、國際合作以及在太空研究中的角色。
  6. SpaceX：涵蓋SpaceX的歷史、成就和目標，這是一家對私人太空飛行產生重大影響的公司。
  7. 朱諾號太空船：關於NASA的朱諾號太空船，該船環繞木星執行並研究木星及其衛星。
  8. 旅行者計畫：關於旅行者任務的詳細資訊，包括它們對我們理解外太陽系和星際空間的貢獻。
  9. 伽利略號太空船：關於研究木星及其衛星的伽利略號任務的概覽。
  10. 開普勒太空望遠鏡：關於開普勒太空望遠鏡的資訊，該望遠鏡旨在發現環繞其他恆星的類別地行星。

資料處理

收集到資料後，我們需要對其進行處理，以確保它們適合用於後續的分析和模型訓練。這個步驟可能包括資料清理、轉換和特徵工程等工作。

import pandas as pd

# 載入資料
data = pd.read_csv("your_data.csv")

# 資料清理和轉換
data = data.dropna()  # 移除缺失值
data = data.apply(pd.to_numeric, errors='coerce')  # 將資料轉換為數值型別

# 特徵工程
# 在這裡進行特徵工程，例如提取有用的特徵、建立新特徵等
內容解密：

在上述過程中，我們使用了 Python 的 os 和 pandas 函式庫來設定環境變數和處理資料。設定環境變數 ACTIVELOOP_TOKEN 是使用 Activeloop API 的必要步驟，而使用 pandas 函式庫可以方便地進行資料操作和分析。

圖表翻譯：

以下是使用 Mermaid 語法繪製的資料收集與準備流程圖：

  flowchart TD
    A[設定環境變數] --> B[收集資料]
    B --> C[資料處理]
    C --> D[模型訓練]
    D --> E[模型評估]

這個圖表展示了從設定環境變數開始，到收集資料、資料處理、模型訓練和模型評估的整個流程。每一步驟都對應著特定的工作，最終目的是完成一個可靠的機器學習模型。

太空探索文章資料收集與準備

在進行太空探索文章分析之前，我們需要收集和準備相關資料。這個過程涉及從網際網路上抓取文章內容、清理和格式化資料，以便於後續分析。

收集資料

首先，我們需要定義要收集的文章來源。在這個例子中，我們使用維基百科的太空探索相關文章。為了簡化資料收集過程，我們可以使用Python的requests和BeautifulSoup函式庫來抓取和解析HTML內容。

import requests
from bs4 import BeautifulSoup
import re

# 定義要收集的文章URL
urls = [
    # 在這裡新增要收集的文章URL
]

# 定義一個函式來抓取和清理文章內容
def fetch_and_clean(url):
    # 傳送HTTP請求並取得回應
    response = requests.get(url)
    
    # 解析HTML內容
    soup = BeautifulSoup(response.content, 'html.parser')
    
    # 找到文章的主要內容
    content = soup.find('div', {'class': 'mw-parser-output'})
    
    # 移除參考文獻和外部連結等不需要的部分
    for section_title in ['References', 'Bibliography', 'External links']:
        section = content.find('span', id=section_title)
        if section:
            for sib in section.parent.find_next_siblings():
                sib.decompose()
    
    # 清理文章內容，移除數字參考等
    cleaned_content = clean_text(content.get_text())
    
    return cleaned_content

# 定義一個函式來清理文章內容
def clean_text(content):
    # 移除數字參考，如[1]、[2]等
    content = re.sub(r'\[\d+\]', '', content)
    
    return content

# 收集和清理所有文章內容
articles = [fetch_and_clean(url) for url in urls]

準備資料

收集和清理完文章內容後，我們可以將其儲存為適合分析的格式，例如JSON或CSV檔案，以便於後續分析。

import json

# 將文章內容儲存為JSON檔案
with open('articles.json', 'w', encoding='utf-8') as f:
    json.dump(articles, f, ensure_ascii=False)

資料預處理與儲存

在資料科學和人工智慧應用中，資料的預處理和儲存是非常重要的步驟。這一步驟不僅確保了資料的品質，也為後續的分析和模型訓練提供了基礎。以下是關於資料預處理和儲存的詳細介紹。

資料預處理

資料預處理是指對原始資料進行清洗、轉換和篩選，以使其變得適合分析和模型訓練的過程。這個過程通常包括以下步驟：

  1. 資料清洗：移除資料中的錯誤、重複和遺失值。
  2. 資料轉換：將資料轉換為適合分析和模型訓練的格式。
  3. 資料篩選：根據特定的條件篩選出有用的資料。

資料儲存

資料儲存是指將預處理後的資料儲存在適合的儲存系統中，以便於後續的分析和模型訓練。常用的儲存系統包括關係式資料函式庫、NoSQL資料函式庫和雲端儲存服務。

實作資料預處理和儲存

以下是使用Python實作資料預處理和儲存的範例：

import requests
from bs4 import BeautifulSoup
import re

# 下載網頁內容
url = "https://example.com"
response = requests.get(url)
soup = BeautifulSoup(response.content, "html.parser")

# 清洗和轉換資料
text = soup.get_text()
text = re.sub(r"\s+", " ", text)

# 儲存資料
with open("data.txt", "w") as f:
    f.write(text)

這個範例下載了一個網頁的內容，清洗和轉換了資料，然後儲存了資料在一個文字檔中。

使用玄貓的資料環境

玄貓的資料環境提供了一個方便的方式來管理和儲存資料。使用玄貓的資料環境，可以輕鬆地實作資料預處理和儲存，並且可以與其他工具和服務整合。

檔案準備與向量儲存建立

在開始探索空間、行星和衛星的旅程之前，我們需要準備好資料。首先，我們定義了源檔案的路徑和名稱，分別為 directory 和 filename。然後，我們下載了名為 llm.txt 的檔案到指定的目錄中。

source_text = "llm.txt"
directory = "Chapter02"
filename = "llm.txt"

# 下載檔案到指定目錄
download(directory, filename)

為了確保檔案的正確性，我們開啟檔案並讀取了前20行的內容。

# 開啟檔案並讀取前20行
with open('llm.txt', 'r', encoding='utf-8') as file:
    lines = file.readlines()

# 列印前20行
for line in lines[:20]:
    print(line.strip())

接下來，我們需要將檔案進行分塊（chunking），以最佳化資料處理。這裡，我們定義了塊大小（CHUNK_SIZE）為1000，但在後面的章節中，我們會進一步探討自動化的分塊策略。

# 定義塊大小
CHUNK_SIZE = 1000

# 開啟檔案並讀取內容
with open(source_text, 'r') as f:
    text = f.read()

# 對檔案進行分塊
chunked_text = [text[i:i+CHUNK_SIZE] for i in range(0, len(text), CHUNK_SIZE)]

現在，我們已經準備好建立一個向量儲存（vector store）來向量化資料或新增到現有的儲存中。首先，我們需要定義向量儲存的路徑。

# 定義向量儲存路徑
vector_store_path = "hub://denis76/space_exploration_v1"

確保向量儲存存在，如果不存在則建立它。這一步驟對於後續的資料處理和查詢至關重要。

內容解密：

  • download 函式用於下載指定的檔案到目錄中。
  • open 函式用於開啟檔案，readlines 方法讀取檔案的所有行。
  • CHUNK_SIZE 變數定義了分塊的大小，影響著資料處理的效率。
  • chunked_text 列表包含了分塊後的檔案內容，每個塊的大小為 CHUNK_SIZE。
  • vector_store_path 變數定義了向量儲存的路徑，用於儲存和管理向量化的資料。

圖表翻譯：

  flowchart TD
    A[開始] --> B[下載檔案]
    B --> C[開啟檔案]
    C --> D[讀取檔案內容]
    D --> E[分塊檔案]
    E --> F[建立向量儲存]
    F --> G[向量化資料]

此圖表展示了從下載檔案到建立向量儲存的整個過程，包括開啟檔案、讀取內容、分塊檔案和最終建立向量儲存。每一步驟都對應著特定的操作，最終實作了資料的向量化和儲存。

建立向量儲存和嵌入函式

在本文中，我們將建立一個向量儲存並定義一個嵌入函式，以便將文字資料轉換為向量。

建立向量儲存

首先，我們需要建立一個向量儲存。如果它不存在，我們將自動建立一個。以下是建立向量儲存的程式碼：

import deeplake.util

vector_store_path = "hub://denis76/space_exploration_v1"

try:
    # 嘗試載入向量儲存
    vector_store = VectorStore(path=vector_store_path)
    print("向量儲存已存在")
except FileNotFoundError:
    print("向量儲存不存在。您可以建立它。")
    create_vector_store = True

如果向量儲存不存在，我們將建立一個新的向量儲存。

定義嵌入函式

接下來，我們需要定義一個嵌入函式，以便將文字資料轉換為向量。在這個例子中，我們使用的是 OpenAI 的 “text-embedding-3-small” 嵌入模型。

def embedding_function(texts, model="text-embedding-3-small"):
    if isinstance(texts, str):
        texts = [texts]
    texts = [t.replace("\n", " ") for t in texts]
    return [data.embedding for data in openai.embeddings.create(i)]

這個嵌入函式接受一個文字列表作為輸入，並傳回一個向量列表。

新增資料到向量儲存

現在，我們可以開始新增資料到向量儲存了。首先，我們需要設定新增資料的旗標：

add_to_vector_store = True

接下來，我們將使用嵌入函式將文字資料轉換為向量，並新增到向量儲存中。

內容解密：

在上面的程式碼中，我們使用了 VectorStore 類別來建立和載入向量儲存。 embedding_function 函式使用 OpenAI 的嵌入模型將文字資料轉換為向量。 add_to_vector_store 旗標用於控制是否新增資料到向量儲存中。

圖表翻譯：

以下是建立向量儲存和嵌入函式的流程圖：

  flowchart TD
    A[建立向量儲存] --> B[定義嵌入函式]
    B --> C[新增資料到向量儲存]
    C --> D[傳回向量列表]

這個流程圖展示了建立向量儲存、定義嵌入函式和新增資料到向量儲存的過程。

文字向量儲存與查詢

在自然語言處理（NLP）中，文字向量儲存是一種將文字轉換為數值向量的技術，以便於電腦理解和處理。這種技術在資訊檢索、文字分類別、情感分析等領域中具有重要的應用價值。

從技術架構視角來看，本文深入淺出地介紹了構建根據 RAG 的生成式 AI 管道的核心概念和實務步驟，涵蓋了資料收集、準備、嵌入、儲存和查詢等關鍵環節。透過模組化設計，將複雜的 RAG 管道分解成三個獨立的元件：資料收集和準備、資料嵌入和儲存、以及增強生成，有效提升了系統的可擴充套件性和可維護性，也方便不同團隊的協同開發。然而，文章並未深入探討不同向量資料函式庫的效能比較和選型策略，例如 Pinecone、Weaviate、Faiss 等，這對於實際應用中的技術選型至關重要。此外，自動化分塊策略的細節也略顯不足，這對於處理大規模文字資料的效率提升至關重要。展望未來，隨著向量資料函式庫技術的持續發展和更多自動化工具的出現，根據 RAG 的生成式 AI 管道將更加便捷高效，並在更多領域展現其應用價值。對於追求高效能和可擴充套件性的 AI 應用開發者而言，深入理解和掌握 RAG 管道的構建方法至關重要。

玄貓

玄貓

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