重要提醒，最后强调一次，换模型等于重建集合：
集合的维度由第一条数据决定，之后锁死。如第一条为 768，后面所有数据必须是 768，否则报错。
换模型可能会更换维度，导致向集合添加数据直接报错。即使维度不变，也不能这么做。因为 A 模型和 B 模型训练数据、分布不同，向量距离没有可比性。最终导致相似度计算错乱，召回结果完全不可控。
所以，换模型 = 新建集合 + 全量重灌
重点：在选择模型时一定要谨慎，选择符合自己的模型，不然后期更换，代价巨大。
下面是一个函数，用来重灌数据：
def rebuild(src_col, dst_col, ef, batch_size=500):
"""用新的嵌入函数重建集合"""
offset = 0
total = 0
while True:
batch = src_col.get(
limit=batch_size, offset=offset, include=["documents", "metadatas"]
)
if not batch["ids"]:
break
docs = [d for d in batch["documents"] if d]
dst_col.upsert(
ids=batch["ids"],
embeddings=ef(docs),
documents=batch["documents"],
metadatas=batch["metadatas"],
)
total += len(batch["ids"])
offset += batch_size
if len(batch["ids"]) < batch_size:
break
return total完整的用法如下：
import os
# 使用国内端点地址
os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"
import chromadb
from sentence_transformers import SentenceTransformer
class BgeZhEmbedding:
def __init__(self, model_path: str = "BAAI/bge-small-zh-v1.5"):
self.model = SentenceTransformer(model_path)
def __call__(self, texts: list[str]) -> list[list[float]]:
"""符合Chroma embedding function调用规范"""
emb = self.model.encode(texts, normalize_embeddings=True)
return emb.tolist()
def rebuild(src_col, dst_col, ef, batch_size=500):
"""用新的嵌入函数重建集合"""
offset = 0
total = 0
while True:
batch = src_col.get(
limit=batch_size, offset=offset, include=["documents", "metadatas"]
)
if not batch["ids"]:
break
docs = [d for d in batch["documents"] if d]
dst_col.upsert(
ids=batch["ids"],
embeddings=ef(docs),
documents=batch["documents"],
metadatas=batch["metadatas"],
)
total += len(batch["ids"])
offset += batch_size
if len(batch["ids"]) < batch_size:
break
return total
if __name__ == "__main__":
client = chromadb.EphemeralClient()
# 源集合：仅存储文档、元数据，向量由重建时重新生成
col_source = client.get_or_create_collection(
name="col_source",
embedding_function=None,
metadata={"hnsw:space": "cosine"}
)
col_source.add(
ids=["q1", "q2", "q3", "q4"],
documents=[
"订单付款后多久发货？一般 48 小时内出库",
"如何申请退款？在订单详情页点击申请退款",
"发票怎么开？在个人中心-发票管理里申请电子发票",
"快递丢了怎么办？联系客服补发或全额退款",
],
metadatas=[
{"category": "物流"},
{"category": "售后"},
{"category": "财务"},
{"category": "售后"},
],
)
# 目标集合
col_target = client.get_or_create_collection(
name="col_target",
embedding_function=None,
metadata={"hnsw:space": "cosine"}
)
# 初始化嵌入函数，传入重建函数
new_ef = BgeZhEmbedding()
total_count = rebuild(col_source, col_target, new_ef, batch_size=10)
print(f"\n重建完成！一共迁移 {total_count} 条数据")
# 简单测试检索
res = col_target.query(query_embeddings=new_ef(["买的东西不想要了，钱能退回来吗"]), n_results=2)
print("\n检索测试结果：")
print(res["documents"])运行代码，输出如下：
Loading weights: 100%|██████████| 71/71 [00:00<00:00, 1430.34it/s]
重建完成！一共迁移 4 条数据
检索测试结果：
[['快递丢了怎么办？联系客服补发或全额退款', '如何申请退款？在订单详情页点击申请退款']]注意了，重灌之前，先备份数据目录。 别问我为什么强调这个。
川公网安备51010802032098