Agent中的向量数据库选型：Milvus、Qdrant与Weaviate的创业场景对比

Agent中的向量数据库选型Milvus、Qdrant与Weaviate的创业场景对比一、RAG架构下的向量检索瓶颈Agent系统的核心能力之一是通过RAG检索增强生成获取外部知识。向量数据库作为RAG的检索层其性能直接决定了Agent的响应质量与延迟表现。在创业场景中选择合适的向量数据库不仅是技术问题更是资源分配与长期演进的战略决策。当前开源向量数据库市场已形成三足鼎立格局Milvus以高性能和分布式架构见长Qdrant以易用性和Rust实现的高性能著称Weaviate以原生多模态支持和GraphRAG集成取胜。三者各有侧重没有绝对的优劣只有适配度的差异。创业团队在选型时面临的核心矛盾是早期需要快速验证倾向于选择上手简单的方案但随着业务增长查询QPS上升、数据量突破千万级早期的简单方案可能成为性能瓶颈。理解三款数据库的设计哲学与适用边界是做出正确选型的前提。二、三大向量数据库的架构差异与能力矩阵三款向量数据库在存储架构、检索算法、分布式能力、生态集成四个维度上存在系统性差异。理解这些差异才能匹配到最合适的业务场景。Milvus采用计算存储分离架构Proxy层负责请求路由QueryNode和DataNode分别处理查询和写入底层依赖消息队列和对象存储。这种设计使其天然支持水平扩展适合数据量在亿级以上的场景。代价是部署复杂度高完整集群至少需要6个以上容器。Qdrant采用单机优化的设计哲学整个引擎用Rust实现依赖HNSW索引和WAL持久化无需外部依赖即可运行。分布式模式通过Raft共识协议实现但功能相对简化。适合数据量在千万级以下、追求低延迟和高开发效率的场景。Weaviate的差异化在于原生集成了多模态模型CLIP等和知识图谱能力。其模块系统允许热插拔不同类型的向量化模型GraphRAG能力使其在多跳推理场景中表现突出。缺点是默认配置下内存消耗较大需要仔细调优。三、生产级向量数据库接口封装与性能测试以下是统一的向量数据库操作接口封装屏蔽三款数据库的差异并附带性能基准测试套件。 向量数据库统一操作接口与性能测试 支持Milvus、Qdrant、Weaviate三款数据库 import time import numpy as np from abc import ABC, abstractmethod from typing import List, Dict, Optional, Tuple, Any from dataclasses import dataclass, field import logging from datetime import datetime import json import threading logging.basicConfig(levellogging.WARNING) logger logging.getLogger(__name__) dataclass class VectorRecord: 向量记录 id: str vector: List[float] metadata: Dict[str, Any] field(default_factorydict) dataclass class SearchResult: 检索结果 id: str score: float metadata: Dict[str, Any] vector: Optional[List[float]] None dataclass class IndexStats: 索引统计信息 total_vectors: int dim: int index_type: str memory_usage_mb: float avg_query_latency_ms: float class VectorDatabase(ABC): 向量数据库抽象接口 abstractmethod def connect(self) - None: 建立连接 pass abstractmethod def create_collection(self, name: str, dim: int, index_type: str hnsw) - None: 创建集合/索引 pass abstractmethod def insert(self, collection: str, records: List[VectorRecord]) - List[str]: 批量插入向量返回写入ID列表 pass abstractmethod def search(self, collection: str, query_vector: List[float], top_k: int 10, filters: Optional[Dict] None) - List[SearchResult]: 向量检索 pass abstractmethod def delete(self, collection: str, ids: List[str]) - None: 删除向量 pass abstractmethod def get_stats(self, collection: str) - IndexStats: 获取索引统计信息 pass abstractmethod def close(self) - None: 关闭连接 pass # Milvus实现 class MilvusAdapter(VectorDatabase): Milvus适配器 def __init__(self, host: str localhost, port: int 19530): self._host host self._port port self._client None self._collection_cache: Dict[str, Any] {} def connect(self) - None: try: from pymilvus import connections, Collection connections.connect(hostself._host, portself._port) self._client connections logger.info(fMilvus连接成功: {self._host}:{self._port}) except ImportError: raise RuntimeError(请先安装pymilvus: pip install pymilvus) def create_collection(self, name: str, dim: int, index_type: str hnsw) - None: from pymilvus import Collection, FieldSchema, CollectionSchema, DataType fields [ FieldSchema(nameid, dtypeDataType.VARCHAR, is_primaryTrue, max_length64), FieldSchema(namevector, dtypeDataType.FLOAT_VECTOR, dimdim), FieldSchema(namemetadata, dtypeDataType.JSON), ] schema CollectionSchema(fieldsfields, descriptionfCollection {name}) collection Collection(namename, schemaschema) # 创建索引 index_params { hnsw: {metric_type: L2, index_type: HNSW, params: {M: 16, efConstruction: 200}}, ivf: {metric_type: L2, index_type: IVF_FLAT, params: {nlist: 1024}}, } params index_params.get(index_type, index_params[hnsw]) collection.create_index(field_namevector, index_paramsparams) collection.load() self._collection_cache[name] collection def insert(self, collection: str, records: List[VectorRecord]) - List[str]: from pymilvus import Collection coll self._get_collection(collection) ids [r.id for r in records] vectors [r.vector for r in records] metadata [r.metadata for r in records] coll.insert([ids, vectors, metadata]) coll.flush() return ids def search(self, collection: str, query_vector: List[float], top_k: int 10, filters: Optional[Dict] None) - List[SearchResult]: coll self._get_collection(collection) results coll.search( data[query_vector], anns_fieldvector, param{metric_type: L2, params: {ef: 50}}, limittop_k, exprself._build_filter_expr(filters) if filters else None ) return [ SearchResult( idhit.id, scorehit.distance, metadatahit.entity.get(metadata, {}) ) for hits in results for hit in hits ] def _get_collection(self, name: str): from pymilvus import Collection if name not in self._collection_cache: coll Collection(name) coll.load() self._collection_cache[name] coll return self._collection_cache[name] def _build_filter_expr(self, filters: Dict) - str: 构建Milvus过滤表达式 expressions [] for key, value in filters.items(): if isinstance(value, (int, float)): expressions.append(fmetadata[{key}] {value}) else: expressions.append(fmetadata[{key}] {value}) return and .join(expressions) def delete(self, collection: str, ids: List[str]) - None: coll self._get_collection(collection) expr fid in {ids} coll.delete(expr) def get_stats(self, collection: str) - IndexStats: coll self._get_collection(collection) return IndexStats( total_vectorscoll.num_entities, dim0, # Milvus需要额外查询 index_typeunknown, memory_usage_mb0.0, avg_query_latency_ms0.0 ) def close(self) - None: from pymilvus import connections connections.disconnect(default) # Qdrant实现 class QdrantAdapter(VectorDatabase): Qdrant适配器 def __init__(self, host: str localhost, port: int 6333): self._host host self._port port self._client None def connect(self) - None: try: from qdrant_client import QdrantClient self._client QdrantClient(hostself._host, portself._port) logger.info(fQdrant连接成功: {self._host}:{self._port}) except ImportError: raise RuntimeError(请先安装qdrant-client: pip install qdrant-client) def create_collection(self, name: str, dim: int, index_type: str hnsw) - None: from qdrant_client.models import VectorParams, Distance self._client.create_collection( collection_namename, vectors_configVectorParams( sizedim, distanceDistance.COSINE ) ) def insert(self, collection: str, records: List[VectorRecord]) - List[str]: from qdrant_client.models import PointStruct points [ PointStruct( idr.id if isinstance(r.id, int) else abs(hash(r.id)) % (2**63), vectorr.vector, payloadr.metadata ) for r in records ] self._client.upsert(collection_namecollection, pointspoints) return [str(p.id) for p in points] def search(self, collection: str, query_vector: List[float], top_k: int 10, filters: Optional[Dict] None) - List[SearchResult]: from qdrant_client.models import Filter, FieldCondition, MatchValue qdrant_filter None if filters: conditions [ FieldCondition(keyk, matchMatchValue(valuev)) for k, v in filters.items() ] qdrant_filter Filter(mustconditions) results self._client.search( collection_namecollection, query_vectorquery_vector, limittop_k, query_filterqdrant_filter ) return [ SearchResult( idstr(r.id), scorer.score, metadatar.payload ) for r in results ] def delete(self, collection: str, ids: List[str]) - None: int_ids [int(i) if i.isdigit() else abs(hash(i)) % (2**63) for i in ids] self._client.delete(collection_namecollection, points_selectorint_ids) def get_stats(self, collection: str) - IndexStats: info self._client.get_collection(collection) return IndexStats( total_vectorsinfo.vectors_count, diminfo.config.params.size, index_typehnsw, memory_usage_mb0.0, avg_query_latency_ms0.0 ) def close(self) - None: pass # QdrantClient不需要显式关闭 # Weaviate实现 class WeaviateAdapter(VectorDatabase): Weaviate适配器 def __init__(self, url: str http://localhost:8080): self._url url self._client None def connect(self) - None: try: import weaviate self._client weaviate.connect_to_local() logger.info(fWeaviate连接成功: {self._url}) except ImportError: raise RuntimeError(请先安装weaviate-client: pip install weaviate-client) def create_collection(self, name: str, dim: int, index_type: str hnsw) - None: import weaviate.classes as wvc self._client.collections.create( namename, vector_index_configwvc.Configure.VectorIndex.hnsw(), properties[ wvc.Property(namemetadata, data_typewvc.DataType.OBJECT) ] ) def insert(self, collection: str, records: List[VectorRecord]) - List[str]: coll self._client.collections.get(collection) # Weaviate的insert接口 ids [] for r in records: result coll.data.insert( properties{metadata: r.metadata}, vectorr.vector ) ids.append(result) return ids def search(self, collection: str, query_vector: List[float], top_k: int 10, filters: Optional[Dict] None) - List[SearchResult]: coll self._client.collections.get(collection) results coll.query.near_vector( near_vectorquery_vector, limittop_k ) return [ SearchResult( idr.uuid, score1.0 - r.metadata.distance, # 转换为相似度 metadatar.properties.get(metadata, {}) ) for r in results.objects ] def delete(self, collection: str, ids: List[str]) - None: coll self._client.collections.get(collection) for id in ids: coll.data.delete_by_id(id) def get_stats(self, collection: str) - IndexStats: coll self._client.collections.get(collection) agg coll.aggregate.over_all() return IndexStats( total_vectorsagg.total_count, dim0, index_typehnsw, memory_usage_mb0.0, avg_query_latency_ms0.0 ) def close(self) - None: self._client.close() # 性能基准测试 class VectorDBBenchmark: 向量数据库性能基准测试 def __init__(self, dimensions: int 768, num_vectors: int 100000): self.dimensions dimensions self.num_vectors num_vectors self._results: Dict[str, Dict] {} def generate_random_vectors(self, n: int) - List[VectorRecord]: 生成随机向量数据 np.random.seed(42) records [] for i in range(n): vector np.random.randn(self.dimensions).tolist() records.append(VectorRecord( idfvec_{i}, vectorvector, metadata{index: i, category: fcat_{i % 10}} )) return records def run_insert_benchmark(self, db: VectorDatabase, batch_size: int 1000) - Dict: 插入性能测试 records self.generate_random_vectors(self.num_vectors) latencies [] for i in range(0, len(records), batch_size): batch records[i:i batch_size] start time.monotonic() db.insert(benchmark, batch) latency (time.monotonic() - start) * 1000 latencies.append(latency) return { total_vectors: self.num_vectors, batch_size: batch_size, avg_latency_ms: sum(latencies) / len(latencies), p99_latency_ms: sorted(latencies)[int(len(latencies) * 0.99)], throughput_qps: (self.num_vectors / batch_size) / (sum(latencies) / 1000 / len(latencies)), } def run_search_benchmark(self, db: VectorDatabase, num_queries: int 100) - Dict: 检索性能测试 np.random.seed(123) queries np.random.randn(num_queries, self.dimensions).tolist() latencies [] for query in queries: start time.monotonic() results db.search(benchmark, query, top_k10) latency (time.monotonic() - start) * 1000 latencies.append(latency) return { num_queries: num_queries, avg_latency_ms: sum(latencies) / len(latencies), p50_latency_ms: sorted(latencies)[len(latencies) // 2], p99_latency_ms: sorted(latencies)[int(len(latencies) * 0.99)], qps: num_queries / (sum(latencies) / 1000), } def compare_all(self, db_configs: List[Tuple[str, VectorDatabase]]) - Dict: 对比三款数据库的性能 results {} for name, db in db_configs: print(f\n测试 {name}...) db.connect() db.create_collection(benchmark, self.dimensions) insert_result self.run_insert_benchmark(db) print(f 插入性能: {insert_result[throughput_qps]:.1f} QPS) search_result self.run_search_benchmark(db) print(f 检索性能: {search_result[qps]:.1f} QPS, fP99延迟: {search_result[p99_latency_ms]:.1f}ms) db.close() results[name] { insert: insert_result, search: search_result } return results四、选型决策的边界条件与场景匹配三款向量数据库的选型本质上是在功能完整性运维复杂度查询性能三者之间做权衡。Milvus适合数据规模大、需要完整分布式能力的场景。当向量数据量超过5000万、或需要跨IDC部署时Milvus是唯一成熟的选择。但代价是运维成本高需要专门的平台工程师维护。对于10人以下的创业团队不建议将Milvus作为第一选择除非已有专职运维资源。Qdrant是目前创业团队最平衡的选择。单机性能优异Rust实现API设计简洁Docker一键启动文档质量高。在千万级向量、100 QPS以内的场景下Qdrant的表现通常优于Milvus。其分布式模式虽然功能不如Milvus完整但对于大多数创业公司的需求已经足够。Weaviate的差异化价值在于多模态和GraphRAG。如果Agent需要处理图片、音频等非结构化数据或者需要构建知识图谱增强的检索能力Weaviate是唯一能提供原生支持的开源方案。对于纯文本的RAG场景Weaviate的优势不明显且内存消耗偏高。混合检索能力是生产环境的关键需求。纯向量检索的召回率通常在70%至80%之间引入标量过滤metadata过滤和全文检索的混合检索可以将召回率提升至90%以上。三款数据库都支持混合检索但实现方式不同Milvus通过标量字段索引实现Qdrant通过丰富的过滤条件实现Weaviate通过倒排索引实现。选型时还有一个容易被忽视的因素客户端SDK的成熟度。Milvus的Python SDK功能最完整但API变动相对频繁Qdrant的SDK设计最简洁上手成本最低Weaviate的SDK支持多语言最全面Python/TypeScript/Go/Java。五、总结向量数据库选型是Agent系统基础设施的核心决策之一。核心要点归纳如下创业早期千万级向量以下首选Qdrant部署简单、性能优异、SDK友好。数据规模超过5000万或需要跨IDC部署时Milvus的分布式架构是必然选择。多模态检索或GraphRAG场景Weaviate的原生支持使其具有不可替代性。混合检索能力向量标量过滤全文检索是生产环境的必备特性选型时必须验证。性能基准测试应在自己的数据集上重新运行公开benchmark的结果与实际业务场景可能存在显著差异。落地建议在架构设计的早期引入向量数据库的抽象接口层如本文的VectorDatabase抽象类使底层数据库可以在不影响业务逻辑的情况下替换。这为未来的技术演进预留了空间是创业团队在技术选型上的最佳保险策略。