Milvus 使用方式与典型场景

使用方式与典型场景

下面两个例子用 pymilvus 的 MilvusClient 接口，它比旧的 ORM 风格 API 更简洁，推荐新项目使用。代码可以直接复制运行，前提是本地已经有一个 Milvus 实例，或者换成 Milvus Lite 的文件路径。

示例一：建 collection、建索引、写入与检索

# pip install pymilvus
from pymilvus import MilvusClient, DataType

client = MilvusClient(uri="http://localhost:19530", token="root:Milvus")

# 1. 定义 schema
schema = client.create_schema(auto_id=False, enable_dynamic_field=True)
schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
schema.add_field(field_name="tenant_id", datatype=DataType.INT64)
schema.add_field(field_name="category", datatype=DataType.VARCHAR, max_length=64)
schema.add_field(field_name="text", datatype=DataType.VARCHAR, max_length=4096)
schema.add_field(field_name="embedding", datatype=DataType.FLOAT_VECTOR, dim=768)

# 2. 定义索引：HNSW + COSINE
index_params = client.prepare_index_params()
index_params.add_index(
    field_name="embedding",
    index_type="HNSW",
    metric_type="COSINE",
    params={"M": 16, "efConstruction": 200},
)

# 3. 建 collection，schema 与索引一次传进去
client.create_collection(
    collection_name="kb_chunks",
    schema=schema,
    index_params=index_params,
)

# 4. 写入，向量由外部 embedding 服务生成
rows = [
    {"id": 1, "tenant_id": 1001, "category": "guide",
     "text": "Milvus 部署步骤", "embedding": [0.01] * 768},
    {"id": 2, "tenant_id": 1001, "category": "faq",
     "text": "连接超时怎么办", "embedding": [0.02] * 768},
]
client.insert(collection_name="kb_chunks", data=rows)

# 5. 检索：带标量过滤，只在这个租户的 guide 与 faq 里找
res = client.search(
    collection_name="kb_chunks",
    data=[[0.015] * 768],
    limit=5,
    filter='tenant_id == 1001 and category in ["guide", "faq"]',
    output_fields=["text", "category"],
    search_params={"metric_type": "COSINE", "params": {"ef": 64}},
)
for hit in res[0]:
    print(hit["id"], hit["distance"], hit["entity"]["text"])

几个容易踩的点：ef 只在查询期生效，写进建索引参数里不起作用；output_fields 不写的话返回结果里只有 id 和距离，拿不到正文；insert 之后数据先进入可搜索状态需要一个短暂过程，如果紧接着就要读到，把一致性级别设成 Strong 或先做一次 flush。

示例二：upsert、delete 与分区检索

from pymilvus import MilvusClient

client = MilvusClient(uri="http://localhost:19530", token="root:Milvus")

# upsert：主键已存在则整行覆盖，不存在则插入
client.upsert(
    collection_name="kb_chunks",
    data=[{"id": 1, "tenant_id": 1001, "category": "guide",
           "text": "更新后的正文", "embedding": [0.03] * 768}],
)

# 按主键删除
client.delete(collection_name="kb_chunks", ids=[2])

# 按表达式批量删除，适合清理某个租户下的一批数据
client.delete(
    collection_name="kb_chunks",
    filter='tenant_id == 1002 and category == "draft"',
)

# 分区：先建分区，再往指定分区写入
client.create_partition(collection_name="kb_chunks", partition_name="tenant_1001")
client.insert(
    collection_name="kb_chunks",
    data=[{"id": 3, "tenant_id": 1001, "category": "faq",
           "text": "分区内写入", "embedding": [0.04] * 768}],
    partition_name="tenant_1001",
)

# 检索时指定分区，扫描范围更小
res = client.search(
    collection_name="kb_chunks",
    data=[[0.035] * 768],
    limit=3,
    partition_names=["tenant_1001"],
    consistency_level="Strong",
    search_params={"metric_type": "COSINE", "params": {"ef": 32}},
)
print(res[0][0]["id"])

关于 upsert 要理解它的语义：Milvus 里 upsert 本质是「删旧 + 插新」的组合，而不是原地更新，因此频繁 upsert 会积累删除记录并触发 compaction，写放大比较明显。如果业务是高频更新同一批主键，要么攒批合并后再写，要么改用「只插入不更新、用版本号过滤取最新」的追加式模型。delete 也是标记删除，数据在 compaction 之后才真正释放空间。

适合与不适合的场景

适合：语义检索与问答召回、以图搜图与相似商品推荐、多模态检索、去重与聚类前的近邻查找、推荐系统的召回通道、以及任何「给定一个向量，要在海量向量里快速找出最像的若干条」的需求。判断标准很简单——如果业务里出现了「相似」「相关」「最像」这类词，并且候选集大到无法全量比对，向量数据库就有位置。

不适合：需要强事务与复杂多表 join 的场景（向量库不做这些）、以精确条件查询为主的业务（那是关系库的活）、数据量只有几千条且没有增长预期的场景（暴力扫描就够了，引入一个数据库是负担）、以及对结果要求绝对精确不容许任何近似误差的场景（ANN 天生是近似的，只能通过调参把召回推向极限，但无法保证 100%）。