Vector Search Designer

向量搜索设计器

向量搜索设计器技能可帮助你设计和实现向量相似性搜索系统，为语义搜索、推荐引擎和 AI 应用提供支持。它会指导你选择合适的向量数据库、设计索引结构、优化查询性能，并扩展至数百万或数十亿个向量。

从 RAG 工作流到产品推荐，向量搜索已成为现代 AI 系统的基础。本技能涵盖完整技术栈：了解近似最近邻（ANN）算法、在数据库选项之间做出选择、调整召回率与延迟之间的权衡，以及实现可用于生产环境的搜索基础设施。

无论你是在 Pinecone、Weaviate、Qdrant、pgvector 上构建系统，还是实现自己的解决方案，本技能都能确保向量搜索系统满足性能和准确性要求。

核心工作流

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

1. 收集需求：
   • 规模：有多少个向量？
   • 查询模式：单条还是批量？是否需要筛选？
   • 延迟要求：实时还是批处理？
   • 更新频率：静态还是动态？
   • 基础设施：托管还是可自托管？
2. 比较选项：

   数据库	规模	托管	功能	最适合
   Pinecone	10 亿以上	是	混合搜索、命名空间	生产环境、无需运维
   Weaviate	1 亿以上	两者皆可	GraphQL、模块	多模态、复杂查询
   Qdrant	1 亿以上	两者皆可	Rust 性能、筛选	高性能、可自托管
   Milvus	10 亿以上	两者皆可	GPU 支持、集群	大规模场景、ML 团队
   pgvector	1000 万	否	PostgreSQL 原生	已有 Postgres、小规模场景
   Chroma	100 万	否	简单 API	原型开发、嵌入式

3. 使用自己的数据进行评估
4. 记录决策依据

工作流 2：设计索引架构

1. 选择ANN 算法：
   • HNSW：召回率与速度之间的权衡最佳，但内存占用较高
   • IVF：适用于超大型数据集，需要训练
   • PQ：通过压缩支持扩展，但会损失一定准确性
   • Flat：精确搜索，仅适用于小型数据集
2. 配置索引参数：

   # 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. 规划适用于扩展的分片策略
4. 设计用于筛选的元数据架构

工作流 3：优化搜索性能

1. 测试基准性能：
   • 每秒查询数（QPS）
   • 延迟（p50、p95、p99）
   • Recall@k 准确率
2. 识别瓶颈：
   • 索引加载时间
   • 搜索延迟
   • 筛选计算
   • 网络开销
3. 应用优化：
   • 调整索引参数（ef、nprobe）
   • 实现查询缓存
   • 优化筛选表达式
   • 考虑使用量化以节省内存
4. 验证召回率与延迟之间的权衡
5. 记录最佳配置

快速参考

操作	命令/触发语
选择数据库	“为 [use case] 选择哪个向量数据库”
设计索引	“为 [scale] 设计向量索引”
优化搜索	“加快向量搜索”
添加筛选	“为向量搜索添加元数据筛选条件”
扩展向量规模	“扩展到 [N] 百万个向量”
测试搜索基准	“测试向量搜索性能”

最佳实践

• 为数据库选择合适的规模：不要为不需要的扩展能力过度设计

  • <1M 个向量：pgvector 或 Chroma 通常就足够了
  • 1-100M 个向量：Qdrant、Weaviate 或 Pinecone
  • 100M+ 个向量：Milvus 或经过谨慎设计的 Pinecone

• 了解召回率与速度之间的权衡：ANN 从设计上就是近似搜索

  • 召回率越高，查询速度越慢
  • 根据准确率要求进行调优
  • 测量实际召回率，不要想当然

• 在需要时使用混合搜索：结合向量搜索和关键词搜索

  • 向量：语义相似度
  • 关键词（BM25）：精确术语、名称、代码
  • 对于混合查询，通常能改善结果

• 设计用于筛选的元数据：提前规划筛选策略

  • 为常用筛选字段建立索引
  • 避免按高基数字段筛选
  • 权衡预筛选与后筛选

• 尽可能批量操作：减少网络开销

  • 批量写入数据
  • 在延迟允许时批量查询
  • 使用异步操作

• 监控并设置告警：生产环境中的搜索需要可观测性

  • 查询延迟百分位数
  • 索引大小和内存使用量
  • 召回率随时间的下降情况

高级技术

多向量搜索

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

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 = []

应避免的常见问题

• 在大规模场景下使用精确（扁平）搜索，而不是 ANN
• 不测量实际召回率，想当然地认为 ANN“足够好”
• 为过多元数据字段建立索引，导致更新变慢
• 忽视 HNSW 索引的内存需求
• 未规划索引重建和维护时段
• 认为向量数据库能处理所有用例（有时 Elasticsearch 更合适）
• 忘记为余弦相似度归一化向量
• 在同一索引中混用不同模型生成的嵌入