RAG向量儲存結合DeepLake與OpenAI實作

隨著資料量的增長，向量儲存技術在生成式 AI 專案中扮演著越來越重要的角色。相較於傳統關鍵字搜尋，嵌入技術能捕捉更深層的語義關聯，提供更精準的檢索結果。本文將探討如何利用 Activeloop Deep Lake 建立向量儲存，並結合 OpenAI 的嵌入模型，開發高效的 RAG 管道。實務上，我們會將整個流程拆分成資料收集與準備、嵌入生成與儲存、以及查詢與增強三個元件，讓團隊得以分工合作，提升開發效率。同時，我們也會探討跨平台環境下的套件安裝與依賴性管理，確保專案的穩定性。

RAG Embedding Vector Stores：結合 Deep Lake 與 OpenAI 的實作

在實作 RAG 驅動的生成式 AI 專案時，我們不可避免地會遇到複雜度增加的問題。嵌入（Embeddings）技術能夠將龐大且結構化或非結構化的文字轉換為精簡且高維度的向量，這些向量捕捉了文字的語義精髓，使得資訊檢索變得更快、更有效率。然而，隨著資料集的擴增，建立和儲存檔案嵌入（Document Embeddings）成為了一項挑戰。

為何選擇嵌入而非關鍵字？

雖然嵌入需要更多的儲存空間，但它們能夠捕捉文字更深層的語義含義，提供比剛性且經常匹配的關鍵字更細緻和上下文感知的檢索結果。因此，我們選擇使用向量儲存（Vector Stores）來組織和快速存取嵌入。

從原始資料到向量儲存的旅程

嵌入技術的本質

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

處理流程

1. 資料收集與清理：收集原始資料（例如書籍、文章、部落格、圖片或歌曲），並清理以去除雜訊。
2. 嵌入模型：將準備好的資料輸入嵌入模型，如 OpenAI 的 text-embedding-3-small，以生成嵌入。
3. 向量儲存：使用 Activeloop Deep Lake 等向量儲存解決方案，將文字分解為預定義的區塊（Chunks），並將這些區塊的嵌入儲存在向量資料函式庫中。

RAG 管道的組織

一個典型的 RAG 管道包括資料收集、處理、檢索和增強輸入生成等步驟。為了提高開發效率，建議將 RAG 管道分解為獨立的元件，以便多個團隊可以同時進行開發。

管道元件

1. 資料收集與處理：收集和清理資料，將資料分解為適當大小的區塊，並生成嵌入。
2. 向量儲存管理：將生成的嵌入儲存在向量儲存中，並對其進行索引以便快速檢索。
3. 查詢與增強：查詢向量儲存以檢索相關的嵌入，增強使用者輸入，並將增強後的輸入提供給生成式 AI 模型。

實作 RAG 管道

本章節將使用 Python、Activeloop Deep Lake 和 OpenAI，從零開始建立一個包含三個元件的 RAG 管道。我們將面臨跨平台環境中的套件和依賴性問題，以及資料分塊、嵌入向量生成和向量儲存載入等挑戰。

程式碼實作範例

import os
import numpy as np
from deeplake import Dataset
from openai import OpenAI

# 初始化 OpenAI 使用者端
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# 定義嵌入函式
def generate_embedding(text):
    response = client.embeddings.create(
        input=text,
        model="text-embedding-3-small"
    )
    return response.data[0].embedding

# 載入資料集並生成嵌入
def process_data(data_path):
    dataset = Dataset(data_path)
    embeddings = []
    for data in dataset:
        text = data["text"]
        embedding = generate_embedding(text)
        embeddings.append(embedding)
    return embeddings

# 將嵌入儲存在向量儲存中
def store_embeddings(embeddings, store_path):
    # 省略實作細節
    pass

# 查詢向量儲存
def query_vector_store(query, store_path):
    # 省略實作細節
    pass

# 主程式
if __name__ == "__main__":
    data_path = "path/to/your/dataset"
    store_path = "path/to/your/vector/store"
    
    embeddings = process_data(data_path)
    store_embeddings(embeddings, store_path)
    
    query = "你的查詢內容"
    results = query_vector_store(query, store_path)
    print(results)

內容解密：

1. 嵌入生成：使用 OpenAI 的 text-embedding-3-small 模型生成文字的嵌入向量。
2. 資料處理：將原始資料集中的文字資料提取出來，並生成對應的嵌入向量。
3. 向量儲存管理：將生成的嵌入向量儲存在 Activeloop Deep Lake 向量儲存中。
4. 查詢：根據使用者輸入的查詢內容，檢索向量儲存中的相關嵌入向量。

圖表翻譯：

此圖示呈現了 RAG 管道的整體架構，包括資料收集、嵌入生成、向量儲存管理和查詢增強等步驟。

建構RAG流程的必要性與優勢

在運用向量儲存（vector store）的過程中，將流程劃分為三個主要元件是至關重要的：

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

元件化方法的優勢

1. 專業分工：使團隊成員能夠專注於其擅長的領域，如資料收集與清理、執行嵌入模型、管理向量儲存或調整生成式AI模型。
2. 可擴充套件性：便於隨著技術演進升級各個元件，並使用專門的方法擴充套件不同的元件。例如，原始資料的儲存可以在不同的伺服器上擴充套件，而嵌入向量則儲存在雲端平台的向量化資料集中。
3. 平行開發：允許各團隊按照自己的節奏推進，而不必等待其他團隊。可以在一個元件上持續改進，而不會干擾其他元件的流程。
4. 獨立維護：各元件的維護是獨立的，一個團隊可以專注於一個元件的改進，而不會影響系統的其他部分。例如，當RAG流程投入生產時，使用者可以繼續查詢並執行生成式AI，同時團隊修復資料收集元件。
5. 最小化安全與隱私問題：由於各團隊可以獨立工作，並根據每個元件的具體授權、存取許可權和角色進行工作，從而最小化安全與隱私問題。

RAG驅動的生成式AI流程

在實際的生產環境或大型專案中，很少有單一程式或團隊能夠管理端對端的流程。現在，我們準備繪製本章將在Python中構建的RAG流程藍圖。

RAG流程的真實呈現

假設我們是一個團隊，需要在幾週內交付整個系統。我們會被諸如以下的問題轟炸：

• 誰來收集和清理所有資料？
• 誰來負責設定OpenAI的嵌入模型？
• 誰來編寫程式碼以執行這些嵌入並管理向量儲存？
• 誰來負責實施GPT-4並管理其輸出？

解決方案：分工

我們將團隊分成三個小組，每個小組負責流程中的不同部分，如圖2.3所示：

• 資料收集與準備（D1和D2）：一個團隊負責收集資料並進行清理。
• 資料嵌入與儲存（D2和D3）：另一個團隊負責將資料透過OpenAI的嵌入模型處理，並將這些向量儲存在Activeloop Deep Lake資料集中。
• 增強生成（D4、G1-G4和E1）：最後一個團隊負責根據使用者輸入和檢索查詢生成內容。他們使用GPT-4來完成這項任務。

建構RAG流程

我們將按照前一節描述的流程，並參照圖2.3，實作三個元件，分別由三個團隊（團隊#1、團隊#2和團隊#3）平行開發：

1. 團隊#1負責資料收集與準備。
2. 團隊#2負責資料嵌入與儲存。
3. 團隊#3負責增強生成。

設定環境

安裝跨平台、跨函式庫的套件及其依賴項可能會非常具有挑戰性。必須考慮到這種複雜性，並準備好正確地設定環境。每個套件都有可能與其他套件版本衝突的依賴項。

安裝套件和函式庫

為了在本文中構建RAG流程，我們需要安裝特定的套件並凍結這些套件的版本，以防止依賴項衝突和函式庫功能問題。

# 安裝必要的套件
!pip install deeplake openai

內容解密：

上述程式碼展示瞭如何安裝deeplake和openai這兩個必要的Python套件。deeplake用於管理向量儲存，而openai則用於呼叫OpenAI的嵌入模型和生成式AI模型。

# 凍結套件版本
import pkg_resources
installed_packages = pkg_resources.working_set
installed_packages_list = sorted(["%s==%s" % (i.key, i.version)
   for i in installed_packages])
print(installed_packages_list)

內容解密：

這段程式碼用於列出並列印當前Python環境中所有已安裝的套件及其版本。透過凍結這些版本，可以確保環境的穩定性，避免因套件更新導致的相容性問題。

環境安裝流程中的元件解析

在探討每個筆記本（notebook）的「安裝環境」章節時，我們首先需要了解所涉及的各個元件。這些元件並非在所有筆記本中都會被安裝；本章節旨在提供一個全面的套件清單。

資料收集與準備

在第一個流程段「資料收集與準備」中，我們僅需要安裝 Beautiful Soup 和 Requests：

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

這個流程段的工作非常適合喜歡建立介面以與網頁互動的開發者，同時也非常適合想要參與資料收集與分析的初級開發者。

其他流程元件的安裝需求

後續的兩個流程元件「資料嵌入與儲存」及「增強生成」需要安裝 requirements01.txt 中列出的套件。安裝步驟將在後續章節中逐步進行。

掛載 Google Drive

在本案例中，我們將 Google Drive 掛載到 Google Colab 中，以安全地讀取 OpenAI API 金鑰和 Activeloop API 令牌，用於存取 OpenAI 模型和 Activeloop Deep Lake 資料集：

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

您可以選擇將金鑰和令牌儲存在其他安全的位置。

從 GitHub 下載檔案的子程式

我們的目標是編寫一個函式，從 GitHub 下載 grequests.py 檔案。該程式使用 curl 下載檔案，並可選擇性地新增私人令牌：

import subprocess
url = "https://raw.githubusercontent.com/Denis2054/RAG-Driven-Ge"
output_file = "grequests.py"
curl_command = [
    "curl",
    "-o", output_file,
    url
]
try:
    subprocess.run(curl_command, check=True)
    print("Download successful.")
except subprocess.CalledProcessError:
    print("Failed to download the file.")

內容解密：

1. 使用 subprocess 模組執行 curl 命令下載檔案。
2. curl_command 列表中包含了下載檔案所需的引數，包括輸出檔案名稱和 URL。
3. 使用 try-except 區塊捕捉 subprocess.run 可能丟出的例外，確保下載失敗時能夠給予適當的錯誤訊息。

安裝必要套件

接下來，我們將安裝與 Activeloop Deep Lake 和 OpenAI 合作所需的套件：

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

內容解密：

1. 安裝指定版本的 deeplake 和 openai 套件。
2. 由於 Google Colab 的 Pillow 版本可能與 deeplake 套件衝突，安裝 deeplake 時會自動處理此問題。
3. 安裝完成後，需要重新啟動 Colab 工作階段以確保正確運作。

設定 Activeloop 的 DNS 伺服器

安裝完成後，需要執行以下程式碼以啟用 Activeloop 的公用 DNS 伺服器：

with open('/etc/resolv.conf', 'w') as file:
    file.write("nameserver 8.8.8.8")

內容解密：

1. 將 DNS 伺服器設定為 8.8.8.8，這是 Google 的公用 DNS 伺服器之一。
2. 這個步驟確保 Activeloop 的服務能夠正確解析網域名稱。

認證流程

首先，您需要註冊 OpenAI 以取得 API 金鑰，並啟用它：

f = open("drive/MyDrive/files/api_key.txt", "r")
API_KEY = f.readline().strip()
f.close()
import os
import openai
os.environ['OPENAI_API_KEY'] = API_KEY
openai.api_key = os.getenv("OPENAI_API_KEY")

內容解密：

1. 從指設定檔案中讀取 OpenAI API 金鑰。
2. 將金鑰設定為環境變數 OPENAI_API_KEY。
3. 將 openai.api_key 設定為該環境變數的值。

接下來，啟用 Activeloop 的 API 令牌：

f = open("drive/MyDrive/files/activeloop.txt", "r")
API_token = f.readline().strip()
f.close()
ACTIVELOOP_TOKEN = API_token
os.environ['ACTIVELOOP_TOKEN'] = ACTIVELOOP_TOKEN

內容解密：

1. 從指設定檔案中讀取 Activeloop API 令牌。
2. 將令牌設定為環境變數 ACTIVELOOP_TOKEN。

完成環境安裝後，您可以隱藏「安裝環境」的程式區塊，以專注於流程元件的內容。

資料收集與準備

資料收集與準備是第一個流程元件。團隊 #1 將專注於此元件，如圖 2.6 所示。

  graph LR
    A[資料收集] --> B[資料準備]
    B --> C[資料嵌入]
    C --> D[增強生成]

圖表翻譯： 此圖示展示了資料處理流程的主要步驟，從資料收集到資料準備，最後到資料嵌入和增強生成。

太空探索資料收集與處理

在太空探索領域，眾多重要的歷史事件與科技發展都值得我們探討。本文將聚焦於十個與太空探索相關的重要維基百科文章，並透過程式設計方式收集和處理這些資料。

資料收集準備

首先，我們需要準備必要的程式函式庫來進行HTTP請求、HTML解析和正規表示式操作。我們將使用requests、BeautifulSoup和re模組來完成這些任務。

import requests
from bs4 import BeautifulSoup
import re

接著，我們定義了需要處理的維基百科文章URL列表：

# 維基百科文章的URL列表
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"
]

內容解密：

上述程式碼片段定義了我們需要處理的維基百科文章列表。這些文章涵蓋了太空探索的多個重要主題，包括歷史計畫、現代技術進展和重要的太空任務。

資料準備

為了清理文章內容，我們定義了一個clean_text函式來移除文字中的數字參考（如[1]、[2]等）。

def clean_text(content):
    # 移除數字參考，如[1]、[2]等
    content = re.sub(r'\[\d+\]', '', content)
    return content

接著，我們定義了一個fetch_and_clean函式來抓取和清理文章內容。

def fetch_and_clean(url):
    # 抓取URL內容
    response = requests.get(url)
    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()
            section.parent.decompose()
    # 提取和清理文字
    text = content.get_text(separator=' ', strip=True)
    text = clean_text(text)
    return text

內容解密：

fetch_and_clean函式首先抓取指定URL的內容，接著使用BeautifulSoup解析HTML並提取主要內容。然後，它會移除參考文獻段落，並清理文字中的數字參考。最終傳回清理後的文字。

寫入清理後的內容

我們將清理後的文章內容寫入llm.txt檔案中，以便後續處理。

# 寫入清理後的文字到檔案
with open('llm.txt', 'w', encoding='utf-8') as file:
    for url in urls:
        clean_article_text = fetch_and_clean(url)
        file.write(clean_article_text + '\n')
print("內容已寫入llm.txt")

內容解密：

上述程式碼將清理後的文章內容逐一寫入llm.txt檔案中，每篇文章之間以換行符號分隔。

驗證檔案內容

最後，我們可以讀取llm.txt檔案的前20行來驗證內容。

# 開啟檔案並讀取前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檔案的內容，確保資料正確寫入。