2026年9月,Qdrant发布了Qdrant-FineWeb-10B基准测试,在100.7亿条文档上执行了超过1千万亿次距离计算。这不是一个跑在笔记本上的Demo数字,而是用24.47TB向量数据集硬刚出来的生产级结果。与此同时,向量搜索创业公司Turbopuffer在10月初突然宣布关停其旗舰数据库Vectra,60多家客户被迫在30天内完成数据迁移。这两件事放在一起看,暴露出向量数据库在大规模落地时一个被严重低估的问题:索引和查询只是冰山一角,水面下是内存管理、更新策略、压缩方案和容灾架构的系统性工程挑战。
我在黑箭科技做舆情监控系统的时候也踩过类似的坑——知识库从100万向量扩到3000万的时候,检索延迟直接从5ms飙到50ms,才意识到Demo和生产完全是两码事。这篇文章把这些坑系统地拆一遍。
你的向量数据库为什么扛不住——从千万级到亿级的性能断崖
我做知识库项目的朋友老李,上个月跟我吐槽了一个事。他的系统用了某开源向量数据库,最初100万条数据的时候跑得飞起,P99延迟5ms,他很满意。后来业务增长到5000万条,P99延迟飙到40ms。再后来上到2亿条,直接OOM了。
他问我:"数据量翻了200倍,延迟不应该只涨8倍吧?"
这个问题问得好。因为向量数据库的性能衰减,从来不是线性的。
核心矛盾在内存。我们算一笔账:2亿条向量,每条768维float32,光原始向量数据就是:
2亿 × 768 × 4字节 = 588.8GB这还没算HNSW图索引的边连接数据、元数据、副本。按经验值,HNSW索引的额外开销大约是原始向量的1.2-1.5倍。也就是说,你需要将近1.5TB的内存来存储这些数据。
一台128GB内存的服务器?不够。两台?勉强。但还得留空间给操作系统、业务进程和查询缓冲。
据Qdrant官方博客(2026年9月)的数据,在1000万向量(768维)的基准测试中,Qdrant的P99延迟约12ms,内存占用约2.5GB。当数据量放大到亿级,内存不是线性增长——HNSW图的多层结构会导致高层节点的连接数指数级增加,内存增长曲线比数据量曲线更陡。
内存问题的三层解法
第一层,向量量化。把float32(32位浮点)压缩到更小的表示:
import numpy as np from dataclasses import dataclass@dataclass
class QuantizationConfig:
"""量化配置"""
method: str # scalar | binary | product
target_bits: int # 8 | 4 | 1
class ScalarQuantizer:
"""标量量化:float32 -> uint8,内存缩减到1/4"""
def __init__(self, dim: int): self.dim = dim self.min_val = None self.max_val = None def fit(self, vectors: np.ndarray): """基于校准数据计算量化范围""" self.min_val = vectors.min(axis=0) self.max_val = vectors.max(axis=0) def encode(self, vectors: np.ndarray) -> np.ndarray: """编码:float32 -> uint8""" range_val = self.max_val - self.min_val range_val = np.where(range_val == 0, 1, range_val) # 防除零 normalized = (vectors - self.min_val) / range_val return (normalized * 255).astype(np.uint8) def decode(self, quantized: np.ndarray) -> np.ndarray: """解码:uint8 -> float32(有损)""" range_val = self.max_val - self.min_val return (quantized.astype(np.float32) / 255) * range_val + self.min_val使用示例
original = np.random.randn(10000, 768).astype(np.float32)
sq = ScalarQuantizer(dim=768)
sq.fit(original)
encoded = sq.encode(original)
decoded = sq.decode(encoded)
内存对比
print(f"原始: {original.nbytes / 1024 / 1024:.1f}MB") # 29.3MB
print(f"量化后: {encoded.nbytes / 1024 / 1024:.1f}MB") # 7.3MB
print(f"压缩比: {original.nbytes / encoded.nbytes:.0f}x") # 4x
精度评估
cos_sim = np.sum(original * decoded, axis=1) / (
np.linalg.norm(original, axis=1) * np.linalg.norm(decoded, axis=1)
)
print(f"平均余弦相似度: {cos_sim.mean():.4f}") # 通常 > 0.99
标量量化把每条向量从3072字节(768维 × 4字节)压缩到768字节,内存直接缩减到1/4。代价是精度轻微损失,但实测余弦相似度通常能保持99%以上。
第二层,Product Quantization(乘积量化)。更激进的压缩方案:
class ProductQuantizer: """乘积量化:将向量切分为子空间,每个子空间独立量化"""def __init__(self, dim: int, n_subquantizers: int = 96, n_bits: int = 8): self.dim = dim self.M = n_subquantizers # 子空间数量 self.n_bits = n_bits self.K = 2 ** n_bits # 每个子空间的聚类数 self.codebooks = None # 聚类中心 assert dim % n_subquantizers == 0 self.sub_dim = dim // n_subquantizers def fit(self, vectors: np.ndarray, n_iter: int = 20): """训练每个子空间的聚类中心""" n_samples = len(vectors) # 切分子空间 sub_vectors = vectors.reshape(n_samples, self.M, self.sub_dim) self.codebooks = np.zeros((self.M, self.K, self.sub_dim)) for m in range(self.M): # 对每个子空间做KMeans data = sub_vectors[:, m, :] # 简化版KMeans centers = data[np.random.choice(n_samples, self.K, replace=False)] for _ in range(n_iter): # 分配 dists = np.linalg.norm( data[:, np.newaxis, :] - centers[np.newaxis, :, :], axis=2 ) labels = dists.argmin(axis=1) # 更新 for k in range(self.K): mask = labels == k if mask.sum() > 0: centers[k] = data[mask].mean(axis=0) self.codebooks[m] = centers def encode(self, vectors: np.ndarray) -> np.ndarray: """编码为量化码""" n_samples = len(vectors) sub_vectors = vectors.reshape(n_samples, self.M, self.sub_dim) codes = np.zeros((n_samples, self.M), dtype=np.uint8) for m in range(self.M): dists = np.linalg.norm( sub_vectors[:, m:m+1, :] - self.codebooks[m][np.newaxis, :, :], axis=2 ) codes[:, m] = dists.argmin(axis=1) return codes def compute_distance_table(self, query: np.ndarray) -> np.ndarray: """预计算查询向量到每个聚类中心的距离表""" # 这是PQ快速检索的核心:提前算好距离表,查询时只需查表求和 sub_query = query.reshape(self.M, self.sub_dim) table = np.zeros((self.M, self.K)) for m in range(self.M): table[m] = np.sum( (self.codebooks[m] - sub_query[m]) ** 2, axis=1 ) return table
压缩效果
pq = ProductQuantizer(dim=768, n_subquantizers=96, n_bits=8)
pq.fit(original[:1000]) # 用1000条做训练
codes = pq.encode(original)
print(f"PQ编码后: {codes.nbytes / 1024:.1f}KB") # 960KB
print(f"压缩比: {original.nbytes / codes.nbytes:.0f}x") # ~32x
Product Quantization能做到32倍压缩。768维向量从3072字节压缩到96字节。亿级数据的内存需求从1.5TB降到不到50GB。
但PQ不是万能的——它的距离计算是近似的,recall会下降。Qdrant的做法是两阶段检索:用PQ做粗召回,再用原始向量做精排。
class TwoStageRetriever: """两阶段检索:PQ粗召回 + 原始向量精排"""def __init__(self, pq_quantizer, raw_vectors): self.pq = pq_quantizer self.raw = raw_vectors def search(self, query: np.ndarray, coarse_k: int = 100, final_k: int = 10): """ 阶段1:用PQ距离表快速召回coarse_k个候选 阶段2:用原始向量精确计算距离,取top final_k """ # 阶段1:PQ粗召回 distance_table = self.pq.compute_distance_table(query) codes = self.pq.encode(self.raw) # 实际生产中codes是预计算好的 # 对每个向量,用查表法快速算距离 n = len(codes) approx_dists = np.zeros(n) for m in range(self.pq.M): approx_dists += distance_table[m][codes[:, m]] # 取最近的coarse_k个 coarse_indices = np.argpartition(approx_dists, coarse_k)[:coarse_k] # 阶段2:原始向量精排 candidates = self.raw[coarse_indices] exact_dists = np.sum((candidates - query) ** 2, axis=1) top_local = np.argsort(exact_dists)[:final_k] return coarse_indices[top_local], exact_dists[top_local]
这套方案,在Qdrant 1.19版本中已经内置为TurboQuant数据类型。据Qdrant官方博客(2026年8月)的测试数据,TurboQuant能在内存减少8倍的情况下保持99%以上的召回率。
另外值得一提的是,Qdrant在2026年10月6日发布的EmbeddingGemma 2集成案例显示,Google新推出的小模型EmbeddingGemma 2能在向量RAM减少30倍的情况下保持99%的检索质量,或者77倍减少时保持94.5%。从Embedding模型侧直接降低存储需求,是另一个值得关注的路径。
亿级数据的索引策略:HNSW不是万能药
HNSW(Hierarchical Navigable Small World)是目前最主流的向量索引算法,几乎所有向量数据库都把它当默认选项。但到了亿级数据量,HNSW的问题开始暴露。
问题一:构建时间。HNSW建索引的时间复杂度是O(N×log N),每插入一个向量,都要在图中找到合适的邻居位置。2亿条数据,建一次索引可能要几个小时甚至几天。
问题二:内存开销。HNSW需要在内存中维护多层图结构。根据Qdrant的基准测试,HNSW索引的额外内存开销大约是原始向量的1.2倍。亿级数据下,这个开销是百GB级的。
问题三:更新困难。HNSW是一个相对静态的索引结构。频繁插入新向量会导致图的质量下降,但重建整个索引的成本太高。
来看一个索引构建的模拟实验:
import time import numpy as npclass SimpleHNSW:
"""简化版HNSW实现,用于演示索引构建过程"""
def __init__(self, dim: int, M: int = 16, ef_construction: int = 200, max_level: int = 4): self.dim = dim self.M = M # 每个节点的最大连接数 self.ef_construction = ef_construction self.max_level = max_level self.graphs = [[] for _ in range(max_level)] self.entry_point = None self._vector_store = {} # id -> vector 映射 def _random_level(self) -> int: """随机分配层级,指数衰减""" level = 0 while level < self.max_level - 1 and np.random.random() < 0.5: level += 1 return level def _distance(self, a: np.ndarray, b: np.ndarray) -> float: return float(np.sum((a - b) ** 2)) def insert(self, vector: np.ndarray, vector_id: int): """插入一个向量到索引中""" self._vector_store[vector_id] = vector level = self._random_level() if self.entry_point is None: self.entry_point = vector_id for l in range(level + 1): self.graphs[l].append({ 'id': vector_id, 'vector': vector, 'neighbors': [] }) return # 从最高层贪心搜索 current = self.entry_point for l in range(self.max_level - 1, -1, -1): if l > level: # 在高层快速定位 while True: best_dist = self._distance( vector, self._vector_store[current] ) improved = False current_node = self._find_node(current, l) if current_node: for neighbor in current_node['neighbors']: d = self._distance(vector, neighbor['vector']) if d < best_dist: best_dist = d current = neighbor['id'] improved = True if not improved: break # 在目标层级建立连接 for l in range(min(level, self.max_level - 1), -1, -1): new_node = { 'id': vector_id, 'vector': vector, 'neighbors': [] } # 搜索最近的M个节点作为邻居 candidates = self._search_layer(vector, self.M, l) new_node['neighbors'] = candidates # 建立双向连接 self.graphs[l].append(new_node) for neighbor in candidates: node = self._find_node(neighbor['id'], l) if node: node['neighbors'].append(new_node) def _find_node(self, vector_id, level): for node in self.graphs[level]: if node['id'] == vector_id: return node return None def _search_layer(self, query, k, level): """搜索最近的k个节点""" results = [] for node in self.graphs[level]: dist = self._distance(query, node['vector']) results.append({ 'id': node['id'], 'vector': node['vector'], 'distance': dist }) results.sort(key=lambda x: x['distance']) return results[:k]模拟不同数据量下的索引构建
注意:简化实现,性能不代表真实HNSW
data_sizes = [10000, 50000, 100000]
for size in data_sizes:
vectors = np.random.randn(size, 768).astype(np.float32)
hnsw = SimpleHNSW(dim=768, M=16)
start = time.time() for i in range(min(size, 5000)): # 演示用,最多5000条 hnsw.insert(vectors[i], i) elapsed = time.time() - start print(f"数据量: {min(size, 5000):>8,} | 构建时间: {elapsed:>8.2f}s | " f"平均每向量: {elapsed/min(size, 5000)*1000:.2f}ms")注意每向量平均时间随数据量增长而增加。这是因为随着图变大,搜索最近邻的开销增加。到亿级数据,这个增长会更加明显。
实际生产中的索引策略
在Qdrant 10B基准测试中,他们选择了一个很关键的决策:使用500万条数据作为代表性样本做标准化评估,而不是在整个100亿数据上反复跑测试。这是一个很实际的做法——你不需要在最终规模上测试一切,用统计上有效的样本来推断整体行为。
据Qdrant官方博客的工程师建议,生产环境中的索引配置应该是:
# Qdrant API示例 - 亿级数据推荐配置 from qdrant_client import QdrantClient from qdrant_client.models import ( VectorParams, Distance, HnswConfigDiff, ScalarQuantizationConfig, ScalarType, QuantizationType, )client = QdrantClient(host="localhost", port=6333)
创建带量化的collection - 适用于5000万到5亿数据量
client.create_collection(
collection_name="production_kb",
vectors_config=VectorParams(
size=768,
distance=Distance.COSINE,
on_disk=True, # 向量存磁盘,节省内存
),
hnsw_config=HnswConfigDiff(
m=16,
ef_construct=200,
on_disk=True, # 索引也存磁盘
),
quantization_config=ScalarQuantizationConfig(
type=QuantizationType.SCALAR,
scalar=ScalarType(
type="int8",
quantile=0.99, # 用99分位数做量化范围
always_ram=True, # 量化后的向量放内存(比原始小4倍)
),
),
# 分段配置:数据量越大,分段越多
optimizer_config={
"memmap_threshold": 20000, # 超过2万向量的段用mmap
"indexing_threshold": 20000, # 超过2万向量才建HNSW索引
},
)
关键参数解读:
on_disk=True:HNSW索引存到NVMe SSD而不是内存,内存需求直降70%以上,查询延迟增加约10-20%always_ram=True(标量量化):量化后的uint8向量常驻内存,768维只需768字节(原来3072字节),在内存和磁盘之间找了个平衡点ef_construct=200:建索引时的搜索宽度,越大索引质量越高但构建越慢。亿级数据建议150-200
写入、更新与删除——向量数据库的工程脏活
向量数据库最让人头疼的不是查询,而是更新。
场景:你的知识库有1000万条文档向量。每天新增5万条,修改1万条,删除2000条。看起来不多?放在HNSW索引里试试。
HNSW的设计初衷是一次构建,多次查询。频繁插入会破坏图的平衡性,导致查询质量下降。而删除操作更麻烦——你不能简单地从图中移除一个节点,因为其他节点可能通过它来导航。
Qdrant的分段存储架构
Qdrant的做法很有意思。它把数据分成两类段:
class SegmentManager: """Qdrant风格的分段存储管理"""def __init__(self): self.mutable_segment = MutableSegment(capacity=100_000) self.immutable_segments = [] def upsert(self, vectors, ids): """写入新数据""" # 新数据先写入mutable segment self.mutable_segment.insert(vectors, ids) # 如果mutable segment满了,sealed并创建新的 if self.mutable_segment.is_full(): sealed = self.mutable_segment.seal() self.immutable_segments.append(sealed) self.mutable_segment = MutableSegment(capacity=100_000) # 后台触发compaction self._trigger_background_compaction() def delete(self, ids): """删除数据""" # 逻辑删除:标记为deleted,不立即物理删除 for segment in [self.mutable_segment] + self.immutable_segments: segment.mark_deleted(ids) def search(self, query, top_k=10): """查询:搜索所有段并合并结果""" candidates = [] # 搜索mutable segment(暴力搜索,因为数据量小) mutable_results = self.mutable_segment.search(query, top_k * 2) candidates.extend(mutable_results) # 搜索immutable segments(用HNSW索引) for seg in self.immutable_segments: seg_results = seg.search(query, top_k * 2) candidates.extend(seg_results) # 全局排序取top_k candidates.sort(key=lambda x: x.score, reverse=True) return candidates[:top_k] def _trigger_background_compaction(self): """后台compaction:合并小段、物理清理已删除数据、重建索引""" pass
class MutableSegment:
"""可变段:接收新写入,数据量小,用暴力搜索"""
def __init__(self, capacity): self.capacity = capacity self.vectors = [] self.ids = [] self.deleted = set() def insert(self, vectors, ids): self.vectors.extend(vectors) self.ids.extend(ids) def is_full(self): return len(self.vectors) >= self.capacity def seal(self): """封段:构建HNSW索引,变为不可变段""" return ImmutableSegment(self.vectors, self.ids) def search(self, query, top_k): import numpy as np results = [] for i, vec in enumerate(self.vectors): if self.ids[i] not in self.deleted: score = np.dot(query, vec) / ( np.linalg.norm(query) * np.linalg.norm(vec) + 1e-9 ) results.append({'id': self.ids[i], 'score': float(score)}) results.sort(key=lambda x: x['score'], reverse=True) return results[:top_k]class ImmutableSegment:
"""不可变段:已构建HNSW索引,只读"""
def __init__(self, vectors, ids): import numpy as np self.vectors = np.array(vectors) self.ids = ids self.deleted = set() # 实际生产中这里构建HNSW索引 self.index = "HNSW_INDEX_PLACEHOLDER" def search(self, query, top_k): # 用HNSW索引快速搜索 pass def mark_deleted(self, ids): self.deleted.update(ids)这个架构的核心思想是:写入和搜索走不同的路径。新数据先进入小的可变段(暴力搜索),积累到一定量后封段、构建HNSW索引,变成不可变段。删除操作只做逻辑标记,物理清理交给后台compaction。
这意味着:
- 刚写入的数据立即可搜(在mutable segment中暴力搜索)
- 不会因为没有建好索引而搜不到新数据
- 后台compaction不影响在线查询
Embedding模型变更——最痛苦的迁移
比日常增删更头疼的是换Embedding模型。
假设你从text-embedding-ada-002迁移到gte-multilingual-base。即使输出维度相同(都是768维),两个模型的向量空间完全不兼容。混在一个索引里,相似度计算就废了。
正确的做法是蓝绿迁移:
class EmbeddingMigration: """Embedding模型蓝绿迁移"""def __init__(self, old_model, new_model, qdrant_client): self.old_model = old_model self.new_model = new_model self.client = qdrant_client self.old_collection = "kb_v1" self.new_collection = "kb_v2" def prepare(self): """阶段1:创建新collection""" vector_size = self.new_model.get_dimension() self.client.create_collection( collection_name=self.new_collection, vectors_config=VectorParams( size=vector_size, distance=Distance.COSINE, ), ) def migrate_batch(self, old_documents): """阶段2:批量迁移""" # 用新模型重新编码 new_vectors = self.new_model.encode( [doc.text for doc in old_documents] ) # 写入新collection from qdrant_client.models import PointStruct self.client.upsert( collection_name=self.new_collection, points=[ PointStruct( id=doc.id, vector=new_vector.tolist(), payload=doc.metadata, ) for doc, new_vector in zip(old_documents, new_vectors) ], ) def validate(self, test_queries) -> bool: """阶段3:对比验证""" results = {} for query in test_queries: old_vec = self.old_model.encode(query) new_vec = self.new_model.encode(query) old_results = self.client.search( self.old_collection, old_vec.tolist(), limit=10 ) new_results = self.client.search( self.new_collection, new_vec.tolist(), limit=10 ) # 比较top-10的重叠度 old_ids = {r.id for r in old_results} new_ids = {r.id for r in new_results} overlap = len(old_ids & new_ids) / 10 results[query] = overlap avg_overlap = sum(results.values()) / len(results) print(f"平均top-10重叠率: {avg_overlap:.2%}") return avg_overlap > 0.7 # 70%以上重叠认为可以切换 def switch_read(self): """阶段4:切换读流量到新collection""" # 应用层切换读取的collection名称 pass def rollback(self): """回滚:切回旧collection""" pass
关键原则:
- 新旧collection并行运行一段时间
- 用测试集对比检索质量
- 确认新模型效果达标后再切换读流量
- 保留旧collection至少两周,出问题随时回滚
容灾与高可用:Turbopuffer宕机事件的教训
2026年10月2日,向量搜索公司Turbopuffer宣布关停旗舰产品Vectra。60多家客户的推荐系统、语义搜索、欺诈检测管道一夜之间失去底层存储。据报道,公司此前获得了4500万美元A轮融资,给了客户30天迁移期限。
这不是一个遥远的商业八卦。对工程师来说,这暴露了一个实实在在的风险:你把你最核心的向量数据,交给了一个可能随时消失的托管服务。
向量数据库容灾架构
class VectorDBDisasterRecovery: """向量数据库容灾方案"""def __init__(self): self.primary = "qdrant-cluster-a" self.replica = "qdrant-cluster-b" self.backup_bucket = "s3://vector-backups" def configure_replication(self): """配置主从复制""" # Qdrant原生支持分片级复制 replication_config = { "collection": "production_kb", "shards": { "shard_0": { "primary": "node-a1", "replicas": ["node-b1", "node-c1"], }, "shard_1": { "primary": "node-a2", "replicas": ["node-b2", "node-c2"], }, }, "write_consistency": "quorum", # 写入需多数确认 } return replication_config def scheduled_backup(self, collection_name): """定时快照备份""" from datetime import datetime snapshot = { "collection": collection_name, "timestamp": datetime.utcnow().isoformat(), "type": "full_snapshot", } # 导出到对象存储 # 包含:向量数据 + 索引 + payload + collection配置 backup_path = ( f"{self.backup_bucket}/{collection_name}/" f"{datetime.utcnow().strftime('%Y%m%d_%H%M')}" ) return { "snapshot": snapshot, "backup_path": backup_path, } def failover(self): """主集群故障时的切换流程""" steps = [ "1. 检测主集群不可用(心跳超时 > 30s)", "2. 确认数据同步状态(检查replication lag)", "3. 将备集群提升为主集群", "4. 更新DNS/负载均衡指向新主集群", "5. 通知应用层切换完成", "6. 记录故障时间和数据丢失窗口", ] return steps def estimate_data_loss_window(self): """评估可能的数据丢失窗口""" # 同步复制:零数据丢失,但写入延迟增加 # 异步复制:可能有几秒到几分钟的数据丢失 return "异步复制模式下,典型数据丢失窗口为10-30秒"
Turbopuffer事件的核心教训:
- 不要只依赖一个托管向量数据库。即使是用Pinecone、Zilliz Cloud这样的托管服务,也要有本地备份和迁移方案
- 定期做迁移演练。不是做备份,是做完整的从备份恢复到另一个环境的全流程测试
- 数据导出能力。确保你能随时把向量数据+元数据+索引配置完整导出,格式是通用的(Parquet、HDF5等)
- 自建+托管混合部署。核心数据用自建Qdrant/Milvus,弹性需求用托管服务,避免单点依赖
据Qdrant 10B基准测试的工程实践,他们选择了开源、可自部署的方案,并把完整的评估数据集以Parquet格式公开。这种做法让任何团队都能独立验证和复现结果,而不是被锁定在某个厂商的生态里。这也是为什么黑箭科技在做技术选型时优先选择开源可自部署方案——数据主权和迁移自由度,比一时的性能数字重要得多。
2026年向量数据库选型决策框架
写了这么多工程细节,最后给一个实用的选型决策树。不吹任何产品,只说场景匹配:
def vector_db_selection( total_vectors: int, daily_insertions: int, filter_complexity: str, # none | simple | complex latency_sla_ms: int, team_size: int, budget: str, # low | medium | high ) -> dict: """向量数据库选型决策"""recommendation = {} # 数据规模判断 if total_vectors < 1_000_000: recommendation["primary"] = "pgvector" recommendation["reason"] = "已有PostgreSQL生态,运维成本最低" recommendation["caution"] = "超过500万向量后性能衰减明显" elif total_vectors < 100_000_000: if filter_complexity == "complex": recommendation["primary"] = "Qdrant" recommendation["reason"] = ( "原生payload过滤+向量检索一体化," "P99延迟最低(约12ms@10M向量)" ) else: recommendation["primary"] = "Qdrant 或 Milvus" recommendation["reason"] = ( "中等规模两者都能胜任," "Qdrant运维更简单,Milvus扩展性更强" ) elif total_vectors < 1_000_000_000: recommendation["primary"] = "Milvus" recommendation["reason"] = "十亿级唯一正解,GPU加速+多索引类型+存算分离" recommendation["caution"] = "运维复杂度高,需专职DBA" recommendation["alternative"] = "Qdrant集群模式(如果过滤需求强)" else: recommendation["primary"] = "Milvus(多集群分片)" recommendation["reason"] = "百亿级必须分布式,Milvus生态最成熟" # 团队规模影响 if team_size < 5 and total_vectors > 10_000_000: recommendation["warning"] = ( "团队规模偏小,优先选Qdrant(运维成本低)" ) # 预算影响 if budget == "low": recommendation["tip"] = "优先开源自部署,量化压缩节省60-80%存储成本" return recommendation
使用示例
result = vector_db_selection(
total_vectors=50_000_000,
daily_insertions=50000,
filter_complexity="complex",
latency_sla_ms=20,
team_size=8,
budget="medium",
)
for k, v in result.items():
print(f"{k}: {v}")
据2026年10月的最新基准测试数据(来源:Qdrant官方博客、独立测评文章),各引擎的典型表现:
| 引擎 | p50延迟 | p99延迟 | 适合规模 | 运维复杂度 |
|---|---|---|---|---|
| pgvector | 取决于PG负载 | 变化大 | < 1000万 | 极低 |
| Qdrant | ~4ms | ~25ms | 1000万-5亿 | 中 |
| Milvus | ~18ms | ~30ms | 5亿-百亿 | 高 |
| Pinecone | < 10ms | < 30ms | 弹性 | 极低(托管) |
注意:以上数据有方向性参考价值,但不能直接作为选型依据。不同硬件、不同数据分布、不同过滤条件,结果差异很大。唯一的正确做法:用你自己的数据,在你自己的硬件上,跑你自己的查询模式。
总结
从千万到百亿,向量数据库的工程挑战不在查询本身,而在查询之外的系统性问题:
- 内存管理:量化压缩(Scalar PQ / Product PQ)能把内存需求砍掉4-32倍,代价是微小的精度损失
- 索引策略:HNSW不是万能药,分段存储+磁盘索引+后台compaction才是亿级数据的正道
- 更新维护:写入走mutable segment、查询走immutable segment,两条路径互不干扰
- 容灾备份:Turbopuffer的关停是2026年最响的警钟,永远要有迁移Plan B
Qdrant 10B基准测试的真正价值,不只是"我们跑到了100亿"这个数字。它提供了一套开放的、可复现的评估方法论——完整Parquet分片、12万条评估查询、精确到top-1000的ground truth。任何团队都能基于这个数据集测试自己的索引策略、量化方案和检索质量。
在向量数据库这个领域,2026年最大的变化不是某个产品有多快,而是工程实践从"能跑Demo"进化到了"能扛生产"。这中间差的,就是本文讨论的这些脏活累活。
你在生产环境中用向量数据库踩过什么坑?内存爆了?索引重建太慢?还是被托管服务商坑过?评论区聊聊,互相避坑。
本文基于Qdrant 10B基准测试(2026年9月发布)、Turbopuffer关停事件(2026年10月报道)及多个独立基准测试数据撰写,数据来源已在文中标注。