基于Python将MongoDB文本数据通过text2vec-large-chinese模型向量化并存储到Milvus数据库的完整实现方案
以下是基于Python将MongoDB文本数据通过text2vec-large-chinese模型向量化并存储到Milvus数据库的完整实现方案，结合了文本向量化、数据转换与数据库操作的核心技术要点：
一、实现流程与代码解析
- 连接MongoDB并提取文本数据
使用pymongo库查询MongoDB中的目标文本字段：
from pymongo import MongoClient# 连接MongoDB
client = MongoClient('mongodb://localhost:27017/')
db = client['my_database']
collection = db['documents']# 提取所有文档中的文本字段（假设字段名为"content"）
cursor = collection.find({}, {"content": 1})
text_data = [doc['content'] for doc in cursor] # 示例格式: ["文本1", "文本2", ...]
- 加载text2vec-large-chinese模型生成向量
使用text2vec库加载中文模型并进行向量化：
import torch
from text2vec import SentenceModel# 初始化模型
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = SentenceModel("text2vec-large-chinese", device=device)# 批量生成向量（支持自动分块处理）
embeddings = model.encode(text_data, convert_to_numpy=True, normalize_embeddings=True)
- 向量预处理与格式转换
将向量数据转换为Milvus支持的格式：
import numpy as np# 转换为32位浮点数组（Milvus要求）
vectors = embeddings.astype(np.float32).tolist()
ids = [str(i) for i in range(len(vectors))] # 生成唯一主键（可替换为MongoDB的_id）
- 连接Milvus并创建集合
定义Schema并创建存储结构：
from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection# 连接Milvus
connections.connect(alias="default", host="localhost", port="19530")# 定义字段（主键+向量+元数据）
fields = [FieldSchema(name="doc_id", dtype=DataType.VARCHAR, is_primary=True, max_length=64),FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=1024), # text2vec-large-chinese输出1024维FieldSchema(name="source", dtype=DataType.VARCHAR, max_length=256) # 可选元数据字段
]
schema = CollectionSchema(fields, description="MongoDB文本向量存储")
collection_milvus = Collection("mongo_text_vectors", schema)
- 批量插入数据并创建索引
分批次插入优化性能：
# 分批次插入（每批5000条）
batch_size = 5000
for i in range(0, len(vectors), batch_size):batch_ids = ids[i:i+batch_size]batch_vectors = vectors[i:i+batch_size]batch_sources = ["mongo" for _ in range(len(batch_ids))] # 示例元数据# 插入数据entities = [batch_ids, batch_vectors, batch_sources]collection_milvus.insert(entities)# 刷新数据并创建IVF_FLAT索引
collection_milvus.flush()
collection_milvus.create_index(field_name="vector",index_params={"index_type": "IVF_FLAT", "metric_type": "L2", "params": {"nlist": 128}}
)
二、关键注意事项
- 模型与向量维度对齐
•text2vec-large-chinese输出1024维向量，需确保Milvus集合的dim参数匹配。
• 若使用其他模型（如BAAI/bge-base-zh），需调整dim值（例如768维）。
- 性能优化策略
• 批量插入：单次插入数据量建议5000-10000条，避免内存溢出。
• 异步操作：使用insert(entities, async_insert=True)提升吞吐量。
• GPU加速：启用CUDA加速模型推理（需NVIDIA GPU环境）。
- 数据一致性验证
抽样检查MongoDB与Milvus的数据对应关系：
# 随机抽取文档验证
sample_id = ids[100]
result = collection_milvus.query(expr=f"doc_id == '{sample_id}'", output_fields=["vector", "source"]
)
print(f"Milvus中数据: {result[0]}")
print(f"MongoDB原始文本: {text_data[100][:50]}...")
三、扩展功能
- 增量同步
通过MongoDB的Change Stream监听数据变更，实现实时向量更新：
# 监听插入/更新操作
pipeline = [{'$match': {'operationType': {'$in': ['insert', 'update']}}}]
with collection.watch(pipeline) as stream:for change in stream:new_text = change['fullDocument']['content']new_vector = model.encode([new_text])[0].tolist()# 插入新数据到Milvus
- 混合检索模式
• 向量+关键词联合查询：结合Milvus的ANN搜索与MongoDB的全文检索。
• 元数据过滤：在Milvus搜索时添加expr参数过滤条件（如source == 'mongo'）。
四、异常处理建议
try:collection_milvus.insert(entities)
except Exception as e:# 重试逻辑或回滚机制print(f"插入失败: {str(e)}")# 记录失败批次至日志文件
通过上述方案，可实现MongoDB文本数据向Milvus的高效迁移。如需处理超大规模数据（百万级以上），建议采用Milvus的Bulk Insert功能直接导入预处理好的Parquet文件。