Jersey's Blog

学习时长：3-4 周

向量数据库是 RAG 系统的核心基础设施，选择合适的向量数据库和 Embedding 模型能显著影响检索质量和系统性能。

4.2.1 Embedding 模型深入

Embedding 模型将文本转换为高维向量（通常 768-3072 维），语义相似的文本在向量空间中距离更近。

主流 Embedding 模型详细对比

模型	开发者	维度	语言	MTEB 得分	价格	部署方式
text-embedding-3-small	OpenAI	1536	多语言	62.3	$0.02/1M tokens	API 调用
text-embedding-3-large	OpenAI	3072	多语言	64.6	$0.13/1M tokens	API 调用
BGE-large-zh-v1.5	智源（BAAI）	1024	中文优化	64.5	免费	本地部署
BGE-M3	智源	1024	100+语言	66.1	免费	本地部署
Jina-embeddings-v2	Jina AI	768	多语言	60.4	免费（有限）	API/本地
Cohere Embed v3	Cohere	1024	多语言	62.8	$0.10/1M tokens	API 调用

选型建议

快速原型开发：OpenAI text-embedding-3-small（API 稳定，文档完善）
中文场景：BGE-large-zh-v1.5（专门优化中文语义）
隐私要求高：BGE-M3 本地部署（完全私有化）
多语言混合：BGE-M3 或 Jina-embeddings-v2
预算充足：OpenAI text-embedding-3-large（最高精度）

1. OpenAI Embedding（推荐起步）

python

from openai import OpenAI

client = OpenAI(api_key="your-api-key")

# 单个文本向量化
response = client.embeddings.create(
    model="text-embedding-3-small",
    input="人工智能正在改变世界"
)
embedding = response.data[0].embedding

print(f"维度：{len(embedding)}")  # 1536
print(f"前5个值：{embedding[:5]}")

# 批量向量化（推荐：减少API调用次数）
texts = [
    "深度学习是机器学习的子领域",
    "神经网络模拟人脑结构",
    "Transformer 架构革新了 NLP"
]

response = client.embeddings.create(
    model="text-embedding-3-small",
    input=texts
)

embeddings = [data.embedding for data in response.data]
print(f"批量生成 {len(embeddings)} 个向量")

降维优化（减少存储成本）：

python

# text-embedding-3-small 支持降维（1536 → 512）
response = client.embeddings.create(
    model="text-embedding-3-small",
    input="测试文本",
    dimensions=512  # 降维到 512，存储成本降低 70%
)

# 注意：降维会轻微损失精度（约 1-2%），但显著减少存储和计算成本

2. BGE 本地部署（隐私优先）

python

from sentence_transformers import SentenceTransformer
import numpy as np

# 下载并加载模型（首次运行会自动下载到 ~/.cache）
model = SentenceTransformer('BAAI/bge-large-zh-v1.5')

# 单个文本
text = "RAG 技术结合了检索和生成"
embedding = model.encode(text)

print(f"维度：{embedding.shape}")  # (1024,)

# 批量文本
texts = [
    "向量数据库存储高维向量",
    "相似度检索使用余弦距离",
    "ANN 算法加速大规模检索"
]

embeddings = model.encode(texts, batch_size=32, show_progress_bar=True)
print(f"批量生成：{embeddings.shape}")  # (3, 1024)

# 计算相似度
from sklearn.metrics.pairwise import cosine_similarity

similarity = cosine_similarity([embeddings[0]], [embeddings[1]])[0][0]
print(f"文本1和文本2的相似度：{similarity:.4f}")  # 0.72

BGE 特殊技巧：查询文本加前缀

python

# BGE 模型推荐为查询（Query）添加前缀
query = "为用户输入的指令添加前缀"
passage = "知识库中的文档内容"

# 查询文本加前缀（提升检索效果 2-3%）
query_embedding = model.encode(f"为这个句子生成表示以用于检索相关文章：{query}")
passage_embedding = model.encode(passage)  # 文档不加前缀

similarity = cosine_similarity([query_embedding], [passage_embedding])[0][0]

3. 多语言 Embedding（BGE-M3）

python

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('BAAI/bge-m3')

# 多语言文本混合
texts = [
    "Machine learning is a subset of AI",  # 英文
    "机器学习是人工智能的子领域",  # 中文
    "機械学習は人工知能のサブセット",  # 日文
    "El aprendizaje automático es un subconjunto de IA"  # 西班牙文
]

embeddings = model.encode(texts)

# 跨语言语义相似度检索
from sklearn.metrics.pairwise import cosine_similarity

# 英文查询，检索中文文档
similarity_matrix = cosine_similarity(embeddings)
print("跨语言相似度矩阵：")
print(similarity_matrix)
# 英文和中文语义相似度 > 0.85

4.2.2 向量数据库选型与实战

主流向量数据库对比

数据库	类型	性能	易用性	成本	适用场景
ChromaDB	嵌入式	⭐⭐⭐	⭐⭐⭐⭐⭐	免费	原型开发、小规模应用（<100万向量）
FAISS	库	⭐⭐⭐⭐⭐	⭐⭐	免费	高性能本地检索、科研项目
Milvus	分布式	⭐⭐⭐⭐⭐	⭐⭐⭐	免费（自建）	生产环境、大规模应用（亿级向量）
Pinecone	云服务	⭐⭐⭐⭐	⭐⭐⭐⭐⭐	付费	快速上线、无运维需求
Weaviate	分布式	⭐⭐⭐⭐	⭐⭐⭐⭐	免费（自建）	混合检索、GraphQL 查询
Qdrant	分布式	⭐⭐⭐⭐	⭐⭐⭐⭐	免费（自建）	Rust 实现，性能优秀

决策树

是否需要云托管？
├─ 是 → Pinecone（简单）/ Weaviate Cloud（开源）
└─ 否 → 是否需要分布式部署？
    ├─ 是 → Milvus（亿级数据）/ Qdrant（性能优先）
    └─ 否 → 向量规模？
        ├─ <10万 → ChromaDB（最简单）
        └─ >10万 → FAISS（最快）

1. ChromaDB 快速上手（推荐入门）

python

import chromadb
from chromadb.config import Settings

# 1. 创建客户端（持久化到本地）
client = chromadb.PersistentClient(path="./chroma_db")

# 2. 创建集合（Collection）
collection = client.get_or_create_collection(
    name="my_documents",
    metadata={"description": "技术文档知识库"}
)

# 3. 添加文档
documents = [
    "Python 是一门高级编程语言",
    "JavaScript 用于前端开发",
    "SQL 是数据库查询语言",
    "Docker 用于容器化部署"
]

collection.add(
    documents=documents,
    ids=[f"doc_{i}" for i in range(len(documents))],
    metadatas=[{"category": "编程语言"} for _ in documents]
)

# 4. 查询（自动向量化）
results = collection.query(
    query_texts=["什么是 Python？"],
    n_results=2
)

print("检索结果：")
for i, doc in enumerate(results['documents'][0]):
    print(f"{i+1}. {doc}")
    print(f"   距离：{results['distances'][0][i]:.4f}")

# 5. 带过滤条件的查询
results = collection.query(
    query_texts=["编程语言"],
    n_results=3,
    where={"category": "编程语言"}
)

# 6. 删除文档
collection.delete(ids=["doc_0"])

# 7. 更新文档
collection.update(
    ids=["doc_1"],
    documents=["JavaScript 是最流行的前端开发语言"],
    metadatas=[{"category": "编程语言", "popularity": "high"}]
)

ChromaDB 自定义 Embedding 函数：

python

from sentence_transformers import SentenceTransformer
from chromadb.utils import embedding_functions

# 使用 BGE 模型
bge_model = SentenceTransformer('BAAI/bge-large-zh-v1.5')

# 自定义 Embedding 函数
class BGEEmbeddingFunction(embedding_functions.EmbeddingFunction):
    def __call__(self, texts):
        return bge_model.encode(texts).tolist()

# 创建集合时指定 Embedding 函数
collection = client.create_collection(
    name="bge_collection",
    embedding_function=BGEEmbeddingFunction()
)

collection.add(
    documents=["向量数据库", "语义检索"],
    ids=["doc_1", "doc_2"]
)

2. FAISS 高性能检索（Meta 开源）

python

import faiss
import numpy as np

# 假设已有 embeddings（numpy 数组）
embeddings = np.random.random((10000, 1024)).astype('float32')  # 1万个向量
dimension = embeddings.shape[1]

# 1. 创建索引（Flat = 暴力搜索，精确但慢）
index = faiss.IndexFlatL2(dimension)  # L2 距离
index.add(embeddings)  # 添加向量

print(f"索引包含 {index.ntotal} 个向量")

# 2. 检索 Top-K
query = np.random.random((1, 1024)).astype('float32')
k = 5  # 返回最相似的5个

distances, indices = index.search(query, k)

print(f"Top-{k} 索引：{indices[0]}")
print(f"距离：{distances[0]}")

# 3. 高性能索引（IVF + PQ 压缩）
# 适合百万级以上数据
nlist = 100  # 聚类中心数量
m = 8        # PQ 压缩参数

quantizer = faiss.IndexFlatL2(dimension)
index = faiss.IndexIVFPQ(quantizer, dimension, nlist, m, 8)

# 训练索引（需要训练数据）
index.train(embeddings[:5000])  # 用5000个向量训练
index.add(embeddings)

# 设置搜索范围（nprobe 越大越精确但越慢）
index.nprobe = 10
distances, indices = index.search(query, k)

# 4. 保存和加载索引
faiss.write_index(index, "my_index.faiss")
loaded_index = faiss.read_index("my_index.faiss")

FAISS 索引类型选择：

索引类型	精度	速度	内存占用	适用数据量
IndexFlatL2	100%	慢	高	<10万
IndexIVFFlat	95-99%	快	高	10万-100万
IndexIVFPQ	90-95%	最快	低（压缩）	>100万
IndexHNSW	99%	快	中	10万-1000万

3. Milvus 企业级部署

python

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

# 1. 连接 Milvus（Docker 启动）
connections.connect("default", host="localhost", port="19530")

# 2. 定义 Schema
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535),
    FieldSchema(name="category", dtype=DataType.VARCHAR, max_length=100)
]

schema = CollectionSchema(fields, description="技术文档集合")
collection = Collection("tech_docs", schema)

# 3. 创建索引
index_params = {
    "index_type": "IVF_FLAT",
    "metric_type": "L2",
    "params": {"nlist": 128}
}

collection.create_index(field_name="embedding", index_params=index_params)

# 4. 插入数据
import random

data = [
    [random.random() for _ in range(1024)] for _ in range(1000)
]

entities = [
    data,  # embeddings
    ["文档内容..." for _ in range(1000)],  # text
    ["Python" if i % 2 == 0 else "JavaScript" for i in range(1000)]  # category
]

collection.insert(entities)
collection.load()  # 加载到内存

# 5. 搜索
search_params = {"metric_type": "L2", "params": {"nprobe": 10}}

query_vector = [random.random() for _ in range(1024)](random.random() for _ in range(1024))
results = collection.search(
    data=query_vector,
    anns_field="embedding",
    param=search_params,
    limit=5,
    expr="category == 'Python'"  # 过滤条件
)

for hits in results:
    for hit in hits:
        print(f"ID: {hit.id}, 距离: {hit.distance:.4f}")

4. Pinecone 云托管（零运维）

python

from pinecone import Pinecone, ServerlessSpec

# 1. 初始化（需要 API Key）
pc = Pinecone(api_key="your-api-key")

# 2. 创建索引
index_name = "tech-docs"

if index_name not in pc.list_indexes().names():
    pc.create_index(
        name=index_name,
        dimension=1536,
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1")
    )

index = pc.Index(index_name)

# 3. 插入向量
vectors = [
    ("id_1", [0.1] * 1536, {"text": "Python 教程", "category": "编程"}),
    ("id_2", [0.2] * 1536, {"text": "JavaScript 指南", "category": "编程"}),
]

index.upsert(vectors=vectors)

# 4. 查询
query_vector = [0.15] * 1536

results = index.query(
    vector=query_vector,
    top_k=5,
    include_metadata=True,
    filter={"category": {"$eq": "编程"}}
)

for match in results['matches']:
    print(f"ID: {match['id']}, 分数: {match['score']:.4f}")
    print(f"元数据: {match['metadata']}")

# 5. 删除向量
index.delete(ids=["id_1"])

# 6. 查看统计
stats = index.describe_index_stats()
print(f"总向量数：{stats['total_vector_count']}")

4.2.3 相似度检索算法

1. 余弦相似度（Cosine Similarity）

最常用的相似度度量，范围 [-1, 1]，值越大越相似。

python

import numpy as np

def cosine_similarity(vec1, vec2):
    """计算余弦相似度"""
    dot_product = np.dot(vec1, vec2)
    norm1 = np.linalg.norm(vec1)
    norm2 = np.linalg.norm(vec2)
    return dot_product / (norm1 * norm2)

vec1 = np.array([1, 2, 3])
vec2 = np.array([2, 4, 6])  # vec1 的2倍，方向相同

similarity = cosine_similarity(vec1, vec2)
print(f"余弦相似度：{similarity:.4f}")  # 1.0（完全相同方向）

2. 欧氏距离（L2 Distance）

向量空间中的直线距离，距离越小越相似。

python

def euclidean_distance(vec1, vec2):
    """计算欧氏距离"""
    return np.linalg.norm(vec1 - vec2)

distance = euclidean_distance(vec1, vec2)
print(f"欧氏距离：{distance:.4f}")

3. 点积（Dot Product）

归一化向量的点积等价于余弦相似度，计算速度更快。

python

def dot_product(vec1, vec2):
    """计算点积"""
    return np.dot(vec1, vec2)

# 注意：需要先归一化向量
vec1_norm = vec1 / np.linalg.norm(vec1)
vec2_norm = vec2 / np.linalg.norm(vec2)

dot = dot_product(vec1_norm, vec2_norm)
print(f"点积（归一化后）：{dot:.4f}")  # 等于余弦相似度

选择建议

余弦相似度：推荐，对向量长度不敏感，适合文本语义
欧氏距离：关注绝对差异，图像检索常用
点积：归一化后等价于余弦，但计算更快

4. ANN（近似最近邻）算法

精确检索（Flat Search）在百万级数据时太慢，ANN 算法牺牲少量精度换取巨大速度提升。

主流 ANN 算法：

算法	原理	精度	速度	内存
HNSW	层次化图结构	99%	⭐⭐⭐⭐⭐	高
IVF	倒排索引 + 聚类	95-98%	⭐⭐⭐⭐	中
PQ	乘积量化压缩	90-95%	⭐⭐⭐	低
ScaNN	Google 高性能	98%	⭐⭐⭐⭐⭐	中

HNSW 示例（推荐）：

python

import faiss

dimension = 1024
num_vectors = 100000

# 生成测试数据
vectors = np.random.random((num_vectors, dimension)).astype('float32')

# 创建 HNSW 索引
index = faiss.IndexHNSWFlat(dimension, 32)  # 32 = M参数（邻居数量）

index.add(vectors)

# 搜索
query = np.random.random((1, dimension)).astype('float32')
k = 10

distances, indices = index.search(query, k)

print(f"检索 {num_vectors} 个向量，返回 Top-{k}")
print(f"最近邻索引：{indices[0]}")

4.2.4 性能优化技巧

1. 向量归一化（提升检索速度）

python

# 归一化向量后，余弦相似度 = 点积，计算更快
def normalize_vectors(vectors):
    norms = np.linalg.norm(vectors, axis=1, keepdims=True)
    return vectors / norms

normalized = normalize_vectors(embeddings)

# 使用点积索引（比余弦更快）
index = faiss.IndexFlatIP(dimension)  # IP = Inner Product
index.add(normalized)

2. 批量插入（减少网络开销）

python

# ❌ 糟糕：逐个插入
for doc in documents:
    collection.add(documents=[doc], ids=[doc['id']])

# ✅ 优秀：批量插入
batch_size = 100
for i in range(0, len(documents), batch_size):
    batch = documents[i:i+batch_size]
    collection.add(
        documents=[d['text'] for d in batch],
        ids=[d['id'] for d in batch]
    )

3. 索引参数调优

python

# Milvus IVF 参数调优
index_params = {
    "index_type": "IVF_FLAT",
    "metric_type": "IP",
    "params": {
        "nlist": 1024  # 聚类中心数量
        # nlist 建议值：sqrt(向量总数)
        # 100万向量 → nlist = 1024
        # 1000万向量 → nlist = 4096
    }
}

search_params = {
    "metric_type": "IP",
    "params": {
        "nprobe": 64  # 搜索的聚类中心数量
        # nprobe 越大越精确但越慢
        # 推荐：nprobe = nlist / 16
    }
}

4. 混合精度存储

python

# 使用 FP16 或 INT8 量化减少内存（损失 <1% 精度）
import faiss

# FP32 → FP16（内存减半）
index_fp16 = faiss.IndexFlatL2(dimension)
index_fp16 = faiss.IndexRefineFlat(index_fp16, index)

# 乘积量化（PQ）压缩 10-20 倍
m = 8  # 子向量数量
index_pq = faiss.IndexPQ(dimension, m, 8)
index_pq.train(vectors[:10000])  # 需要训练
index_pq.add(vectors)

4.2.5 实战建议与学习路径

Week 1：基础概念

理解 Embedding 原理，对比不同模型效果
使用 ChromaDB 构建第一个向量检索系统
实现余弦相似度、欧氏距离计算

Week 2：进阶实践

本地部署 BGE 模型，对比 OpenAI Embedding
学习 FAISS，实现高性能检索
测试不同 ANN 算法（HNSW vs IVF）

Week 3-4：生产级优化

Docker 部署 Milvus，实现分布式检索
性能调优：批量插入、索引参数、向量压缩
监控检索质量（精度、召回率、延迟）

最佳实践清单

✅ 选择合适的 Embedding 模型：中文优先 BGE，多语言用 M3
✅ 向量归一化：使用点积代替余弦，性能提升 30%
✅ 批量操作：减少网络往返，批量插入/查询
✅ 索引优化：数据量 >10万必须用 ANN 索引
✅ 监控指标：QPS、P99 延迟、召回率
✅ 定期重建索引：数据变化 >20% 时重建以维持性能

常见错误

❌ 向量维度不匹配（插入 768 维，查询 1536 维）
❌ 未归一化向量就使用点积（结果错误）
❌ FAISS 索引未训练就添加数据（IVF/PQ 需训练）
❌ 生产环境用 Flat 索引（百万级数据会超时）

实战项目

项目1：构建本地文档问答系统

使用 ChromaDB + BGE 模型
支持 PDF、Word、Markdown 文档
实现语义检索和相似度排序

项目2：高性能检索服务

使用 FAISS 构建百万级向量检索
对比不同索引类型的性能
实现 API 服务封装

项目3：企业级知识库

使用 Milvus 部署分布式向量数据库
实现混合检索（向量 + 关键词）
监控系统性能和检索质量

总结

向量数据库是 RAG 系统的核心组件，选择合适的技术栈需要考虑：

数据规模：小规模用 ChromaDB，大规模用 Milvus
部署方式：云托管用 Pinecone，自建用开源方案
性能要求：高性能场景必须使用 ANN 索引
成本预算：API 服务按量付费，自建需要服务器成本

掌握向量数据库技术后，你将能够构建高效的语义检索系统，为 RAG 应用打下坚实基础。

4.2.1 Embedding 模型深入 ​

主流 Embedding 模型详细对比 ​

选型建议 ​

1. OpenAI Embedding（推荐起步） ​

2. BGE 本地部署（隐私优先） ​

3. 多语言 Embedding（BGE-M3） ​

4.2.2 向量数据库选型与实战 ​

主流向量数据库对比 ​

决策树 ​

1. ChromaDB 快速上手（推荐入门） ​

2. FAISS 高性能检索（Meta 开源） ​

3. Milvus 企业级部署 ​

4. Pinecone 云托管（零运维） ​

4.2.3 相似度检索算法 ​

1. 余弦相似度（Cosine Similarity） ​

2. 欧氏距离（L2 Distance） ​

3. 点积（Dot Product） ​

选择建议 ​

4. ANN（近似最近邻）算法 ​

4.2.4 性能优化技巧 ​

1. 向量归一化（提升检索速度） ​

2. 批量插入（减少网络开销） ​

3. 索引参数调优 ​

4. 混合精度存储 ​

4.2.5 实战建议与学习路径 ​

Week 1：基础概念 ​

Week 2：进阶实践 ​

Week 3-4：生产级优化 ​

最佳实践清单 ​

常见错误 ​

推荐资源 ​

实战项目 ​

项目1：构建本地文档问答系统 ​

项目2：高性能检索服务 ​

项目3：企业级知识库 ​

总结 ​

4.2.1 Embedding 模型深入

主流 Embedding 模型详细对比

选型建议

1. OpenAI Embedding（推荐起步）

2. BGE 本地部署（隐私优先）

3. 多语言 Embedding（BGE-M3）

4.2.2 向量数据库选型与实战

主流向量数据库对比

决策树

1. ChromaDB 快速上手（推荐入门）

2. FAISS 高性能检索（Meta 开源）

3. Milvus 企业级部署

4. Pinecone 云托管（零运维）

4.2.3 相似度检索算法

1. 余弦相似度（Cosine Similarity）

2. 欧氏距离（L2 Distance）

3. 点积（Dot Product）

选择建议

4. ANN（近似最近邻）算法

4.2.4 性能优化技巧

1. 向量归一化（提升检索速度）

2. 批量插入（减少网络开销）

3. 索引参数调优

4. 混合精度存储

4.2.5 实战建议与学习路径

Week 1：基础概念

Week 2：进阶实践

Week 3-4：生产级优化

最佳实践清单

常见错误

推荐资源

实战项目

项目1：构建本地文档问答系统

项目2：高性能检索服务

项目3：企业级知识库

总结