vector search designer

Vector Search Designer 是一个帮助架构和实现向量相似度搜索系统的技能，用于支撑语义搜索、推荐引擎和 AI 应用。它引导完成向量数据库选型、索引结构设计、查询性能优化，以及扩展到百万乃至十亿级向量的全过程。

向量搜索已成为现代 AI 系统的基础，从 RAG 流水线到商品推荐均依赖于此。该技能覆盖完整技术栈：理解近似最近邻（ANN）算法、在多种数据库方案间做选择、调优召回率与延迟的权衡，以及实现生产级搜索基础设施。无论基于 Pinecone、Weaviate、Qdrant、pgvector 构建，还是自行实现，该技能都旨在帮助向量搜索系统满足性能与精度要求。

核心工作流

工作流 1：选择向量数据库

先收集需求：规模（多少向量）、查询模式（单条还是批量、是否需要过滤）、延迟要求（实时还是批处理）、更新频率（静态还是动态）、基础设施（托管还是自托管）。

然后比较选项：

数据库	规模	托管	特性	适用场景
Pinecone	十亿+	是	混合、命名空间	生产、零运维
Weaviate	1 亿+	两者皆可	GraphQL、模块	多模态、复杂查询
Qdrant	1 亿+	两者皆可	Rust 性能、过滤	高性能、自托管
Milvus	十亿+	两者皆可	GPU 支持、集群	大规模、ML 团队
pgvector	1000 万	否	PostgreSQL 原生	已有 Postgres、小规模
Chroma	100 万	否	简单 API	原型、嵌入式

最后用自有数据评估，并记录决策依据。

工作流 2：设计索引架构

选择 ANN 算法：HNSW 召回率/速度权衡最佳但内存占用高；IVF 适合超大数据集但需要训练；PQ 用于压缩以扩展规模但有一定精度损失；Flat 为精确搜索，仅适合小数据集。

配置索引参数示例：

# HNSW example configuration
hnsw_config = {
    "M": 16,  # Connections per node (higher = better recall, more memory)
    "efConstruction": 200,  # Build-time search depth
    "efSearch": 100,  # Query-time search depth
}

# IVF example configuration
ivf_config = {
    "nlist": 1024,  # Number of clusters
    "nprobe": 32,  # Clusters to search at query time
}

同时规划分片策略以支持扩展，并设计用于过滤的元数据模式。

工作流 3：优化搜索性能

先建立基准性能：每秒查询数（QPS）、延迟（p50、p95、p99）、Recall@k 精度。识别瓶颈：索引加载时间、搜索延迟、过滤计算、网络开销。再应用优化：调优索引参数（ef、nprobe）、实现查询缓存、优化过滤表达式、考虑量化以降低内存。最后验证召回率与延迟的权衡，并记录最优配置。

快速参考

操作	命令/触发
选择数据库	"Which vector database for [use case]"
设计索引	"Design vector index for [scale]"
优化搜索	"Speed up vector search"
添加过滤	"Add metadata filters to vector search"
扩展向量	"Scale to [N] million vectors"
基准测试	"Benchmark vector search performance"

最佳实践

• 合理选择数据库规模：不要为不需要的规模过度设计。少于 100 万向量时 pgvector 或 Chroma 通常足够；100 万至 1 亿向量可选 Qdrant、Weaviate 或 Pinecone；1 亿以上需 Milvus 或 Pinecone 并谨慎设计。
• 理解召回率与速度的权衡：ANN 本质是近似的，召回率越高查询越慢，需根据精度要求调优，并实际测量召回率而非假设。
• 必要时使用混合搜索：结合向量（语义相似）与关键词（BM25，精确词、名称、代码），通常能改善混合查询结果。
• 为过滤设计元数据：提前规划过滤策略，对高频过滤字段建索引，避免在高基数字段上过滤，权衡预过滤与后过滤。
• 尽可能批量操作：批量 upsert 用于写入，延迟允许时批量查询，使用异步操作，以减少网络开销。
• 监控与告警：生产搜索需要可观测性，包括查询延迟分位数、索引大小与内存使用、召回率随时间的变化。

高级技术

多向量搜索

处理具有多种表示的文档：

class MultiVectorIndex:
    def __init__(self):
        self.title_index = VectorIndex(dim=768)
        self.content_index = VectorIndex(dim=768)
        self.summary_index = VectorIndex(dim=768)

    def search(self, query_embedding, weights=None):
        weights = weights or {"title": 0.3, "content": 0.5, "summary": 0.2}

        results = {}
        for field, weight in weights.items():
            index = getattr(self, f"{field}_index")
            field_results = index.search(query_embedding, k=20)
            for doc_id, score in field_results:
                results[doc_id] = results.get(doc_id, 0) + score * weight

        return sorted(results.items(), key=lambda x: x[1], reverse=True)[:10]

带过滤的向量搜索策略

def filtered_search(query_embedding, filters, k=10):
    # Strategy 1: Pre-filter (for selective filters)
    if estimate_selectivity(filters) < 0.1:
        candidate_ids = apply_filters(filters)
        return vector_search_subset(query_embedding, candidate_ids, k)

    # Strategy 2: Post-filter (for non-selective filters)
    elif estimate_selectivity(filters) > 0.5:
        results = vector_search(query_embedding, k * 3)
        filtered = [r for r in results if matches_filters(r, filters)]
        return filtered[:k]

    # Strategy 3: Hybrid (general case)
    else:
        return vector_search_with_filters(query_embedding, filters, k)

量化以扩展规模

在可接受的精度损失下降低内存：

# Product Quantization configuration
pq_config = {
    "nbits": 8,  # Bits per sub-quantizer
    "m": 16,  # Number of sub-quantizers
    # 768-dim * 4 bytes = 3KB/vector -> 16 * 1 byte = 16 bytes/vector
}

# Binary quantization (extreme compression)
binary_config = {
    "threshold": 0,  # Values > 0 -> 1, else -> 0
    # 768-dim * 4 bytes = 3KB/vector -> 768 bits = 96 bytes/vector
}

增量索引更新

高效处理动态数据：

class DynamicVectorIndex:
    def __init__(self, rebuild_threshold=10000):
        self.main_index = build_optimized_index()
        self.delta_index = []  # Recent additions
        self.rebuild_threshold = rebuild_threshold

    def add(self, vector, metadata):
        self.delta_index.append((vector, metadata))
        if len(self.delta_index) >= self.rebuild_threshold:
            self.rebuild()

    def search(self, query, k):
        main_results = self.main_index.search(query, k)
        delta_results = brute_force_search(self.delta_index, query, k)
        return merge_results(main_results, delta_results, k)

    def rebuild(self):
        all_data = self.main_index.get_all() + self.delta_index
        self.main_index = build_optimized_index(all_data)
        self.delta_index = []

常见陷阱

• 在大规模场景使用精确（Flat）搜索而非 ANN。
• 不测量实际召回率，假设 ANN 已"足够好"。
• 对元数据字段过度建索引，拖慢更新。
• 忽视 HNSW 索引的内存需求。
• 未规划索引重建与维护窗口。
• 假设向量数据库能处理所有用例（有时 Elasticsearch 更合适）。
• 忘记为余弦相似度归一化向量。
• 在同一索引中混用不同模型的嵌入。