在AI应用爆发的今天，非结构化数据——图片、文本、音视频——已经占到了全球数据总量的80%以上。传统关系型数据库擅长处理规整的表格数据，但对这些“模糊”的数据形式却束手无策。于是，向量数据库应运而生。而Milvus，正是这个领域最受关注的开源项目之一。

什么是Milvus?

Milvus是一款开源的云原生向量数据库，专为海量向量数据集的高性能相似性搜索而设计。它诞生于2019年，目前是LF AI & Data Foundation旗下的顶级开源项目，采用Apache 2.0许可协议。

简单来说，Milvus解决的是“模糊查找”问题：在亿级高维向量中，快速找到与目标向量最相似的那些结果。它是AI时代的关键基础设施，服务于推荐系统、图像搜索、RAG（检索增强生成）等场景。

Milvus构建在Faiss、HNSW、DiskANN等流行向量搜索库之上，并进行了深度优化。根据公开的基准测试，Milvus的性能通常是其他向量数据库的2-5倍。

核心概念速览

在深入之前，先理解三个关键概念：

1. 非结构化数据：指图像、文本、音视频等不符合预定义模型的数据。计算机无法直接理解它们。

2. 向量嵌入（Embedding）：通过AI模型将非结构化数据转换成的数字列表（浮点数数组）。它捕捉了数据的语义特征。

3. 向量相似性搜索：计算两个向量之间的距离（如欧氏距离、内积），距离越近代表原始数据越相似。

Milvus核心架构

• 设计核心总结为四个字：“存算分离”
  

1. 接入层：Proxy

• 这是整个Milvus集群的API网关，客户端SDK所有的请求都会先打到Proxy

• 作用：路由开发、鉴权、限流、负载均衡

补充：
路由开发：根据请求的URL或参数，决定把这个请求转发给后端的哪个具体服务。
鉴权：身份认证和权限校验。
限流：当瞬间请求量过大时，主动拒绝一部分请求，防止后端服务被压垮。
负载均衡：当同一个服务有多台服务器时，把请求均匀分配给它们（轮询、随机、一致性hash）
客户端的负载均衡如何做？
从注册中心拉取服务列表，在本地缓存该列表，然后基于某种负载均衡算法（随机、轮询、一致性hash）

2. 协调层：Coordinator

• 相当一个司令，作用是注册和指挥。

• 工作：1. 注册（就是从DCL/DDL这两个元数据存储到etcd里面） 2. 指挥workers建立索引，建立数据以及索引的存储和查询。

3. 执行层：Worker Nodes
   3.1 streaming nodes:
   • 将DML语句进行流式处理（将命令追加到日志文件，类似于redis中的AOF），最终是用kafka实现日志的保存以及分析。
   • 将DML语句保存到minio中。
     • 流式结点内部有两个segment：
     • 1. 全量段。一旦存满后自动刷盘。
     • 2. 增量段。人工干预刷盘。

3.2 query nodes:
- 对接前端关键字查询或语义查询。数据的源头是从minio加载上来的。
- 两路操作：
- 1. DQL直接操作Query Node(冷数据：直接从Object Storage(minio)中加载数据)，然后进行查询。
- 2. DQL还可以走到流式结点（热数据）：写数据的同时查询数据（DML将新数据写到流式结点，然后DQL从新数据中去查询）。此时就会带来一个问题：高并发场景下读写不一致问题。解决方案：写时复刻；读写缓存、只读缓存。

3.3 data node:真正存放数据以及索引数据。

4. 存储层：Storage

• 数据插入流程
• 向量检索流程

部署方式

docker桌面版Windows安装，首先要有milvus_image.tar文件。

1. docker load -i milvus_image.tar

2. Invoke-WebRequest https://raw.githubusercontent.com/milvus-io/milvus/refs/heads/master/scripts/standalone_embed.bat -OutFile standalone.bat

3. .\standalone.bat start

Milvus核心概念

• Database(库) --> Collection（表） --> Entity（行/记录） --> Filed（列/字段） --> Schema（表结构定义） --> Index（索引）

• Milvus支持三大字段：

  • 主键 ：支持自动增长
  • 向量字段 ：语义搜索（稠密向量（HNSW/FLAT）和 稀疏向量（数据类型为SPARSE_FLOAT_VES））
  • 标量字： 关键字搜索（精确匹配）

• 索引：加速搜索的关键

  • 稠密向量搜索：HNSW（分层可导航小世界图，最流行的高校近似算法。上层稀疏、下称稠密。），FLAT（暴力检索：最笨但最准。把库里的每个向量都和你的查询向量算一遍距离。）
  • 稀疏向量索引：SPARSE_INVERTED_INDEX

• 相似度度量

  • COSINE: 余弦相似度，适用场景：语义相似度、归一化向量；值越大越相似。
  • IP: 内积；适用场景是稀疏向量；值越大越相似。
  • L2: 欧氏距离；适配场景：图像检索；值越小越相似。

基本操作流程（原理块）

一、核心存储单元：segment及其生命周期

• 完整生命周期

Growing（增长中） → Sealed（已封存） → Flushing（刷盘中） → Flushed（已刷盘） → Indexed（已索引） → Dropped（已删除）

Growing：正在接收写入数据，位于内存中。

Sealed：停止接收写入，等待持久化。

Flushing：正在将数据写入对象存储（MinIO/S3）。

Flushed：数据已完整写入对象存储，生成 Binlog 文件。

Indexed：Index Node 已为该 Segment 构建好向量索引。

Dropped：Segment 被标记删除，等待 GC 回收存储空间。

二、数据写入流程

客户端 → Proxy（分配TSO、路由） → Log Broker（Pulsar/Kafka） → DataNode（消费、构建Growing Segment、落盘） → 对象存储

**步骤 1：**请求接入与分片路由（Proxy）

客户端 SDK 发起插入请求后，首先到达接入层（Access Layer） 的 Proxy 节点。

1. 身份验证与校验：Proxy 对请求进行鉴权，并校验 Schema 合法性。

2. 分配全局时间戳（TSO）：Proxy 向 Root Coordinator 中的 TSO（Timestamp Oracle） 模块申请一个全局唯一且单调递增的时间戳。这个时间戳决定了数据的可见性顺序，是保证分布式一致性的基础。

3. 数据分片（Sharding）：Proxy 根据主键（Primary Key）的哈希值，按照 Collection 定义的分片数（shards_num）将数据路由到对应的虚拟通道（vchannel）。每个 vchannel 映射到一个物理通道（pchannel），绑定到特定的 Streaming Node（流节点）。

**步骤 2：**日志持久化（Streaming Node & Log Broker）
数据被发送到对应的 Streaming Node 后：

1. 写入 WAL（预写日志）：Streaming Node 立即将数据写入底层的日志中间件（默认 Pulsar 或 Kafka）。这是数据持久化的第一道防线——只要数据成功写入 WAL，即使系统崩溃，也能通过重放 WAL 完全恢复。

2. 异步消费：DataNode 和 QueryNode 作为消费者，异步地从 Log Broker 订阅并拉取数据。这就是为什么 insert() 返回成功时，数据不一定立即可见——它只是被“承诺”到了 WAL 中。

📌 常见误区：很多用户发现插入 100 万条数据后，total_rows 显示少于 100 万。这是因为 DataNode 消费 WAL 存在数百毫秒到数秒的延迟，total_rows 只统计已被 DataNode 消费的行数。

**步骤 3：**构建 Growing Segment（DataNode）
DataNode 从 Log Broker 消费到数据后：

1. 内存缓存：每个分片（shard）在每个分区（partition）下对应一个 Growing Segment。DataNode 将数据缓存在该 Growing Segment 的 InsertRecord 结构中。

2. 增量落盘（Chunk 机制）：DataNode 不会等到 Segment 写满才落盘。当缓存大小达到 16MB（dataNode.segment.insertBufSize）或距离上次落盘超过 600 秒（dataNode.segment.syncPeriod）时，会将缓存数据作为一个 Chunk（数据块） 写入对象存储（MinIO/S3）。每个字段的数据会被分别写入独立的文件。

**步骤 4：**封存与刷盘（Seal & Flush）
当 Growing Segment 累积的数据量达到阈值（默认约 122MB，由 dataCoord.segment.maxSize 和 sealProportion 共同决定）时：

1. 封存（Seal）：DataCoord 将该 Segment 标记为 Sealed，不再接收新数据。同时创建一个新的 Growing Segment 继续服务写入。

2. 刷盘（Flush）：DataNode 将 Sealed Segment 的所有数据完整写入对象存储，生成 Binlog 文件（包含插入日志、统计日志、删除日志等）。

3. 元数据更新：DataCoord 更新 etcd 中的元数据，记录该 Segment 的状态为 Flushed。

**步骤 5：**异步索引构建（IndexNode）
Segment 完成 Flush 后，IndexNode（索引节点） 会异步地为该 Segment 构建向量索引：

1. 从对象存储加载该 Segment 的 Binlog 数据到内存。

2. 根据用户指定的索引类型（如 HNSW、IVF_FLAT）构建索引结构。

3. 将构建好的索引文件序列化，写回对象存储（路径为 index_files/）。

📌 重要：索引构建是异步的，不会阻塞写入。索引未完成时，查询会回退到暴力搜索（Brute Force），待索引加载完成后自动切换。

三、数据查询流程

客户端 → Proxy（分发） → QueryNode（查Sealed数据） + StreamingNode（查Growing数据） → 多级归并 → 返回结果

一次 search 请求的内部执行流程如下：

**步骤 1：**请求接入与路由（Proxy）

1. 客户端发起搜索请求到达 Proxy。

2. Proxy 向 QueryCoord（查询协调器） 请求集群中所有 QueryNode 的 Segment 分布信息（缓存）。

3. Proxy 将搜索请求并行广播给所有相关的 QueryNode（查询节点） 和 StreamingNode（流节点）。

**步骤 2：**分段查询（并行执行）
Milvus 将查询任务拆分为两部分并行处理：

• 针对 Growing Segment（实时热数据）：请求发送给 StreamingNode。StreamingNode 直接在内存中对 Growing Segment 的原始向量进行暴力扫描（或使用临时内存索引）。

• 针对 Sealed Segment（历史冷数据）：请求发送给持有该 Segment 副本的 QueryNode。QueryNode 会从对象存储加载该 Segment 的索引文件到内存（或使用 mmap 内存映射），然后利用索引（如 HNSW）进行近似最近邻搜索（ANN），大幅加速检索。

**步骤 3：**标量过滤与向量检索（QueryNode 内部）
QueryNode 在本地执行查询时，会进行以下操作：

1. 标量预过滤：如果查询带有标量过滤条件（如 price > 100），首先利用标量索引（如倒排索引、Bloom Filter）进行剪枝，缩小候选集范围。

2. ANN 搜索：在剩余的候选集中，使用向量索引执行近似最近邻搜索，返回 Top-K 候选向量。

3. 精确距离计算：对 ANN 返回的候选向量，精确计算与查询向量的距离（L2/IP/Cosine），得到局部 Top-K 结果。

**步骤 4：**多级结果归并
各个节点计算完成后，需要将结果汇总：

1. 第一级归并（QueryNode 内部）：每个 QueryNode 合并自己负责的多个 Segment 的局部结果。

2. 第二级归并（StreamingNode）：StreamingNode 归并所有 QueryNode 返回的局部 Top-K 结果，并与自己处理的 Growing Segment 结果合并。

3. 第三级归并（Proxy）：Proxy 收集所有 StreamingNode 返回的中间结果，进行全局归并排序，剔除重复数据，得到最终的 Global Top-K。

**步骤 5：**结果返回与 Handoff
最终结果通过 Proxy 返回给客户端。当 Growing Segment 被刷盘为 Sealed Segment 并完成索引构建后，QueryCoord 会触发 Handoff（交接） 操作，将新 Sealed Segment 的索引加载到 QueryNode 内存中，并释放对应的 Growing Segment 资源，确保后续查询直接使用索引。

总结

• 写入路径

客户端 → Proxy（TSO + 路由） → Log Broker（WAL 持久化） → DataNode（消费 + Growing Segment + 异步落盘） → IndexNode（异步构建索引）

• 查询路径

客户端 → Proxy（分发） → StreamingNode（查 Growing） + QueryNode（加载索引查 Sealed） → 多级归并 → 返回结果

基本操作流程（代码块）

1. 定义schema

# 1. 构建schema
def build_schema() -> CollectionSchema:
    return (
        MilvusClient.create_schema(auto_id=True)
        # 主键：自动生成ID
        .add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
        # 稠密向量：用于语义检索，维度与嵌入模型一致
        .add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=1024)
        # 稀疏向量：用于关键词检索
        .add_field(field_name="sparse_vector",
        datatype=DataType.SPARSE_FLOAT_VECTOR)
        # 原始文本：存储Chunk内容，检索后返回给用户
        .add_field(field_name="text", datatype=DataType.VARCHAR, max_length=1500)
        # 元数据：存储来源、页码等信息，支持过滤
        .add_field(field_name="metadata", datatype=DataType.JSON)
    )

2. 创建索引

# 2. 配置索引，只针对向量类型（稠密|稀疏）来配置索引
def build_index() -> IndexParams:
    index_params = MilvusClient.prepare_index_params()

    # 稠密向量：使用HNSW索引 + L2度量
    index_params.add_index(
        field_name="vector",
        index_type="HNSW",
        metric_type="L2",
    )

    # 稀疏向量：使用倒排索引 + IP度量
    index_params.add_index(
        field_name="sparse_vector",
        index_type="SPARSE_INVERTED_INDEX",
        metric_type="IP",
    )

    return index_params

def get_milvus_client():
    """
    创建Milvus客户端
    :return:
    """
    return MilvusClient(uri="http://localhost:19530", token="")

def create_collection(client:MilvusClient):
    """
    创建collection（集合）
    :param client:
    :return:
    """
    collection_name = "demo_collection"
    # 删除集合
    client.drop_collection(collection_name=collection_name)
    # 判断是否存在集合，如果不存在则创建
    if not client.has_collection(collection_name=collection_name):
        client.create_collection(
            collection_name=collection_name,
            schema=build_schema(),
            index_params=build_index(),
        )

3. 插入数据

3.1 加载数据

 # 1、加载文件
    doc_list = UnstructuredWordDocumentLoader(
    "./demo.docx", mode="single"
    ).load()

3.2 文档切分

 # 2、切分文件
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=50, chunk_overlap=50, separators=["\n\n", "\n", "。",""]
    )
    splitted_doc_list = text_splitter.split_documents(doc_list)

3.3 文档的向量化

# 3、构建向量（同时生成稠密和稀疏向量）
    model = BGEM3FlagModel(r"D:/bge/bge-m3")
    all_vectors = model.encode(
        [doc.page_content for doc in splitted_doc_list],
        return_dense=True, return_sparse=True
    )

# 4、准备数据：组装成List[Dict]，每个Dict对应Schema中的字段
    insert_data_list = []
    for doc, dense_vector, sparse_vector in zip(
            splitted_doc_list,
            all_vectors["dense_vecs"],
            all_vectors['lexical_weights']
    ):
        insert_data_list.append({
            "vector": dense_vector,
            "sparse_vector": sparse_vector,
            "metadata": doc.metadata,
            "text": doc.page_content
        })

    # 5、插入数据
    res = client.insert(collection_name=collection_name, data=insert_data_list)
    print(res)

4. 搜索相似向量

4.1 稠密向量查询

def encode_query(model,query):
    """使用 BGE-M3 模型同时生成查询的稠密向量和稀疏向量"""
    vectors = model.encode(
        [query],
        return_dense=True,
        return_sparse=True
    )
    # 提取列表中的第一个（也是唯一一个）元素的向量
    dense_vec = vectors["dense_vecs"][0]
    sparse_vec = vectors["lexical_weights"][0]
    return dense_vec, sparse_vec


# 5. 稠密向量查询
def dense_vector_search_example(client, query: str, limit: int = 5):

    model = BGEM3FlagModel(r"D:/bge/bge-m3")

    # 将用户的提问与模型一并传入该方法进行向量化（将用户的提问进行向量化），生成稠密向量
    dense_vec, _ = encode_query(model, query)

    # 用户的提问（提示词）已经向量化了，然后去查询向量数据库，返回相似度匹配的chunk（文档块）
    # 这样返回的文档块还需进行重排（因为还有稀疏向量的查询），也就是说，我们需要将稠密和稀疏做混合查询后再经过reranker(重排模型)进行排序
    # 这样返回的结果才是最终的结果
    results = client.search(
        collection_name="demo_collection",
        data=[dense_vec],
        anns_field="vector",
        limit=5,
        search_params={"metric_type": "L2"},
        output_fields=["id", "text", "metadata"]
    )
    return results[0]

4.2 稀疏向量查询

# 6. 稀疏向量查询
def sparse_vector_search_example(client, query: str, limit: int = 5):
    model = BGEM3FlagModel(r"D:/bge/bge-m3")
    _, sparse_vec = encode_query(model, query)

    results = client.search(
        collection_name="demo_collection",
        data=[sparse_vec],
        anns_field="sparse_vector",
        limit=limit,
        search_params={"metric_type": "IP"},
        output_fields=["id", "text", "metadata"],
    )
    return results[0]

4.3 混合检索与重排序

def hybrid_vector_search_example_rrf(client, query: str, limit: int = 5):

    model = BGEM3FlagModel(r"D:/bge/bge-m3")
    dense_vec, sparse_vec = encode_query(model, query)
    dense_req = AnnSearchRequest(
        data=[dense_vec], anns_field="vector",
        param={"metric_type": "COSINE"}, limit=limit,
    )
    sparse_req = AnnSearchRequest(
        data=[sparse_vec], anns_field="sparse_vector",
        param={"metric_type": "IP"}, limit=limit,
    )
    results = client.hybrid_search(
        collection_name="demo_collection",
        reqs=[dense_req, sparse_req],
        ranker=RRFRanker(k=60),
        limit=limit,
        output_fields=["id", "text", "metadata"],
    )
    return results[0]

4.4 标量查询

# 标量查询
def scalar_query_examples(client, keyword: str = "大模型"):
    # 对text字段进行模糊检索
    like_res = client.query(
        collection_name="demo_collection",
        filter=f'text like "%{keyword}%"',
        output_fields=["id", "text"],
        limit=5,
    )
    # 对metadata的JSON字段进行检索
    json_res = client.query(
        collection_name="demo_collection",
        filter='metadata["source"] like "%sample%"',
        output_fields=["id", "metadata"],
        limit=5,
    )

5. 完整代码

from pymilvus import MilvusClient, DataType, CollectionSchema, AnnSearchRequest
from pymilvus.milvus_client.index import  IndexParams
from langchain_community.document_loaders import UnstructuredWordDocumentLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from pymilvus import AnnSearchRequest, RRFRanker
from FlagEmbedding import BGEM3FlagModel

# 1. 构建schema
def build_schema() -> CollectionSchema:
    return (
        MilvusClient.create_schema(auto_id=True)
        # 主键：自动生成ID
        .add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
        # 稠密向量：用于语义检索，维度与嵌入模型一致
        .add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=1024)
        # 稀疏向量：用于关键词检索
        .add_field(field_name="sparse_vector",
        datatype=DataType.SPARSE_FLOAT_VECTOR)
        # 原始文本：存储Chunk内容，检索后返回给用户
        .add_field(field_name="text", datatype=DataType.VARCHAR, max_length=1500)
        # 元数据：存储来源、页码等信息，支持过滤
        .add_field(field_name="metadata", datatype=DataType.JSON)
    )


# 2. 配置索引，只针对向量类型（稠密|稀疏）来配置索引
def build_index() -> IndexParams:
    index_params = MilvusClient.prepare_index_params()

    # 稠密向量：使用HNSW索引 + L2度量
    index_params.add_index(
        field_name="vector",
        index_type="HNSW",
        metric_type="L2",
    )

    # 稀疏向量：使用倒排索引 + IP度量
    index_params.add_index(
        field_name="sparse_vector",
        index_type="SPARSE_INVERTED_INDEX",
        metric_type="IP",
    )

    return index_params


# 3. 创建collection
def get_milvus_client():
    """
    创建Milvus客户端
    :return:
    """
    return MilvusClient(uri="http://localhost:19530", token="")

def create_collection(client:MilvusClient):
    """
    创建collection（集合）
    :param client:
    :return:
    """
    collection_name = "demo_collection"
    # 删除集合
    client.drop_collection(collection_name=collection_name)
    # 判断是否存在集合，如果不存在则创建
    if not client.has_collection(collection_name=collection_name):
        client.create_collection(
            collection_name=collection_name,
            schema=build_schema(),
            index_params=build_index(),
        )


# 4. 插入数据
def insert_data(client: MilvusClient, collection_name: str):

    # 1、加载文件
    doc_list = UnstructuredWordDocumentLoader(
    "./demo.docx", mode="single"
    ).load()

    # 2、切分文件
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=50, chunk_overlap=50, separators=["\n\n", "\n", "。",""]
    )
    splitted_doc_list = text_splitter.split_documents(doc_list)

    # 3、构建向量（同时生成稠密和稀疏向量）
    model = BGEM3FlagModel(r"D:/bge/bge-m3")
    all_vectors = model.encode(
        [doc.page_content for doc in splitted_doc_list],
        return_dense=True, return_sparse=True
    )

    # 4、准备数据：组装成List[Dict]，每个Dict对应Schema中的字段
    insert_data_list = []
    for doc, dense_vector, sparse_vector in zip(
            splitted_doc_list,
            all_vectors["dense_vecs"],
            all_vectors['lexical_weights']
    ):
        insert_data_list.append({
            "vector": dense_vector,
            "sparse_vector": sparse_vector,
            "metadata": doc.metadata,
            "text": doc.page_content
        })

    # 5、插入数据
    res = client.insert(collection_name=collection_name, data=insert_data_list)
    print(res)




def encode_query(model,query):
    """使用 BGE-M3 模型同时生成查询的稠密向量和稀疏向量"""
    vectors = model.encode(
        [query],
        return_dense=True,
        return_sparse=True
    )
    # 提取列表中的第一个（也是唯一一个）元素的向量
    dense_vec = vectors["dense_vecs"][0]
    sparse_vec = vectors["lexical_weights"][0]
    return dense_vec, sparse_vec


# 5. 稠密向量查询
def dense_vector_search_example(client, query: str, limit: int = 5):

    model = BGEM3FlagModel(r"D:/bge/bge-m3")

    # 将用户的提问与模型一并传入该方法进行向量化（将用户的提问进行向量化），生成稠密向量
    dense_vec, _ = encode_query(model, query)

    # 用户的提问（提示词）已经向量化了，然后去查询向量数据库，返回相似度匹配的chunk（文档块）
    # 这样返回的文档块还需进行重排（因为还有稀疏向量的查询），也就是说，我们需要将稠密和稀疏做混合查询后再经过reranker(重排模型)进行排序
    # 这样返回的结果才是最终的结果
    results = client.search(
        collection_name="demo_collection",
        data=[dense_vec],
        anns_field="vector",
        limit=5,
        search_params={"metric_type": "L2"},
        output_fields=["id", "text", "metadata"]
    )
    return results[0]

# 6. 稀疏向量查询
def sparse_vector_search_example(client, query: str, limit: int = 5):
    model = BGEM3FlagModel(r"D:/bge/bge-m3")
    _, sparse_vec = encode_query(model, query)

    results = client.search(
        collection_name="demo_collection",
        data=[sparse_vec],
        anns_field="sparse_vector",
        limit=limit,
        search_params={"metric_type": "IP"},
        output_fields=["id", "text", "metadata"],
    )
    return results[0]


# 混合检索与重排序
def hybrid_vector_search_example_rrf(client, query: str, limit: int = 5):

    model = BGEM3FlagModel(r"D:/bge/bge-m3")
    dense_vec, sparse_vec = encode_query(model, query)
    dense_req = AnnSearchRequest(
        data=[dense_vec], anns_field="vector",
        param={"metric_type": "COSINE"}, limit=limit,
    )
    sparse_req = AnnSearchRequest(
        data=[sparse_vec], anns_field="sparse_vector",
        param={"metric_type": "IP"}, limit=limit,
    )
    results = client.hybrid_search(
        collection_name="demo_collection",
        reqs=[dense_req, sparse_req],
        ranker=RRFRanker(k=60),
        limit=limit,
        output_fields=["id", "text", "metadata"],
    )
    return results[0]

    # 向量搜索（相似度匹配 | 语义匹配）
    client.search()

    # 标量搜索
    client.query()

# 标量查询
def scalar_query_examples(client, keyword: str = "大模型"):
    # 对text字段进行模糊检索
    like_res = client.query(
        collection_name="demo_collection",
        filter=f'text like "%{keyword}%"',
        output_fields=["id", "text"],
        limit=5,
    )
    # 对metadata的JSON字段进行检索
    json_res = client.query(
        collection_name="demo_collection",
        filter='metadata["source"] like "%sample%"',
        output_fields=["id", "metadata"],
        limit=5,
    )

if __name__ == '__main__':
    # 创建collection表，创建成功就不用再创了
    # create_collection(get_milvus_client())

    # 这是插入数据
    # insert_data()

    # 这是稠密向量查询
    resp = dense_vector_search_example("如何使用Fastapi实现文件上传",get_milvus_client())
    print(resp)

    # 这是稀疏向量查询
    resp1 = sparse_vector_search_example("如何使用Fastapi实现文件上传", get_milvus_client())
    print(resp1)

    # 混合检索与重排序
    resp = hybrid_vector_search_example_rrf(get_milvus_client(), "如何使用FastAPI实现文件上传")
    print(resp)