1. Kubernetes简介 1.1 什么是Kubernetes Kubernetes（简称K8s）是一个开源的容器编排平台，用于自动化部署、扩展和管理容器化应用程序。它最初由Google开发，现在由Cloud Native Computing Foundation（CNCF）维护。 1.2 核心特性 自动化部署和回滚：支持应用的自动化部署和版本回滚 服务发现和负载均衡：自动分配IP地址和DNS名称，并在多个容器间分配负载 存储编排：自动挂载存储系统，包括本地存储、公有云存储等 自我修复：重启失败的容器，替换和重新调度节点上的容器 密钥和配置管理：管理敏感信息和应用配置 水平扩展：根据CPU使用率或其他指标自动扩展应用 1.3 使用场景 微服务架构部署 容器化应用管理 多云和混合云部署 DevOps和CI/CD流水线 大规模应用编排 2. 核心概念与架构 2.1 集群架构 Kubernetes集群由以下组件组成： 2.1.1 Master节点 API Server：集群的统一入口，提供RESTful API etcd：分布式键值存储，保存集群状态数据 Controller Manager：运行控制器进程，管理集群状态 Scheduler：负责Pod的调度决策 2.1.2 Worker节点 kubelet：节点代理，管理Pod生命周期 kube-proxy：网络代理，实现服务的网络规则 Container Runtime：容器运行时（如Docker、containerd） 2.2 核心对象 2.2.1 Pod 最小的部署单元，包含一个或多个容器： ...

1. Milvus简介 什么是Milvus Milvus是一个开源的向量数据库，专为处理大规模向量数据而设计。它支持多种向量相似性搜索算法，能够处理十亿级别的向量数据，广泛应用于AI应用场景，如推荐系统、图像检索、自然语言处理等。 主要特性 高性能：支持十亿级向量的毫秒级检索 多样化索引：支持多种向量索引算法（IVF、HNSW、ANNOY等） 云原生：基于Kubernetes的分布式架构 多语言SDK：支持Python、Java、Go、Node.js等 ACID事务：保证数据一致性 混合搜索：支持向量和标量数据的混合查询 应用场景 推荐系统：基于用户行为向量进行个性化推荐 图像检索：以图搜图、相似图片查找 文本搜索：语义搜索、文档相似性匹配 视频分析：视频内容检索和分析 药物发现：分子结构相似性搜索 异常检测：基于向量距离的异常识别 2. 核心概念 基本术语 Collection（集合） 类似于关系数据库中的表，用于存储向量数据和相关的标量字段。 Field（字段） 集合中的列，包括向量字段和标量字段。 Entity（实体） 集合中的一行数据，包含多个字段的值。 Partition（分区） 集合的子集，用于数据分片和查询优化。 Index（索引） 为加速向量检索而构建的数据结构。 Segment（段） Milvus内部的数据存储单元，用于数据管理和查询优化。 数据类型 向量类型 FloatVector：浮点数向量 BinaryVector：二进制向量 标量类型 Bool：布尔值 Int8/Int16/Int32/Int64：整数 Float/Double：浮点数 String/VarChar：字符串 JSON：JSON对象 3. 安装部署 系统要求 硬件要求 CPU：x86_64架构，支持SSE4.2指令集 内存：8GB以上（推荐16GB+） 存储：SSD硬盘（推荐NVMe） 网络：千兆网卡 软件要求 操作系统：Ubuntu 18.04+、CentOS 7+、macOS 10.14+ Docker：20.10+ Docker Compose：1.28+ Docker安装（推荐） 1. 下载配置文件 1 2 # 下载docker-compose.yml wget https://github.com/milvus-io/milvus/releases/download/v2.3.0/milvus-standalone-docker-compose.yml -O docker-compose.yml 2. 启动Milvus 1 2 3 4 5 # 启动服务 docker-compose up -d # 检查服务状态 docker-compose ps 3. 验证安装 1 2 # 检查Milvus是否正常运行 curl -X GET "http://localhost:9091/health" Kubernetes部署 1. 安装Helm 1 curl https://raw.githubusercontent.com/helm/helm/main/scripts/get-helm-3 | bash 2. 添加Milvus Helm仓库 1 2 helm repo add milvus https://milvus-io.github.io/milvus-helm/ helm repo update 3. 部署Milvus 1 2 3 4 5 # 创建命名空间 kubectl create namespace milvus # 部署Milvus集群 helm install milvus milvus/milvus --namespace milvus 源码编译安装 1. 安装依赖 1 2 3 4 5 6 7 # Ubuntu/Debian sudo apt update sudo apt install -y build-essential cmake libopenblas-dev # CentOS/RHEL sudo yum groupinstall -y "Development Tools" sudo yum install -y cmake openblas-devel 2. 编译安装 1 2 3 4 5 6 7 8 9 # 克隆源码 git clone https://github.com/milvus-io/milvus.git cd milvus # 编译 make build # 启动 ./bin/milvus run standalone 4. 快速开始 安装Python SDK 1 pip install pymilvus 基本操作示例 1. 连接Milvus 1 2 3 4 5 6 7 8 9 10 from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType # 连接到Milvus connections.connect( alias="default", host='localhost', port='19530' ) print("Connected to Milvus") 2. 创建集合 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 # 定义字段 fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=False), FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=128), FieldSchema(name="title", dtype=DataType.VARCHAR, max_length=200), FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=50) ] # 创建集合schema schema = CollectionSchema( fields=fields, description="Document embedding collection" ) # 创建集合 collection = Collection( name="documents", schema=schema ) print("Collection created") 3. 插入数据 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 import random # 准备数据 num_entities = 1000 entities = [ [i for i in range(num_entities)], # id字段 [[random.random() for _ in range(128)] for _ in range(num_entities)], # embedding字段 [f"Document {i}" for i in range(num_entities)], # title字段 [f"Category {i % 10}" for i in range(num_entities)] # category字段 ] # 插入数据 insert_result = collection.insert(entities) print(f"Inserted {len(insert_result.primary_keys)} entities") # 刷新数据到磁盘 collection.flush() 4. 创建索引 1 2 3 4 5 6 7 8 9 10 11 12 13 14 # 定义索引参数 index_params = { "metric_type": "L2", "index_type": "IVF_FLAT", "params": {"nlist": 128} } # 创建索引 collection.create_index( field_name="embedding", index_params=index_params ) print("Index created") 5. 加载集合 1 2 3 # 加载集合到内存 collection.load() print("Collection loaded") 6. 向量检索 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 # 准备查询向量 query_vectors = [[random.random() for _ in range(128)]] # 执行搜索 search_params = {"metric_type": "L2", "params": {"nprobe": 10}} results = collection.search( data=query_vectors, anns_field="embedding", param=search_params, limit=10, output_fields=["title", "category"] ) # 输出结果 for hits in results: for hit in hits: print(f"ID: {hit.id}, Distance: {hit.distance}, Title: {hit.entity.get('title')}") 5. 数据管理 集合管理 创建集合 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 from pymilvus import Collection, FieldSchema, CollectionSchema, DataType # 定义复杂schema fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True), FieldSchema(name="vector", dtype=DataType.FLOAT_VECTOR, dim=256), FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=1000), FieldSchema(name="score", dtype=DataType.FLOAT), FieldSchema(name="timestamp", dtype=DataType.INT64), FieldSchema(name="metadata", dtype=DataType.JSON) ] schema = CollectionSchema( fields=fields, description="Advanced collection with multiple field types", enable_dynamic_field=True # 启用动态字段 ) collection = Collection(name="advanced_collection", schema=schema) 查看集合信息 1 2 3 4 5 6 7 8 9 10 11 12 13 # 获取集合统计信息 stats = collection.get_stats() print(f"Collection stats: {stats}") # 获取集合schema schema = collection.schema for field in schema.fields: print(f"Field: {field.name}, Type: {field.dtype}, Params: {field.params}") # 检查集合是否存在 from pymilvus import utility has_collection = utility.has_collection("advanced_collection") print(f"Collection exists: {has_collection}") 删除集合 1 2 3 4 5 # 删除集合 collection.drop() # 或者使用utility函数 utility.drop_collection("collection_name") 分区管理 创建分区 1 2 3 4 5 6 7 8 # 创建分区 collection.create_partition("partition_2023") collection.create_partition("partition_2024") # 查看所有分区 partitions = collection.partitions for partition in partitions: print(f"Partition: {partition.name}") 分区数据操作 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 # 向特定分区插入数据 entities = [ [1, 2, 3], # ids [[0.1] * 256, [0.2] * 256, [0.3] * 256], # vectors ["text1", "text2", "text3"], # text [0.8, 0.9, 0.7], # scores [1640995200, 1640995300, 1640995400], # timestamps [{"key": "value1"}, {"key": "value2"}, {"key": "value3"}] # metadata ] collection.insert(entities, partition_name="partition_2023") # 在特定分区中搜索 results = collection.search( data=[[0.1] * 256], anns_field="vector", param={"metric_type": "L2", "params": {"nprobe": 10}}, limit=10, partition_names=["partition_2023"] ) 数据插入和更新 批量插入 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 import numpy as np # 大批量数据插入 batch_size = 10000 for i in range(0, 100000, batch_size): ids = list(range(i, min(i + batch_size, 100000))) vectors = np.random.random((len(ids), 256)).tolist() texts = [f"Document {j}" for j in ids] scores = np.random.random(len(ids)).tolist() timestamps = [1640995200 + j for j in ids] metadata = [{"batch": i // batch_size} for _ in ids] entities = [ids, vectors, texts, scores, timestamps, metadata] collection.insert(entities) if i % 50000 == 0: collection.flush() # 定期刷新 print(f"Inserted {i + len(ids)} entities") 数据更新（Upsert） 1 2 3 4 5 6 7 8 9 10 11 # Milvus 2.3+支持upsert操作 update_entities = [ [1, 2, 3], # 更新已存在的ID [[0.5] * 256, [0.6] * 256, [0.7] * 256], # 新的向量 ["Updated text1", "Updated text2", "Updated text3"], # 新的文本 [0.95, 0.96, 0.97], # 新的分数 [1640995500, 1640995600, 1640995700], # 新的时间戳 [{"updated": True}, {"updated": True}, {"updated": True}] # 新的元数据 ] collection.upsert(update_entities) 数据删除 按ID删除 1 2 3 4 5 6 7 8 # 删除指定ID的实体 delete_ids = [1, 2, 3, 4, 5] expr = f"id in {delete_ids}" collection.delete(expr) # 删除满足条件的实体 expr = "score < 0.5" collection.delete(expr) 按条件删除 1 2 3 4 5 6 7 # 复杂删除条件 expr = "score < 0.3 and timestamp < 1640995300" collection.delete(expr) # 使用JSON字段删除 expr = "JSON_CONTAINS(metadata, '\"updated\": true')" collection.delete(expr) 6. 向量检索 基本检索 相似性搜索 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 # 基本向量搜索 query_vectors = [[0.1] * 256, [0.2] * 256] search_params = { "metric_type": "L2", "params": {"nprobe": 16} } results = collection.search( data=query_vectors, anns_field="vector", param=search_params, limit=10, output_fields=["text", "score", "timestamp"] ) for i, hits in enumerate(results): print(f"Query {i} results:") for hit in hits: print(f" ID: {hit.id}, Distance: {hit.distance:.4f}") print(f" Text: {hit.entity.get('text')}") print(f" Score: {hit.entity.get('score')}") 混合搜索 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 # 向量搜索 + 标量过滤 query_vectors = [[0.1] * 256] search_params = {"metric_type": "L2", "params": {"nprobe": 16}} # 添加标量过滤条件 filter_expr = "score > 0.8 and timestamp > 1640995200" results = collection.search( data=query_vectors, anns_field="vector", param=search_params, limit=10, expr=filter_expr, output_fields=["text", "score", "timestamp", "metadata"] ) 高级检索 范围搜索 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 # 搜索距离在指定范围内的向量 from pymilvus import SearchResult query_vectors = [[0.1] * 256] search_params = { "metric_type": "L2", "params": { "nprobe": 16, "radius": 0.1, # 最大距离 "range_filter": 0.05 # 最小距离 } } results = collection.search( data=query_vectors, anns_field="vector", param=search_params, limit=100, output_fields=["text", "score"] ) 多向量搜索 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 # 同时搜索多个向量字段（如果集合有多个向量字段） # 假设有text_vector和image_vector两个字段 # 创建包含多个向量字段的集合 multi_vector_fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True), FieldSchema(name="text_vector", dtype=DataType.FLOAT_VECTOR, dim=128), FieldSchema(name="image_vector", dtype=DataType.FLOAT_VECTOR, dim=256), FieldSchema(name="title", dtype=DataType.VARCHAR, max_length=200) ] multi_schema = CollectionSchema(fields=multi_vector_fields) multi_collection = Collection(name="multi_vector_collection", schema=multi_schema) # 分别在不同向量字段上搜索 text_results = multi_collection.search( data=[[0.1] * 128], anns_field="text_vector", param={"metric_type": "L2", "params": {"nprobe": 16}}, limit=10 ) image_results = multi_collection.search( data=[[0.1] * 256], anns_field="image_vector", param={"metric_type": "L2", "params": {"nprobe": 16}}, limit=10 ) 查询操作 标量查询 1 2 3 4 5 6 7 8 9 # 基于标量字段的查询 query_expr = "score > 0.8" results = collection.query( expr=query_expr, output_fields=["id", "text", "score", "timestamp"] ) for result in results: print(f"ID: {result['id']}, Text: {result['text']}, Score: {result['score']}") 复杂查询 1 2 3 4 5 6 7 8 9 10 11 # 复杂查询表达式 complex_expr = """ (score > 0.8 and timestamp > 1640995200) or (score > 0.9 and JSON_CONTAINS(metadata, '"important": true')) """ results = collection.query( expr=complex_expr, output_fields=["*"], # 输出所有字段 limit=100 ) 分页查询 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 # 分页查询大量数据 page_size = 1000 offset = 0 while True: results = collection.query( expr="score > 0.5", output_fields=["id", "text", "score"], limit=page_size, offset=offset ) if not results: break print(f"Page {offset // page_size + 1}: {len(results)} results") # 处理结果 for result in results: # 处理每个结果 pass offset += page_size 7. 索引管理 索引类型 FLAT索引 1 2 3 4 5 6 7 8 9 10 11 # FLAT索引 - 精确搜索，适合小数据集 flat_index = { "index_type": "FLAT", "metric_type": "L2", "params": {} } collection.create_index( field_name="vector", index_params=flat_index ) IVF索引 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 # IVF_FLAT索引 - 平衡性能和精度 ivf_flat_index = { "index_type": "IVF_FLAT", "metric_type": "L2", "params": { "nlist": 128 # 聚类中心数量 } } # IVF_PQ索引 - 压缩存储，适合大数据集 ivf_pq_index = { "index_type": "IVF_PQ", "metric_type": "L2", "params": { "nlist": 128, "m": 16, # PQ分段数 "nbits": 8 # 每段的位数 } } collection.create_index(field_name="vector", index_params=ivf_pq_index) HNSW索引 1 2 3 4 5 6 7 8 9 10 11 # HNSW索引 - 高性能近似搜索 hnsw_index = { "index_type": "HNSW", "metric_type": "L2", "params": { "M": 16, # 每层的最大连接数 "efConstruction": 200 # 构建时的搜索深度 } } collection.create_index(field_name="vector", index_params=hnsw_index) ANNOY索引 1 2 3 4 5 6 7 8 9 10 # ANNOY索引 - 内存友好 annoy_index = { "index_type": "ANNOY", "metric_type": "L2", "params": { "n_trees": 8 # 树的数量 } } collection.create_index(field_name="vector", index_params=annoy_index) 距离度量 欧几里得距离（L2） 1 2 3 4 5 l2_index = { "index_type": "IVF_FLAT", "metric_type": "L2", # 欧几里得距离 "params": {"nlist": 128} } 内积（IP） 1 2 3 4 5 ip_index = { "index_type": "IVF_FLAT", "metric_type": "IP", # 内积 "params": {"nlist": 128} } 余弦相似度 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 # 余弦相似度需要先归一化向量，然后使用IP import numpy as np def normalize_vectors(vectors): """归一化向量以使用余弦相似度""" vectors = np.array(vectors) norms = np.linalg.norm(vectors, axis=1, keepdims=True) return (vectors / norms).tolist() # 插入归一化后的向量 normalized_vectors = normalize_vectors(original_vectors) entities = [ids, normalized_vectors, texts, scores, timestamps, metadata] collection.insert(entities) # 使用IP度量进行余弦相似度搜索 cosine_index = { "index_type": "IVF_FLAT", "metric_type": "IP", "params": {"nlist": 128} } 索引管理操作 查看索引信息 1 2 3 4 5 6 7 8 9 10 # 获取索引信息 index_info = collection.index() print(f"Index type: {index_info.params['index_type']}") print(f"Metric type: {index_info.params['metric_type']}") print(f"Index params: {index_info.params['params']}") # 检查索引构建进度 from pymilvus import utility index_progress = utility.index_building_progress("collection_name") print(f"Index building progress: {index_progress}") 重建索引 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 # 删除现有索引 collection.drop_index() # 创建新索引 new_index = { "index_type": "HNSW", "metric_type": "L2", "params": {"M": 32, "efConstruction": 400} } collection.create_index(field_name="vector", index_params=new_index) # 等待索引构建完成 import time while True: progress = utility.index_building_progress(collection.name) if progress['pending_index_rows'] == 0: break print(f"Index building progress: {progress}") time.sleep(5) print("Index building completed") 8. 性能优化 搜索参数优化 IVF索引优化 1 2 3 4 5 6 7 8 9 10 11 12 # 根据数据量调整nlist data_size = collection.num_entities optimal_nlist = int(np.sqrt(data_size)) optimal_nlist = max(128, min(optimal_nlist, 4096)) # 限制在合理范围内 # 搜索时调整nprobe search_params = { "metric_type": "L2", "params": { "nprobe": min(optimal_nlist // 4, 64) # 通常设置为nlist的1/4 } } HNSW索引优化 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 # 构建时参数 hnsw_build_params = { "index_type": "HNSW", "metric_type": "L2", "params": { "M": 16, # 连接数，影响精度和内存 "efConstruction": 200 # 构建时搜索深度 } } # 搜索时参数 hnsw_search_params = { "metric_type": "L2", "params": { "ef": 100 # 搜索时的候选数量，越大精度越高但速度越慢 } } 内存管理 集合加载策略 1 2 3 4 5 6 7 8 9 10 11 # 部分加载 - 只加载需要的字段 collection.load(replica_number=1, _resource_groups=["default"]) # 释放不需要的集合 collection.release() # 检查内存使用 from pymilvus import utility memory_info = utility.get_query_segment_info(collection.name) for info in memory_info: print(f"Segment {info.segmentID}: {info.mem_size} bytes") 分区加载 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 # 只加载特定分区 collection.load(partition_names=["partition_2024"]) # 动态加载/释放分区 def load_partition_by_date(date_str): partition_name = f"partition_{date_str}" if partition_name in [p.name for p in collection.partitions]: collection.load(partition_names=[partition_name]) return True return False def release_old_partitions(keep_days=7): from datetime import datetime, timedelta cutoff_date = datetime.now() - timedelta(days=keep_days) for partition in collection.partitions: if partition.name.startswith("partition_"): date_str = partition.name.replace("partition_", "") try: partition_date = datetime.strptime(date_str, "%Y%m%d") if partition_date < cutoff_date: collection.release(partition_names=[partition.name]) print(f"Released partition: {partition.name}") except ValueError: continue 批处理优化 批量插入优化 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 def optimized_batch_insert(collection, data, batch_size=10000): """优化的批量插入函数""" total_entities = len(data[0]) for i in range(0, total_entities, batch_size): end_idx = min(i + batch_size, total_entities) batch_data = [field_data[i:end_idx] for field_data in data] # 插入批次 collection.insert(batch_data) # 定期刷新 if (i + batch_size) % 50000 == 0: collection.flush() print(f"Inserted and flushed {i + batch_size} entities") # 最终刷新 collection.flush() print(f"Completed insertion of {total_entities} entities") 并行搜索 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 import concurrent.futures import threading def parallel_search(collection, query_vectors, search_params, max_workers=4): """并行执行多个搜索请求""" def search_batch(vectors_batch): return collection.search( data=vectors_batch, anns_field="vector", param=search_params, limit=10 ) # 将查询向量分批 batch_size = len(query_vectors) // max_workers batches = [query_vectors[i:i+batch_size] for i in range(0, len(query_vectors), batch_size)] # 并行执行搜索 with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_batch = {executor.submit(search_batch, batch): batch for batch in batches} all_results = [] for future in concurrent.futures.as_completed(future_to_batch): batch_results = future.result() all_results.extend(batch_results) return all_results 连接池管理 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 from pymilvus import connections import threading class MilvusConnectionPool: def __init__(self, host='localhost', port='19530', pool_size=10): self.host = host self.port = port self.pool_size = pool_size self.connections = [] self.lock = threading.Lock() self._initialize_pool() def _initialize_pool(self): for i in range(self.pool_size): alias = f"connection_{i}" connections.connect( alias=alias, host=self.host, port=self.port ) self.connections.append(alias) def get_connection(self): with self.lock: if self.connections: return self.connections.pop() else: # 如果池为空，创建新连接 alias = f"temp_connection_{threading.current_thread().ident}" connections.connect( alias=alias, host=self.host, port=self.port ) return alias def return_connection(self, alias): with self.lock: if len(self.connections) < self.pool_size: self.connections.append(alias) else: connections.disconnect(alias) # 使用连接池 pool = MilvusConnectionPool() def search_with_pool(query_vector): alias = pool.get_connection() try: # 使用指定连接执行搜索 connections.connect(alias=alias) collection = Collection("documents", using=alias) results = collection.search( data=[query_vector], anns_field="vector", param={"metric_type": "L2", "params": {"nprobe": 16}}, limit=10 ) return results finally: pool.return_connection(alias) 9. 集群部署 Kubernetes集群部署 1. 准备配置文件 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 # milvus-cluster-values.yaml cluster: enabled: true image: all: repository: milvusdb/milvus tag: v2.3.0 pullPolicy: IfNotPresent service: type: LoadBalancer port: 19530 portName: milvus nodePort: 30530 rootCoordinator: replicas: 1 resources: limits: cpu: 1 memory: 2Gi requests: cpu: 0.5 memory: 1Gi queryCoordinator: replicas: 1 resources: limits: cpu: 1 memory: 2Gi requests: cpu: 0.5 memory: 1Gi queryNode: replicas: 2 resources: limits: cpu: 2 memory: 8Gi requests: cpu: 1 memory: 4Gi indexNode: replicas: 1 resources: limits: cpu: 2 memory: 4Gi requests: cpu: 1 memory: 2Gi dataNode: replicas: 2 resources: limits: cpu: 1 memory: 4Gi requests: cpu: 0.5 memory: 2Gi proxy: replicas: 2 resources: limits: cpu: 1 memory: 2Gi requests: cpu: 0.5 memory: 1Gi # 存储配置 minio: enabled: true mode: distributed replicas: 4 persistence: enabled: true size: 100Gi storageClass: "fast-ssd" etcd: enabled: true replicaCount: 3 persistence: enabled: true size: 10Gi storageClass: "fast-ssd" pulsar: enabled: true components: broker: true bookkeeper: true zookeeper: true zookeeper: replicaCount: 3 bookkeeper: replicaCount: 3 broker: replicaCount: 2 2. 部署集群 1 2 3 4 5 6 7 8 9 10 11 # 创建命名空间 kubectl create namespace milvus-cluster # 部署Milvus集群 helm install milvus-cluster milvus/milvus \ --namespace milvus-cluster \ --values milvus-cluster-values.yaml # 检查部署状态 kubectl get pods -n milvus-cluster kubectl get services -n milvus-cluster 3. 配置负载均衡 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 # milvus-ingress.yaml apiVersion: networking.k8s.io/v1 kind: Ingress metadata: name: milvus-ingress namespace: milvus-cluster annotations: nginx.ingress.kubernetes.io/backend-protocol: "GRPC" nginx.ingress.kubernetes.io/grpc-backend: "true" spec: ingressClassName: nginx rules: - host: milvus.example.com http: paths: - path: / pathType: Prefix backend: service: name: milvus-cluster port: number: 19530 高可用配置 多副本配置 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 # 连接到集群 connections.connect( alias="cluster", host='milvus.example.com', port='19530' ) # 创建集合时指定副本数 collection = Collection("ha_collection", schema=schema) collection.create_index(field_name="vector", index_params=index_params) # 加载时指定副本数 collection.load(replica_number=2) # 检查副本状态 from pymilvus import utility replica_info = utility.get_replicas(collection.name) for replica in replica_info: print(f"Replica {replica.id}: {replica.node_ids}") 故障转移测试 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 def test_failover(collection): """测试故障转移能力""" import time import random query_vector = [random.random() for _ in range(256)] # 持续查询测试 success_count = 0 total_count = 0 for i in range(100): try: results = collection.search( data=[query_vector], anns_field="vector", param={"metric_type": "L2", "params": {"nprobe": 16}}, limit=10 ) success_count += 1 print(f"Query {i}: Success") except Exception as e: print(f"Query {i}: Failed - {e}") total_count += 1 time.sleep(1) print(f"Success rate: {success_count/total_count*100:.2f}%") 数据分片策略 基于时间的分片 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 from datetime import datetime, timedelta def create_time_based_partitions(collection, start_date, end_date): """创建基于时间的分区""" current_date = start_date while current_date <= end_date: partition_name = f"partition_{current_date.strftime('%Y%m%d')}" try: collection.create_partition(partition_name) print(f"Created partition: {partition_name}") except Exception as e: print(f"Partition {partition_name} already exists or error: {e}") current_date += timedelta(days=1) def insert_with_time_partition(collection, entities, timestamp_field_idx=4): """根据时间戳插入到对应分区""" # 按时间戳分组数据 partition_data = {} for i, timestamp in enumerate(entities[timestamp_field_idx]): date_str = datetime.fromtimestamp(timestamp).strftime('%Y%m%d') partition_name = f"partition_{date_str}" if partition_name not in partition_data: partition_data[partition_name] = [[] for _ in entities] for j, field_data in enumerate(entities): partition_data[partition_name][j].append(field_data[i]) # 分别插入到各个分区 for partition_name, partition_entities in partition_data.items(): try: collection.insert(partition_entities, partition_name=partition_name) print(f"Inserted {len(partition_entities[0])} entities to {partition_name}") except Exception as e: print(f"Failed to insert to {partition_name}: {e}") 基于哈希的分片 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 import hashlib def create_hash_based_partitions(collection, num_partitions=8): """创建基于哈希的分区""" for i in range(num_partitions): partition_name = f"partition_hash_{i}" try: collection.create_partition(partition_name) print(f"Created partition: {partition_name}") except Exception as e: print(f"Partition {partition_name} already exists or error: {e}") def insert_with_hash_partition(collection, entities, key_field_idx=0, num_partitions=8): """根据键值哈希插入到对应分区""" partition_data = {f"partition_hash_{i}": [[] for _ in entities] for i in range(num_partitions)} for i, key in enumerate(entities[key_field_idx]): # 计算哈希值确定分区 hash_value = int(hashlib.md5(str(key).encode()).hexdigest(), 16) partition_idx = hash_value % num_partitions partition_name = f"partition_hash_{partition_idx}" for j, field_data in enumerate(entities): partition_data[partition_name][j].append(field_data[i]) # 插入到各个分区 for partition_name, partition_entities in partition_data.items(): if partition_entities[0]: # 如果分区有数据 collection.insert(partition_entities, partition_name=partition_name) print(f"Inserted {len(partition_entities[0])} entities to {partition_name}") 10. 监控运维 系统监控 Prometheus监控配置 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 # prometheus-config.yaml global: scrape_interval: 15s scrape_configs: - job_name: 'milvus' static_configs: - targets: ['milvus:9091'] metrics_path: /metrics scrape_interval: 15s - job_name: 'milvus-cluster' kubernetes_sd_configs: - role: pod namespaces: names: - milvus-cluster relabel_configs: - source_labels: [__meta_kubernetes_pod_label_app_kubernetes_io_name] action: keep regex: milvus - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_scrape] action: keep regex: true - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_port] action: replace target_label: __address__ regex: (.+) replacement: ${1}:9091 Grafana仪表板 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 { "dashboard": { "title": "Milvus Monitoring", "panels": [ { "title": "QPS (Queries Per Second)", "type": "graph", "targets": [ { "expr": "rate(milvus_proxy_search_vectors_count[5m])", "legendFormat": "Search QPS" }, { "expr": "rate(milvus_proxy_insert_vectors_count[5m])", "legendFormat": "Insert QPS" } ] }, { "title": "Response Time", "type": "graph", "targets": [ { "expr": "histogram_quantile(0.95, rate(milvus_proxy_search_latency_bucket[5m]))", "legendFormat": "Search P95 Latency" }, { "expr": "histogram_quantile(0.99, rate(milvus_proxy_search_latency_bucket[5m]))", "legendFormat": "Search P99 Latency" } ] }, { "title": "Memory Usage", "type": "graph", "targets": [ { "expr": "milvus_querynode_memory_usage_bytes", "legendFormat": "QueryNode Memory" }, { "expr": "milvus_indexnode_memory_usage_bytes", "legendFormat": "IndexNode Memory" } ] }, { "title": "Collection Statistics", "type": "table", "targets": [ { "expr": "milvus_collection_num_entities", "format": "table" } ] } ] } } 性能监控 自定义监控脚本 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 import time import psutil import threading from pymilvus import connections, Collection, utility from datetime import datetime class MilvusMonitor: def __init__(self, collection_name, interval=60): self.collection_name = collection_name self.interval = interval self.running = False self.metrics = [] def start_monitoring(self): self.running = True monitor_thread = threading.Thread(target=self._monitor_loop) monitor_thread.daemon = True monitor_thread.start() def stop_monitoring(self): self.running = False def _monitor_loop(self): while self.running: try: metrics = self._collect_metrics() self.metrics.append(metrics) print(f"[{metrics['timestamp']}] {metrics}") # 保留最近1000条记录 if len(self.metrics) > 1000: self.metrics = self.metrics[-1000:] except Exception as e: print(f"Monitoring error: {e}") time.sleep(self.interval) def _collect_metrics(self): collection = Collection(self.collection_name) # 集合统计信息 stats = collection.get_stats() num_entities = int(stats['row_count']) # 系统资源 cpu_percent = psutil.cpu_percent() memory = psutil.virtual_memory() disk = psutil.disk_usage('/') # 查询性能测试 start_time = time.time() try: test_vector = [0.1] * 256 collection.search( data=[test_vector], anns_field="vector", param={"metric_type": "L2", "params": {"nprobe": 16}}, limit=10 ) query_latency = (time.time() - start_time) * 1000 # ms except Exception as e: query_latency = -1 return { 'timestamp': datetime.now().isoformat(), 'collection_entities': num_entities, 'cpu_percent': cpu_percent, 'memory_percent': memory.percent, 'memory_used_gb': memory.used / (1024**3), 'disk_percent': disk.percent, 'disk_used_gb': disk.used / (1024**3), 'query_latency_ms': query_latency } def get_metrics_summary(self, last_n=100): """获取最近N条记录的统计摘要""" if not self.metrics: return None recent_metrics = self.metrics[-last_n:] latencies = [m['query_latency_ms'] for m in recent_metrics if m['query_latency_ms'] > 0] cpu_usage = [m['cpu_percent'] for m in recent_metrics] memory_usage = [m['memory_percent'] for m in recent_metrics] return { 'avg_query_latency_ms': sum(latencies) / len(latencies) if latencies else 0, 'max_query_latency_ms': max(latencies) if latencies else 0, 'avg_cpu_percent': sum(cpu_usage) / len(cpu_usage), 'max_cpu_percent': max(cpu_usage), 'avg_memory_percent': sum(memory_usage) / len(memory_usage), 'max_memory_percent': max(memory_usage), 'total_entities': recent_metrics[-1]['collection_entities'] if recent_metrics else 0 } # 使用监控器 monitor = MilvusMonitor("documents", interval=30) monitor.start_monitoring() # 运行一段时间后查看摘要 time.sleep(300) # 5分钟 summary = monitor.get_metrics_summary() print(f"Performance Summary: {summary}") monitor.stop_monitoring() 日志管理 日志配置 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 # milvus-log-config.yaml log: level: info file: rootPath: "/var/log/milvus" maxSize: 100 # MB maxAge: 7 # days maxBackups: 10 format: json # 在Kubernetes中配置日志收集 apiVersion: v1 kind: ConfigMap metadata: name: fluent-bit-config namespace: milvus-cluster data: fluent-bit.conf: | [SERVICE] Flush 1 Log_Level info Daemon off Parsers_File parsers.conf [INPUT] Name tail Path /var/log/milvus/*.log Parser json Tag milvus.* Refresh_Interval 5 [OUTPUT] Name es Match milvus.* Host elasticsearch.logging.svc.cluster.local Port 9200 Index milvus-logs Type _doc 日志分析脚本 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 import json import re from datetime import datetime, timedelta from collections import defaultdict def analyze_milvus_logs(log_file_path, hours=24): """分析Milvus日志文件""" cutoff_time = datetime.now() - timedelta(hours=hours) error_counts = defaultdict(int) warning_counts = defaultdict(int) performance_metrics = [] with open(log_file_path, 'r') as f: for line in f: try: log_entry = json.loads(line.strip()) log_time = datetime.fromisoformat(log_entry.get('time', '').replace('Z', '+00:00')) if log_time < cutoff_time: continue level = log_entry.get('level', '').upper() message = log_entry.get('msg', '') # 统计错误和警告 if level == 'ERROR': error_counts[message] += 1 elif level == 'WARN': warning_counts[message] += 1 # 提取性能指标 if 'latency' in message.lower(): latency_match = re.search(r'latency[:\s]+(\d+(?:\.\d+)?)\s*(ms|μs)', message) if latency_match: latency_value = float(latency_match.group(1)) latency_unit = latency_match.group(2) if latency_unit == 'μs': latency_value /= 1000 # 转换为ms performance_metrics.append({ 'timestamp': log_time, 'latency_ms': latency_value, 'operation': extract_operation(message) }) except (json.JSONDecodeError, ValueError) as e: continue return { 'error_summary': dict(error_counts), 'warning_summary': dict(warning_counts), 'performance_metrics': performance_metrics } def extract_operation(message): """从日志消息中提取操作类型""" if 'search' in message.lower(): return 'search' elif 'insert' in message.lower(): return 'insert' elif 'index' in message.lower(): return 'index' else: return 'unknown' def generate_log_report(analysis_result): """生成日志分析报告""" print("=== Milvus Log Analysis Report ===") print(f"Analysis time: {datetime.now()}") print() # 错误摘要 print("Top Errors:") sorted_errors = sorted(analysis_result['error_summary'].items(), key=lambda x: x[1], reverse=True) for error, count in sorted_errors[:10]: print(f" {count:4d} - {error[:100]}...") print() # 警告摘要 print("Top Warnings:") sorted_warnings = sorted(analysis_result['warning_summary'].items(), key=lambda x: x[1], reverse=True) for warning, count in sorted_warnings[:10]: print(f" {count:4d} - {warning[:100]}...") print() # 性能摘要 metrics = analysis_result['performance_metrics'] if metrics: latencies = [m['latency_ms'] for m in metrics] print("Performance Summary:") print(f" Total operations: {len(metrics)}") print(f" Average latency: {sum(latencies)/len(latencies):.2f} ms") print(f" Max latency: {max(latencies):.2f} ms") print(f" Min latency: {min(latencies):.2f} ms") # 按操作类型分组 by_operation = defaultdict(list) for metric in metrics: by_operation[metric['operation']].append(metric['latency_ms']) print("\nPerformance by Operation:") for operation, latencies in by_operation.items(): if latencies: avg_latency = sum(latencies) / len(latencies) print(f" {operation.capitalize()}:") print(f" Count: {len(latencies)}") print(f" Avg latency: {avg_latency:.2f} ms") print(f" Max latency: {max(latencies):.2f} ms") else: print("No performance metrics found.") # 使用示例 if __name__ == "__main__": log_file = "/path/to/milvus.log" result = analyze_milvus_logs(log_file, hours=24) generate_log_report(result) 11. 最佳实践 1. 数据建模最佳实践 Collection设计原则 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 # 良好的Collection设计示例 def create_production_collection(name, vector_dim, expected_size): """生产环境Collection设计""" # 根据数据规模选择分片数 shard_num = min(max(expected_size // 1000000, 2), 16) fields = [ # 主键字段 - 使用有意义的ID FieldSchema( name="id", dtype=DataType.INT64, is_primary=True, auto_id=False, description="Document unique identifier" ), # 时间戳字段 - 便于数据管理 FieldSchema( name="created_at", dtype=DataType.INT64, description="Creation timestamp" ), # 分类字段 - 用于过滤 FieldSchema( name="category", dtype=DataType.VARCHAR, max_length=50, description="Document category" ), # 向量字段 - 核心搜索字段 FieldSchema( name="embedding", dtype=DataType.FLOAT_VECTOR, dim=vector_dim, description="Document embedding vector" ), # 元数据字段 - 存储额外信息 FieldSchema( name="metadata", dtype=DataType.JSON, description="Additional metadata" ) ] schema = CollectionSchema( fields=fields, description=f"Production collection for {expected_size} documents", enable_dynamic_field=True # 允许动态字段 ) collection = Collection( name=name, schema=schema, shards_num=shard_num, consistency_level="Strong" # 生产环境建议强一致性 ) return collection 分区策略 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 # 基于时间的分区策略 def create_time_based_partitions(collection, start_date, end_date): """创建基于时间的分区""" from datetime import datetime, timedelta current_date = start_date while current_date <= end_date: partition_name = f"partition_{current_date.strftime('%Y%m%d')}" try: collection.create_partition(partition_name) print(f"Created partition: {partition_name}") except Exception as e: print(f"Partition {partition_name} already exists or error: {e}") current_date += timedelta(days=1) # 基于类别的分区策略 def create_category_partitions(collection, categories): """创建基于类别的分区""" for category in categories: partition_name = f"category_{category.lower()}" try: collection.create_partition(partition_name) print(f"Created partition: {partition_name}") except Exception as e: print(f"Partition {partition_name} already exists or error: {e}") 2. 性能优化最佳实践 索引选择策略 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 def choose_optimal_index(data_size, memory_budget, latency_requirement, accuracy_requirement): """根据需求选择最优索引""" if data_size < 100000: # 小数据集使用FLAT return { "index_type": "FLAT", "metric_type": "L2", "params": {} } elif latency_requirement == "ultra_low" and memory_budget == "high": # 超低延迟需求使用HNSW return { "index_type": "HNSW", "metric_type": "L2", "params": { "M": 32, "efConstruction": 400 } } elif memory_budget == "low": # 内存受限使用PQ压缩 return { "index_type": "IVF_PQ", "metric_type": "L2", "params": { "nlist": min(4 * int(np.sqrt(data_size)), 4096), "m": 16, "nbits": 8 } } else: # 平衡选择IVF_FLAT return { "index_type": "IVF_FLAT", "metric_type": "L2", "params": { "nlist": min(4 * int(np.sqrt(data_size)), 4096) } } # 动态调整搜索参数 def get_adaptive_search_params(index_type, accuracy_level="medium", data_size=None): """根据索引类型和精度要求动态调整搜索参数""" if index_type == "IVF_FLAT" or index_type == "IVF_PQ": nprobe_map = { "low": max(8, int(np.sqrt(data_size)) // 100) if data_size else 8, "medium": max(16, int(np.sqrt(data_size)) // 50) if data_size else 16, "high": max(32, int(np.sqrt(data_size)) // 25) if data_size else 32 } return {"nprobe": nprobe_map[accuracy_level]} elif index_type == "HNSW": ef_map = {"low": 64, "medium": 128, "high": 256} return {"ef": ef_map[accuracy_level]} elif index_type == "ANNOY": search_k_map = {"low": 100, "medium": 200, "high": 400} return {"search_k": search_k_map[accuracy_level]} return {} 批处理优化 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 class OptimizedBatchProcessor: """优化的批处理器""" def __init__(self, collection, batch_size=10000, flush_interval=50000): self.collection = collection self.batch_size = batch_size self.flush_interval = flush_interval self.total_inserted = 0 def insert_batch(self, data): """批量插入数据""" total_entities = len(data[0]) for i in range(0, total_entities, self.batch_size): end_idx = min(i + self.batch_size, total_entities) batch_data = [field_data[i:end_idx] for field_data in data] try: result = self.collection.insert(batch_data) self.total_inserted += len(result.primary_keys) # 定期刷新 if self.total_inserted % self.flush_interval == 0: self.collection.flush() print(f"Flushed after inserting {self.total_inserted} entities") except Exception as e: print(f"Error inserting batch {i//self.batch_size}: {e}") continue # 最终刷新 self.collection.flush() return self.total_inserted def parallel_search(self, query_vectors, search_params, max_workers=4): """并行搜索""" import concurrent.futures def search_single(query_vector): return self.collection.search( data=[query_vector], anns_field="embedding", param=search_params, limit=10 ) with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: futures = [executor.submit(search_single, qv) for qv in query_vectors] results = [future.result() for future in concurrent.futures.as_completed(futures)] return results 3. 数据质量最佳实践 向量预处理 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 import numpy as np from sklearn.preprocessing import normalize def preprocess_vectors(vectors, normalization="l2", dimension_check=True): """向量预处理管道""" vectors = np.array(vectors, dtype=np.float32) # 维度检查 if dimension_check and len(vectors.shape) != 2: raise ValueError(f"Expected 2D array, got {len(vectors.shape)}D") # 检查NaN和无穷值 if np.any(np.isnan(vectors)) or np.any(np.isinf(vectors)): print("Warning: Found NaN or infinite values, replacing with zeros") vectors = np.nan_to_num(vectors, nan=0.0, posinf=0.0, neginf=0.0) # 归一化 if normalization == "l2": vectors = normalize(vectors, norm='l2', axis=1) elif normalization == "minmax": from sklearn.preprocessing import MinMaxScaler scaler = MinMaxScaler() vectors = scaler.fit_transform(vectors) return vectors.tolist() # 数据验证 def validate_data_quality(data, schema): """验证数据质量""" issues = [] # 检查数据长度一致性 field_lengths = [len(field_data) for field_data in data] if len(set(field_lengths)) > 1: issues.append(f"Inconsistent field lengths: {field_lengths}") # 检查向量维度 for i, field in enumerate(schema.fields): if field.dtype == DataType.FLOAT_VECTOR: vectors = data[i] expected_dim = field.params.get('dim') for j, vector in enumerate(vectors[:100]): # 检查前100个 if len(vector) != expected_dim: issues.append(f"Vector {j} has dimension {len(vector)}, expected {expected_dim}") break return issues 4. 监控和运维最佳实践 健康检查系统 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 import time import logging from datetime import datetime class MilvusHealthMonitor: """Milvus健康监控系统""" def __init__(self, collection_names, alert_thresholds=None): self.collection_names = collection_names self.alert_thresholds = alert_thresholds or { 'query_latency': 1.0, # 秒 'memory_usage': 0.8, # 80% 'error_rate': 0.05 # 5% } self.logger = self._setup_logger() def _setup_logger(self): logger = logging.getLogger('milvus_monitor') logger.setLevel(logging.INFO) handler = logging.FileHandler('milvus_health.log') formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') handler.setFormatter(formatter) logger.addHandler(handler) return logger def check_connection_health(self): """检查连接健康状态""" try: from pymilvus import utility version = utility.get_server_version() self.logger.info(f"Milvus server version: {version}") return True except Exception as e: self.logger.error(f"Connection health check failed: {e}") return False def check_collection_health(self, collection_name): """检查集合健康状态""" try: collection = Collection(collection_name) # 检查集合状态 stats = collection.get_stats() num_entities = collection.num_entities # 执行测试查询 start_time = time.time() test_vector = [[0.1] * 128] # 假设128维向量 results = collection.search( data=test_vector, anns_field="embedding", param={"metric_type": "L2", "params": {"nprobe": 10}}, limit=1 ) query_latency = time.time() - start_time # 记录指标 metrics = { 'collection': collection_name, 'num_entities': num_entities, 'query_latency': query_latency, 'timestamp': datetime.now().isoformat() } self.logger.info(f"Collection health: {metrics}") # 检查告警阈值 if query_latency > self.alert_thresholds['query_latency']: self.logger.warning(f"High query latency: {query_latency:.3f}s") return metrics except Exception as e: self.logger.error(f"Collection health check failed for {collection_name}: {e}") return None def run_continuous_monitoring(self, interval=60): """持续监控""" while True: try: # 检查连接 if not self.check_connection_health(): self.logger.critical("Milvus connection lost!") # 检查所有集合 for collection_name in self.collection_names: self.check_collection_health(collection_name) time.sleep(interval) except KeyboardInterrupt: self.logger.info("Monitoring stopped by user") break except Exception as e: self.logger.error(f"Monitoring error: {e}") time.sleep(interval) 备份和恢复策略 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 import json import os from datetime import datetime class MilvusBackupManager: """Milvus备份管理器""" def __init__(self, backup_dir="./milvus_backups"): self.backup_dir = backup_dir os.makedirs(backup_dir, exist_ok=True) def backup_collection_metadata(self, collection_name): """备份集合元数据""" try: collection = Collection(collection_name) # 收集元数据 metadata = { 'name': collection.name, 'description': collection.description, 'schema': { 'fields': [ { 'name': field.name, 'dtype': str(field.dtype), 'params': field.params, 'is_primary': field.is_primary, 'auto_id': field.auto_id } for field in collection.schema.fields ], 'enable_dynamic_field': collection.schema.enable_dynamic_field }, 'num_entities': collection.num_entities, 'partitions': [p.name for p in collection.partitions], 'indexes': [] } # 获取索引信息 try: index_info = collection.index() if index_info: metadata['indexes'].append({ 'field_name': 'embedding', # 假设向量字段名 'index_params': index_info.params }) except: pass # 保存元数据 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_file = f"{self.backup_dir}/{collection_name}_metadata_{timestamp}.json" with open(backup_file, 'w') as f: json.dump(metadata, f, indent=2) print(f"Metadata backup saved: {backup_file}") return backup_file except Exception as e: print(f"Backup failed: {e}") return None def restore_collection_from_metadata(self, backup_file): """从元数据恢复集合结构""" try: with open(backup_file, 'r') as f: metadata = json.load(f) # 重建字段 fields = [] for field_info in metadata['schema']['fields']: field = FieldSchema( name=field_info['name'], dtype=getattr(DataType, field_info['dtype'].split('.')[-1]), is_primary=field_info.get('is_primary', False), auto_id=field_info.get('auto_id', False), **field_info.get('params', {}) ) fields.append(field) # 重建schema schema = CollectionSchema( fields=fields, description=metadata['description'], enable_dynamic_field=metadata['schema']['enable_dynamic_field'] ) # 创建集合 collection = Collection( name=metadata['name'], schema=schema ) # 重建分区 for partition_name in metadata['partitions']: if partition_name != '_default': collection.create_partition(partition_name) print(f"Collection {metadata['name']} restored from backup") return collection except Exception as e: print(f"Restore failed: {e}") return None 5. 安全最佳实践 访问控制 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 from pymilvus import connections # 安全连接配置 def secure_connect(host, port, username, password, secure=True): """安全连接到Milvus""" try: connections.connect( alias="secure_connection", host=host, port=port, user=username, password=password, secure=secure, server_pem_path="/path/to/server.pem", # TLS证书路径 server_name="milvus-server", timeout=30 ) print("Secure connection established") return True except Exception as e: print(f"Secure connection failed: {e}") return False # 输入验证 def validate_search_input(query_vectors, limit, expr=None): """验证搜索输入""" # 验证向量 if not isinstance(query_vectors, list) or not query_vectors: raise ValueError("Query vectors must be a non-empty list") # 验证limit if not isinstance(limit, int) or limit <= 0 or limit > 10000: raise ValueError("Limit must be a positive integer <= 10000") # 验证表达式（防止注入） if expr: dangerous_keywords = ['DROP', 'DELETE', 'UPDATE', 'INSERT', 'CREATE'] expr_upper = expr.upper() for keyword in dangerous_keywords: if keyword in expr_upper: raise ValueError(f"Dangerous keyword '{keyword}' found in expression") return True 12. 常见问题 Q1: 如何选择合适的索引类型？ A: 索引选择主要考虑以下因素： ...

1. 引言 随着大语言模型（LLM）技术的快速发展，如何有效地构建基于LLM的应用程序成为了开发者面临的重要挑战。LangChain作为一个专门为此设计的开源框架，为开发者提供了一套完整的工具链，使得构建复杂的AI应用变得更加简单和高效。 2. LangChain概述 2.1 什么是LangChain LangChain是一个开源框架，专门用于开发由大语言模型（LLM）驱动的应用程序。它提供了一套标准化的接口、工具和端到端的链式调用机制，使开发者能够轻松地将LLM集成到各种应用场景中。 1 2 3 4 5 6 7 8 graph TD A[开发者] --> B[LangChain框架] B --> C[大语言模型] B --> D[外部数据源] B --> E[工具集成] C --> F[智能应用] D --> F E --> F 2.2 核心设计理念 LangChain的设计基于以下几个核心理念： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 mindmap root((LangChain核心理念)) 模块化设计 独立使用 组合使用 松耦合架构 标准化接口 统一调用方式 多LLM支持 兼容性保证 链式组合 工作流构建 组件串联 复杂任务分解 数据感知 外部数据源 实时数据处理 多格式支持 代理能力 工具交互 环境感知 自主决策 2.3 主要特性 多模型支持：支持OpenAI、Anthropic、Hugging Face等多种LLM 丰富的组件库：提供提示模板、链、代理、记忆等核心组件 灵活的数据处理：支持多种文档格式和数据源 强大的检索能力：集成向量数据库和检索算法 可扩展性：易于添加自定义组件和集成 3. LangChain架构详解 3.1 整体架构 LangChain采用分层架构设计，从底层的模型接口到上层的应用逻辑，形成了一个完整的技术栈： ...

1. MySQL架构 1.1 体系结构 MySQL采用分层架构设计，从上到下分为四层： 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 graph TD %% 客户端层 Client1[MySQL命令行客户端] --> ConnLayer Client2[JDBC/ODBC] --> ConnLayer Client3[GUI工具<br/>Navicat/Workbench] --> ConnLayer Client4[应用程序] --> ConnLayer %% 连接层 - 最上层 subgraph ConnLayer["🔗 连接层 (Connection Layer)"] Conn1[连接管理<br/>TCP/IP, Socket] Conn2[线程池<br/>线程分配/复用] Conn3[权限认证<br/>用户名/密码验证] Conn4[SSL加密<br/>安全连接] end ConnLayer --> ServiceLayer %% 服务层 - SQL层 subgraph ServiceLayer["⚙️ 服务层 (Service Layer) - SQL处理核心"] direction TB Parser[📝 查询解析器<br/>Parser<br/>词法分析/语法分析] Cache[💾 查询缓存<br/>Query Cache<br/>MySQL 8.0已移除] Optimizer[🧠 查询优化器<br/>Optimizer<br/>执行计划/索引选择] Executor[⚡ 执行器<br/>Executor<br/>调用存储引擎API] Functions[🔧 内置函数<br/>日期/字符串/数学函数] Parser --> Cache Cache --> Optimizer Parser --> Optimizer Optimizer --> Executor Functions --> Executor end ServiceLayer --> EngineLayer %% 存储引擎层 subgraph EngineLayer["🗄️ 存储引擎层 (Storage Engine Layer) - 插件式架构"] direction LR subgraph InnoDB["InnoDB (默认)"] InnoDB1[✅ 事务支持 ACID] InnoDB2[🔒 行级锁] InnoDB3[🔗 外键约束] InnoDB4[📊 MVCC多版本] InnoDB5[💥 崩溃恢复] end subgraph MyISAM["MyISAM (传统)"] MyISAM1[❌ 无事务] MyISAM2[🔒 表级锁] MyISAM3[⚡ 查询速度快] MyISAM4[💾 占用空间小] end subgraph Memory["Memory (内存)"] Memory1[⚡ 数据存内存] Memory2[🚀 极快访问速度] Memory3[💥 重启数据丢失] end subgraph Others["其他引擎"] Archive[Archive<br/>压缩存储] CSV[CSV<br/>CSV格式] Federated[Federated<br/>远程表] end end EngineLayer --> StorageLayer %% 存储层 - 文件系统层 subgraph StorageLayer["💾 存储层 (File System Layer) - 持久化"] direction TB subgraph DataFiles["数据文件"] IBD[.ibd文件<br/>InnoDB表空间] FRM[.frm文件<br/>表结构定义] MYD[.MYD文件<br/>MyISAM数据] MYI[.MYI文件<br/>MyISAM索引] end subgraph LogFiles["日志文件"] RedoLog[Redo Log<br/>重做日志<br/>崩溃恢复] UndoLog[Undo Log<br/>回滚日志<br/>事务回滚] BinLog[Binary Log<br/>二进制日志<br/>主从复制] ErrorLog[Error Log<br/>错误日志] SlowLog[Slow Query Log<br/>慢查询日志] end subgraph TableSpace["表空间管理"] SystemTS[系统表空间<br/>ibdata1] FilePerTable[独立表空间<br/>每表一文件] TempTS[临时表空间<br/>ibtmp1] UndoTS[Undo表空间<br/>undo_001] end end StorageLayer --> FileSystem[🗂️ 操作系统文件系统<br/>Linux/Windows/macOS] %% 样式定义 classDef clientStyle fill:#e1f5fe,stroke:#01579b,stroke-width:2px classDef connStyle fill:#f3e5f5,stroke:#4a148c,stroke-width:2px classDef serviceStyle fill:#e8f5e8,stroke:#1b5e20,stroke-width:2px classDef engineStyle fill:#fff3e0,stroke:#e65100,stroke-width:2px classDef storageStyle fill:#fce4ec,stroke:#880e4f,stroke-width:2px class Client1,Client2,Client3,Client4 clientStyle class ConnLayer connStyle class ServiceLayer serviceStyle class EngineLayer engineStyle class StorageLayer storageStyle 各层详细说明： ...

Nginx是一个高性能的HTTP和反向代理服务器，也是一个IMAP/POP3/SMTP服务器。以其高性能、稳定性、丰富的功能集、简单的配置文件和低系统资源消耗而闻名。

⭐️JVM vs JDK vs JRE ⭐️什么是字节码?采用字节码的好处是什么? 在 Java 中，JVM 可以理解的代码就叫做字节码（即扩展名为 .class 的文件），它不面向任何特定的处理器，只面向虚拟机。Java 语言通过字节码的方式，在一定程度上解决了传统解释型语言执行效率低的问题，同时又保留了解释型语言可移植的特点。所以， Java 程序运行时相对来说还是高效的（不过，和 C、 C++，Rust，Go 等语言还是有一定差距的），而且，由于字节码并不针对一种特定的机器，因此，Java 程序无须重新编译便可在多种不同操作系统的计算机上运行。 ⭐️为什么说 Java 语言“编译与解释并存”？ 这是因为 Java 语言既具有编译型语言的特征，也具有解释型语言的特征。因为 Java 程序要经过先编译，后解释两个步骤，由 Java 编写的程序需要先经过编译步骤，生成字节码（.class 文件），这种字节码必须由 Java 解释器来解释执行。 ⭐️自增自减运算符 1 2 3 4 5 int a = 9; int b = a++; int c = ++a; int d = c--; int e = --d; 答案：a = 11 、b = 9 、 c = 10 、 d = 10 、 e = 10。 ...

1. 基础语法 功能 语法 效果 行内公式 $a^2+b^2=c^2$ $a^2+b^2=c^2$ 公式块 $$ a^2+b^2=c^2 $$ $a^2+b^2=c^2$ 强制大号公式 \dfrac{a}{b} $\dfrac{a}{b}$ 普通分数（小号） \tfrac{a}{b} $\tfrac{a}{b}$ 2. 上下标 语法 效果 x^2 $x^2$ x_1 $x_1$ x_{ij} $x_{ij}$ e^{i\pi}+1=0 $e^{i\pi}+1=0$ 3. 分数与根号 语法 效果 \frac{a}{b} $\frac{a}{b}$ \dfrac{1}{2} $\dfrac{1}{2}$ （大号，推荐公式块） \tfrac{1}{2} $\tfrac{1}{2}$ （小号，推荐行内） \sqrt{x} $\sqrt{x}$ \sqrt[3]{x} $\sqrt[3]{x}$ 4. 求和/积分/极限 语法 效果 \sum_{i=1}^n i $\sum_{i=1}^n i$ \prod_{i=1}^n i $\prod_{i=1}^n i$ \int_0^1 x^2 dx $\int_0^1 x^2 dx$ \iint_D f(x,y)dA $\iint_D f(x,y)\,dA$ \lim_{x \to 0} \frac{\sin x}{x} $\lim_{x \to 0} \frac{\sin x}{x}$ 5. 矩阵 1 2 3 4 \begin{bmatrix} a & b \\ c & d \end{bmatrix} $$ \begin{bmatrix} a & b \\ c & d \end{bmatrix} $$常见： ...

文章来源 Strengthening AI Agent Hijacking Evaluations Large AI models are increasingly used to power agentic systems, or “agents,” which can automate complex tasks on behalf of users. AI agents could have a wide range of potential benefits, such as automating scientific research or serving as personal assistants. However, to fully realize the potential of AI agents, it is essential to identify and measure — in order to ultimately mitigate — the security risks these systems could introduce. ...

一、安装与配置 一键安装 1 2 3 4 5 # macOS/Linux curl -LsSf https://astral.sh/uv/install.sh | sh # Windows irm https://astral.sh/uv/install.ps1 | iex 支持通过 pip 安装：pip install uv 版本验证与更新 1 2 uv --version # 检查版本 uv self update # 自动更新到最新版本 二、项目管理 初始化项目 1 2 uv init my_project # 创建项目目录并生成配置文件 uv init . --python 3.12 # 在当前目录初始化并指定 Python 版本 生成文件：pyproject.toml（主依赖）、uv.lock（精确锁文件） 依赖管理 1 2 3 4 5 uv add requests # 添加主依赖 uv add --dev pytest # 添加开发依赖 uv remove flask # 移除依赖 uv lock # 生成/更新锁文件 uv sync # 同步依赖到虚拟环境 运行脚本 ...