前言
Milvus作为使用最广泛的向量数据库之一，在市场中占据了很大的份额，我们想要学习大模型RAG，向量数据库是逃不开的一环，根据上一篇文章，我们已经配置好了milvus，接下来我们来学习一下如何向Milvus中批量导入数据。

一、准备工作

1、准备环境

博主用的是linux中的ubuntu系统，此处推荐使用linux系统进行操作

2、下载embedding模型

我们在modelscope社区下载对应模型
首先安装modelscope包方便下载

pip install modelscope

下载到当前文件夹，此处下载的是bge模型，如需其他模型请自行更改

mkdir bge-m3
modelscope download --model BAAI/bge-m3  --local_dir ./dir

等待下载完成即可

二、主要代码讲解

1、连接数据库

首先，连接服务器，创建database

client = MilvusClient(
    uri="http://xx.xx.xx.xx:19530",
    token="root:Milvus" #此处使用token登录方式，默认用户名密码
)
client.create_database(db_name="tests_db") #此命令只在第一次运行添加，后续运行注释掉
client.use_database(db_name="tests_db") #使用该database进行后续操作

2、创建集合

#schema 参数编写 定义表结构
schema = MilvusClient.create_schema(auto_id=False)
schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True) #ID
schema.add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=1024) #存入内容展示
schema.add_field(field_name="content", datatype=DataType.VARCHAR, max_length=65535)  #内容转换成的向量结果展示

if client.has_collection(collection_name="demo_collection"):
    client.drop_collection(collection_name="demo_collection")
#根据编写的schema进行集合创建  只第一次使用，后续再运行将会进行覆盖操作
client.create_collection(
    collection_name="demo_collection",
    schema=schema
)

3、embedding模型引入

bge_m3_ef = BGEM3EmbeddingFunction(
    model_name='/data/models/embedding/bge-m3/', #此处为本地模型地址，也可使用BAAI/bge-m3地址格式，会在huggingface上下载
    device='cpu', #CPU或者cuda:0
    use_fp16=False 
)

4、数据准备

docs = [
    "无边落木萧萧下，不尽长江滚滚来。",
    "在天愿作比翼鸟，在地愿为连理枝。",
    "千山鸟飞绝，万径人踪灭。",
]
docs_embeddings = bge_m3_ef.encode_documents(docs) #将上述文本向量化处理
time_number = datetime.now().strftime("%Y%m%d%H%M%S") #id编写

#按照集合创建格式进行内容对齐
data = [  
    {        
        "id": int(str(time_number)+str(i)), 
        "vector": docs_embeddings["dense"][i],
        "content":docs[i]       
    }
    for i in range(len(docs_embeddings["dense"]))
]

5、批量导入数据

# 创建向量索引 只需在第一次集合创建时使用，给新的集合创建向量索引
index_params = MilvusClient.prepare_index_params()
index_params.add_index(
    field_name="vector",
    index_type="HNSW",
    metric_type="L2",
    params={"M": 16, "efConstruction": 200}
)
client.create_index(collection_name="demo_collection", index_params=index_params)

# 插入数据 
client.insert(collection_name="demo_collection", data=data)

# 加载集合到内存（需在创建索引后执行）
client.load_collection(collection_name="demo_collection")

完成之后即可在Attu中查看到导入的数据

三、完整代码

复制 填写空白参数即可运行

from pymilvus import MilvusClient, DataType
from pymilvus.model.hybrid import BGEM3EmbeddingFunction
from datetime import datetime


client = MilvusClient(
    uri="", # http://xx.xx.xx.xx:19530
    token="" # root:Milvus
)
client.create_database(db_name="tests_db")
client.use_database(db_name="tests_db")


schema = MilvusClient.create_schema(auto_id=False)
schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
schema.add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=1024) 
schema.add_field(field_name="content", datatype=DataType.VARCHAR, max_length=65535) 

if client.has_collection(collection_name="demo_collection"):
    client.drop_collection(collection_name="demo_collection")
client.create_collection(
    collection_name="", #demo_collection
    schema=schema
)

bge_m3_ef = BGEM3EmbeddingFunction(
    model_name='', #本地bge-m3 Embedding模型绝对路径
    device='cpu',
    use_fp16=False
)


docs = [
    "无边落木萧萧下，不尽长江滚滚来。",
    "在天愿作比翼鸟，在地愿为连理枝。",
    "千山鸟飞绝，万径人踪灭。",
]
docs_embeddings = bge_m3_ef.encode_documents(docs)
time_number = datetime.now().strftime("%Y%m%d%H%M%S")

data = [
    {        
        "id": int(str(time_number)+str(i)), 
        "vector": docs_embeddings["dense"][i],
        "content":docs[i]       
    }
    for i in range(len(docs_embeddings["dense"]))
]

# 创建向量索引
index_params = MilvusClient.prepare_index_params()
index_params.add_index(
    field_name="vector",
    index_type="HNSW",
    metric_type="L2",
    params={"M": 16, "efConstruction": 200}
)
client.create_index(collection_name="demo_collection", index_params=index_params)

# 插入数据
client.insert(collection_name="demo_collection", data=data)

# 加载集合到内存（需在创建索引后执行）
client.load_collection(collection_name="demo_collection")
    

结语

运行上述代码后，即可成功实现批量导入数据到milvus向量数据库。如果在运行中遇到问题可以在评论区留言，我看到会积极回复，后续我还会出更多关于dify与大模型相关的实用教程，记得关注我，我们下次再见！