Skip to main content
Open In ColabOpen on GitHub

Milvus

Milvus is a database that stores, indexes, and manages massive embedding vectors generated by deep neural networks and other machine learning (ML) models.

本笔记本展示了如何使用与 Milvus 向量数据库相关的功能。

设置​

你需要安装 langchain-milvus 并使用 pip install -qU langchain-milvus 来使用此集成。

pip install -qU langchain_milvus
Note: you may need to restart the kernel to use updated packages.

凭据​

使用 Milvus 向量存储时无需凭据。

初始化​

pip install -qU langchain-openai
import getpass
import os

if not os.environ.get("OPENAI_API_KEY"):
  os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter API key for OpenAI: ")

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

LangChain大型语言模型开发框架​

最简单的原型设计方法是使用 Milvus Lite，所有内容都存储在本地向量数据库文件中。只能使用 Flat 索引。

from langchain_milvus import Milvus

URI = "./milvus_example.db"

vector_store = Milvus(
    embedding_function=embeddings,
    connection_args={"uri": URI},
    index_params={"index_type": "FLAT", "metric_type": "L2"},
)
API 参考：Milvus

Milvus 服务器​

如果您有大量的数据（例如，超过一百万个向量），我们建议在 Docker 或 Kubernetes 上设置一个性能更高的 Milvus 服务器。

Milvus 服务器支持多种 索引。利用这些不同的索引可以显著提升检索能力并加快检索速度，以满足您的特定需求。

例如，考虑 Milvus 独立版的情况。要启动 Docker 容器，可以运行以下命令：

!curl -sfL https://raw.githubusercontent.com/milvus-io/milvus/master/scripts/standalone_embed.sh -o standalone_embed.sh

!bash standalone_embed.sh start
Password:

这里我们创建一个 Milvus 数据库：

from pymilvus import Collection, MilvusException, connections, db, utility

conn = connections.connect(host="127.0.0.1", port=19530)

# Check if the database exists
db_name = "milvus_demo"
try:
    existing_databases = db.list_database()
    if db_name in existing_databases:
        print(f"Database '{db_name}' already exists.")

        # Use the database context
        db.using_database(db_name)

        # Drop all collections in the database
        collections = utility.list_collections()
        for collection_name in collections:
            collection = Collection(name=collection_name)
            collection.drop()
            print(f"Collection '{collection_name}' has been dropped.")

        db.drop_database(db_name)
        print(f"Database '{db_name}' has been deleted.")
    else:
        print(f"Database '{db_name}' does not exist.")
        database = db.create_database(db_name)
        print(f"Database '{db_name}' created successfully.")
except MilvusException as e:
    print(f"An error occurred: {e}")
Database 'milvus_demo' does not exist.
Database 'milvus_demo' created successfully.

请注意以下URI的变化。实例初始化后，导航至 http://127.0.0.1:9091/webui 查看本地Web界面。

以下是一个使用 Milvus 数据库服务创建向量存储实例的示例：

from langchain_milvus import BM25BuiltInFunction, Milvus

URI = "http://localhost:19530"

vectorstore = Milvus(
    embedding_function=embeddings,
    connection_args={"uri": URI, "token": "root:Milvus", "db_name": "milvus_demo"},
    index_params={"index_type": "FLAT", "metric_type": "L2"},
    consistency_level="Strong",
    drop_old=False,  # set to True if seeking to drop the collection with that name if it exists
)

If you want to use Zilliz Cloud, the fully managed cloud service for Milvus, please adjust the uri and token, which correspond to the Public Endpoint and Api key in Zilliz Cloud.

使用 Milvus 集合对数据进行分隔​

你可以在同一个 Milvus 实例中的不同集合里存储不相关的文档。

以下是创建新集合的方法：

from langchain_core.documents import Document

vector_store_saved = Milvus.from_documents(
    [Document(page_content="foo!")],
    embeddings,
    collection_name="langchain_example",
    connection_args={"uri": URI},
)
API 参考：文档

这就是如何检索存储的集合：

vector_store_loaded = Milvus(
    embeddings,
    connection_args={"uri": URI},
    collection_name="langchain_example",
)

管理向量存储​

创建向量存储后，我们可以通过添加和删除不同条目来与其交互。

将项目添加到向量存储​

我们可以通过使用add_documents函数将项目添加到我们的向量存储中。

from uuid import uuid4

from langchain_core.documents import Document

document_1 = Document(
    page_content="I had chocolate chip pancakes and scrambled eggs for breakfast this morning.",
    metadata={"source": "tweet"},
)

document_2 = Document(
    page_content="The weather forecast for tomorrow is cloudy and overcast, with a high of 62 degrees.",
    metadata={"source": "news"},
)

document_3 = Document(
    page_content="Building an exciting new project with LangChain - come check it out!",
    metadata={"source": "tweet"},
)

document_4 = Document(
    page_content="Robbers broke into the city bank and stole $1 million in cash.",
    metadata={"source": "news"},
)

document_5 = Document(
    page_content="Wow! That was an amazing movie. I can't wait to see it again.",
    metadata={"source": "tweet"},
)

document_6 = Document(
    page_content="Is the new iPhone worth the price? Read this review to find out.",
    metadata={"source": "website"},
)

document_7 = Document(
    page_content="The top 10 soccer players in the world right now.",
    metadata={"source": "website"},
)

document_8 = Document(
    page_content="LangGraph is the best framework for building stateful, agentic applications!",
    metadata={"source": "tweet"},
)

document_9 = Document(
    page_content="The stock market is down 500 points today due to fears of a recession.",
    metadata={"source": "news"},
)

document_10 = Document(
    page_content="I have a bad feeling I am going to get deleted :(",
    metadata={"source": "tweet"},
)

documents = [
    document_1,
    document_2,
    document_3,
    document_4,
    document_5,
    document_6,
    document_7,
    document_8,
    document_9,
    document_10,
]
uuids = [str(uuid4()) for _ in range(len(documents))]

vector_store.add_documents(documents=documents, ids=uuids)
API 参考：文档

从向量存储中删除项目​

vector_store.delete(ids=[uuids[-1]])
(insert count: 0, delete count: 1, upsert count: 0, timestamp: 0, success count: 0, err count: 0, cost: 0)

查询向量存储​

一旦你的向量存储创建完成并且相关文档已添加，你很可能希望在链或代理运行时对其进行查询。

直接查询

通过元数据过滤进行简单的相似性搜索可以如下操作：

results = vector_store.similarity_search(
    "LangChain provides abstractions to make working with LLMs easy",
    k=2,
    expr='source == "tweet"',
)
for res in results:
    print(f"* {res.page_content} [{res.metadata}]")
* Building an exciting new project with LangChain - come check it out! [{'pk': '9905001c-a4a3-455e-ab94-72d0ed11b476', 'source': 'tweet'}]
* LangGraph is the best framework for building stateful, agentic applications! [{'pk': '1206d237-ee3a-484f-baf2-b5ac38eeb314', 'source': 'tweet'}]

带分数的相似性搜索​

您也可以按分数搜索：

results = vector_store.similarity_search_with_score(
    "Will it be hot tomorrow?", k=1, expr='source == "news"'
)
for res, score in results:
    print(f"* [SIM={score:3f}] {res.page_content} [{res.metadata}]")
* [SIM=21192.628906] bar [{'pk': '2', 'source': 'https://example.com'}]

有关使用 Milvus 向量存储时可用的所有搜索选项的完整列表，您可以访问 API 参考。

通过转换为检索器进行查询​

您还可以将向量存储转换为检索器，以便在链中更轻松地使用。

retriever = vector_store.as_retriever(search_type="mmr", search_kwargs={"k": 1})
retriever.invoke("Stealing from the bank is a crime", filter={"source": "news"})
[Document(metadata={'pk': 'eacc7256-d7fa-4036-b1f7-83d7a4bee0c5', 'source': 'news'}, page_content='Robbers broke into the city bank and stole $1 million in cash.')]

最常见的混合搜索场景是密集 + 稀疏混合搜索，其中候选结果通过语义向量相似性和精确关键词匹配两种方式检索。这些方法的结果被合并、重新排序，并传递给大型语言模型（LLM）以生成最终答案。这种方法在精度和语义理解之间取得了平衡，使其在各种查询场景中非常有效。

全文搜索​

自 Milvus 2.5 以来，通过将 BM25 算法表示为稀疏向量的方式，原生支持了全文搜索。Milvus 接受原始文本作为输入，并自动将其转换为存储在指定字段中的稀疏向量，从而无需手动生成稀疏嵌入。

对于全文搜索，Milvus VectorStore 接受一个 builtin_function 参数。通过此参数，您可以传入 BM25BuiltInFunction 的实例。这与语义搜索不同，后者通常将密集嵌入传递给 VectorStore，

这是一个使用 Milvus 的混合搜索简单示例，其中 OpenAI 稠密嵌入用于语义搜索，BM25 用于全文搜索：

from langchain_milvus import BM25BuiltInFunction, Milvus
from langchain_openai import OpenAIEmbeddings

vectorstore = Milvus.from_documents(
    documents=documents,
    embedding=OpenAIEmbeddings(),
    builtin_function=BM25BuiltInFunction(),
    # `dense` is for OpenAI embeddings, `sparse` is the output field of BM25 function
    vector_field=["dense", "sparse"],
    connection_args={
        "uri": URI,
    },
    consistency_level="Strong",
    drop_old=True,
)
  • When you use BM25BuiltInFunction, please note that the full-text search is available in Milvus Standalone and Milvus Distributed, but not in Milvus Lite, although it is on the roadmap for future inclusion. It will also be available in Zilliz Cloud (fully-managed Milvus) soon. Please reach out to support@zilliz.com for more information.

在上面的代码中，我们定义了一个 BM25BuiltInFunction 的实例，并将其传递给 Milvus 对象。BM25BuiltInFunction 是 Milvus 中 Function 的一个轻量级包装类。我们可以使用它与 OpenAIEmbeddings 一起初始化一个密集 + 稀疏混合搜索的 Milvus 向量存储实例。

BM25BuiltInFunction 不需要客户端传递语料库或进行训练，所有操作都在 Milvus 服务器端自动完成，因此用户无需关心任何词汇和语料库。此外，用户还可以自定义 分析器 来实现 BM25 中的自定义文本处理。

重新排序候选者​

在检索的第一阶段之后，我们需要对候选结果进行重新排序以获得更好的结果。您可以参考 重新排序 以获取更多信息。

这是一个加权重排序的例子：

query = "What are the novels Lila has written and what are their contents?"

vectorstore.similarity_search(
    query, k=1, ranker_type="weighted", ranker_params={"weights": [0.6, 0.4]}
)

有关全文搜索和混合搜索的更多信息，请参阅 使用 LangChain 和 Milvus 进行全文搜索 和 使用 LangChain 和 Milvus 进行混合检索。

检索增强生成的用法​

有关如何使用此向量存储进行检索增强生成 (RAG) 的指南，请参阅以下部分：

按用户检索​

在构建检索应用程序时，通常需要考虑到多个用户的需求。这意味着你可能不仅要为一个用户存储数据，还要为许多不同的用户存储数据，并且他们不应该能够看到彼此的数据。

Milvus 建议使用 partition_key 来实现多租户。以下是一个示例：

The Partition key feature is not available in Milvus Lite, if you want to use it, you need to start Milvus server, as mentioned above.

from langchain_core.documents import Document

docs = [
    Document(page_content="i worked at kensho", metadata={"namespace": "harrison"}),
    Document(page_content="i worked at facebook", metadata={"namespace": "ankush"}),
]
vectorstore = Milvus.from_documents(
    docs,
    embeddings,
    connection_args={"uri": URI},
    drop_old=True,
    partition_key_field="namespace",  # Use the "namespace" field as the partition key
)
API 参考：文档

要使用分区键进行搜索，您应该在搜索请求的布尔表达式中包含以下任一项：

search_kwargs={"expr": '<partition_key> == "xxxx"'}

search_kwargs={"expr": '<partition_key> == in ["xxx", "xxx"]'}

请将 <partition_key> 替换为被指定为主分区键的字段名称。

Milvus 根据指定的分区键进行分区，根据分区键过滤实体，并在过滤后的实体中进行搜索。

# This will only get documents for Ankush
vectorstore.as_retriever(search_kwargs={"expr": 'namespace == "ankush"'}).invoke(
    "where did i work?"
)
[Document(page_content='i worked at facebook', metadata={'namespace': 'ankush'})]
# This will only get documents for Harrison
vectorstore.as_retriever(search_kwargs={"expr": 'namespace == "harrison"'}).invoke(
    "where did i work?"
)
[Document(page_content='i worked at kensho', metadata={'namespace': 'harrison'})]

API 参考​

有关 __ModuleName__VectorStore 的所有功能和配置的详细文档，请访问 API 参考： https://python.langchain.com/api_reference/milvus/vectorstores/langchain_milvus.vectorstores.milvus.Milvus.html