使用 Milvus 和 EmbedAnything 建立 RAG

EmbedAnything是用 Rust 建立的快速、輕量嵌入管道，支援文字、PDF、圖片、音訊等。

在本教程中，我們將示範如何使用 EmbedAnything 與Milvus 來建立一個 Retrieval-Augmented Generation (RAG) 管道。EmbedAnything 並未與任何特定資料庫緊密結合，而是使用可插拔的適配器系統 - 適配器可作為封裝程式，定義嵌入式資料如何格式化、編入索引，以及如何儲存在目標向量儲存庫中。

將 EmbedAnything 與 Milvus 轉接器搭配使用，您只需要幾行程式碼就可以從多種檔案類型產生嵌入式資料，並將它們有效地儲存在 Milvus 中。

⚠️ 注意：EmbedAnything 中的適配器可處理插入到 Milvus 中，但不支援開箱即用的搜尋功能。若要建立完整的 RAG 管道，您還需要單獨實體化 MilvusClient，並實作擷取邏輯 (例如，向量的相似性搜尋) 作為應用程式的一部分。

準備工作

相依性與環境

$ pip install -qU pymilvus milvus-lite openai embed_anything

複製儲存庫並載入適配器

接下來，我們要克隆EmbedAnythingrepo，並將examples/adapters 目錄加入 Python 路徑。這是我們儲存自訂 Milvus 適配器實作的地方，它允許 EmbedAnything 與 Milvus 通訊以插入向量。

import sys

# Clone the EmbedAnything repository if not already cloned
![ -d "EmbedAnything" ] || git clone https://github.com/StarlightSearch/EmbedAnything.git

# Add the `examples/adapters` directory to the Python path
sys.path.append("EmbedAnything/examples/adapters")
print("✅ EmbedAnything cloned and adapter path added.")

✅ EmbedAnything cloned and adapter path added.

我們將在此 RAG 管道中使用 OpenAI 作為 LLM。您應該準備api key OPENAI_API_KEY 作為環境變數。

import os
from openai import OpenAI

os.environ["OPENAI_API_KEY"] = "sk-***********"

openai_client = OpenAI()

建立 RAG

初始化 Milvus

在嵌入任何檔案之前，我們需要準備兩個與 Milvus 互動的元件：

1. MilvusVectorAdapter - 這是 EmbedAnything 的 Milvus 適配器，僅用於向量攝取(即插入嵌入和建立索引)。目前不支援搜尋作業。
2. MilvusClient - 這是來自pymilvus 的官方用戶端，可完整存取Milvus 功能，包括向量搜尋、過濾和集合管理。

為了避免混淆：

• 將MilvusVectorAdapter 視為您儲存向量的「只寫」工具。
• 將MilvusClient 視為您的「讀取與搜尋」引擎，實際執行查詢並擷取 RAG 的文件。

import embed_anything
from embed_anything import (
    WhichModel,
    EmbeddingModel,
)
from milvus_db import MilvusVectorAdapter
from pymilvus import MilvusClient

# Official Milvus client for full operations
milvus_client = MilvusClient(uri="./milvus.db", token="")

# EmbedAnything adapter for pushing embeddings into Milvus
index_name = "embed_anything_milvus_collection"
milvus_adapter = MilvusVectorAdapter(
    uri="./milvus.db", token="", collection_name=index_name
)

# Delete existing collection if it exists
if milvus_client.has_collection(index_name):
    milvus_client.drop_collection(index_name)

# Create a new collection with dimension matching the embedding model later used
milvus_adapter.create_index(dimension=384)

Ok - Milvus DB connection established.
Collection 'embed_anything_milvus_collection' created with index.

初始化嵌入模型和嵌入 PDF 文件

現在我們要初始化嵌入模型。我們將使用來自 sentence-transformers 函式庫的all-MiniLM-L12-v2 model ，這是一個輕量級但功能強大的模型，用來產生文字嵌入。它會產生 384 維的嵌入模型，因此這與我們的 Milvus 套件維度設定為 384 是一致的。這種對齊是非常重要的，可以確保儲存在 Milvus 中的向量尺寸與模型產生的向量尺寸之間的相容性。

EmbedAnything 支援更多的嵌入模型。如需詳細資訊，請參閱官方文件。

# Initialize the embedding model
model = EmbeddingModel.from_pretrained_hf(
    WhichModel.Bert, model_id="sentence-transformers/all-MiniLM-L12-v2"
)

現在，讓我們嵌入一個 PDF 檔案。EmbedAnything 可以輕鬆處理 PDF（以及更多）文件，並直接在 Milvus 中儲存它們的嵌入。

# Embed a PDF file
data = embed_anything.embed_file(
    "./pdf_files/WhatisMilvus.pdf",
    embedder=model,
    adapter=milvus_adapter,
)

Converted 12 embeddings for insertion.
Successfully inserted 12 embeddings.

擷取並產生回應

同樣地，EmbedAnything 的MilvusVectorAdapter 目前只是一個輕量級的抽象，僅用於向量攝取和索引。它不支援搜尋或檢索查詢。因此，為了搜尋相關文件以建立我們的 RAG 管道，我們必須直接使用MilvusClient 實例 (milvus_client) 來查詢我們的 Milvus 向量儲存庫。

定義一個函式從 Milvus 擷取相關文件。

def retrieve_documents(question, top_k=3):
    query_vector = list(
        embed_anything.embed_query([question], embedder=model)[0].embedding
    )
    search_res = milvus_client.search(
        collection_name=index_name,
        data=[query_vector],
        limit=top_k,
        output_fields=["text"],
    )
    docs = [(res["entity"]["text"], res["distance"]) for res in search_res[0]]
    return docs

定義一個函式，使用 RAG 管道中擷取的文件產生回應。

def generate_rag_response(question):
    retrieved_docs = retrieve_documents(question)
    context = "\n".join([f"Text: {doc[0]}\n" for doc in retrieved_docs])
    system_prompt = (
        "You are an AI assistant. Provide answers based on the given context."
    )
    user_prompt = f"""
    Use the following pieces of information to answer the question. If the information is not in the context, say you don't know.
    
    Context:
    {context}
    
    Question: {question}
    """
    response = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
    )
    return response.choices[0].message.content

讓我們用一個範例問題來測試 RAG 管道。

question = "How does Milvus search for similar documents?"
answer = generate_rag_response(question)
print(f"Question: {question}")
print(f"Answer: {answer}")

Question: How does Milvus search for similar documents?
Answer: Milvus searches for similar documents primarily through Approximate Nearest Neighbor (ANN) search, which finds the top K vectors closest to a given query vector. It also supports various other types of searches, such as filtering search under specified conditions, range search within a specified radius, hybrid search based on multiple vector fields, and keyword search based on BM25. Additionally, it can perform reranking to adjust the order of search results based on additional criteria, refining the initial ANN search results.