FAISS向量資料函式庫持久化與合併策略

– 11 分鐘閱讀

在建構向量檢索系統時，每次查詢都重新計算嵌入向量會造成效能瓶頸。為了提升效率，可以將 FAISS 索引持久化到檔案中，以便重複使用。利用 faiss.write_index 函式可以將索引儲存至檔案，而 faiss.read_index 則可將其載入回記憶體。此外，IndexFlatL2 的 add 方法允許合併多個 FAISS 索引，方便分批更新資料。然而，合併後原向量 ID 將被重新編號，需額外維護 ID 對映關係。LangChain 框架簡化了 FAISS 的整合，提供便捷的向量儲存、檢索和問答系統建構功能。文章也介紹了雲端向量資料函式庫 Pinecone，其優勢包括免維護、高擴充套件性、可靠性和效能最佳化，但也需考量成本、供應商鎖定和資料隱私等因素。程式碼範例示範瞭如何在 Pinecone 中建立索引、儲存向量資料和執行查詢，包含批次處理、錯誤重試和中繼資料管理等技巧。

最佳化向量資料函式庫的持久化與合併：以FAISS為例

在前面的章節中，我們已經成功地建立了一個端對端的向量檢索系統。然而，直接在每次查詢時重新計算嵌入向量（embeddings）並建立向量資料函式庫並不是一個高效的做法。即使用開源的嵌入模型，也會帶來計算成本和延遲問題。為瞭解決這個問題，我們可以將FAISS索引儲存到檔案中，以便在不同會話或環境中重複使用。

儲存與載入FAISS索引

FAISS提供了faiss.write_index函式來將索引儲存到檔案中：

# 儲存索引到檔案
faiss.write_index(index, "data/my_index_file.index")

這將在當前目錄下建立一個名為my_index_file.index的檔案，其中包含了序列化的索引。稍後，我們可以使用faiss.read_index將這個索引載入回記憶體：

# 從檔案載入索引
index = faiss.read_index("data/my_index_file.index")

內容解密：

  1. faiss.write_index(index, "data/my_index_file.index")：這個函式呼叫將FAISS索引物件index序列化並儲存到指定的檔案路徑"data/my_index_file.index"中。這樣做的好處是可以在後續的使用中直接載入已建立的索引，而無需每次都重新計算。
  2. faiss.read_index("data/my_index_file.index")：這個函式用於從檔案中讀取先前儲存的FAISS索引，並將其載入到記憶體中的index變數。這使得我們可以在不同的程式執行階段或不同的機器上重複使用相同的索引。

合併多個FAISS索引

在某些情況下，我們可能需要合併多個向量資料函式庫。例如，當我們分批更新或載入檔案時，可以先為每個批次建立獨立的索引，然後再將它們合併。FAISS允許我們透過IndexFlatL2索引的add方法來合併兩個索引：

# 假設index1和index2是兩個IndexFlatL2索引
index1.add(index2.reconstruct_n(0, index2.ntotal))

內容解密：

  1. index2.reconstruct_n(0, index2.ntotal)：這個方法呼叫用於從index2中檢索所有的向量。reconstruct_n方法根據指定的範圍（從0到index2.ntotal）重建向量。
  2. index1.add()：將從index2中檢索到的向量新增到index1中，從而實作兩個索引的合併。
  3. 需要注意的是，這種合併方式不會保留原始的向量ID。合併後，原本在index2中的向量將在index1中獲得新的ID。如果需要保留ID，需要外部管理向量ID與資料項之間的對映關係。

在LangChain中使用FAISS進行RAG

LangChain是一個流行的AI工程框架，支援多種RAG（檢索增強生成）技術。我們可以使用LangChain結合FAISS來實作高效的向量檢索和問答系統。以下是一個範例：

from langchain_community.vectorstores.faiss import FAISS
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

# 1. 建立檔案列表
documents = [
    "James Phoenix worked at JustUnderstandingData.",
    "James Phoenix currently is 31 years old.",
    """Data engineering is the designing and building systems for collecting,
    storing, and analyzing data at scale.""",
]

# 2. 建立向量儲存
vectorstore = FAISS.from_texts(texts=documents, embedding=OpenAIEmbeddings())
retriever = vectorstore.as_retriever()

# 3. 建立提示範本
template = """Answer the question based only on the following context:
---
Context: {context}
---
Question: {question}
"""
prompt = ChatPromptTemplate.from_template(template)

# 4. 建立聊天模型
model = ChatOpenAI()

# 5. 建立LCEL鏈
chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

# 測試查詢
print(chain.invoke("What is data engineering?"))
print(chain.invoke("Who is James Phoenix?"))
print(chain.invoke("What is the president of the US?"))

內容解密：

  1. 建立檔案列表：首先定義了一個包含三個文字的檔案列表。
  2. 建立向量儲存：使用FAISS和OpenAI的嵌入模型將文字轉換為向量，並建立檢索器。
  3. 建立提示範本：定義了一個提示範本，用於格式化輸入LLM（大語言模型）的查詢和上下文。
  4. 建立聊天模型：初始化了一個根據OpenAI的聊天模型，用於生成回應。
  5. 建立LCEL鏈：構建了一個LCEL鏈，將檢索器、提示範本、聊天模型和輸出解析器串聯起來，以實作RAG功能。

使用Pinecone等託管向量資料函式庫

除了FAISS之外，還有多種託管向量資料函式庫服務可供選擇，如Pinecone、Chroma和Weaviate等。這些服務提供了便捷的向量檢索和管理功能，可以與LangChain等框架結合使用，以實作更複雜的AI應用。

  graph LR
    A[使用者查詢] --> B[檢索器]
    B --> C[向量資料函式庫]
    C --> D[傳回相關檔案]
    D --> E[LLM生成回應]
    E --> F[最終答案]

圖表翻譯： 此圖示展示了使用向量資料函式庫進行檢索增強生成（RAG）的流程。首先，使用者的查詢被傳遞給檢索器，檢索器在向量資料函式庫中查詢相關檔案。找到的檔案隨後被用來生成最終的回應，並傳回給使用者。

雲端向量資料函式庫的優勢與實作

雲端向量資料函式庫的優勢

雲端向量資料函式庫（Hosted Vector Databases）相較於開源的本地向量儲存提供了多項優勢。這些優勢主要體現在以下幾個方面：

維護簡便

使用雲端向量資料函式庫，無需自行設定、管理及維護資料函式庫。這對於缺乏專職DevOps或資料函式倉管理團隊的企業來說，可以節省大量的時間和資源。

擴充套件性

雲端向量資料函式庫設計上能夠隨著需求進行擴充套件。當資料量增長時，資料函式庫可以自動擴充套件以處理增加的負載，確保應用程式持續高效運作。

可靠性

雲端服務通常提供高用性，並伴有服務水平協定（SLA）、自動備份及災難還原功能。這不僅提供了安心保障，也避免了潛在的資料遺失風險。

效能

雲端向量資料函式庫通常具備最佳化的基礎設施和演算法，能夠提供比自行管理的開源解決方案更好的效能。對於依賴即時或近即時向量搜尋功能的應用程式而言，這一點尤為重要。

支援服務

使用雲端服務時，通常能夠獲得來自服務提供商的支援。這對於遇到問題或需要最佳化資料函式庫使用方式的情況非常有幫助。

安全性

雲端服務通常具備健全的安全措施來保護資料，包括加密、存取控制及監控等。主要雲端服務提供商更可能具備必要的合規證書，並符合歐盟等地區的隱私法規。

當然，這些額外的功能是以成本為代價的，並且存在超支的風險。就像使用Amazon Web Services、Microsoft Azure或Google Cloud一樣，開發者因組態錯誤或程式碼錯誤而意外花費數千美元的案例屢見不鮮。此外，還存在供應商鎖定的風險，因為儘管各供應商的功能相似，但在某些方面仍存在差異，因此在不同供應商之間遷移並不簡單。另一個重要的考慮因素是隱私問題，因為與第三方分享資料伴隨著安全風險和潛在的法律後果。

使用雲端向量資料函式庫的步驟

使用雲端向量資料函式庫的步驟與設定開源FAISS向量儲存的步驟相同。首先，將檔案分塊並檢索向量；然後，在向量資料函式庫中索引檔案塊，以便檢索與查詢相似的記錄，以作為上下文插入提示中。

首先，讓我們在Pinecone（一個流行的商業向量資料函式庫供應商）中建立一個索引。然後登入Pinecone並檢索API金鑰（造訪側邊選單中的API Keys並點選“create API Key”）。本例的程式碼已在該書的GitHub儲存函式庫中提供。

程式碼範例

from pinecone import Pinecone, ServerlessSpec
import os

# 初始化連線（在app.pinecone.io取得API金鑰）：
os.environ["PINECONE_API_KEY"] = "insert-your-api-key-here"
index_name = "employee-handbook"
environment = "us-east-1"

pc = Pinecone()  # 這會讀取PINECONE_API_KEY環境變數

# 檢查索引是否已存在：
# （如果是第一次執行，不應該存在）
if index_name not in pc.list_indexes().names():
    # 如果不存在，則建立索引
    pc.create_index(
        index_name,
        # 使用與text-embedding-ada-002相同的向量維度
        dimension=1536,
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region=environment),
    )

# 連線到索引：
index = pc.Index(index_name)

# 檢視索引統計資訊：
index.describe_index_stats()

輸出結果：

{
    'dimension': 1536,
    'index_fullness': 0.0,
    'namespaces': {},
    'total_vector_count': 0
}

#### 內容解密：

  1. 匯入函式庫：首先，指令碼匯入了必要的模組。from pinecone import Pinecone, ServerlessSpec 和 import os 用於存取和設定環境變數。

  2. 設定Pinecone API金鑰：Pinecone API金鑰對於身份驗證至關重要，它透過 os.environ["PINECONE_API_KEY"] = "insert-your-api-key-here" 被設定為環境變數。務必將 "insert-your-api-key-here" 替換為實際的Pinecone API金鑰。

  3. 定義索引名稱和環境：設定了 index_name 和 environment 變數。index_name 被指定為 "employee-handbook"，這是在Pinecone資料函式庫中要建立或存取的索引名稱。environment 變數被指定為 "us-east-1"，指明瞭伺服器的位置。

  4. 初始化Pinecone連線：透過 Pinecone() 建構函式初始化與Pinecone的連線。該建構函式自動從環境變數中讀取 PINECONE_API_KEY。

  5. 檢查現有索引：指令碼檢查是否已存在名稱為 index_name 的索引。這透過 pc.list_indexes().names() 函式實作，該函式傳回所有現有索引名稱的列表。

  6. 建立索引：如果索引不存在，則使用 pc.create_index() 函式建立索引。該函式被呼叫時帶有多個引數，用於組態新索引：

    • index_name：指定索引的名稱。
    • dimension=1536：設定要儲存在索引中的向量的維度。
    • metric='cosine'：決定使用餘弦相似度指標進行向量比較。
    • spec=ServerlessSpec(cloud="aws", region=environment)：指定索引的伺服器組態。

  7. 連線到索引：在驗證或建立索引後，指令碼透過 pc.Index(index_name) 連線到索引。此連線對於後續的操作（如插入或查詢資料）是必要的。

Pinecone 向量資料函式庫的操作與查詢最佳化

在前面的章節中，我們已經建立了一個 Pinecone 索引並將向量資料儲存其中。本章節將探討如何有效地儲存向量資料到 Pinecone，以及如何進行高效的向量查詢。

向量資料儲存至 Pinecone

首先，我們需要將文字片段（text chunks）和對應的向量嵌入（vector embeddings）儲存到 Pinecone 中。這可以透過迴圈遍歷所有的文字片段和向量，並將它們作為記錄（records）插入到 Pinecone 中來實作。

from tqdm import tqdm 
from time import sleep

# 設定批次大小和重試次數
batch_size = 10
retry_limit = 5 

for i in tqdm(range(0, len(chunks), batch_size)):
    # 計算批次結束索引
    i_end = min(len(chunks), i+batch_size)
    meta_batch = chunks[i:i_end]
    
    # 取得 ID 清單
    ids_batch = [str(j) for j in range(i, i_end)]
    
    # 取得需要編碼的文字
    texts = [x for x in meta_batch]
    
    # 建立向量嵌入
    done = False
    try:
        embeds = []
        for text in texts:
            embedding = get_vector_embeddings(text)
            embeds.append(embedding)
        done = True
    except:
        retry_count = 0
        while not done and retry_count < retry_limit:
            try:
                for text in texts:
                    embedding = get_vector_embeddings(text)
                    embeds.append(embedding)
                done = True
            except:
                sleep(5)
                retry_count += 1
        if not done:
            print(f"Failed to get embeddings after {retry_limit} retries.")
            continue
    
    # 清理中繼資料
    meta_batch = [{
        'batch': i,
        'text': x
    } for x in meta_batch]
    
    # 組合要插入的資料
    to_upsert = list(zip(ids_batch, embeds, meta_batch))
    
    # 插入或更新至 Pinecone
    index.upsert(vectors=to_upsert)

內容解密：

  1. 批次處理：程式碼使用 batch_size 變數來控制每次處理的資料量，避免一次性處理大量資料導致效能問題。
  2. 重試機制：在取得向量嵌入的過程中，程式碼加入了重試機制，以處理可能的速率限制錯誤（RateLimitError）。
  3. 中繼資料處理：程式碼將原始的 meta_batch 資料轉換為包含額外中繼資料（如批次號碼）的字典格式，這有助於後續的查詢和資料管理。
  4. upsert 操作：Pinecone 的 upsert 方法用於插入或更新向量資料，如果指定的 ID 已存在，則更新其對應的向量和中繼資料；否則，插入新的記錄。

向量查詢

儲存向量資料到 Pinecone 後，我們可以根據使用者的查詢請求進行向量相似度查詢。

# 從 Pinecone 檢索資料
user_query = "do we get free unicorn rides?"
def pinecone_vector_search(user_query, k):
    xq = get_vector_embeddings(user_query)
    res = index.query(vector=xq, top_k=k, include_metadata=True)
    return res

pinecone_vector_search(user_query, k=1)

內容解密：

  1. 查詢向量生成：首先，使用 get_vector_embeddings 函式將使用者的查詢文字轉換為向量表示。
  2. Pinecone 查詢：呼叫 Pinecone 索引的 query 方法，傳入查詢向量 xq，並指定傳回最相似的前 k 個結果。
  3. 包含中繼資料：設定 include_metadata=True 以傳回匹配結果的中繼資料，這些資訊對於理解查詢結果非常有幫助。

結果分析

對於查詢 “do we get free unicorn rides?"，Pinecone 傳回的最相似結果如下：

{'matches': 
    [{'id': '15',
      'metadata': {'batch': 10.0,
                   'text': "You'll enjoy a treasure chest of perks, including unlimited unicorn rides, a bottomless cauldron of coffee and potions, and access to our company library filled with spellbinding books. We also offer competitive health and dental plans, ensuring your physical well-being is as "}
    }]
}

圖表翻譯：

  graph LR;
    A[使用者查詢] --> B[生成查詢向量];
    B --> C[Pinecone 向量查詢];
    C --> D[傳回最相似結果];
    D --> E[包含中繼資料];

此圖表呈現了從使用者查詢到傳回最相似結果的整個流程。使用者輸入查詢請求後，系統首先將查詢文字轉換為向量表示，接著使用該向量在 Pinecone 中進行相似度查詢，最後傳回最相似的結果及其相關的中繼資料。

隨著人工智慧和機器學習技術的發展，向量資料函式庫的需求將越來越大。未來，我們可以預見 Pinecone 這類別向量資料函式庫將在更多的應用場景中發揮重要作用，例如影像搜尋、語音識別、自然語言處理等領域。同時，如何進一步最佳化向量資料函式庫的效能、提高查詢效率、降低成本，將是未來研究的重要方向。

透過本章節的學習，我們對 Pinecone 向量資料函式庫的操作有了更深入的瞭解，並掌握瞭如何有效地儲存和查詢向量資料。這些知識對於開發根據向量的應用具有重要的參考價值。

玄貓

玄貓

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