Milvus 向量数据库混合搜索实现¶

混合搜索通过结合语义检索与关键词匹配的优势，能够提供更准确且符合上下文关联的搜索结果。这种同时融合语义搜索和关键词匹配特性的方法，在复杂信息检索任务中表现尤为出色。

本指南将展示如何在 LlamaIndex RAG 流程中使用 Milvus 实现混合搜索。我们将从推荐的默认混合搜索方案（语义搜索+BM25算法）开始，继而探索其他稀疏嵌入方法的替代方案，以及混合重排序器的自定义配置。

前提条件¶

安装依赖项

在开始之前，请确保已安装以下依赖项：

In [ ]:

Copied!

! pip install llama-index-vector-stores-milvus
! pip install llama-index-embeddings-openai
! pip install llama-index-llms-openai
! pip install llama-index-vector-stores-milvus
! pip install llama-index-embeddings-openai
! pip install llama-index-llms-openai

如果您使用的是 Google Colab，可能需要重启运行时环境（在界面顶部的"Runtime"菜单中选择下拉菜单中的"Restart session"选项。）

设置账户

本教程使用 OpenAI 进行文本嵌入和答案生成。您需要准备 OpenAI API 密钥。

In [ ]:

Copied!

import openai

openai.api_key = "sk-"
import openai

openai.api_key = "sk-"

要使用 Milvus 向量数据库，请指定您的 Milvus 服务器 URI（可选添加 TOKEN）。要启动 Milvus 服务器，您可以按照 Milvus 安装指南进行设置，或直接免费试用 Zilliz Cloud。

全文检索功能目前支持 Milvus 单机版、Milvus 分布式版和 Zilliz Cloud，但暂不支持 Milvus Lite 版本（计划未来实现）。如需了解更多信息，请联系 support@zilliz.com。

In [ ]:

Copied!

URI = "http://localhost:19530"
# TOKEN = ""
URI = "http://localhost:19530"
# TOKEN = ""

加载示例数据

运行以下命令将示例文档下载到 "data/paul_graham" 目录中：

In [ ]:

Copied!

! mkdir -p 'data/paul_graham/'
! wget 'https://raw.githubusercontent.com/run-llama/llama_index/main/docs/docs/examples/data/paul_graham/paul_graham_essay.txt' -O 'data/paul_graham/paul_graham_essay.txt'
! mkdir -p 'data/paul_graham/'
! wget 'https://raw.githubusercontent.com/run-llama/llama_index/main/docs/docs/examples/data/paul_graham/paul_graham_essay.txt' -O 'data/paul_graham/paul_graham_essay.txt'

然后使用 SimpleDirectoryReaderLoad 加载保罗·格雷厄姆的文章《我的工作历程》：

In [ ]:

Copied!

from llama_index.core import SimpleDirectoryReader

documents = SimpleDirectoryReader("./data/paul_graham/").load_data()

# Let's take a look at the first document
print("Example document:\n", documents[0])
from llama_index.core import SimpleDirectoryReader

documents = SimpleDirectoryReader("./data/paul_graham/").load_data()

# Let's take a look at the first document
print("Example document:\n", documents[0])

Example document:
 Doc ID: f9cece8c-9022-46d8-9d0e-f29d70e1dbbe
Text: What I Worked On  February 2021  Before college the two main
things I worked on, outside of school, were writing and programming. I
didn't write essays. I wrote what beginning writers were supposed to
write then, and probably still are: short stories. My stories were
awful. They had hardly any plot, just characters with strong feelings,
which I ...

混合检索（BM25）¶

本节演示如何使用BM25算法实现混合检索。开始前，我们将初始化MilvusVectorStore并为示例文档创建索引。默认配置包含：

来自默认嵌入模型（OpenAI的text-embedding-ada-002）的稠密向量
若启用稀疏检索（enable_sparse=True）则使用BM25进行全文搜索
启用混合检索时采用RRFRanker（k=60）进行结果融合

In [ ]:

Copied!





# Create an index over the documnts
from llama_index.vector_stores.milvus import MilvusVectorStore
from llama_index.core import StorageContext, VectorStoreIndex


vector_store = MilvusVectorStore(
    uri=URI,
    # token=TOKEN,
    dim=1536,  # vector dimension depends on the embedding model
    enable_sparse=True,  # enable the default full-text search using BM25
    overwrite=True,  # drop the collection if it already exists
)
storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex.from_documents(
    documents, storage_context=storage_context
)
# Create an index over the documnts
from llama_index.vector_stores.milvus import MilvusVectorStore
from llama_index.core import StorageContext, VectorStoreIndex


vector_store = MilvusVectorStore(
    uri=URI,
    # token=TOKEN,
    dim=1536,  # vector dimension depends on the embedding model
    enable_sparse=True,  # enable the default full-text search using BM25
    overwrite=True,  # drop the collection if it already exists
)
storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex.from_documents(
    documents, storage_context=storage_context
)

2025-04-17 03:38:16,645 [DEBUG][_create_connection]: Created new connection using: cf0f4df74b18418bb89ec512063c1244 (async_milvus_client.py:547)
Sparse embedding function is not provided, using default.
Default sparse embedding function: BM25BuiltInFunction(input_field_names='text', output_field_names='sparse_embedding').

以下是关于配置 MilvusVectorStore 中稠密字段和稀疏字段参数的更多信息：

稠密字段

enable_dense (bool): 布尔标志，用于启用或禁用稠密嵌入。默认为 True。
dim (int, 可选): 集合中嵌入向量的维度。
embedding_field (str, 可选): 集合中稠密嵌入字段的名称，默认为 DEFAULT_EMBEDDING_KEY。
index_config (dict, 可选): 用于构建稠密嵌入索引的配置。默认为 None。
search_config (dict, 可选): 用于搜索 Milvus 稠密索引的配置。注意该配置必须与 index_config 指定的索引类型兼容。默认为 None。
similarity_metric (str, 可选): 稠密嵌入使用的相似度度量标准，当前支持 IP、COSINE 和 L2。

稀疏字段

enable_sparse (bool): 布尔标志，用于启用或禁用稀疏嵌入。默认为 False。
sparse_embedding_field (str): 稀疏嵌入字段的名称，默认为 DEFAULT_SPARSE_EMBEDDING_KEY。
sparse_embedding_function (Union[BaseSparseEmbeddingFunction, BaseMilvusBuiltInFunction], 可选): 若 enable_sparse 为 True，需提供此对象以将文本转换为稀疏嵌入。若为 None，将使用默认稀疏嵌入函数 (BM25BuiltInFunction)，或对已存在且无内置函数的集合使用 BGEM3SparseEmbedding。
sparse_index_config (dict, 可选): 用于构建稀疏嵌入索引的配置。默认为 None。

要在查询阶段启用混合搜索，请将 vector_store_query_mode 设置为 "hybrid"。这将结合并重新排序来自语义搜索和全文搜索的结果。让我们用一个示例查询进行测试："作者在 Viaweb 学到了什么？":

In [ ]:

Copied!





import textwrap

query_engine = index.as_query_engine(
    vector_store_query_mode="hybrid", similarity_top_k=5
)
response = query_engine.query("What did the author learn at Viaweb?")
print(textwrap.fill(str(response), 100))
import textwrap

query_engine = index.as_query_engine(
    vector_store_query_mode="hybrid", similarity_top_k=5
)
response = query_engine.query("What did the author learn at Viaweb?")
print(textwrap.fill(str(response), 100))

The author learned about retail, the importance of user feedback, and the significance of growth
rate as the ultimate test of a startup at Viaweb.

自定义文本分析器¶

分析器在全文搜索中扮演着关键角色，通过将句子拆分为词元并进行词法处理（如词干提取和停用词移除）。它们通常具有语言特异性。更多细节请参阅 Milvus 分析器指南。

Milvus 支持两种分析器类型：内置分析器和自定义分析器。默认情况下，若 enable_sparse 设为 True，MilvusVectorStore 会使用默认配置的 BM25BuiltInFunction，采用基于标点符号分词的标准内置分析器。

如需使用不同分析器或自定义现有分析器，可在构建 BM25BuiltInFunction 时向 analyzer_params 参数传入配置值，然后将该函数设为 MilvusVectorStore 中的 sparse_embedding_function。

In [ ]:

Copied!





from llama_index.vector_stores.milvus.utils import BM25BuiltInFunction

bm25_function = BM25BuiltInFunction(
    analyzer_params={
        "tokenizer": "standard",
        "filter": [
            "lowercase",  # Built-in filter
            {"type": "length", "max": 40},  # Custom cap size of a single token
            {"type": "stop", "stop_words": ["of", "to"]},  # Custom stopwords
        ],
    },
    enable_match=True,
)

vector_store = MilvusVectorStore(
    uri=URI,
    # token=TOKEN,
    dim=1536,
    enable_sparse=True,
    sparse_embedding_function=bm25_function,  # BM25 with custom analyzer
    overwrite=True,
)
from llama_index.vector_stores.milvus.utils import BM25BuiltInFunction

bm25_function = BM25BuiltInFunction(
    analyzer_params={
        "tokenizer": "standard",
        "filter": [
            "lowercase",  # Built-in filter
            {"type": "length", "max": 40},  # Custom cap size of a single token
            {"type": "stop", "stop_words": ["of", "to"]},  # Custom stopwords
        ],
    },
    enable_match=True,
)

vector_store = MilvusVectorStore(
    uri=URI,
    # token=TOKEN,
    dim=1536,
    enable_sparse=True,
    sparse_embedding_function=bm25_function,  # BM25 with custom analyzer
    overwrite=True,
)

2025-04-17 03:38:48,085 [DEBUG][_create_connection]: Created new connection using: 61afd81600cb46ee89f887f16bcbfe55 (async_milvus_client.py:547)

混合检索与其他稀疏嵌入的结合¶

除了将语义搜索与BM25相结合外，Milvus还支持使用稀疏嵌入函数（如BGE-M3）进行混合检索。以下示例使用内置的BGEM3SparseEmbeddingFunction来生成稀疏嵌入。

首先，我们需要安装FlagEmbedding包：

In [ ]:

Copied!

! pip install -q FlagEmbedding
! pip install -q FlagEmbedding

接下来让我们使用默认的 OpenAI 模型进行稠密嵌入（dense embedding），并内置 BGE-M3 进行稀疏嵌入（sparse embedding）来构建向量存储和索引：

In [ ]:

Copied!





from llama_index.vector_stores.milvus.utils import BGEM3SparseEmbeddingFunction

vector_store = MilvusVectorStore(
    uri=URI,
    # token=TOKEN,
    dim=1536,
    enable_sparse=True,
    sparse_embedding_function=BGEM3SparseEmbeddingFunction(),
    overwrite=True,
)

storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex.from_documents(
    documents, storage_context=storage_context
)
from llama_index.vector_stores.milvus.utils import BGEM3SparseEmbeddingFunction

vector_store = MilvusVectorStore(
    uri=URI,
    # token=TOKEN,
    dim=1536,
    enable_sparse=True,
    sparse_embedding_function=BGEM3SparseEmbeddingFunction(),
    overwrite=True,
)

storage_context = StorageContext.from_defaults(vector_store=vector_store)
index = VectorStoreIndex.from_documents(
    documents, storage_context=storage_context
)

Fetching 30 files: 100%|██████████| 30/30 [00:00<00:00, 68871.99it/s]
2025-04-17 03:39:02,074 [DEBUG][_create_connection]: Created new connection using: ff4886e2f8da44e08304b748d9ac9b51 (async_milvus_client.py:547)
Chunks: 100%|██████████| 1/1 [00:00<00:00,  1.07it/s]

现在让我们通过一个示例问题执行混合搜索查询：

In [ ]:

Copied!





query_engine = index.as_query_engine(
    vector_store_query_mode="hybrid", similarity_top_k=5
)
response = query_engine.query("What did the author learn at Viaweb??")
print(textwrap.fill(str(response), 100))
query_engine = index.as_query_engine(
    vector_store_query_mode="hybrid", similarity_top_k=5
)
response = query_engine.query("What did the author learn at Viaweb??")
print(textwrap.fill(str(response), 100))

Chunks: 100%|██████████| 1/1 [00:00<00:00, 17.29it/s]

The author learned about retail, the importance of user feedback, the value of growth rate in a
startup, the significance of pricing strategy, the benefits of working on things that weren't
prestigious, and the challenges and rewards of running a startup.

自定义稀疏嵌入函数¶

只要继承自 BaseSparseEmbeddingFunction，您同样可以自定义稀疏嵌入函数，需要实现以下方法：

encode_queries：该方法将文本转换为查询的稀疏嵌入列表
encode_documents：该方法将文本转换为文档的稀疏嵌入列表

每个方法的输出应遵循稀疏嵌入的格式规范，即返回字典组成的列表。每个字典应包含一个键（整数类型）表示维度，以及对应的值（浮点类型）表示该维度上的嵌入强度（例如 {1: 0.5, 2: 0.3}）。

以下是通过 BGE-M3 实现自定义稀疏嵌入函数的示例：

In [ ]:

Copied!





from FlagEmbedding import BGEM3FlagModel
from typing import List
from llama_index.vector_stores.milvus.utils import BaseSparseEmbeddingFunction


class ExampleEmbeddingFunction(BaseSparseEmbeddingFunction):
    def __init__(self):
        self.model = BGEM3FlagModel("BAAI/bge-m3", use_fp16=False)

    def encode_queries(self, queries: List[str]):
        outputs = self.model.encode(
            queries,
            return_dense=False,
            return_sparse=True,
            return_colbert_vecs=False,
        )["lexical_weights"]
        return [self._to_standard_dict(output) for output in outputs]

    def encode_documents(self, documents: List[str]):
        outputs = self.model.encode(
            documents,
            return_dense=False,
            return_sparse=True,
            return_colbert_vecs=False,
        )["lexical_weights"]
        return [self._to_standard_dict(output) for output in outputs]

    def _to_standard_dict(self, raw_output):
        result = {}
        for k in raw_output:
            result[int(k)] = raw_output[k]
        return result
from FlagEmbedding import BGEM3FlagModel
from typing import List
from llama_index.vector_stores.milvus.utils import BaseSparseEmbeddingFunction


class ExampleEmbeddingFunction(BaseSparseEmbeddingFunction):
    def __init__(self):
        self.model = BGEM3FlagModel("BAAI/bge-m3", use_fp16=False)

    def encode_queries(self, queries: List[str]):
        outputs = self.model.encode(
            queries,
            return_dense=False,
            return_sparse=True,
            return_colbert_vecs=False,
        )["lexical_weights"]
        return [self._to_standard_dict(output) for output in outputs]

    def encode_documents(self, documents: List[str]):
        outputs = self.model.encode(
            documents,
            return_dense=False,
            return_sparse=True,
            return_colbert_vecs=False,
        )["lexical_weights"]
        return [self._to_standard_dict(output) for output in outputs]

    def _to_standard_dict(self, raw_output):
        result = {}
        for k in raw_output:
            result[int(k)] = raw_output[k]
        return result

自定义混合重排序器¶

Milvus 支持两种类型的重排序策略：互逆排序融合（RRF）和加权评分。MilvusVectorStore混合搜索默认使用k=60的RRF排序器。要自定义混合排序器，请修改以下参数：

hybrid_ranker (str)：指定混合搜索查询使用的排序器类型。当前仅支持["RRFRanker", "WeightedRanker"]。默认为"RRFRanker"。
hybrid_ranker_params (dict, 可选)：混合排序器的配置参数。该字典结构取决于所使用的具体排序器：
- 对于"RRFRanker"，应包含：
  - "k" (int)：用于互逆排序融合（RRF）算法的参数。该值用于计算RRF算法中的排序分数，该算法将多种排序策略合并为单一分数以提高搜索相关性。若未指定，默认值为60。
- 对于"WeightedRanker"，需要提供：
  - "weights" (float列表)：必须包含两个权重的列表：
    1. 稠密嵌入向量的权重
    2. 稀疏嵌入向量的权重这些权重用于平衡混合检索过程中稠密和稀疏嵌入向量组件的重要性。若未指定，默认权重为[1.0, 1.0]。

In [ ]:

Copied!





vector_store = MilvusVectorStore(
    uri=URI,
    # token=TOKEN,
    dim=1536,
    overwrite=False,  # Use the existing collection created in the previous example
    enable_sparse=True,
    hybrid_ranker="WeightedRanker",
    hybrid_ranker_params={"weights": [1.0, 0.5]},
)
index = VectorStoreIndex.from_vector_store(vector_store)
query_engine = index.as_query_engine(
    vector_store_query_mode="hybrid", similarity_top_k=5
)
response = query_engine.query("What did the author learn at Viaweb?")
print(textwrap.fill(str(response), 100))
vector_store = MilvusVectorStore(
    uri=URI,
    # token=TOKEN,
    dim=1536,
    overwrite=False,  # Use the existing collection created in the previous example
    enable_sparse=True,
    hybrid_ranker="WeightedRanker",
    hybrid_ranker_params={"weights": [1.0, 0.5]},
)
index = VectorStoreIndex.from_vector_store(vector_store)
query_engine = index.as_query_engine(
    vector_store_query_mode="hybrid", similarity_top_k=5
)
response = query_engine.query("What did the author learn at Viaweb?")
print(textwrap.fill(str(response), 100))

2025-04-17 03:44:00,419 [DEBUG][_create_connection]: Created new connection using: 09c051fb18c04f97a80f07958856587b (async_milvus_client.py:547)
Sparse embedding function is not provided, using default.
No built-in function detected, using BGEM3SparseEmbeddingFunction().
Fetching 30 files: 100%|██████████| 30/30 [00:00<00:00, 136622.28it/s]
Chunks: 100%|██████████| 1/1 [00:00<00:00,  1.07it/s]

The author learned several valuable lessons at Viaweb, including the importance of understanding
growth rate as the ultimate test of a startup, the significance of user feedback in shaping the
software, and the realization that web applications were the future of software development.
Additionally, the experience at Viaweb taught the author about the challenges and rewards of running
a startup, the value of simplicity in software design, and the impact of pricing strategies on
attracting customers.