基于Milvus 3.0的Embedding模型平滑迁移教程

2026-10-20 16:39:30

文章摘要

针对AI应用更换Embedding模型需重生成历史向量、需兼顾服务连续性与回滚空间的痛点，本文介绍借助Milvus 3.0在线增删向量字段能力实现平滑迁移的方案，分阶段完成迁移并保留回滚能力，文中给出具体模型迁移的实操实验与完整代码，验证了方案的可行性。

在RAG、多模态搜索、推荐系统等AI应用中，项目运行时更换Embedding模型是司空见惯的操作。但模型升级并非仅需修改查询端配置：历史文档需重新生成向量，同时用户查询持续进行、新文档不断写入，新版本还可能出现延迟或检索结果不符合预期的情况，因此系统需为新模型的效果验证和回滚预留空间。

如何实现Embedding模型的平稳升级，并判断最终升级效果？借助Milvus 3.0最新推出的在线新增和删除向量字段能力，可实现平滑迁移：旧模型继续服务的同时，完成新向量回填和效果验证，允许新字段暂时为空；确定效果后逐步切换流量，最后删除旧字段。

本文以知识库系统为例，在Milvus 3.0中完成从BGE-small-en-v1.5到BGE-base-en-v1.5的Embedding模型切换，实验基于Milvus 3.0.2、PyMilvus 3.0.2和公开数据集SciFact，按六个阶段展开。

第一阶段：用旧模型构建检索服务

实验采用BEIR中的SciFact数据集，包含5183篇科学文献、300条测试查询及对应的相关性标注，用于构建文献知识库并对比迁移前后的检索效果。旧模型为BGE-small-en-v1.5，候选模型为BGE-base-en-v1.5，两者维度不同（分别为384维和768维），适合验证跨维度迁移，模型规格可参考相关模型文档。

为控制实验变量，需固定文献主键、分块和文本处理方式：每篇文献的标题与摘要拼接为一个文本片段，对应一条实体记录，迁移期间保持分块方式和主键不变。两个模型的最大输入长度均为512 tokens，超出部分由分词器截断，本文指标均基于该文本处理方式。

本例通过Milvus模型集成库pymilvus.model（0.3.4）调用Sentence Transformers，在CPU上以FP32运行上述两个BGE模型。准备可连接的Milvus 3.0.2实例，从相关代码仓库获取代码后，以Linux x86_64、Python 3.12和CPU推理为例，执行以下命令即可运行完整实验：

uv venv --python 3.12 .venv
uv pip install --python .venv/bin/python "torch==2.14.1+cpu" \
 --index-url https://download.pytorch.org/whl/cpu
uv pip install --python .venv/bin/python -r requirements.lock.txt
HF_HUB_DISABLE_XET=1 .venv/bin/python run_public.py \
 --uri http://localhost:19530 --model-cache ./models

脚本会自动下载数据，并在独立的测试集合中完成整个迁移流程，部署新实例可使用仓库中的Docker Compose配置。首先初始化Milvus客户端，并通过SentenceTransformerEmbeddingFunction加载两个BGE模型，代码中model_a、vector_a对应BGE-small-en-v1.5，model_b、vector_b对应BGE-base-en-v1.5：

import torch
from pymilvus import DataType, MilvusClient
from pymilvus.model.dense import SentenceTransformerEmbeddingFunction
from run_public import MODEL_REVISIONS
torch.set_num_threads(2)

client = MilvusClient(uri=“http://localhost:19530”)

name = “scifact_model_upgrade”

prefix = “Represent this sentence for searching relevant passages: "

options = dict(

device=“cpu”, batch_size=8, normalize_embeddings=True,

query_instruction=prefix, doc_instruction=”“, cache_folder=”./models",

)

model_a = SentenceTransformerEmbeddingFunction(

model_name=“BAAI/bge-small-en-v1.5”,

revision=MODEL_REVISIONS[“A”], **options,

)

model_b = SentenceTransformerEmbeddingFunction(

model_name=“BAAI/bge-base-en-v1.5”,

revision=MODEL_REVISIONS[“B”], **options,

)

model_a.model.max_seq_length = 512

model_b.model.max_seq_length = 512

初始集合结构包含文献主键、标题、原始文本，以及BGE-small-en-v1.5对应的向量字段vector_a，代码如下：

schema = client.create_schema(auto_id=False, enable_dynamic_field=False)
schema.add_field("id", DataType.VARCHAR, max_length=64, is_primary=True)
schema.add_field("title", DataType.VARCHAR, max_length=2048)
schema.add_field("text", DataType.VARCHAR, max_length=65535)
schema.add_field("vector_a", DataType.FLOAT_VECTOR, dim=384)
client.create_collection(name, schema=schema, consistency_level="Strong")

load_data()从相关官方地址加载完整语料，将文献的_id、title、text转换为实体记录后，使用BGE-small-en-v1.5生成向量，并分批写入Milvus：

from run_public import load_data
docs, queries, qrels = load_data()
rows = [
    {"id": d["_id"], "title": d["title"],
     "text": d["title"] + "\n" + d["text"]}
    for d in docs
]
for start in range(0, len(rows), 32):
    batch = rows[start:start + 32]
    vectors = model_a.encode_documents([r["text"] for r in batch])
    client.insert(name, data=[
        {**r, "vector_a": v.tolist()} for r, v in zip(batch, vectors)
    ])

向量写入后，为vector_a创建HNSW索引并加载集合，索引创建函数也用于后续的vector_b字段：

def create_vector_index(field):
    params = client.prepare_index_params()
    params.add_index(
        field_name=field, index_name=field + "_hnsw",
        index_type="HNSW", metric_type="COSINE",
        params={"M": 16, "efConstruction": 100},
    )
    client.create_index(name, index_params=params)
client.flush(name)
create_vector_index("vector_a")
client.load_collection(name)

此时，旧模型已可提供检索服务，新向量准备期间，查询继续使用这套配置。

第二阶段：为新模型添加向量字段

BGE-small-en-v1.5输出384维向量，BGE-base-en-v1.5输出768维向量，因此需要为新模型增加独立的向量字段和索引，两个字段分别定义维度和索引，检索时通过anns_field指定目标字段。将不同维度的新向量写入原字段会触发维度校验错误，即使维度相同，不同模型生成的向量通常也不具备直接比较的语义基础，因此不应在同一检索空间中混用。本例保留vector_a，新增vector_b并创建索引：

client.add_collection_field(
    name, field_name="vector_b",
    data_type=DataType.FLOAT_VECTOR, dim=768, nullable=True,
)
create_vector_index("vector_b")

新增字段必须设置nullable=True，以表示历史记录尚未生成的新向量，字段创建操作本身不触发Embedding推理，新向量需要由应用计算并写入，此阶段的查询仍由BGE-small-en-v1.5和vector_a处理。

第三阶段：用新模型重新生成文档向量

从Milvus的text字段分批读取原文，使用BGE-base-en-v1.5重新生成向量，再按主键写入vector_b，设置partial_update=True，保留未提交的标题、正文和旧向量，若原文存储在其他系统中，则按主键读取并关联：

iterator = client.query_iterator(
    name, filter="", output_fields=["id", "text"],
    batch_size=32, consistency_level="Strong",
)
first_batch = True
try:
    while True:
        batch = iterator.next()
        if not batch:
            break
        vectors = model_b.encode_documents([r["text"] for r in batch])
        client.upsert(name, data=[
            {"id": r["id"], "vector_b": v.tolist()}
            for r, v in zip(batch, vectors)
        ], partial_update=True)
        if first_batch:
            client.flush(name)
            client.load_collection(name, load_fields=[
                "id", "title", "text", "vector_a", "vector_b",
            ])
            client.refresh_load(name)
            first_batch = False
finally:
    iterator.close()

第四阶段：检查回填结果，对比检索效果

切换流量前，应确认历史回填与增量同步已完成、新索引已构建、目标字段已加载，并通过实际检索验证可用性，完整脚本还会逐条核对原文与向量，任何检查未通过，都继续使用旧模型：

import time
client.flush(name)
client.refresh_load(name)
deadline = time.monotonic() + 180
while True:
    state = client.describe_index(name, "vector_b_hnsw")
    if (
        state.get("state") == "Finished"
        and state.get("pending_index_rows", 0) == 0
    ):
        break
    if time.monotonic() > deadline:
        raise TimeoutError(str(state))
    time.sleep(1)

首批回填后的检索检查表明，已回填的记录可以被召回，向量值为NULL的记录不参与该字段的检索。检查通过后，在相同语料和查询集上比较两个模型的检索效果，本实验使用SciFact的全部300条测试查询，比较Recall@10和nDCG@10，Recall@10衡量前10条结果对相关文献的覆盖比例，nDCG@10进一步考虑相关文献的排序位置，两个指标均对测试查询取平均。

评测时，每个模型分别编码查询文本，并检索由自身生成的文档向量，encode_queries()会自动添加初始化时配置的查询指令，因此直接传入查询原文，避免重复拼接前缀，以下代码通过BGE-base-en-v1.5编码查询，再检索vector_b：

query_ids = sorted(qrels, key=int)
query_vectors_b = model_b.encode_queries([
    queries[qid] for qid in query_ids
])
results_b = client.search(
    name, anns_field="vector_b",
    data=[v.tolist() for v in query_vectors_b], limit=10,
    search_params={"metric_type": "COSINE", "params": {"ef": 128}},
    consistency_level="Strong",
)

使用BGE-small-en-v1.5编码同一组查询并检索vector_a，即可得到替换前的评测结果，代码包保存了逐查询的相关文献标注、召回文献ID和指标，可据此分析具体查询的改善与退化，替换前后的效果对比如图所示。

第五阶段：逐步切换流量，保留回滚能力

离线验证通过后，先将部分请求交给BGE-base-en-v1.5，观察检索质量、延迟和错误率，再逐步扩大流量比例，观察期内保留BGE-small-en-v1.5的推理能力和最新向量，以支持回滚。将模型与目标字段绑定为一个检索版本，并在请求进入时固定该配置，随后，向量生成和检索均使用同一版本，即使默认版本在请求执行期间发生变化，也不影响当前请求的配置：

routes = {
    "A": (model_a, "vector_a"),
    "B": (model_b, "vector_b"),
}
def search(question, version):
    model, field = routes[version]  # 请求开始时固定这一对配置
    vector = model.encode_queries([question])[0].tolist()
    return client.search(
        name, data=[vector], anns_field=field, limit=10,
        output_fields=["title"],
        search_params={"metric_type": "COSINE", "params": {"ef": 128}},
        consistency_level="Strong",
    )

以下命令展示实验HTTP接口的调用方式，通过version="B"选择BGE-base-en-v1.5，自动实验由脚本控制路由，不需要额外手动执行：

curl http://localhost:8080/search \
 -H 'Content-Type: application/json' \
 -d '{"query":"0-dimensional biomaterials show inductive properties.","version":"B"}'

实验通过30个请求验证混合版本路由，其中6个使用新模型、24个使用旧模型，该检查不等同于真实用户的效果对照实验，生产环境可按用户、租户或会话稳定分流，灰度验证通过后，切换默认检索版本：

curl http://localhost:8080/route \
 -H 'Content-Type: application/json' -d '{"version":"B"}'

若观察期内新版本的检索效果、延迟或错误率未达到要求，可将后续请求切回BGE-small-en-v1.5，并保留两列向量，修正问题后，重新执行第四阶段的验证，再尝试切换，回滚请求如下：

curl http://localhost:8080/route \
 -H 'Content-Type: application/json' -d '{"version":"A"}'

第六阶段：删除不再使用的旧向量字段

观察期结束、确认不再需要回滚后，将全部请求切换至新模型，停止旧向量写入，并等待旧版本请求完成，再删除vector_a，删除后，将无法仅通过切换路由恢复旧模型检索。本实验采用低并发请求和固定观察间隔，生产部署应通过请求计数或排空机制，确认旧版本请求已经结束：

client.drop_collection_field(name, field_name="vector_a")
print(client.describe_collection(name))
print(client.list_indexes(name))

本次实验中，删除vector_a后，其关联索引也被移除，vector_b及文献主键、标题和正文均保留，仅写入BGE-base-en-v1.5向量的新记录仍可正常检索，字段删除后，对象存储中的物理空间由后续后台过程回收，迁移期间需要为两套向量及索引预留资源，并承担额外的模型推理成本。

总结

借助Milvus 3.0的在线加减列能力，Embedding模型升级可以分步平滑完成：旧模型继续服务，新模型独立回填并验证效果，再同步切换查询模型和向量字段，观察期内保留并更新旧向量以支持回滚，确认无需回滚后再删除旧字段。本次SciFact实验中，替换为BGE-base-en-v1.5后，Recall@10从0.8362提升至0.8742，完整代码与复现步骤见相关代码仓库。

以上内容不代表本平台立场，仅供读者参考