向量数据库调优

优化向量数据库：HNSW参数，分块优化，混合搜索评分，性能调优

执行摘要

向量数据库（Pinecone，Qdrant，Weaviate，ChromaDB）为RAG和语义搜索提供动力。性能在很大程度上取决于调优参数，分块策略和混合搜索配置。这项技能涵盖了生产优化。

关键概念

向量搜索性能

索引类型（HNSW，IVF，基于磁盘）- 速度与准确性的权衡
参数（ef construction，ef search，M）- 微调索引
分块句子，段落，语义分块 - 检索质量
混合搜索 密集+稀疏向量 - 两全其美
扩展复制，分片 - 处理生产负载

流行的向量数据库

数据库	最适合	管理？	定价
Pinecone	简单设置，生产	✅ 是	基于使用
Qdrant	性能，OSS	✅ 是	免费+付费
Weaviate	混合搜索，GraphQL	✅ 是	免费+付费
ChromaDB	本地开发，OSS	❌ 否	自托管
pgvector	PostgreSQL原生	❌ 否	自托管

HNSW调优（最常用的索引）

HNSW参数解释

HNSW（层次导航小世界）是最广泛使用的向量索引。

参数	它的作用	默认	调优指南
M	每个节点的最大连接数（邻居）	16	↑准确性，↓速度，↑内存
ef_construction	索引构建期间的搜索深度	200	↑准确性，↓构建速度
ef_search	查询时的搜索深度	10	↑准确性，↓查询速度
dim	向量维度	1536	必须与嵌入模型匹配

Pinecone HNSW配置

import pinecone

# 初始化
pc = pinecone.Pinecone(api_key="...")

# 创建索引并进行HNSW调优
pc.create_index(
    name="my-index",
    dimension=1536,  # OpenAI text-embedding-3-small
    metric="cosine",
    spec={
        "pod_type": "p1.x1",  # 成本优化
        "pods": 1,
        "replicas": 1,
        "pod_type": "p1",
        "shards": 1,
        "metadata_config": {
            "indexed": ["category", "tags"]  # 为过滤索引元数据
        }
    }
)

Qdrant HNSW调优

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, HnswConfigDiff

client = QdrantClient(url="http://localhost:6333")

# 创建集合并进行HNSW调优
client.create_collection(
    collection_name="documents",
    vectors_config=VectorParams(
        size=1536,
        distance=Distance.COSINE,
        hnsw_config=HnswConfigDiff(
            m=32,              # 更多邻居=更好的召回率，更多内存
            ef_construct=200,  # 构建时搜索深度
            full_scan_threshold=10000,  # 小型集合的暴力搜索
            max_indexing_threads=4      # 并行索引构建
        )
    )
)

Weaviate HNSW调优

import weaviate

client = weaviate.Client("http://localhost:8080")

# 创建集合（类）并进行HNSW调优
class_obj = {
    "class": "Document",
    "vectorizer": "none",  # 带来自己的嵌入
    "vectorIndexConfig": {
        "type": "hnsw",
        "skip": False,
        "dynamicEfFactor": 8,
        "efConstruction": 256,
        "maxConnections": 32,
        "ef": -1,  # 查询时使用动态ef
        "vectorCacheMaxObjects": 100000
    }
}

client.schema.create_class(class_obj)

分块策略

1. 固定长度分块

def fixed_length_chunking(text: str, chunk_size: int = 500):
    """将文本分割成固定长度的块"""

    chunks = []
    for i in range(0, len(text), chunk_size):
        chunk = text[i:i+chunk_size]
        chunks.append(chunk)

    return chunks

2. 基于句子的分块

import nltk

nltk.download('punkt')

def sentence_chunking(text: str, max_sentences: int = 5):
    """将文本分成句子组"""

    sentences = nltk.sent_tokenize(text)
    chunks = []

    for i in range(0, len(sentences), max_sentences):
        chunk_sentences = sentences[i:i+max_sentences]
        chunk = ' '.join(chunk_sentences)
        chunks.append(chunk)

    return chunks

3. 语义分块（推荐）

按语义相似性分组（相似主题保持在一起）。

from sentence_transformers import SentenceTransformer
import numpy as np

class SemanticChunker:
    """按语义相似性分块文本"""

    def __init__(self):
        self.model = SentenceTransformer('all-MiniLM-L6-v2')

    def chunk(self, text: str, max_chunk_size: int = 500, similarity_threshold: float = 0.5):
        """将文本分成语义上连贯的块"""

        # 分成句子
        sentences = nltk.sent_tokenize(text)

        # 获取所有句子的嵌入
        embeddings = self.model.encode(sentences)

        chunks = []
        current_chunk = []

        for i, sentence in enumerate(sentences):
            if not current_chunk:
                current_chunk.append(sentence)
                continue

            # 获取当前块的嵌入
            chunk_text = ' '.join(current_chunk)
            chunk_embedding = self.model.encode([chunk_text])[0]

            # 获取与新句子的相似性
            sentence_embedding = embeddings[i]
            similarity = np.dot(chunk_embedding, sentence_embedding) / (
                np.linalg.norm(chunk_embedding) * np.linalg.norm(sentence_embedding)
            )

            # 检查是否足够相似且不太长
            if similarity > similarity_threshold and len(chunk_text) < max_chunk_size:
                current_chunk.append(sentence)
            else:
                # 开始新块
                chunks.append(' '.join(current_chunk))
                current_chunk = [sentence]

        # 添加最后一个块
        if current_chunk:
            chunks.append(' '.join(current_chunk))

        return chunks

分块比较

策略	优点	缺点	何时使用
固定长度	简单，可预测	打破句子	统一文档
基于句子	保持含义	可能将不相关的组合在一起	一般文本
语义	最佳检索质量	更慢，更复杂	RAG系统

混合搜索（密集+稀疏）

密集向量（语义）

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')

query = "How do I reset my password?"
query_embedding = model.encode(query)

# 通过向量数据库搜索
results = vector_db.search(
    vector=query_embedding,
    top_k=10
)

稀疏向量（BM25/关键词）

from rank_bm25 import BM25Okapi

# 标记文档
tokenized_corpus = [doc.split() for doc in documents]

# 构建BM25索引
bm25 = BM25Okapi(tokenized_corpus)

# 查询
query_tokens = query.split()
sparse_scores = bm25.get_scores(query_tokens)

混合评分（互惠排名融合）

def reciprocal_rank_fusion(dense_results, sparse_results, k: float = 60.0):
    """
    组合密集和稀疏搜索结果

    k：排名常数（越高=更倾向于第一排名）
    """

    # 创建RRF分数字典
    rrf_scores = {}

    # 处理密集结果
    for i, (doc_id, score) in enumerate(dense_results):
        rank = i + 1
        if doc_id not in rrf_scores:
            rrf_scores[doc_id] = 0
        rrf_scores[doc_id] += 1.0 / (k + rank)

    # 处理稀疏结果
    for i, (doc_id, score) in enumerate(sparse_results):
        rank = i + 1
        if doc_id not in rrf_scores:
            rrf_scores[doc_id] = 0
        rrf_scores[doc_id] += 1.0 / (k + rank)

    # 按总RRF分数排序
    sorted_results = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)

    return sorted_results

Weaviate混合搜索

# Weaviate内置混合搜索
query_results = (
    client.query
    .get("Document", ["content", "title"])
    .with_hybrid(
        query="reset password",
        alpha=0.7  # 0 = 纯BM25，1 = 纯向量
    )
    .with_limit(10)
    .do()
)

Qdrant混合搜索

# 使用向量和关键词搜索
class SearchResult:
    def __init__(self, id, dense_score, sparse_score):
        self.id = id
        self.dense_score = dense_score
        self.sparse_score = sparse_score

# 获取密集结果
dense_results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    limit=10
)

# 获取稀疏结果（通过Qdrant使用BM25）
sparse_results = client.search(
    collection_name="documents",
    query_filter=Must([
        Field(key="content", match=MatchText(text="reset password"))
    ]),
    limit=10
)

# 使用RRF组合
combined = reciprocal_rank_fusion(dense_results, sparse_results)

查询时间调优

ef_search（召回率与速度权衡）

# 低ef = 快速但准确性较低
results_low_recall = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    limit=10,
    search_params={"hnsw_ef": 10}  # 快速
)

# 高ef = 速度慢但更准确
results_high_recall = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    limit=10,
    search_params={"hnsw_ef": 100}  # 准确
)

动态ef_search（自适应）

def adaptive_ef_search(num_results_needed: int) -> int:
    """根据需要的结果数量计算ef_search"""

    # 经验法则：ef_search ≈ 10× num_results
    return min(100, num_results_needed * 10)

# 使用
ef = adaptive_ef_search(10)
results = client.search(..., search_params={"hnsw_ef": ef})

使用预过滤器过滤

如果可能，总是在向量搜索之前进行过滤（元数据过滤器）。

# 不好：后过滤（浪费资源）
results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    limit=1000  # 获取很多，然后过滤
)
filtered = [r for r in results if r.metadata["category"] == "tech"]

# 好：预过滤（在子集中搜索）
results = client.search(
    collection_name="documents",
    query_vector=query_embedding,
    query_filter=Filter(
        must=[Field(key="category", match=MatchValue(value="tech"))]
    ),
    limit=10  # 只获取你需要的
)

生产监控

跟踪这些指标

class VectorDBMonitor:
    """监控向量数据库性能"""

    def __init__(self):
        self.stats = {
            "query_latency": [],
            "recall": [],
            "throughput": 0,
            "index_build_time": []
        }

    def log_query(self, latency_ms: float, num_results: int):
        """记录查询指标"""
        self.stats["query_latency"].append(latency_ms)
        self.stats["throughput"] += 1

    def log_recall(self, retrieved_relevant: int, total_relevant: int):
        """计算召回率@k"""
        recall = retrieved_relevant / total_relevant if total_relevant > 0 else 0
        self.stats["recall"].append(recall)

    def get_summary(self):
        """获取性能摘要"""
        import statistics

        return {
            "avg_query_latency_ms": statistics.mean(self.stats["query_latency"]),
            "p95_query_latency_ms": statistics.quantiles(self.stats["query_latency"], n=20)[18],
            "avg_recall": statistics.mean(self.stats["recall"]),
            "total_queries": self.stats["throughput"]
        }

重新索引策略

def reindex_with_new_params(collection_name: str, new_configs: dict):
    """用新参数重建索引"""

    # 导出数据
    existing_data = client.scroll(collection_name, limit=10000)

    # 删除旧集合
    client.delete_collection(collection_name)

    # 创建新配置
    client.create_collection(
        collection_name=collection_name,
        vectors_config=VectorParams(size=1536, distance=Distance.COSINE),
        hnsw_config=HnswConfigDiff(**new_configs)
    )

    # 重新插入数据
    client.upsert(
        collection_name=collection_name,
        points=existing_data
    )

生产清单

✅ 选择索引类型（大多数情况下选择HNSW）
✅ 调整HNSW参数（M，ef_construction，ef_search）
✅ 定义分块策略（推荐语义）
✅ 配置混合搜索（密集+稀疏）
✅ 在搜索之前应用查询过滤器
✅ 监控就位（延迟，召回率）
✅ 定义重新索引策略
✅ 备份和灾难恢复

常见陷阱

❌ 使用默认HNSW参数 → 未针对您的用例进行优化 ❌ 后过滤 → 浪费资源；先预过滤 ❌ 固定长度分块 → 打破语义含义 ❌ 纯密集搜索 → 错过精确匹配；使用混合搜索 ❌ 不监控召回率 → 无法判断调优是否有效