一、向量检索的核心算法
向量数据库的本质是在高维空间（通常1024~1536维）中快速找到与查询向量最相似的N个向量。暴力遍历的复杂度是O(n)，向量索引将这个过程加速到O(log n)甚至O(1)。
1.1 HNSW算法
# HNSW（Hierarchical Navigable Small World）
# 一种基于图的近似最近邻搜索算法（ANN）
# 核心思想：多层跳表 + 可导航小世界图
# HNSW结构：
# Layer 2: 0 —— ———— 5 ← 最稀疏，跳得远
# Layer 1: 0 —— 2 —— 3 —— 5 ← 中间层
# Layer 0: 0 — 1 — 2 — 3 — 4 — 5 ← 最密集，邻居多
# 搜索过程（从顶层到低层）：
# ① 从Layer2入口开始，找到该层最近的邻居
# ② 沿边移动到最近邻居，如果当前位置不比任何邻居好，停止
# ③ 下降到下一层，从当前位置继续搜索
# ④ 重复直到Layer0
# HNSW参数调优：
# M（每层邻居数）：
# M=16: 建索引时间适中，精度高，内存适中
# M=32: 精度更高，内存增加约20%
# efConstruction（搜索宽度）：
# 值越大精度越高，索引时间越长（平方级增长）
# efConstruction=256: 大多数场景的最佳平衡
# 经验公式：efConstruction = M × 2^layers
# HNSW的内存占用：
# 索引大小 ≈ 原始数据大小 × (1 + M×2 × 4字节/指针)
# 1024维，100万条，FP16，HNSW索引 ≈ 原始数据2GB × 2.5 ≈ 5GB二、向量量化与压缩
2.1 Product Quantization（乘积量化）
# Product Quantization（PQ）：将高维向量压缩存储
# 原理：将D维向量拆分成m个子向量，每个子向量独立量化
# 示例：
# 向量维度D=128，分成m=8个子向量
# 每个子向量维度d'=D/m=16
# 训练阶段：从数据中聚类k=256个中心点（k=2^8）
# 存储阶段：存储子向量所属中心点ID（8字节/向量）
# 压缩比：(128×2B) / 8B = 32倍！
# PQ编码过程：
# 1. 将向量x分成m个段：x = [x¹, x², ..., x^m]
# 2. 每段x^i独立查表，找到最近的中心c_i
# 3. 存储：code = [code¹, code², ..., code^m]
# 其中code^i是中心点ID（8bit）
# 距离计算（对称PQ距离）：
# d(x, y) ≈ Σ d_lookup(x^i, y^i)
# 其中d_lookup是预计算的段间距离表
# IVF-PQ（倒排索引+PQ）：
# ① 先用IVF聚类将向量分配到V个桶
# ② 搜索时只扫描目标向量所在的桶
# ③ 桶内用PQ压缩计算距离
# 减少PQ的计算量（从全量到部分桶）2.2 相似度度量选择
# 三大相似度度量：
# ① 余弦相似度（Cosine）
# cos_sim = (A·B) / (||A|| × ||B||)
# 值范围：[-1, 1]，1表示完全相同
# 适用：归一化向量，语义相似性（推荐中文Embedding）
# ② 内积（IP/Dot Product）
# ip = A·B
# 值范围：无界
# 适用：归一化后等价于余弦；未归一化时反映向量"长度"信息
# ③ 欧氏距离（L2）
# l2 = ||A - B||
# 值范围：[0, +∞)
# 适用：几何距离有意义的场景
# ⚠️ 重要：Embedding模型输出是否归一化？
# text-embedding-ada-002: 已归一化 → 用余弦或IP等价
# bge-large-zh-v1.5: 已归一化 → 用IP即可
# 验证归一化：
import numpy as np
norms = np.linalg.norm(embeddings, axis=1)
print(f"均值: {norms.mean():.6f}, 标准差: {norms.std():.6f}")
# 如果均值≈1.0，标准差≈0，则已归一化三、分段（Chunking）策略
# Chunking = 将文档切成小块再向量化
# 策略选择直接影响检索质量
# ① 固定长度分块（简单但有风险）
CHUNK_SIZE = 512 # token数
CHUNK_OVERLAP = 50 # 重叠token数（保持上下文连贯）
# 缺点：可能在句子中间断开，语义割裂
# ② 语义分块（按段落/句子切分）
def semantic_chunk(text: str, max_tokens: int = 512) -> list[str]:
sentences = text.split('。') # 按句子分隔
chunks, current = [], ""
for sent in sentences:
sent = sent.strip() + '。'
if len(current) + len(sent) <= max_tokens:
current += sent
else:
if current: chunks.append(current)
current = sent
if current: chunks.append(current)
return chunks
# ③ 层级分块（父子关系，支持回溯）
# - chunks表：子块（512 tokens）
# - chunks_parent表：父块（4096 tokens，包含多个子块）
# 检索返回子块，生成时用父块提供完整上下文
# ④ Markdown/HTML结构分块（保留标题层级）
from markdown import Markdown
from bs4 import BeautifulSoup
def chunk_by_heading(content: str) -> list[dict]:
soup = BeautifulSoup(content, 'html.parser')
chunks = []
for heading in soup.find_all(['h1', 'h2', 'h3']):
section_text = heading.get_text()
sibling = heading.find_next_sibling()
while sibling:
if sibling.name in ['h1', 'h2', 'h3']:
break
section_text += sibling.get_text()
sibling = sibling.find_next_sibling()
chunks.append({"heading": heading.text, "content": section_text})
return chunks四、混合检索架构
# 纯向量检索的局限：专有名词/精确匹配召回差
# 解决方案：向量检索 + 关键词检索（BM25）混合
# ① Reciprocal Rank Fusion（RRF）
# 将不同检索方法的结果合并排序
def reciprocal_rank_fusion(results_list: list[list[tuple]], k=60) -> list:
"""融合多个检索结果"""
fused_scores = {}
for results in results_list:
for rank, (doc_id, score) in enumerate(results):
# RRF公式：1 / (k + rank)
# k=60是常用平滑参数
fused_scores[doc_id] = fused_scores.get(doc_id, 0) + 1 / (k + rank)
# 按融合分数排序
sorted_docs = sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
return sorted_docs
# ② 稀疏检索（BM25）补充
# BM25是传统关键词检索算法，解决TF-IDF的缺陷
# 特点：惩罚高频词，奖励罕见词（IDF因子）
# 实现：使用rank_bm25库或Elasticsearch
# ③ 混合检索配置（Milvus + BM25）
# 稀疏向量 = BM25权重（词频统计）
# 稠密向量 = Embedding模型输出（语义向量）
# 融合：RRF或加权求和
# 效果对比（测试集：500条企业文档问答）
# 纯向量检索: Recall@5 = 68%
# 纯BM25: Recall@5 = 52%
# 混合(RRF): Recall@5 = 81% ← 提升13%