Milvus 向量数据库学习 Demo 涵盖 1. 连接 MilvusMilvusClient 2. 创建 Collectionschema 索引 3. 插入数据含向量 标量字段 4. 向量相似度搜索search 5. 标量字段过滤查询query 6. 按 ID 查询get 7. 删除数据 8. 删除 Collection import time from typing import Optional, List #向量归一化 import numpy as np from pymilvus import MilvusClient, DataType #向量归一化 def normalize(v): norm np.linalg.norm(v) return [x / norm for x in v] # # 配置按项目实际地址修改 # MILVUS_HOST 127.0.0.1 MILVUS_PORT 19530 库 MILVUS_DB default 表 COLLECTION_NAME t_employee 维度信息 VECTOR_DIM 4 # # 1. 连接 Milvus # def connect_milvus() - Optional[MilvusClient]: 创建 MilvusClient 连接 项目模式MilvusClient(urihttp://host:port, db_name...) try: client MilvusClient( urifhttp://{MILVUS_HOST}:{MILVUS_PORT}, db_nameMILVUS_DB, ) # ping 验证连接 client.list_collections() print(f[OK] 连接 Milvus 成功: {MILVUS_HOST}:{MILVUS_PORT}, db{MILVUS_DB}) return client except Exception as e: print(f[ERROR] 连接 Milvus 失败: {e}) return None # # 2. 创建 Collection # def create_collection(client: MilvusClient): 创建 t_employee 集合 字段设计 - id: VARCHAR 主键 - name: VARCHAR 员工姓名 - department: VARCHAR 部门 - salary: INT64 薪资 - vector: FLOAT_VECTOR 向量4维demo 用 - metadata: JSON 额外信息 # 如果已存在先删除demo 需要 if client.has_collection(collection_nameCOLLECTION_NAME): print(f[INFO] 集合已存在先删除: {COLLECTION_NAME}) client.drop_collection(collection_nameCOLLECTION_NAME) # 创建 Schema # 手动指定主键 # 允许动态字段项目用 True schema client.create_schema( auto_idFalse, enable_dynamic_fieldTrue, ) # 添加字段对标项目 add_field() schema.add_field( field_nameid, datatypeDataType.VARCHAR, max_length100, is_primaryTrue, nullableFalse, ) schema.add_field( field_namename, datatypeDataType.VARCHAR, max_length200, nullableTrue, ) schema.add_field( field_namedepartment, datatypeDataType.VARCHAR, max_length200, nullableTrue, ) schema.add_field( field_namesalary, datatypeDataType.INT64, nullableTrue, ) # 项目里可用大模型解析成1024 维 schema.add_field( field_namevector, datatypeDataType.FLOAT_VECTOR, dimVECTOR_DIM, ) # 代表扩展字段 schema.add_field( field_namemetadata, datatypeDataType.JSON, nullableTrue, ) # 创建集合 client.create_collection( collection_nameCOLLECTION_NAME, schemaschema, ) print(f[OK] 创建集合成功: {COLLECTION_NAME}) # 创建索引 index_params client.prepare_index_params() # 标量字段索引用于过滤加速 index_params.add_index( field_namedepartment, index_typeINVERTED, index_namedepartment_index, params{}, ) # 向量字段索引用于相似度搜索 # 配置AUTOINDEX IP内积AUTOINDEX 自动选择最优索引算法 index_params.add_index( field_namevector, index_typeAUTOINDEX, index_namevector_index, metric_typeIP, # 内积Inner Product ) client.create_index( collection_nameCOLLECTION_NAME, index_paramsindex_params, ) # 加载集合到内存查询前必须加载 client.load_collection(collection_nameCOLLECTION_NAME) # 新增阻塞等待加载完成 while True: state client.get_load_state(collection_nameCOLLECTION_NAME) if state[state] 3: # 3 Loaded break print(等待集合加载中...) time.sleep(0.2) print(f[OK] 索引创建并加载完成) # # 3. 插入数据 # def insert_employees(client: MilvusClient): 插入员工数据 项目模式client.insert(collection_name..., data[dict1, dict2, ...]) employees [ { id: emp_001, name: 张三, department: 技术部, salary: 25000, vector: normalize([0.12, 0.34, 0.56, 0.78]), # 模拟向量 metadata: {level: 高级, skills: [Python, ML]}, }, { id: emp_002, name: 李四, department: 技术部, salary: 20000, vector: normalize([0.15, 0.32, 0.54, 0.80]), metadata: {level: 中级, skills: [Java, Spring]}, }, { id: emp_003, name: 王五, department: 市场部, salary: 18000, vector: normalize([0.90, 0.78, 0.12, 0.34]), metadata: {level: 高级, skills: [营销, 品牌]}, }, { id: emp_004, name: 赵六, department: 市场部, salary: 15000, vector: normalize([0.85, 0.80, 0.15, 0.30]), metadata: {level: 中级, skills: [广告, 策划]}, }, { id: emp_005, name: 陈七, department: 人事部, salary: 16000, vector: normalize([0.45, 0.55, 0.65, 0.75]), metadata: {level: 初级, skills: [招聘, 培训]}, }, ] result client.insert( collection_nameCOLLECTION_NAME, dataemployees, ) print(f[OK] 插入 {len(employees)} 条员工数据, insert_count{result[insert_count]}) # # 4. 向量相似度搜索 # def search_similar_employees(client: MilvusClient, query_vector: List[float], top_k: int 3): 向量相似度搜索核心功能 项目模式client.search(collection_name..., data[vectors], ...) 搜索逻辑 1. 用 query_vector用户问题向量化后的结果去 Milvus 中找最相似的向量 2. 返回 top_k 条最匹配的结果 data # 查询向量支持批量传入列表套列表 anns_field # 在哪个向量字段上搜索 search_params # 搜索参数 output_fields # 返回的标量字段 result client.search( collection_nameCOLLECTION_NAME, data[query_vector], # 查询向量支持批量传入列表套列表 anns_fieldvector, # 在哪个向量字段上搜索 search_params{nprobe: 10}, # 搜索参数 limittop_k, # 返回 top-K output_fields[name, department, salary, metadata], # 返回的标量字段 ) print(f\n--- 向量搜索 (top_k{top_k}) ---) print(f 查询向量: {query_vector}) for i, hit in enumerate(result[0]): print(f [{i1}] id{hit[id]}, 距离{hit[distance]:.4f}) print(f 姓名: {hit[entity][name]}, 部门: {hit[entity][department]}, f薪资: {hit[entity][salary]}) return result # # 5. 标量字段过滤查询 # def query_by_department(client: MilvusClient, department: str): 按部门过滤查询标量字段过滤 项目模式client.query(collection_name..., filter..., output_fields...) result client.query( collection_nameCOLLECTION_NAME, filterfdepartment {department}, output_fields[id, name, department, salary, metadata], ) print(f\n--- 按部门查询: {department} ---) print(f 找到 {len(result)} 人:) for item in result: print(f - {item[name]} | 薪资 {item[salary]} | {item[metadata]}) return result def query_by_salary_range(client: MilvusClient, min_salary: int, max_salary: int): 按薪资范围过滤查询 result client.query( collection_nameCOLLECTION_NAME, filterfsalary {min_salary} salary {max_salary}, output_fields[id, name, department, salary], ) print(f\n--- 按薪资范围查询: {min_salary} ~ {max_salary} ---) print(f 找到 {len(result)} 人:) for item in result: print(f - {item[name]} ({item[department]}) 薪资 {item[salary]}) return result # # 6. 按 ID 查询 # def get_by_ids(client: MilvusClient, ids: List[str]): 按主键 ID 批量查询 result client.get( collection_nameCOLLECTION_NAME, idsids, output_fields[id, name, department, salary, vector], ) print(f\n--- 按 ID 查询: {ids} ---) for item in result: print(f - {item[id]}: {item[name]} | {item[department]} | 向量前3维: {item[vector][:3]}) return result # # 7. 删除数据 # def delete_by_filter(client: MilvusClient, filter_expr: str): 按条件删除对标项目中的 filter 删除 项目模式client.delete(collection_name..., filter...) result client.delete( collection_nameCOLLECTION_NAME, filterfilter_expr, ) print(f\n--- 删除: filter{filter_expr} ---) print(f 删除数量: {result[delete_count]}) return result # # 8. 主程序 # def main(): print( * 60) print(Milvus 向量数据库 Demo — 基于项目模式) print( * 60) # 1. 连接 client connect_milvus() if not client: return # 2. 创建 Collection create_collection(client) # 3. 插入数据 insert_employees(client) 等待数据加载到内存 time.sleep(1) # 4. 向量搜索找最像张三的人 # 张三的向量是 [0.12, 0.34, 0.56, 0.78] # 李四和他最像同技术部王五市场部最不像 search_similar_employees(client, query_vectornormalize([0.12, 0.34, 0.56, 0.78]), top_k3) # 5. 向量搜索找最像王五的人 search_similar_employees(client, query_vectornormalize([0.90, 0.78, 0.12, 0.34]), top_k3) # 6. 按部门过滤 query_by_department(client, 技术部) # 7. 按薪资范围过滤 query_by_salary_range(client, 18000, 30000) # 8. 按 ID 查询 get_by_ids(client, [emp_001, emp_003, emp_005]) # 9. 删除数据 delete_by_filter(client, department 人事部) # 验证删除 remaining client.query( collection_nameCOLLECTION_NAME, filterid ! , output_fields[id, name], ) print(f\n当前剩余数据: {len(remaining)} 条) for item in remaining: print(f - {item[id]}: {item[name]}) # 10. 清理 print(\n--- 清理 ---) client.drop_collection(collection_nameCOLLECTION_NAME) print(f[OK] 删除集合: {COLLECTION_NAME}) print(\n * 60) print(Demo 运行完毕) print( * 60) if __name__ __main__: main()