向量数据库与嵌入
大约 13 分钟约 3840 字
向量数据库与嵌入
简介
向量数据库的核心作用，是让系统能够根据"语义相似度"而不是"关键词是否完全匹配"来检索内容。它通常和嵌入模型（Embedding Model）一起使用：先把文本、图像、音频等内容编码成高维向量，再通过 ANN（近似最近邻）索引快速查找相似内容。这是 RAG、语义搜索、推荐系统和多模态检索的基础设施之一。
向量数据库的发展源于对大规模相似性搜索的需求。在高维向量空间中，精确的 K 近邻搜索（KNN）需要线性扫描所有向量，时间复杂度为 O(n)，无法满足百万级以上数据集的实时检索需求。近似最近邻（ANN）搜索通过牺牲少量精度换取数量级的速度提升，使得大规模语义检索成为可能。
从信息检索的角度看，传统关键词搜索（BM25、TF-IDF）基于词汇匹配，无法理解语义。例如搜索"如何提升代码性能"时，包含"优化程序速度"但不含"性能"一词的文档不会被召回。向量检索通过将文本映射到语义空间，使得语义相近的文本在向量空间中距离更近，从而实现真正的语义检索。
特点
向量检索的核心概念
- 嵌入（Embedding）：将离散的文本/图像映射为连续的浮点向量
- 距离度量：余弦相似度、欧氏距离、内积
- ANN 索引：HNSW、IVF、PQ 等近似最近邻算法
- 元数据过滤：在向量检索的同时进行属性条件过滤
- 混合检索：向量检索 + 关键词检索的组合
实现
嵌入基础：文本转向量
from openai import OpenAI
import numpy as np
client = OpenAI(api_key="your-api-key")
def get_embedding(text: str, model: str = "text-embedding-3-small"):
response = client.embeddings.create(
input=[text],
model=model
)
return response.data[0].embedding
def cosine_similarity(v1, v2):
v1 = np.array(v1)
v2 = np.array(v2)
return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))texts = [
"ASP.NET Core 是跨平台 Web 框架",
".NET 可以在 Linux 上运行",
"今天天气很好"
]
vectors = [get_embedding(t) for t in texts]
print(cosine_similarity(vectors[0], vectors[1]))
print(cosine_similarity(vectors[0], vectors[2]))嵌入模型输出的是"向量表示"，
而不是人类直接可读的标签。
相似内容在向量空间中距离更近。距离度量对比
import numpy as np
def compare_distance_metrics():
"""距离度量对比
1. 余弦相似度 (Cosine Similarity):
cos(a, b) = (a · b) / (||a|| × ||b||)
范围: [-1, 1]，越大越相似
最常用，对向量长度不敏感
2. 欧氏距离 (L2 Distance):
d(a, b) = sqrt(Σ(a_i - b_i)^2)
范围: [0, +∞)，越小越相似
对向量长度敏感
3. 内积 (Inner Product / Dot Product):
ip(a, b) = a · b
范围: (-∞, +∞)，越大越相似
当向量已归一化时，等价于余弦相似度
选择建议：
- 默认用余弦相似度（最鲁棒）
- 向量已归一化时用内积（计算更快）
- 需要考虑向量长度差异时用欧氏距离
"""
a = np.array([1.0, 2.0, 3.0])
b = np.array([1.1, 2.1, 3.1])
c = np.array([0.0, 0.0, 0.0])
cosine_ab = np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
l2_ab = np.linalg.norm(a - b)
ip_ab = np.dot(a, b)
print(f"余弦相似度(a,b): {cosine_ab:.6f}")
print(f"欧氏距离(a,b): {l2_ab:.6f}")
print(f"内积(a,b): {ip_ab:.6f}")
compare_distance_metrics()Chroma：原型与本地实验
import chromadb
client = chromadb.PersistentClient(path="./chroma_db")
collection = client.get_or_create_collection(
name="tech_docs",
metadata={"hnsw:space": "cosine"}
)
collection.add(
ids=["doc1", "doc2", "doc3"],
documents=[
"ASP.NET Core 支持依赖注入和中间件管道",
"Docker 可以把应用及其依赖一起容器化部署",
"Redis 常用于缓存、排行榜和分布式锁"
],
metadatas=[
{"category": "dotnet", "source": "docs"},
{"category": "devops", "source": "docs"},
{"category": "database", "source": "docs"}
]
)results = collection.query(
query_texts=["如何处理 HTTP 请求"],
n_results=2,
where={"category": "dotnet"}
)
print(results["documents"])
print(results["distances"])Chroma 适合：
- 本地原型验证
- 小规模知识库
- 开发阶段快速试验Milvus / Qdrant 等生产型思路
from pymilvus import MilvusClient, DataType
client = MilvusClient(uri="http://localhost:19530")
schema = client.create_schema(auto_id=True, enable_dynamic_field=True)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=1536)
schema.add_field("text", DataType.VARCHAR, max_length=1000)
schema.add_field("category", DataType.VARCHAR, max_length=100)
index_params = client.prepare_index_params()
index_params.add_index(
field_name="embedding",
index_type="IVF_FLAT",
metric_type="COSINE",
params={"nlist": 128}
)
client.create_collection("knowledge", schema=schema, index_params=index_params)query_vector = get_embedding("如何拆分微服务")
results = client.search(
collection_name="knowledge",
data=[query_vector],
limit=5,
output_fields=["text", "category"],
filter='category == "architecture"'
)
print(results)常见生产向量库：
- Chroma：轻量原型
- Qdrant：工程化较友好，使用广泛
- Milvus：大规模向量检索能力强
- Pinecone：云托管方案
- Weaviate：带更丰富语义层支持ANN 索引算法详解
def explain_ann_algorithms():
"""ANN（近似最近邻）索引算法
1. HNSW (Hierarchical Navigable Small World):
- 基于图的多层索引结构
- 查询速度快，召回率高
- 内存占用较大
- 适合: 百万级以内数据，实时检索
- 参数: M (每层连接数), efConstruction (构建精度), ef (查询精度)
2. IVF (Inverted File Index):
- 将向量聚类为多个桶，查询时只搜索最近的桶
- 需要预先训练聚类中心
- 适合: 百万到千万级数据
- 参数: nlist (桶数), nprobe (搜索桶数)
3. PQ (Product Quantization):
- 将高维向量压缩为低维编码
- 大幅减少内存占用
- 会有精度损失
- 适合: 亿级以上数据，内存受限场景
4. IVF_PQ:
- IVF + PQ 的组合
- 先用 IVF 缩小搜索范围，再用 PQ 压缩存储
精度与性能的权衡：
- 精度最高: 暴力搜索 (100% 召回率)
- HNSW: 95-99% 召回率，延迟 ~1ms
- IVF_FLAT: 90-95% 召回率
- IVF_PQ: 80-90% 召回率，内存最小
"""
print("ANN 索引选择:")
print(" < 10万向量: HNSW (简单高效)")
print(" 10万-100万: HNSW 或 IVF_FLAT")
print(" 100万-1亿: IVF_FLAT 或 IVF_PQ")
print(" > 1亿: IVF_PQ + 分布式部署")
explain_ann_algorithms()RAG 场景中的实际链路
# 简化示意：RAG 不是"只把文档塞进向量库"
rag_pipeline = {
"step1": "文档清洗与切分",
"step2": "embedding 向量化",
"step3": "写入向量数据库",
"step4": "用户提问向量化",
"step5": "相似检索 + 元数据过滤",
"step6": "重排（可选）",
"step7": "拼上下文给 LLM 生成答案"
}
print(rag_pipeline)# 元数据过滤通常非常重要
results = collection.query(
query_texts=["退款政策是什么"],
n_results=3,
where={
"$and": [
{"category": "policy"},
{"lang": "zh"}
]
}
)向量数据库只解决"相似内容找出来"，
不自动解决：
- 文档怎么切
- 检索是否真的命中
- 排序是否合理
- 最终回答是否可信向量数据库选型对比
def vector_db_comparison():
"""向量数据库选型对比"""
print("向量数据库对比:")
print(" Chroma: 原型开发, Python 友好, 功能简单")
print(" Qdrant: 生产推荐, Rust 实现, 混合检索")
print(" Milvus: 大规模, 分布式, 功能最全")
print(" Pinecone: 云托管, 零运维, 按量付费")
print(" Weaviate: 多模态, GraphQL, 内置向量化")
print(" FAISS: Meta 开源库, 非 DB, 纯计算")
print("\n选择建议:")
print(" PoC: Chroma / FAISS")
print(" 中小规模生产: Qdrant")
print(" 大规模: Milvus")
print(" 不想运维: Pinecone")
vector_db_comparison()优点
缺点
总结
向量数据库最值得理解的不是产品名称，而是"向量检索在系统中的位置"。它适合做语义召回，但不能替代分块策略、元数据设计、重排和生成控制。真正做 RAG 或语义搜索时，向量数据库只是基础设施中的一层，而不是整套方案本身。
关键知识点
- 嵌入质量决定了语义空间质量。
- 向量检索常常要和元数据过滤一起使用。
- ANN 是性能与精度之间的折中，不是绝对精确检索。
- RAG 的效果不好，问题不一定在向量库本身，也可能在切分和召回策略。
项目落地视角
- 企业知识问答、FAQ 检索、产品文档搜索都很适合向量检索。
- 推荐系统可把用户行为与内容向量一起组合使用。
- 多模态检索中，图像和文本也能统一映射到向量空间。
- 生产环境通常要同时考虑向量库、对象存储、元数据数据库和模型服务。
常见误区
- 以为接入向量数据库就自然变成"高质量 RAG"。
- 不做 chunk 策略设计，直接整篇文档 embedding。
- 只看 Demo 效果，不建立评估集和失败样例库。
- 把所有责任都交给向量检索，忽略重排和输出校验。
进阶路线
- 深入学习 HNSW、IVF、PQ 等 ANN 索引原理。
- 研究混合检索（BM25 + 向量）和 reranker。
- 建立向量检索评估集，量化 recall / precision / mrr。
- 结合知识图谱、权限过滤和在线反馈做更完整的检索体系。
适用场景
- RAG 知识问答。
- 语义搜索。
- 推荐系统召回层。
- 多模态检索和内容相似度匹配。
落地建议
- 先明确你的目标是搜索、推荐还是 RAG，不同目标设计不同。
- 优先建立评估集，而不是只看偶发成功案例。
- 元数据过滤要和向量检索一起设计，不要单独看。
- 对 embedding 模型、分块策略和向量库索引参数做版本管理。
排错清单
- 搜索不准：先看 chunk 是否合理、embedding 是否匹配语言/领域。
- 召回太慢：检查索引类型、向量维度和候选数设置。
- 成本过高：检查是否存了太多重复 chunk、是否频繁重建索引。
- RAG 回答差：检查是召回问题、重排问题，还是生成问题。
复盘问题
- 当前使用的向量库和索引类型是什么？为什么选择它？
- 检索召回率是否经过量化评估？
- embedding 模型的维度和语言是否与业务匹配？
Qdrant 生产实践
from qdrant_client import QdrantClient
from qdrant_client.models import (
Distance, VectorParams, PointStruct,
Filter, FieldCondition, MatchValue, Range
)
client = QdrantClient(host="localhost", port=6333)
# 创建集合
client.create_collection(
collection_name="knowledge_base",
vectors_config=VectorParams(
size=1536, # OpenAI text-embedding-3-small 维度
distance=Distance.COSINE
),
# 可选：启用 HNSW 参数调优
hnsw_config={
"m": 16, # 每层最大连接数
"ef_construct": 100 # 构建时搜索宽度
}
)
# 批量写入
from openai import OpenAI
openai_client = OpenAI(api_key="your-api-key")
documents = [
{"id": 1, "text": "ASP.NET Core 中间件管道", "category": "dotnet"},
{"id": 2, "text": "Docker 容器化部署", "category": "devops"},
{"id": 3, "text": "Redis 缓存策略", "category": "database"},
]
points = []
for doc in documents:
embedding = openai_client.embeddings.create(
input=[doc["text"]],
model="text-embedding-3-small"
).data[0].embedding
points.append(PointStruct(
id=doc["id"],
vector=embedding,
payload={
"text": doc["text"],
"category": doc["category"]
}
))
client.upsert(
collection_name="knowledge_base",
points=points
)
# 带过滤的搜索
query_embedding = openai_client.embeddings.create(
input=["如何处理 HTTP 请求"],
model="text-embedding-3-small"
).data[0].embedding
results = client.search(
collection_name="knowledge_base",
query_vector=query_embedding,
limit=5,
query_filter=Filter(
must=[
FieldCondition(key="category", match=MatchValue(value="dotnet"))
]
),
score_threshold=0.7 # 相似度阈值
)
for hit in results:
print(f"score={hit.score:.4f}, text={hit.payload['text']}")向量数据库性能调优
def explain_vector_db_tuning():
"""向量数据库性能调优
1. HNSW 参数调优：
- M（每层连接数）：
默认 16，增大到 32-64 可提高召回率，但增加内存
- efConstruction（构建时搜索宽度）：
默认 100，增大可提高索引质量，但构建更慢
- ef（查询时搜索宽度）：
默认 128，增大可提高召回率，但查询更慢
生产环境建议 64-256
2. 索引选择：
- < 10 万向量：HNSW（简单高效，内存足够）
- 10-100 万：HNSW 或 IVF_FLAT
- 100 万-1 亿：IVF_FLAT 或 IVF_PQ
- > 1 亿：IVF_PQ + 分布式部署
3. 写入优化：
- 批量写入优于逐条写入（10-100 倍性能差距）
- 建议每批 100-1000 条
- 写入时暂时关闭索引构建，写完后再重建
4. 内存优化：
- PQ 量化可减少 70-90% 内存
- 标量量化（SQ）精度损失更小
- 二值量化（BQ）最省内存，但精度损失最大
5. 查询优化：
- 使用 score_threshold 过滤低分结果
- 利用 metadata 过滤减少搜索范围
- 调整 ef 参数平衡精度和速度
"""
print("调优建议:")
print(" 小规模: HNSW, M=16, ef=128")
print(" 大规模: IVF_PQ, nlist=4096, nprobe=64")
print(" 写入: 批量写入，每批 100+ 条")
print(" 查询: metadata 过滤 + score_threshold")
explain_vector_db_tuning()向量检索评估体系
def evaluate_vector_search():
"""向量检索质量评估
评估指标：
1. Recall@K: 相关文档是否在 Top-K 中被召回
2. Precision@K: Top-K 中相关文档的比例
3. MRR (Mean Reciprocal Rank): 第一个相关文档的排名倒数均值
4. NDCG: 考虑位置权重的归一化折损累计增益
评估流程：
1. 准备评估集（query + 相关文档列表）
2. 对每个 query 执行检索
3. 对比检索结果与标注的相关文档
4. 计算 Recall、Precision、MRR、NDCG
"""
# 示例评估代码
def recall_at_k(retrieved_ids, relevant_ids, k):
retrieved_top_k = set(retrieved_ids[:k])
relevant_set = set(relevant_ids)
return len(retrieved_top_k & relevant_set) / len(relevant_set)
def precision_at_k(retrieved_ids, relevant_ids, k):
retrieved_top_k = set(retrieved_ids[:k])
relevant_set = set(relevant_ids)
return len(retrieved_top_k & relevant_set) / k
def mrr(retrieved_ids, relevant_ids):
relevant_set = set(relevant_ids)
for i, doc_id in enumerate(retrieved_ids):
if doc_id in relevant_set:
return 1.0 / (i + 1)
return 0.0
# 模拟评估
retrieved = [1, 5, 3, 8, 2, 7, 4, 6, 9, 10]
relevant = [1, 2, 3, 7]
print(f"Recall@5: {recall_at_k(retrieved, relevant, 5):.2f}") # 2/4 = 0.50
print(f"Recall@10: {recall_at_k(retrieved, relevant, 10):.2f}") # 4/4 = 1.00
print(f"P@5: {precision_at_k(retrieved, relevant, 5):.2f}") # 2/5 = 0.40
print(f"MRR: {mrr(retrieved, relevant):.2f}") # 1/1 = 1.00
# 评估集建议：
# - 至少 200-500 条查询
# - 每条查询标注 3-10 个相关文档
# - 覆盖不同类型的问题（事实型、推理型、模糊型）
# - 定期更新评估集以反映真实查询分布
evaluate_vector_search()向量数据库运维实践
def explain_vector_db_operations():
"""向量数据库生产运维
1. 数据备份：
- 定期快照（Qdrant 支持 snapshot API）
- 跨区域复制（Milvus 支持跨集群同步）
- 元数据备份（独立于向量数据的 payload）
2. 索引重建：
- 大量写入后建议重建索引
- 批量删除后重建索引恢复性能
- 索引重建期间服务可用性（后台重建）
3. 监控指标：
- 查询延迟 P50/P95/P99
- 查询 QPS 和吞吐量
- 索引内存占用
- 集合大小和向量数量
- 写入延迟和写入 QPS
4. 容量规划：
- 内存 = 向量数量 × 维度 × 4 字节 × HNSW 开销倍数(约 1.5-2x)
- 磁盘 = 向量数量 × 维度 × 4 字节 + 元数据大小
- 例：100 万条 1536 维向量 ≈ 100 万 × 1536 × 4 × 2 ≈ 12 GB 内存
5. 故障排查：
- 查询慢：检查 ef 参数、索引类型、过滤条件
- 召回差：检查 embedding 模型、分块策略、维度匹配
- 内存高：考虑 PQ 量化、删除无用集合
"""
print("运维建议:")
print(" 备份: 每日快照 + 异地存储")
print(" 监控: 延迟 + QPS + 内存 + 磁盘")
print(" 容量: 100万×1536维 ≈ 12GB 内存")
print(" 重建: 大量写入/删除后重建索引")
explain_vector_db_operations()