基本流程：

创建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_MODE=revision >> - ETCD_AUTO_COMPACTION_RETENTION=1000 >> - ETCD_QUOTA_BACKEND_BYTES=4294967296 >> - ETCD_SNAPSHOT_COUNT=50000 >> volumes: >> - ./volumes/etcd:/etcd >> command: etcd -advertise-client-urls=http://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(uri="http://localhost:19530") # [reference:14][reference:15] # 2. 创建一个名为 "demo_collection" 的集合 # 指定向量的维度为 5 (此处仅为演示，实际维度视模型而定) if client.has_collection(collection_name="demo_collection"): client.drop_collection(collection_name="demo_collection") client.create_collection( collection_name="demo_collection", dimension=5, # [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_name="demo_collection", data=data) print(f"插入了 {res['insert_count']} 条数据") # 4. 执行一次向量搜索 # 搜索与向量 [0.1, 0.2, 0.3, 0.4, 0.5] 最相似的 2 个向量 search_res = client.search( collection_name="demo_collection", data=[[0.1, 0.2, 0.3, 0.4, 0.5]], limit=2, # 返回最相似的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_name="demo_collection")

1. 在向量数据库中创建表collection 和表结构 Schema：

from pymilvus import MilvusClient, DataType, FieldSchema, CollectionSchema # 配置 VECTOR_DIM = 4096 COLLECTION_NAME = "customer_service_chunks" URI = "http://localhost:19530" def main(): # 1. 连接 Milvus client = MilvusClient(uri=URI) print("已连接到 Milvus") # 2. 如果集合已存在，删除它（避免参数冲突） if client.has_collection(COLLECTION_NAME): client.drop_collection(COLLECTION_NAME) print(f"已删除旧集合：{COLLECTION_NAME}") # 3. 定义字段 Schema fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True), FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=VECTOR_DIM), FieldSchema(name="chunk_text", dtype=DataType.VARCHAR, max_length=8192), FieldSchema(name="doc_id", dtype=DataType.VARCHAR, max_length=64), FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=32), ] schema = CollectionSchema(fields, description="电商客服知识库chunk集合") # 4. 创建 Collection（仅一次） client.create_collection(COLLECTION_NAME, schema=schema) print(f"Collection 创建成功：{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(uri=MILVUS_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": f"Bearer {SILICONFLOW_API_KEY}", "Content-Type": "application/json" } payload = { "model": EMBEDDING_MODEL, "input": texts } response = requests.post(EMBEDDING_URL, headers=headers, json=payload) 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_id=True，插入数据时不需要提供 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_name=COLLECTION_NAME, data=data_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(uri="http://localhost:19530") COLLECTION_NAME = "customer_service_chunks" # ========== 1. 向量字段 HNSW 索引 ========== vector_index_params = client.prepare_index_params() # 替代 IndexParams() vector_index_params.add_index( field_name="vector", index_type="HNSW", metric_type="COSINE", params={"M": 16, "efConstruction": 256} ) client.create_index(COLLECTION_NAME, index_params=vector_index_params) print("✅ 向量索引（HNSW）创建成功") # ========== 2. category 标量字段 TRIE 索引 ========== category_index_params = client.prepare_index_params() category_index_params.add_index( field_name="category", index_type="TRIE" ) client.create_index(COLLECTION_NAME, index_params=category_index_params) print("✅ 标量索引（TRIE）创建成功")

将索引加载到内存中，才能使用

from pymilvus import MilvusClient # 沿用你之前初始化好的客户端对象 client = MilvusClient(uri="http://localhost:19530") # 加载集合到内存（等价Java LoadCollectionReq.builder().collectionName().build() + client.loadCollection()） client.load_collection( collection_name="customer_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(uri=MILVUS_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_name=COLLECTION_NAME, data=query_vectors, # 查询向量列表（这里只有一个） anns_field="vector", # 在哪个向量字段上检索 search_params=SEARCH_PARAMS, # 搜索参数（如 ef） limit=TOP_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(f"Top-{idx+1}：") 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(uri=MILVUS_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_name=COLLECTION_NAME, data=query_vectors, # 查询向量列表 anns_field="vector", # 向量字段名 search_params=SEARCH_PARAMS, # 搜索参数（HNSW 的 ef） limit=TOP_K, # 返回 Top-K output_fields=["chunk_text", "doc_id", "category"], # 需要返回的字段 filter="category == 'return_policy'" # 标量过滤条件（注意用单引号括起字符串） ) # 4. 输出过滤后的结果 print("\n=== 过滤检索结果（仅退货政策） ===") for idx, hit in enumerate(filtered_search_result[0]): print(f"Top-{idx+1}：") print(f" 相似度分数：{hit['distance']:.4f}") print(f" 内容：{hit['entity']['chunk_text']}") print()