RAG 系列 7：使用 Milvus 实现持久化向量检索

FAISS 和 Chroma 都可以直接嵌入 Python 程序。Milvus Standalone 不同，它是一个独立运行的向量数据库服务，应用通过网络端口访问。

本文章启动 Milvus 2.6.9 Standalone，使用 PyMilvus 2.6.9 完成新增、读取、过滤查询、修改、删除和重启持久化验证。

为了准确比较三种向量存储，本篇与 FAISS、Chroma 统一使用：

• 相同的四条测试文档；
• 相同的本地 Qwen3 Embedding 模型；
• 相同的 HuggingFaceEmbeddings 参数；
• 文档统一调用 embed_documents()；
• 查询统一调用带 Qwen3 query prompt 的 embed_query()。

数据库操作全部使用原生 MilvusClient。HuggingFaceEmbeddings 只负责生成向量，没有使用 LangChain 的 Milvus VectorStore、Retriever 或 RAG 封装。

1. Milvus 的三种部署形态

Milvus 2.6 提供三种部署方式：

形态	特点	适合场景
Lite	嵌入 Python 进程，使用本地文件	学习、快速实验、小规模数据
Standalone	单机数据库服务	本地开发、单机服务、完整功能验证
Distributed	多节点分布式集群	更大数据规模和生产负载

本篇选择 Standalone。四条演示数据不需要三个容器，但 Standalone 能完整展示数据库服务、元数据、对象存储、健康检查和重启恢复。

2. Standalone 组件

Milvus v2.6.9 Standalone Compose 启动三个容器：

容器	作用
milvus-standalone	提供 Collection、写入、查询、索引和 WebUI
milvus-etcd	保存 Collection Schema、Segment 状态等元数据
milvus-minio	保存日志快照、标量、向量和索引等对象数据

本篇 Compose 还设置 MQ_TYPE=woodpecker。Woodpecker 是 Milvus 2.6 使用的 WAL 实现，运行在 Standalone 内部并复用 MinIO，不是第四个容器。

主要端口：

19530 -> PyMilvus
9091  -> 健康检查和 WebUI
9000  -> MinIO API
9001  -> MinIO Console

四个端口全部绑定到 127.0.0.1，避免把未启用生产级鉴权的教学环境暴露到局域网。

Milvus 的数据层次可以理解为：

Database
└── Collection
    └── Partition
        └── Entity

本篇使用默认 Database、默认 Partition 和 life_notes Collection。

3. 创建独立 Colima profile

本机默认 Colima 已经运行其他项目，而且只有 6 GiB 内存。Milvus 使用名为 milvus 的独立 profile：

colima start milvus \
  --cpus 4 \
  --memory 12 \
  --disk 60 \
  --runtime docker \
  --activate=false

--activate=false 不修改默认 Docker context。后续所有容器命令都显式指定：

docker --context colima-milvus ...

这样 Milvus 的启动、停止和资源配置不会作用到默认 Colima。

4. 启动 Milvus Standalone

项目中的 Compose 文件为：

rag/p07_milvus_native/docker-compose.yml

启动并查看状态：

cd /Users/bianhn/git/llm-learning

docker --context colima-milvus compose \
  -f rag/p07_milvus_native/docker-compose.yml \
  up -d

docker --context colima-milvus compose \
  -f rag/p07_milvus_native/docker-compose.yml \
  ps

容器完全启动后：

NAME                STATUS                  PORTS
milvus-etcd         Up (healthy)            2379-2380/tcp
milvus-minio        Up (healthy)            127.0.0.1:9000-9001->9000-9001/tcp
milvus-standalone   Up (healthy)            127.0.0.1:9091->9091/tcp, 127.0.0.1:19530->19530/tcp

健康检查：

curl http://127.0.0.1:9091/healthz

返回：

OK

WebUI 地址为：

http://127.0.0.1:9091/webui/

5. Python 环境

继续使用 Chroma 文章创建的 .venv_vector_db：

source .venv_vector_db/bin/activate
python -m pip install -r rag/p07_milvus_native/requirements.txt
python -m pip check

p07_milvus_native/requirements.txt 复用 p06 的依赖，其中包括：

langchain-huggingface==1.2.2
pymilvus==2.6.9
sentence-transformers==5.1.2
transformers==4.57.6

6. 检查服务

# 这个文件连接 Milvus Standalone，检查服务版本、数据库和 Collection。
from pymilvus import MilvusClient, __version__ as pymilvus_version

client = MilvusClient(uri="http://127.0.0.1:19530")

print(f"PyMilvus 版本：{pymilvus_version}")
print(f"Milvus 服务版本：{client.get_server_version()}")
print(f"数据库：{client.list_databases()}")
print(f"default 数据库中的 Collection：{client.list_collections()}")

运行：

python rag/p07_milvus_native/01_check_milvus_server.py

输出：

PyMilvus 版本：2.6.9
Milvus 服务版本：2.6.9
数据库：['default']
default 数据库中的 Collection：[]

7. 共用代码

common.py 中的 Embedding 初始化与 FAISS、Chroma 完全相同：

"""Milvus 示例共用的模型、测试数据和连接配置。"""

from pathlib import Path

import torch
from langchain_huggingface import HuggingFaceEmbeddings

MODEL_PATH = Path("/Users/bianhn/git/llm/qwen3-embedding/model")
MILVUS_URI = "http://127.0.0.1:19530"
COLLECTION_NAME = "life_notes"

# 与 FAISS 示例使用相同的文本和分类；额外字段用于演示标量过滤。
DOCUMENTS = [
    {
        "id": 1,
        "text": "西湖边有步道和树荫，很适合周末散步。",
        "category": "travel",
        "rating": 4.8,
        "year": 2025,
        "is_public": True,
    },
    {
        "id": 2,
        "text": "苹果富含膳食纤维，是常见的健康水果。",
        "category": "food",
        "rating": 4.6,
        "year": 2024,
        "is_public": True,
    },
    {
        "id": 3,
        "text": "汽车需要定期更换机油并检查轮胎。",
        "category": "car",
        "rating": 4.2,
        "year": 2023,
        "is_public": False,
    },
    {
        "id": 4,
        "text": "杭州植物园环境安静，适合慢慢游览。",
        "category": "travel",
        "rating": 4.7,
        "year": 2025,
        "is_public": True,
    },
]


def create_embeddings_model() -> tuple[HuggingFaceEmbeddings, str]:
    """创建与 FAISS 示例配置完全相同的 Embedding 模型。"""
    device = "mps" if torch.backends.mps.is_available() else "cpu"
    embeddings_model = HuggingFaceEmbeddings(
        model_name=str(MODEL_PATH),
        model_kwargs={"device": device},
        encode_kwargs={"normalize_embeddings": True},
        query_encode_kwargs={
            "prompt_name": "query",
            "normalize_embeddings": True,
        },
    )
    return embeddings_model, device


def build_rows(items: list[dict], vectors: list[list[float]]) -> list[dict]:
    """把文档、metadata 和向量组装成 Milvus Entity。"""
    return [
        {
            "id": item["id"],
            "vector": vector,
            "text": item["text"],
            "category": item["category"],
            "rating": item["rating"],
            "year": item["year"],
            "is_public": item["is_public"],
        }
        for item, vector in zip(items, vectors)
    ]

向量生成路径固定为：

Document → embed_documents() → insert/upsert
Query    → embed_query()     → search

normalize_embeddings=True 把向量归一化。本篇 Collection 使用 COSINE 度量，与 Chroma 的 cosine space 和 FAISS 归一化向量保持可比。

测试文本和 category 与 FAISS 相同，同时增加三个过滤字段：

ID	category	rating	year	is_public	text
1	travel	4.8	2025	true	西湖边有步道和树荫，很适合周末散步。
2	food	4.6	2024	true	苹果富含膳食纤维，是常见的健康水果。
3	car	4.2	2023	false	汽车需要定期更换机油并检查轮胎。
4	travel	4.7	2025	true	杭州植物园环境安静，适合慢慢游览。

9. 新增：创建 Collection 并插入数据

入口为 create_collection.py。先使用 embed_documents() 生成四条文档向量：

embeddings_model, device = create_embeddings_model()
document_vectors = embeddings_model.embed_documents(
    [item["text"] for item in DOCUMENTS]
)
rows = build_rows(DOCUMENTS, document_vectors)

创建 Collection 时直接从真实向量读取维度，不把 1024 写死：

client.create_collection(
    collection_name=COLLECTION_NAME,
    dimension=len(document_vectors[0]),
    metric_type="COSINE",
    consistency_level="Strong",
)
result = client.insert(
    collection_name=COLLECTION_NAME,
    data=rows,
)
client.flush(collection_name=COLLECTION_NAME)

完整代码:

"""重建 Collection，并新增与 FAISS 示例一致的测试数据。"""

import sys
from pathlib import Path

# 支持 `python -m ...` 与 IDE 直接运行脚本。
if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from pymilvus import MilvusClient

from rag.p07_milvus_native.common import (
    COLLECTION_NAME,
    DOCUMENTS,
    MILVUS_URI,
    build_rows,
    create_embeddings_model,
)

# 1. 离线计算文档向量（与 p06 相同，使用 embed_documents）。
embeddings_model, device = create_embeddings_model()
document_vectors = embeddings_model.embed_documents(
    [item["text"] for item in DOCUMENTS]
)
rows = build_rows(DOCUMENTS, document_vectors)

# 2. 连接 Milvus；若 Collection 已存在则先删除，保证可重复运行恢复初始数据。
client = MilvusClient(uri=MILVUS_URI)
if client.has_collection(COLLECTION_NAME):
    client.drop_collection(COLLECTION_NAME)

# 3. 创建 Collection：MilvusClient 会根据 insert 的数据自动推断 schema。
#    metric_type=COSINE 与 normalize_embeddings=True 配套；Strong 保证写入立即可查。
client.create_collection(
    collection_name=COLLECTION_NAME,
    dimension=len(document_vectors[0]),
    metric_type="COSINE",
    consistency_level="Strong",
)
result = client.insert(collection_name=COLLECTION_NAME, data=rows)
# flush 将内存中的 segment 持久化，后续 query / search 才能看到新数据。
client.flush(collection_name=COLLECTION_NAME)

print(f"运行设备：{device}")
print(f"Collection：{COLLECTION_NAME}")
print(f"向量维度：{len(document_vectors[0])}")
print(f"新增记录数：{result['insert_count']}")

运行：

python -m rag.p07_milvus_native.create_collection

输出：

运行设备：mps
Collection：life_notes
向量维度：1024
新增记录数：4

本节使用 insert() 表示新增。由于脚本会先重建 Collection，不会遇到重复主键。

10. 读取：按主键获取 Entity

入口为 read_records.py：

rows = client.get(
    collection_name=COLLECTION_NAME,
    ids=[1, 4],
    output_fields=["text", "category", "rating", "year", "is_public"],
)

完整代码:

"""按主键读取 Milvus Entity。"""

import sys
from pathlib import Path

# 支持 `python -m ...` 与 IDE 直接运行脚本。
if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from pymilvus import MilvusClient

from rag.p07_milvus_native.common import COLLECTION_NAME, MILVUS_URI

client = MilvusClient(uri=MILVUS_URI)
# get 按主键精确读取；output_fields 指定返回哪些标量字段（不含 vector）。
rows = client.get(
    collection_name=COLLECTION_NAME,
    ids=[1, 4],
    output_fields=["text", "category", "rating", "year", "is_public"],
)

print("按主键读取 id=1 和 id=4：")
for row in rows:
    print(
        f"id={row['id']} category={row['category']} "
        f"rating={row['rating']} text={row['text']}"
    )

运行：

python -m rag.p07_milvus_native.read_records

输出：

按主键读取 id=1 和 id=4：
id=1 category=travel rating=4.8 text=西湖边有步道和树荫，很适合周末散步。
id=4 category=travel rating=4.7 text=杭州植物园环境安静，适合慢慢游览。

已知主键时使用 get()，不需要生成查询向量，也不会发生相似度计算。

11. 过滤查询

代码目录为 filter/，入口为 filter_queries.py。

11.1 标量与文本过滤

Milvus 的 filter 使用字符串表达式：

类型	表达式
等值	category == "travel"
比较	rating >= 4.7
集合	category in ["food", "car"]
AND	category == "travel" and rating >= 4.8
OR	category == "food" or is_public == false
文本匹配	text like "%步道%"

标量过滤调用 query()：

rows = client.query(
    collection_name=COLLECTION_NAME,
    filter='category == "travel" and rating >= 4.8',
    output_fields=["id"],
)

11.2 带过滤条件的向量搜索

查询向量必须使用 embed_query()，让 Qwen3 应用 query prompt：

query = "周末想找一个适合散步的地方"
query_vector = embeddings_model.embed_query(query)

results = client.search(
    collection_name=COLLECTION_NAME,
    data=[query_vector],
    anns_field="vector",
    filter='category == "travel"',
    limit=2,
    output_fields=["text", "category", "rating"],
    search_params={"metric_type": "COSINE"},
)

Milvus 先应用标量条件，再在 travel 候选中执行 COSINE Top-K。

完整代码:

"""演示标量过滤、文本匹配和带过滤条件的向量搜索。"""

import sys
from pathlib import Path

# 支持 `python -m ...` 与 IDE 直接运行脚本。
if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from pymilvus import MilvusClient

from rag.p07_milvus_native.common import (
    COLLECTION_NAME,
    MILVUS_URI,
    create_embeddings_model,
)


def print_filtered_ids(
    client: MilvusClient,
    title: str,
    filter_expression: str,
) -> None:
    """执行标量过滤并稳定输出匹配主键。

    Milvus 使用布尔表达式字符串（类似 SQL WHERE），
    与 p06 Chroma 的 where={"category": {"$eq": "travel"}} 语法不同。
    """
    rows = client.query(
        collection_name=COLLECTION_NAME,
        filter=filter_expression,
        output_fields=["id"],
    )
    print(f"{title}：{sorted(row['id'] for row in rows)}")


# 查询向量使用 embed_query（带 Qwen3 query prompt），与文档向量计算方式不同。
embeddings_model, _ = create_embeddings_model()
query = "周末想找一个适合散步的地方"
query_vector = embeddings_model.embed_query(query)

client = MilvusClient(uri=MILVUS_URI)

# --- 标量过滤（client.query，不涉及向量相似度）---
print_filtered_ids(client, "category = travel", 'category == "travel"')
print_filtered_ids(client, "rating >= 4.7", "rating >= 4.7")
print_filtered_ids(
    client,
    "category in [food, car]",
    'category in ["food", "car"]',
)
print_filtered_ids(
    client,
    "travel 且 rating >= 4.8",
    'category == "travel" and rating >= 4.8',
)
print_filtered_ids(
    client,
    "food 或非公开记录",
    'category == "food" or is_public == false',
)
# text like 对标量字段做子串匹配，类似 p06 的 where_document $contains。
print_filtered_ids(client, "text 包含“步道”", 'text like "%步道%"')

# --- 带过滤条件的向量搜索（client.search，先 filter 再 ANN）---
results = client.search(
    collection_name=COLLECTION_NAME,
    data=[query_vector],
    anns_field="vector",
    filter='category == "travel"',
    limit=2,
    output_fields=["text", "category", "rating"],
    search_params={"metric_type": "COSINE"},
)

print(f"\n只在 category=travel 中执行向量查询：{query}")
for rank, hit in enumerate(results[0], start=1):
    entity = hit["entity"]
    # distance 在 COSINE 度量下越大表示越相似（IP 内积，向量已归一化时等价于余弦相似度）。
    print(
        f"{rank}. score={hit['distance']:.4f} id={hit['id']} "
        f"rating={entity['rating']} text={entity['text']}"
    )

运行：

python -m rag.p07_milvus_native.filter_queries

关键输出：

category = travel：[1, 4]
rating >= 4.7：[1, 4]
category in [food, car]：[2, 3]
travel 且 rating >= 4.8：[1]
food 或非公开记录：[2, 3]
text 包含“步道”：[1]

只在 category=travel 中执行向量查询：周末想找一个适合散步的地方
1. score=0.6498 id=1 rating=4.8 text=西湖边有步道和树荫，很适合周末散步。
2. score=0.4660 id=4 rating=4.7 text=杭州植物园环境安静，适合慢慢游览。

PyMilvus 的命中字段名是 distance，但 COSINE 模式下这里表示 cosine similarity，数值越大越相似。

它与 Chroma cosine distance 的关系为：

milvus_cosine_score = 1 - chroma_cosine_distance

FAISS 使用单位向量和 IndexFlatL2 时，三者对本例可以得到相同排序，但返回数值和方向不同。

12. 修改：使用 upsert 更新 Entity

Milvus 没有对整条 Entity 单独命名为 update 的接口。本例使用相同主键 upsert() 覆盖 id=4。文本改变后必须同步重新生成文档向量：入口为 update_record.py。

updated_document = {
    "id": 4,
    "text": "杭州植物园有安静的林间步道，适合周末散步和慢慢游览。",
    "category": "travel",
    "rating": 4.9,
    "year": 2026,
    "is_public": True,
}
updated_vector = embeddings_model.embed_documents([updated_document["text"]])
updated_rows = build_rows([updated_document], updated_vector)

client.upsert(
    collection_name=COLLECTION_NAME,
    data=updated_rows,
)

完整代码:

"""使用相同主键 upsert 修改 id=4，并同步更新向量。"""

import sys
from pathlib import Path

# 支持 `python -m ...` 与 IDE 直接运行脚本。
if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from pymilvus import MilvusClient

from rag.p07_milvus_native.common import (
    COLLECTION_NAME,
    MILVUS_URI,
    build_rows,
    create_embeddings_model,
)

# 修改文本后必须重新 embed，否则向量与 text 字段不一致，检索结果会偏差。
updated_document = {
    "id": 4,
    "text": "杭州植物园有安静的林间步道，适合周末散步和慢慢游览。",
    "category": "travel",
    "rating": 4.9,
    "year": 2026,
    "is_public": True,
}
embeddings_model, _ = create_embeddings_model()
updated_vector = embeddings_model.embed_documents([updated_document["text"]])
updated_rows = build_rows([updated_document], updated_vector)

client = MilvusClient(uri=MILVUS_URI)
# upsert：主键存在则覆盖，不存在则插入（Milvus 无 Chroma 单独的 update API）。
result = client.upsert(
    collection_name=COLLECTION_NAME,
    data=updated_rows,
)
row = client.get(
    collection_name=COLLECTION_NAME,
    ids=[4],
    output_fields=["text", "category", "rating", "year"],
)[0]

print(f"修改记录数：{result['upsert_count']}")
print(
    f"id={row['id']} rating={row['rating']} "
    f"year={row['year']} text={row['text']}"
)

运行：

python -m rag.p07_milvus_native.update_record

输出：

修改记录数：1
id=4 rating=4.9 year=2026 text=杭州植物园有安静的林间步道，适合周末散步和慢慢游览。

insert() 适合新增；upsert() 在主键存在时替换记录，不存在时新增记录。

13. 删除：按主键和表达式删除

代码目录为 delete/，入口为 delete_records.py。

按主键删除：

client.delete(
    collection_name=COLLECTION_NAME,
    ids=[3],
)

按表达式删除：

client.delete(
    collection_name=COLLECTION_NAME,
    filter='category == "food"',
)

完整代码:

"""分别演示按主键删除和按标量表达式删除。"""

import sys
from pathlib import Path

# 支持 `python -m ...` 与 IDE 直接运行脚本。
if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from pymilvus import MilvusClient

from rag.p07_milvus_native.common import COLLECTION_NAME, MILVUS_URI

client = MilvusClient(uri=MILVUS_URI)

# 按主键删除单条记录。
delete_by_id = client.delete(
    collection_name=COLLECTION_NAME,
    ids=[3],
)
print(f"按主键删除数量：{delete_by_id['delete_count']}")

# 按标量表达式批量删除（语法与 filter_queries 中的 filter 相同）。
delete_by_filter = client.delete(
    collection_name=COLLECTION_NAME,
    filter='category == "food"',
)
print(f"按 category=food 删除数量：{delete_by_filter['delete_count']}")

# 删除后 flush，再 query 才能看到最新剩余记录。
client.flush(collection_name=COLLECTION_NAME)
remaining_rows = client.query(
    collection_name=COLLECTION_NAME,
    filter="id >= 0",
    output_fields=["id"],
)
print(f"剩余 ID：{sorted(row['id'] for row in remaining_rows)}")

运行：

python -m rag.p07_milvus_native.delete_records

输出：

按主键删除数量：1
按 category=food 删除数量：1
剩余 ID：[1, 4]

表达式删除可能影响多条 Entity。生产代码应先用同样的 filter 调用 query()，确认主键和数量后再执行。

14. 验证重启持久化

重启三个容器：

docker --context colima-milvus compose \
  -f rag/p07_milvus_native/docker-compose.yml \
  restart

等待容器恢复 healthy，再运行：

python rag/p07_milvus_native/01_check_milvus_server.py
python -m rag.p07_milvus_native.read_records

Collection 仍然存在，读取结果也与重启前一致，说明 etcd、MinIO 和 Milvus 的挂载目录都被保留。

如果已经运行删除脚本，重启后仍然只剩 id=1 和 id=4；若想恢复四条初始数据，重新运行 create/create_collection.py 对应的模块命令。

15. 查看资源与停止服务

查看一次容器资源占用：

docker --context colima-milvus stats --no-stream \
  milvus-standalone milvus-etcd milvus-minio

四条数据的实测采样：

NAME                CPU %     MEM USAGE / LIMIT
milvus-standalone   8.93%     304.9MiB / 11.66GiB
milvus-etcd         1.83%     53.84MiB / 11.66GiB
milvus-minio        0.06%     162.3MiB / 11.66GiB

这只是瞬时开发环境数据，不能替代生产容量规划。

测试完成后停止独立 profile：

colima stop milvus

再次启动：

colima start milvus --activate=false

docker --context colima-milvus compose \
  -f rag/p07_milvus_native/docker-compose.yml \
  up -d