使用 Milvus 和 Haystack 進行全文檢索

全文檢索是一種透過匹配文字中特定關鍵字或短語來檢索文件的傳統方法。它會根據詞彙頻率等因素計算出的相關性分數對結果進行排序。語意搜尋更擅長於理解意義和上下文，而全文搜尋則擅長於精確的關鍵字比對，因此是語意搜尋的有效補充。BM25 演算法廣泛用於全文檢索的排序，並在檢索增強世代 (Retrieval-Augmented Generation, RAG) 中扮演關鍵角色。

Milvus 2.5引入了使用 BM25 的原生全文搜尋功能。此方法可將文字轉換成代表 BM25 分數的稀疏向量。您只需輸入原始文字，Milvus 即會自動產生並儲存稀疏向量，無需手動產生稀疏嵌入。

Haystack現在支援 Milvus 的這項功能，讓您可以輕鬆地在 RAG 應用程式中加入全文檢索功能。您可以將全文檢索與密集向量語意檢索結合，以獲得混合方法，從語意理解和關鍵字匹配精確度中獲益。這種結合可提高搜尋準確度，並提供使用者更好的結果。

本教學示範如何使用 Haystack 和 Milvus 在應用程式中實作全文和混合搜尋。

要使用 Milvus 向量存儲，請指定您的 Milvus 伺服器URI (也可選擇使用TOKEN)。若要啟動 Milvus 伺服器，您可以依照Milvus 安裝指南設定 Milvus 伺服器，或直接免費試用 Zilliz Cloud(完全管理的 Milvus)。

先決條件

在執行本筆記本之前，請確認您已安裝下列依賴項目：

$ pip install --upgrade --quiet pymilvus milvus-haystack

我們將使用 OpenAI 的模型。您應該準備api key OPENAI_API_KEY 作為環境變數。

import os

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

準備資料

匯入本筆記本中所需的套件。然後準備一些範例文件。

from haystack import Pipeline
from haystack.components.embedders import OpenAIDocumentEmbedder, OpenAITextEmbedder
from haystack.components.writers import DocumentWriter
from haystack.utils import Secret
from milvus_haystack import MilvusDocumentStore, MilvusSparseEmbeddingRetriever
from haystack.document_stores.types import DuplicatePolicy
from milvus_haystack.function import BM25BuiltInFunction
from milvus_haystack import MilvusDocumentStore
from milvus_haystack.milvus_embedding_retriever import MilvusHybridRetriever

from haystack.utils import Secret
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator
from haystack import Document

documents = [
    Document(content="Alice likes this apple", meta={"category": "fruit"}),
    Document(content="Bob likes swimming", meta={"category": "sport"}),
    Document(content="Charlie likes white dogs", meta={"category": "pets"}),
]

將全文檢索整合到 RAG 系統中，可以平衡語意檢索與精確且可預測的關鍵字檢索。您也可以選擇只使用全文檢索，但建議結合全文檢索與語意檢索，以獲得更好的檢索結果。為了示範的目的，我們將在此展示單獨的全文檢索和混合檢索。

無嵌入的 BM25 搜尋

建立索引管道

對於全文搜尋，Milvus MilvusDocumentStore 接受builtin_function 參數。透過這個參數，您可以傳入BM25BuiltInFunction 的一個實例，它在 Milvus 伺服器端實作 BM25 演算法。設定builtin_function 指定為 BM25 函式實例。例如

connection_args = {"uri": "http://localhost:19530"}
# connection_args = {"uri": YOUR_ZILLIZ_CLOUD_URI, "token": Secret.from_env_var("ZILLIZ_CLOUD_API_KEY")}

document_store = MilvusDocumentStore(
    connection_args=connection_args,
    sparse_vector_field="sparse_vector",  # The sparse vector field.
    text_field="text",
    builtin_function=[
        BM25BuiltInFunction(  # The BM25 function converts the text into a sparse vector.
            input_field_names="text",
            output_field_names="sparse_vector",
        )
    ],
    consistency_level="Bounded",  # Supported values are (`"Strong"`, `"Session"`, `"Bounded"`, `"Eventually"`).
    drop_old=True,  # Drop the old collection if it exists and recreate it.
)

對於 connection_args：

• 您可以在docker 或 kubernetes 上設定效能更高的 Milvus 伺服器。在此設定中，請使用伺服器位址，例如http://localhost:19530 ，作為您的uri 。
• 如果您想使用Zilliz Cloud（Milvus 的完全管理雲端服務），請調整uri 和token ，與 Zilliz Cloud 中的Public Endpoint 和 Api key對應。

建立索引管道，將文件寫入 Milvus 文件儲存庫。

writer = DocumentWriter(document_store=document_store, policy=DuplicatePolicy.NONE)

indexing_pipeline = Pipeline()
indexing_pipeline.add_component("writer", writer)
indexing_pipeline.run({"writer": {"documents": documents}})

{'writer': {'documents_written': 3}}

建立檢索管道

建立檢索管道，使用MilvusSparseEmbeddingRetriever 從 Milvus 文件儲存庫檢索文件，這是document_store 的包裝程式。

retrieval_pipeline = Pipeline()
retrieval_pipeline.add_component(
    "retriever", MilvusSparseEmbeddingRetriever(document_store=document_store)
)

question = "Who likes swimming?"

retrieval_results = retrieval_pipeline.run({"retriever": {"query_text": question}})

retrieval_results["retriever"]["documents"][0]

Document(id=bd334348dd2087c785e99b5a0009f33d9b8b8198736f6415df5d92602d81fd3e, content: 'Bob likes swimming', meta: {'category': 'sport'}, score: 1.2039074897766113)

具有語意搜尋和全文搜尋的混合搜尋

建立索引管道

在混合搜尋中，我們使用 BM25 函式來執行全文搜尋，並指定密集向量場vector ，來執行語意搜尋。

document_store = MilvusDocumentStore(
    connection_args=connection_args,
    vector_field="vector",  # The dense vector field.
    sparse_vector_field="sparse_vector",  # The sparse vector field.
    text_field="text",
    builtin_function=[
        BM25BuiltInFunction(  # The BM25 function converts the text into a sparse vector.
            input_field_names="text",
            output_field_names="sparse_vector",
        )
    ],
    consistency_level="Bounded",  # Supported values are (`"Strong"`, `"Session"`, `"Bounded"`, `"Eventually"`).
    drop_old=True,  # Drop the old collection and recreate it.
)

建立索引管道，將文件轉換成嵌入式。然後將文件寫入 Milvus 文件儲存庫。

writer = DocumentWriter(document_store=document_store, policy=DuplicatePolicy.NONE)

indexing_pipeline = Pipeline()
indexing_pipeline.add_component("dense_doc_embedder", OpenAIDocumentEmbedder())
indexing_pipeline.add_component("writer", writer)
indexing_pipeline.connect("dense_doc_embedder", "writer")
indexing_pipeline.run({"dense_doc_embedder": {"documents": documents}})

print("Number of documents:", document_store.count_documents())

Calculating embeddings: 100%|██████████| 1/1 [00:01<00:00,  1.15s/it]


Number of documents: 3

建立檢索管道

建立檢索管道，使用MilvusHybridRetriever 從 Milvus 文件儲存庫檢索文件，其中包含document_store ，並接收有關混合搜尋的參數。

# from pymilvus import WeightedRanker
retrieval_pipeline = Pipeline()
retrieval_pipeline.add_component("dense_text_embedder", OpenAITextEmbedder())
retrieval_pipeline.add_component(
    "retriever",
    MilvusHybridRetriever(
        document_store=document_store,
        # top_k=3,
        # reranker=WeightedRanker(0.5, 0.5),  # Default is RRFRanker()
    ),
)

retrieval_pipeline.connect("dense_text_embedder.embedding", "retriever.query_embedding")

<haystack.core.pipeline.pipeline.Pipeline object at 0x3383ad990>
🚅 Components
  - dense_text_embedder: OpenAITextEmbedder
  - retriever: MilvusHybridRetriever
🛤️ Connections
  - dense_text_embedder.embedding -> retriever.query_embedding (List[float])

當使用MilvusHybridRetriever 執行混合搜尋時，我們可以選擇性地設定 topK 和 reranker 參數。它會自動處理向量嵌入和內建函數，最後再使用 reranker 來精煉結果。搜尋過程的底層實作細節是對使用者隱藏的。

關於混合搜尋的更多資訊，您可以參考混合搜尋介紹。

question = "Who likes swimming?"

retrieval_results = retrieval_pipeline.run(
    {
        "dense_text_embedder": {"text": question},
        "retriever": {"query_text": question},
    }
)

retrieval_results["retriever"]["documents"][0]

Document(id=bd334348dd2087c785e99b5a0009f33d9b8b8198736f6415df5d92602d81fd3e, content: 'Bob likes swimming', meta: {'category': 'sport'}, score: 0.032786883413791656, embedding: vector of size 1536)

自訂分析器

分析器是全文檢索中不可或缺的工具，它可將句子分割成字元，並執行詞彙分析，如詞幹分析和停止詞移除。分析器通常是特定語言的。您可以參考本指南以瞭解更多關於 Milvus 分析器的資訊。

Milvus 支援兩種類型的分析器：內建分析器和自訂分析器。在預設情況下，BM25BuiltInFunction 會使用標準的內建分析器，這是最基本的分析器，會用標點符號來標記文字。

如果您想使用不同的分析器或自訂分析器，可以在BM25BuiltInFunction 初始化時傳入analyzer_params 參數。

analyzer_params_custom = {
    "tokenizer": "standard",
    "filter": [
        "lowercase",  # Built-in filter
        {"type": "length", "max": 40},  # Custom filter
        {"type": "stop", "stop_words": ["of", "to"]},  # Custom filter
    ],
}

document_store = MilvusDocumentStore(
    connection_args=connection_args,
    vector_field="vector",
    sparse_vector_field="sparse_vector",
    text_field="text",
    builtin_function=[
        BM25BuiltInFunction(
            input_field_names="text",
            output_field_names="sparse_vector",
            analyzer_params=analyzer_params_custom,  # Custom analyzer parameters.
            enable_match=True,  # Whether to enable match.
        )
    ],
    consistency_level="Bounded",
    drop_old=True,
)

# write documents to the document store
writer = DocumentWriter(document_store=document_store, policy=DuplicatePolicy.NONE)
indexing_pipeline = Pipeline()
indexing_pipeline.add_component("dense_doc_embedder", OpenAIDocumentEmbedder())
indexing_pipeline.add_component("writer", writer)
indexing_pipeline.connect("dense_doc_embedder", "writer")
indexing_pipeline.run({"dense_doc_embedder": {"documents": documents}})

Calculating embeddings: 100%|██████████| 1/1 [00:00<00:00,  1.39it/s]





{'dense_doc_embedder': {'meta': {'model': 'text-embedding-ada-002-v2',
   'usage': {'prompt_tokens': 11, 'total_tokens': 11}}},
 'writer': {'documents_written': 3}}

我們可以看看 Milvus 套件的 schema，並確保自訂的分析器設定正確。

document_store.col.schema

{'auto_id': False, 'description': '', 'fields': [{'name': 'text', 'description': '', 'type': <DataType.VARCHAR: 21>, 'params': {'max_length': 65535, 'enable_match': True, 'enable_analyzer': True, 'analyzer_params': {'tokenizer': 'standard', 'filter': ['lowercase', {'type': 'length', 'max': 40}, {'type': 'stop', 'stop_words': ['of', 'to']}]}}}, {'name': 'id', 'description': '', 'type': <DataType.VARCHAR: 21>, 'params': {'max_length': 65535}, 'is_primary': True, 'auto_id': False}, {'name': 'vector', 'description': '', 'type': <DataType.FLOAT_VECTOR: 101>, 'params': {'dim': 1536}}, {'name': 'sparse_vector', 'description': '', 'type': <DataType.SPARSE_FLOAT_VECTOR: 104>, 'is_function_output': True}], 'enable_dynamic_field': True, 'functions': [{'name': 'bm25_function_7b6e15a4', 'description': '', 'type': <FunctionType.BM25: 1>, 'input_field_names': ['text'], 'output_field_names': ['sparse_vector'], 'params': {}}]}

更多關於概念的細節，例如analyzer,tokenizer,filter,enable_match,analyzer_params ，請參考分析器文件。

在 RAG 管道中使用混合搜尋

我們已學會如何在 Haystack 和 Milvus 中使用基本的 BM25 內建函式，並準備了載入的document_store 。讓我們介紹使用混合搜尋的最佳化 RAG 實作。

這個圖表顯示了 Hybrid Retrieve & Reranking 流程，結合了用於關鍵字匹配的 BM25 和用於語意檢索的密集向量搜尋。來自這兩種方法的結果會合併、重新排序，並傳送到 LLM 以產生最終答案。

混合搜尋平衡了精確度與語意理解，針對不同的查詢提高了精確度與穩健性。它利用 BM25 全文檢索和向量檢索來擷取候選項目，同時確保語意、上下文感知和精確的檢索。

讓我們嘗試使用混合搜尋的最佳化 RAG 實作。

prompt_template = """Answer the following query based on the provided context. If the context does
                     not include an answer, reply with 'I don't know'.\n
                     Query: {{query}}
                     Documents:
                     {% for doc in documents %}
                        {{ doc.content }}
                     {% endfor %}
                     Answer:
                  """

rag_pipeline = Pipeline()
rag_pipeline.add_component("text_embedder", OpenAITextEmbedder())
rag_pipeline.add_component(
    "retriever", MilvusHybridRetriever(document_store=document_store, top_k=1)
)
rag_pipeline.add_component("prompt_builder", PromptBuilder(template=prompt_template))
rag_pipeline.add_component(
    "generator",
    OpenAIGenerator(
        api_key=Secret.from_token(os.getenv("OPENAI_API_KEY")),
        generation_kwargs={"temperature": 0},
    ),
)
rag_pipeline.connect("text_embedder.embedding", "retriever.query_embedding")
rag_pipeline.connect("retriever.documents", "prompt_builder.documents")
rag_pipeline.connect("prompt_builder", "generator")

results = rag_pipeline.run(
    {
        "text_embedder": {"text": question},
        "retriever": {"query_text": question},
        "prompt_builder": {"query": question},
    }
)
print("RAG answer:", results["generator"]["replies"][0])

RAG answer: Bob likes swimming.

有關如何使用 milvus-haystack 的詳細資訊，請參閱milvus-haystack 官方資源庫。