零基础学AI大模型之Milvus DML实战

# 安装PyMilvus（若未安装）
pip install pymilvus==2.5.5

# Python连接Milvus服务（核心代码，后续操作均基于此连接）
from pymilvus import MilvusClient, DataType

# 实例化客户端，连接远程Milvus服务（替换为你的服务地址）
client = MilvusClient(uri="http://192.168.229.128:19530")

# 验证连接（列出所有数据库，无报错则连接成功）
databases = client.list_databases()
print("当前Milvus数据库列表：", databases)  # 输出：['default', ...]

# 步骤1：定义Schema（字段结构）
schema = client.create_schema(
    auto_id=False,  # 关闭自动主键，手动指定ID（也可设为True自动生成）
    enable_dynamic_field=True  # 开启动态字段，支持灵活扩展
)

# 添加字段（主键字段+向量字段，动态字段无需预定义）
schema.add_field(
    field_name="id",
    datatype=DataType.INT64,
    is_primary=True  # 主键字段（必须唯一，不可为向量类型）
)
schema.add_field(
    field_name="vector",
    datatype=DataType.FLOAT_VECTOR,
    dim=128  # 向量维度（需与实际插入数据一致，如768维BERT向量）
)

# 验证Schema合法性（避免字段定义错误）
schema.verify()

# 步骤2：定义索引参数（向量字段需创建索引，否则查询低效）
index_params = client.prepare_index_params()
index_params.add_index(
    field_name="vector",  # 索引字段（向量字段）
    index_type="IVF_FLAT",  # 索引类型（中大数据量首选，性价比高）
    metric_type="L2",  # 距离度量方式（欧氏距离）
    params={"nlist": 1024}  # 索引参数（聚类中心数，建议sqrt(数据量)）
)

# 步骤3：创建集合（集合名自定义，需唯一）
client.create_collection(
    collection_name="dml_demo_collection",  # 集合名
    schema=schema,  # 字段结构
    index_params=index_params  # 索引配置
)

print("集合创建成功！")

# 场景1：单条数据插入
single_data = {
    "id": 1,
    "vector": [0.1]*128,  # 128维向量（简化为全0.1，实际需替换为真实向量）
    "text": "这是第一条测试数据",  # 动态字段（未预定义，直接插入）
    "category": "test"  # 动态字段（灵活扩展）
}

# 单条插入
single_insert_result = client.insert(
    collection_name="dml_demo_collection",
    data=single_data
)
print("单条插入成功，主键ID：", single_insert_result["ids"])  # 输出：[1]

# 场景2：批量数据插入（推荐，效率更高）
batch_data = [
    {"id": 2, "vector": [0.2]*128, "text": "批量插入-测试数据2", "category": "test"},
    {"id": 3, "vector": [0.3]*128, "text": "批量插入-测试数据3", "category": "formal"},
    {"id": 4, "vector": [0.4]*128, "text": "批量插入-测试数据4", "category": "formal"}
]

# 批量插入
batch_insert_result = client.insert(
    collection_name="dml_demo_collection",
    data=batch_data
)
print("批量插入成功，主键ID列表：", batch_insert_result["ids"])  # 输出：[2,3,4]

# 场景1：按主键删除（精准删除，推荐优先使用）
delete_ids = [1, 2]  # 要删除的主键ID列表
client.delete(
    collection_name="dml_demo_collection",
    ids=delete_ids
)
print(f"主键ID为{delete_ids}的数据已删除")

# 场景2：按条件删除（灵活筛选，支持标量字段/动态字段过滤）
# 示例：删除category为"formal"且text包含"测试"的数据
client.delete(
    collection_name="dml_demo_collection",
    filter='category == "formal" and text like "%测试%"'  # 过滤条件（类似SQL语法）
)
print('category为"formal"且text包含"测试"的数据已删除')

# 需求：更新主键ID=3的数据（vector和text字段）
old_id = 3  # 要更新的旧数据主键ID

# 步骤1：删除旧数据
client.delete(
    collection_name="dml_demo_collection",
    ids=[old_id]
)
print(f"主键ID={old_id}的旧数据已删除")

# 步骤2：插入新数据（复用旧主键ID，实现“更新”效果）
updated_data = {
    "id": old_id,  # 保持主键ID不变
    "vector": [0.35]*128,  # 新的向量值
    "text": "更新后的测试数据3",  # 新的文本内容
    "category": "formal",  # 保留原有的动态字段值
    "update_time": "2024-01-01"  # 新增动态字段（更新时间）
}

client.insert(
    collection_name="dml_demo_collection",
    data=updated_data
)
print(f"主键ID={old_id}的数据已更新完成")

# 加载集合到内存（查询/删除前必须执行）
client.load_collection(collection_name="dml_demo_collection")
print("集合加载成功，可执行查询/删除操作")

# 释放集合内存（操作完成后执行，节省资源）
client.release_collection(collection_name="dml_demo_collection")
print("集合已释放内存")