🎯 课程主题

Milvus向量数据库的 DML（数据操作语言：增删改）和 DQL（数据查询语言：扫描、主键查询、向量相似度检索）操作，掌握如何通过 Milvus Client 在 Collection 上进行数据的增、查、检索全流程。

📝 核心知识点

  1. Milvus Collection 元数据（Metadata）：通过 client.describe_collection() 查看 Collection 的结构，类似关系型数据库中 DESC table。默认自动包含 id（主键）和 vector（嵌入向量，维度由创建时指定，如 1024）两个字段；支持 enable_dynamic_field，允许动态接收未预先声明的字段。

  2. DML — 数据插入（Upsert）：

    • Milvus 使用 upsert 而非 insert，具有幂等性：若主键已存在则更新，不存在则插入
    • 插入前需准备数据，构建为列表，每条记录为字典格式，包含 id、vector、text、source 等字段
    • 手动 flush 将数据落盘

  3. DQL — 扫描数据：使用 client.query_iterator() 遍历 Collection 中所有数据，支持 filter 过滤条件和 output_fields 指定返回字段（"*" 表示全部字段）。通过 iterator.next() 逐批获取，最后需手动 close 防止内存泄漏。

  4. DQL — 主键查询：使用 client.get() 通过 ids 参数指定要查询的主键列表，批量返回对应记录。

  5. DQL — 向量相似度检索：

    • 将查询文本通过嵌入模型转为向量，调用 client.search() 进行相似度搜索
    • 关键参数：data（向量需包装为二维列表 [query_vector]）、limit（返回条数）、output_fields（返回字段）
    • 返回值中包含 distance（距离/相似度值，cosine 下越接近 1 越相似）；结果本身是二维结构，取 [0] 对应第一个查询向量的命中文档列表

🏗️ 架构与工作流

嵌入模型初始化 → 准备文本数据(列表) → embed_documents() 生成向量 → 
构建 data 列表 [{id, vector, text, source}] → 
client.upsert(collection_name, data) → client.flush(collection_name) →
client.get_collection_statistics() 验证 → 
DQL 操作: query_iterator / get(ids) / search(query_vector)

💻 代码实战

# ============ 嵌入模型 ============
from langchain_openai import OpenAIEmbeddings

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

# ============ 创建 Collection ============
from pymilvus import MilvusClient

client = MilvusClient(uri="http://localhost:19530")
client.create_collection(
    collection_name="dogs",
    dimension=1024,
    metric_type="COSINE"
)

# 查看元数据
metadata = client.describe_collection("dogs")
from rich import print as rprint
rprint(metadata)

# ============ 准备数据并向量化 ============
texts = [
    "什么是向量数据库？",
    "Milvus是一个开源的向量数据库",
    "向量相似度搜索是一种高效的检索方式",
    "嵌入式模型可以将文本转化为向量"
]
vectors = embed_model.embed_documents(texts)  # 返回 List[List[float]]

# 构建 data
data = []
for i in range(len(texts)):
    data.append({
        "id": i,
        "vector": vectors[i],
        "text": texts[i],
        "source": "demo"
    })

# ============ DML: Upsert 插入数据 ============
insert_result = client.upsert(
    collection_name="dogs",
    data=data
)
print("insert result:", insert_result)

# 手动刷新落盘
client.flush(collection_name="dogs")

# 查看统计信息
stats = client.get_collection_statistics(collection_name="dogs")
print(stats)  # row_count: 4

# ============ DQL: 扫描所有数据 (query_iterator) ============
iterator = client.query_iterator(
    collection_name="dogs",
    filter="",              # 空字符串表示不过滤
    output_fields=["*"]     # 返回所有字段
)

i = 0
while True:
    rows = iterator.next()
    if not rows:
        break
    for row in rows:
        i += 1
        print(f"第{i}条数据: id={row['id']}, text={row['text']}, source={row['source']}")
iterator.close()

# ============ DQL: 通过主键查询 ============
res = client.get(
    collection_name="dogs",
    ids=[0, 1, 2]
)
for i in range(len(res)):
    r = res[i]
    print(f"第{i+1}条数据: id={r['id']}, text={r['text']}")

# ============ DQL: 向量相似度检索 ============
query = "什么是向量数据库"
query_vector = embed_model.embed_query(query)

results = client.search(
    collection_name="dogs",
    data=[query_vector],       # 注意：二维列表
    limit=3,
    output_fields=["id", "text", "source"]
)

for res in results[0]:        # 取第一个查询向量的结果
    print(res)
    # 输出包含: id, distance, entity: {id, text, source}

⚠️ 常见问题与避坑指南

  1. data 传参格式：client.search() 的 data 参数必须是二维列表 [query_vector]，因为一次可传入多个查询向量。如果直接传一维向量会报错。

  2. results 是二维结构：search() 返回的 results 是外层列表（每个元素对应一个查询向量），需要 results[0] 再遍历才能拿到内部命中文档列表。

  3. embed_model 必须先执行初始化：嵌入模型变量要在使用前执行初始化代码块，否则报 not defined。

  4. iterator 需手动关闭：query_iterator 遍历完成后必须调用 iterator.close()，否则可能导致内存泄漏。

💡 个人总结与延伸

本节是 Milvus 操作的核心实战，涵盖从数据写入到多种查询方式的完整链路。Upsert 的幂等性设计在实际项目中非常实用——重复执行不会产生脏数据。query_iterator 适合全量扫描场景，get 适合精确主键查询，search 才是 RAG 架构中真正频繁使用的检索入口。

需特别留意 results 的二维结构，这是初学时容易踩的坑。理解 distance 的含义也很关键：cosine 度量下值越大（越接近 1）表示向量越相似。

最近更新:
Contributors: yangwenguang