基本流程创建collection → 定义 Schema → 插入数据 → 创建索引 → 加载 Collection → 执行检索部署阶段从docker中拉取milvus 用管理员身份打开powershell输入docker run -d --name milvus-standalone -p 19530:19530 -p 9091:9091 milvusdb/milvus:latest如果下载成功后如图看到一串中英文的字符串或者通过创建yaml文件推荐在本地创建一个milvus文件夹比如路径为E:\Mydocument\milvus通过命令创建docker-compose.yml文件完整命令如下 version: 3.5 services: etcd: container_name: milvus-etcd image: quay.io/coreos/etcd:v3.5.5 environment: - ETCD_AUTO_COMPACTION_MODErevision - ETCD_AUTO_COMPACTION_RETENTION1000 - ETCD_QUOTA_BACKEND_BYTES4294967296 - ETCD_SNAPSHOT_COUNT50000 volumes: - ./volumes/etcd:/etcd command: etcd -advertise-client-urlshttp://127.0.0.1:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd minio: container_name: milvus-minio image: minio/minio:RELEASE.2023-03-20T20-16-18Z environment: MINIO_ACCESS_KEY: minioadmin MINIO_SECRET_KEY: minioadmin volumes: - ./volumes/minio:/minio_data command: minio server /minio_data healthcheck: test: [CMD, curl, -f, http://localhost:9000/minio/health/live] interval: 30s timeout: 20s retries: 3 standalone: container_name: milvus-standalone image: milvusdb/milvus:latest command: [milvus, run, standalone] environment: ETCD_ENDPOINTS: etcd:2379 MINIO_ADDRESS: minio:9000 volumes: - ./volumes/milvus:/var/lib/milvus ports: - 19530:19530 - 9091:9091 depends_on: - etcd - minio | Out-File -FilePath docker-compose.yml -Encoding utf8验证milvus是否启动成功成功界面如下图所示如果输入命令后没有任何容器只出现字段名通过查询log内容看哪里出错了将log复制到ai聊天找解决方案。或者打开docker desktop删除现有的容器重新run一次。电商客服知识库实践 代码用jupyer 演示首先在虚拟环境中安装好pymilvus pip install -U pymilvus下载attuReleases · zilliztech/attu (github.com)并安装测试连接也可以用python代码测试连接# 1. 导入库并连接 from pymilvus import MilvusClient # 连接到你的本地 Milvus 服务 client MilvusClient(urihttp://localhost:19530) # [reference:14][reference:15] # 2. 创建一个名为 demo_collection 的集合 # 指定向量的维度为 5 (此处仅为演示实际维度视模型而定) if client.has_collection(collection_namedemo_collection): client.drop_collection(collection_namedemo_collection) client.create_collection( collection_namedemo_collection, dimension5, # [reference:16] ) print(集合创建成功) # 3. 准备并插入一些数据 (向量) data [ {id: 1, vector: [0.1, 0.2, 0.3, 0.4, 0.5], color: red}, {id: 2, vector: [0.6, 0.7, 0.8, 0.9, 1.0], color: blue}, {id: 3, vector: [0.11, 0.22, 0.33, 0.44, 0.55], color: red}, ] res client.insert(collection_namedemo_collection, datadata) print(f插入了 {res[insert_count]} 条数据) # 4. 执行一次向量搜索 # 搜索与向量 [0.1, 0.2, 0.3, 0.4, 0.5] 最相似的 2 个向量 search_res client.search( collection_namedemo_collection, data[[0.1, 0.2, 0.3, 0.4, 0.5]], limit2, # 返回最相似的2个结果 output_fields[color], # 同时返回 color 字段 ) print(搜索结果) for result in search_res[0]: print(f ID: {result[id]}, 距离: {result[distance]}, Color: {result[entity][color]}) # 5. 清理 (可选) # client.drop_collection(collection_namedemo_collection)1. 在向量数据库中创建表collection 和表结构 Schemafrom pymilvus import MilvusClient, DataType, FieldSchema, CollectionSchema # 配置 VECTOR_DIM 4096 COLLECTION_NAME customer_service_chunks URI http://localhost:19530 def main(): # 1. 连接 Milvus client MilvusClient(uriURI) print(已连接到 Milvus) # 2. 如果集合已存在删除它避免参数冲突 if client.has_collection(COLLECTION_NAME): client.drop_collection(COLLECTION_NAME) print(f已删除旧集合{COLLECTION_NAME}) # 3. 定义字段 Schema fields [ FieldSchema(nameid, dtypeDataType.INT64, is_primaryTrue, auto_idTrue), FieldSchema(namevector, dtypeDataType.FLOAT_VECTOR, dimVECTOR_DIM), FieldSchema(namechunk_text, dtypeDataType.VARCHAR, max_length8192), FieldSchema(namedoc_id, dtypeDataType.VARCHAR, max_length64), FieldSchema(namecategory, dtypeDataType.VARCHAR, max_length32), ] schema CollectionSchema(fields, description电商客服知识库chunk集合) # 4. 创建 Collection仅一次 client.create_collection(COLLECTION_NAME, schemaschema) print(fCollection 创建成功{COLLECTION_NAME}) if __name__ __main__: main()2. 插入向量数据到向量数据库中此处要用到silicon平台的key申请地址硅基流动 SiliconFlow - 致力于成为全球领先的 AI 能力提供商import requests import json from pymilvus import MilvusClient # 配置 SILICONFLOW_API_KEY sk-你的key # 请替换为真实 Key EMBEDDING_URL https://api.siliconflow.cn/v1/embeddings EMBEDDING_MODEL Qwen/Qwen3-Embedding-8B COLLECTION_NAME customer_service_chunks MILVUS_URI http://localhost:19530 # 1. 连接 Milvus client MilvusClient(uriMILVUS_URI) print(已连接到 Milvus) # 2. 模拟电商客服知识库的 chunk 数据 chunk_texts [ 退货政策自签收之日起 7 天内商品未拆封、不影响二次销售的情况下支持无理由退货。退货运费由买家承担质量问题除外。, 退货政策生鲜食品、定制商品、贴身衣物等特殊商品不支持无理由退货。如有质量问题请在签收后 48 小时内联系客服并提供照片凭证。, 物流规则普通商品下单后 48 小时内发货预售商品以商品详情页标注的发货时间为准。偏远地区新疆、西藏、青海等可能需要额外 2~3 天。, 物流规则支持顺丰、中通、圆通、韵达等主流快递。默认使用中通快递如需指定快递公司请在下单时备注可能产生额外运费。, 促销活动2026 年春节大促全场满 300 减 50满 500 减 100。活动时间2026 年 1 月 20 日至 2 月 5 日。优惠券不可叠加使用。 ] doc_ids [doc_return_001, doc_return_001, doc_logistics_001, doc_logistics_001, doc_promo_001] categories [return_policy, return_policy, logistics, logistics, promotion] # 3. 调用 Embedding API 生成向量 def get_embeddings(texts): 批量获取文本的向量返回 List[List[float]] headers { Authorization: fBearer {SILICONFLOW_API_KEY}, Content-Type: application/json } payload { model: EMBEDDING_MODEL, input: texts } response requests.post(EMBEDDING_URL, headersheaders, jsonpayload) response.raise_for_status() # 如果状态码不是 200抛出异常 data response.json() embeddings [] for item in data[data]: embeddings.append(item[embedding]) # item[embedding] 是一个 float 列表 return embeddings print(正在生成向量...) vectors get_embeddings(chunk_texts) print(f成功生成 {len(vectors)} 个向量每个维度为 {len(vectors[0])}) # 4. 组装插入数据 # 注意因为 id 字段设置了 auto_idTrue插入数据时不需要提供 id data_rows [] for i, text in enumerate(chunk_texts): row { chunk_text: text, doc_id: doc_ids[i], category: categories[i], vector: vectors[i] # 列表 } data_rows.append(row) # 5. 插入 Milvus print(正在插入数据...) insert_result client.insert(collection_nameCOLLECTION_NAME, datadata_rows) # print(f插入成功共 {insert_result[insert_count]} 条) # print(f插入的主键 ID: {insert_result[primary_keys]}) print(f插入成功共 {insert_result[insert_count]} 条) # 安全获取主键 primary_keys insert_result.get(primary_keys) if primary_keys is None: primary_keys insert_result.get(ids) # 某些版本可能用 ids print(f插入的主键 ID: {primary_keys})3. 创建索引没有索引是无法做高效检索的等于MySQL暴力检索# from pymilvus import MilvusClient, IndexParams # 正确写法不需要导入IndexParams通过client.prepare_index_params()创建 from pymilvus import MilvusClient # 初始化客户端你原有连接逻辑保留 client MilvusClient(urihttp://localhost:19530) COLLECTION_NAME customer_service_chunks # 1. 向量字段 HNSW 索引 vector_index_params client.prepare_index_params() # 替代 IndexParams() vector_index_params.add_index( field_namevector, index_typeHNSW, metric_typeCOSINE, params{M: 16, efConstruction: 256} ) client.create_index(COLLECTION_NAME, index_paramsvector_index_params) print(✅ 向量索引HNSW创建成功) # 2. category 标量字段 TRIE 索引 category_index_params client.prepare_index_params() category_index_params.add_index( field_namecategory, index_typeTRIE ) client.create_index(COLLECTION_NAME, index_paramscategory_index_params) print(✅ 标量索引TRIE创建成功)将索引加载到内存中才能使用from pymilvus import MilvusClient # 沿用你之前初始化好的客户端对象 client MilvusClient(urihttp://localhost:19530) # 加载集合到内存等价Java LoadCollectionReq.builder().collectionName().build() client.loadCollection() client.load_collection( collection_namecustomer_service_chunks ) print(Collection 已加载到内存)4. 执行向量检索from pymilvus import MilvusClient # 配置 COLLECTION_NAME customer_service_chunks MILVUS_URI http://localhost:19530 QUERY 买了东西不想要了怎么退货 TOP_K 3 SEARCH_PARAMS {ef: 128} # HNSW 搜索时的动态搜索宽度 # 1. 连接 Milvus client MilvusClient(uriMILVUS_URI) # 2. 确保集合已加载如果还没有加载的话 # 如果已经加载过重复加载也无妨 client.load_collection(COLLECTION_NAME) print(✅ 集合已加载) # 3. 调用 Embedding API 将问题向量化复用您已有的 get_embeddings 函数 query_vectors get_embeddings([QUERY]) # 返回 List[List[float]] print(f✅ 查询向量生成成功维度{len(query_vectors[0])}) # 4. 执行向量检索 search_result client.search( collection_nameCOLLECTION_NAME, dataquery_vectors, # 查询向量列表这里只有一个 anns_fieldvector, # 在哪个向量字段上检索 search_paramsSEARCH_PARAMS, # 搜索参数如 ef limitTOP_K, # 返回 Top-K 条 output_fields[chunk_text, doc_id, category] # 返回这些字段 ) # 5. 输出检索结果 print(\n 检索结果 ) for idx, hit in enumerate(search_result[0]): # search_result[0] 是第一个查询的结果列表 print(fTop-{idx1}) print(f 相似度分数{hit[distance]:.4f}) # distance 是相似度距离 print(f 分类{hit[entity][category]}) print(f 文档ID{hit[entity][doc_id]}) print(f 内容{hit[entity][chunk_text]}) print()也可以增加元数据过滤条件的检索from pymilvus import MilvusClient # 配置 COLLECTION_NAME customer_service_chunks MILVUS_URI http://localhost:19530 QUERY 买了东西不想要了怎么退货 TOP_K 3 SEARCH_PARAMS {ef: 128} # 1. 连接 Milvus client MilvusClient(uriMILVUS_URI) # 2. 确保集合已加载 client.load_collection(COLLECTION_NAME) print(✅ 集合已加载) # 3. 用户问题向量化复用 get_embeddings 函数 query_vectors get_embeddings([QUERY]) # 返回 List[List[float]] print(f✅ 查询向量生成成功维度{len(query_vectors[0])}) # 混合检索向量相似度 标量过滤 # 只在退货政策类的 chunk 里检索 filtered_search_result client.search( collection_nameCOLLECTION_NAME, dataquery_vectors, # 查询向量列表 anns_fieldvector, # 向量字段名 search_paramsSEARCH_PARAMS, # 搜索参数HNSW 的 ef limitTOP_K, # 返回 Top-K output_fields[chunk_text, doc_id, category], # 需要返回的字段 filtercategory return_policy # 标量过滤条件注意用单引号括起字符串 ) # 4. 输出过滤后的结果 print(\n 过滤检索结果仅退货政策 ) for idx, hit in enumerate(filtered_search_result[0]): print(fTop-{idx1}) print(f 相似度分数{hit[distance]:.4f}) print(f 内容{hit[entity][chunk_text]}) print()