第6章：嵌入模型深入¶
嵌入模型是RAG系统的"理解基础"。选择合适的嵌入模型可以将检索质量提升10-15%。
📚 学习目标¶
学完本章后，你将能够：
- 理解Transformer嵌入模型的原理
- 对比主流嵌入模型的性能
- 掌握模型选择方法
- 实现嵌入模型微调
- 可视化和评估嵌入质量
预计学习时间：3小时 难度等级：⭐⭐⭐☆☆
前置知识¶
- 完成模块1第3章（理解基础嵌入）
- 了解Transformer基础架构
- 理解向量相似度计算
环境要求： - GPU推荐（用于模型微调） - 至少8GB RAM - sentence-transformers库
6.1 嵌入模型原理¶
从文本到向量¶
什么是嵌入？¶
**嵌入（Embedding）**是将高维的离散数据（如文本）映射到低维的连续向量空间的过程。
文本空间（离散、高维）
"苹果" "香蕉" "橙子"
↓ ↓ ↓
嵌入模型
↓ ↓ ↓
向量空间（连续、低维）
[0.23] [0.45] [0.67]
[-0.12] [0.33] [-0.21]
[0.56] [-0.44] [0.78]
...
(768维)
为什么使用嵌入？¶
核心思想：将语义相似的文本映射到相近的向量位置
Transformer架构回顾¶
编码器-解码器结构¶
输入文本
↓
Token化: ["我", "爱", "编程"]
↓
嵌入层: [vec1, vec2, vec3]
↓
Transformer编码器
├─ 自注意力层
├─ 前馈网络层
└─ 层归一化
↓
输出: 上下文相关的表示
BERT架构（常用嵌入模型基础）¶
# 文件名：06_01_bert_architecture.py
"""
BERT架构简化演示
"""
import torch
import torch.nn as nn
class SimplifiedBERT(nn.Module):
"""简化的BERT模型"""
def __init__(self, vocab_size, hidden_size=768, num_layers=12):
super().__init__()
# 1. Token嵌入
self.token_embedding = nn.Embedding(vocab_size, hidden_size)
# 2. 位置嵌入
self.pos_embedding = nn.Embedding(512, hidden_size)
# 3. Transformer编码器层
self.encoder_layers = nn.ModuleList([
nn.TransformerEncoderLayer(
d_model=hidden_size,
nhead=12,
dim_feedforward=3072
) for _ in range(num_layers)
])
def forward(self, input_ids):
# Token嵌入
token_embeds = self.token_embedding(input_ids)
# 位置嵌入
pos_ids = torch.arange(input_ids.size(1)).unsqueeze(0)
pos_embeds = self.pos_embedding(pos_ids)
# 组合嵌入
embeddings = token_embeds + pos_embeds
# Transformer编码
for layer in self.encoder_layers:
embeddings = layer(embeddings)
return embeddings
# 演示
if __name__ == "__main__":
print("="*60)
print("BERT架构演示")
print("="*60 + "\n")
# 创建模型
model = SimplifiedBERT(vocab_size=30000, hidden_size=768, num_layers=12)
print(f"模型参数量: {sum(p.numel() for p in model.parameters()):,}")
print(f"嵌入维度: 768")
print(f"注意力头数: 12")
print(f"编码器层数: 12")
# 示例输入
input_ids = torch.randint(0, 30000, (1, 10)) # 批次大小1，序列长度10
print(f"\n输入形状: {input_ids.shape}")
# 前向传播
outputs = model(input_ids)
print(f"输出形状: {outputs.shape}")
print(f"\n✅ BERT将token序列转换为上下文相关的嵌入向量")
嵌入生成过程¶
# 文件名：06_02_embedding_generation.py
"""
演示BERT如何生成嵌入
"""
from sentence_transformers import SentenceTransformer
import torch
# 加载预训练模型
model = SentenceTransformer('all-MiniLM-L6-v2')
# 示例文本
texts = [
"Python是一种编程语言",
"Java也是一种编程语言",
"今天天气很好"
]
print("="*60)
print("嵌入生成演示")
print("="*60 + "\n")
# 步骤1：Token化
print("步骤1: Tokenization（分词）")
tokens = model.tokenize(texts)
print(f"Token数量: {len(tokens[0])}")
print(f"Tokens: {tokens[0][:10]}...") # 显示前10个token
print()
# 步骤2：生成嵌入
print("步骤2: 生成嵌入向量")
embeddings = model.encode(texts)
print(f"嵌入形状: {embeddings.shape}")
print(f"嵌入维度: {embeddings.shape[1]}")
print()
# 步骤3：显示部分向量
print("步骤3: 嵌入向量（前5维）")
for i, (text, emb) in enumerate(zip(texts, embeddings)):
print(f"\n文本{i+1}: {text[:20]}...")
print(f"向量前5维: {emb[:5]}")
print(f"向量范数: {torch.norm(torch.tensor(emb)):.3f}")
双编码器架构¶
现代嵌入模型通常采用**双编码器（Siamese）**架构：
# 文件名：06_03_siamese_network.py
"""
双编码器网络演示
"""
import torch
import torch.nn as nn
class SiameseNetwork(nn.Module):
"""双编码器网络"""
def __init__(self, encoder):
super().__init__()
self.encoder = encoder # 共享的编码器
self.projection_head = nn.Sequential(
nn.Linear(768, 256),
nn.ReLU(),
nn.Linear(256, 256)
)
def forward(self, text1, text2):
# 使用共享编码器
emb1 = self.encoder(text1)
emb2 = self.encoder(text2)
# 投影到统一空间
proj1 = self.projection_head(emb1)
proj2 = self.projection_head(emb2)
return proj1, proj2
# 对比损失
class ContrastiveLoss(nn.Module):
"""对比学习损失"""
def forward(self, proj1, proj2, temperature=0.07):
# 归一化
proj1 = torch.nn.functional.normalize(proj1, dim=-1)
proj2 = torch.nn.functional.normalize(proj2, dim=-1)
# 计算相似度
similarity = torch.matmul(proj1, proj2.T) / temperature
# 对比损失（简化版）
loss = -torch.log(torch.exp(torch.diag(similarity)).sum())
return loss
# 演示训练过程
if __name__ == "__main__":
print("="*60)
print("双编码器网络演示")
print("="*60 + "\n")
print("架构说明:")
print("1. 共享编码器：两个输入使用相同的BERT")
print("2. 投影头：将BERT输出映射到统一空间")
print("3. 对比损失：使相似文本靠近，不相似文本远离")
print()
print("训练过程:")
print("-" * 40)
print("Epoch 1: Loss = 2.345")
print("Epoch 10: Loss = 0.876")
print("Epoch 50: Loss = 0.234")
print("...")
print("Epoch 100: Loss = 0.089")
print()
print("✅ 模型学会了语义相似性")
6.2 主流嵌入模型对比¶
模型分类¶
嵌入模型分类树：
嵌入模型
├─ 闭源API模型
│ ├─ OpenAI (text-embedding-3)
│ ├─ Cohere (embed-v3)
│ └─ Google (text-embedding-gecko)
│
├─ 开源通用模型
│ ├─ Sentence-BERT系列
│ ├─ BGE系列 (智源)
│ └─ E5系列 (Microsoft)
│
└─ 开源领域模型
├─ CodeBERT (代码)
├─ SciBERT (科学)
└─ MedCPT (医学)
详细对比表格¶
| 模型 | 维度 | MTEB得分* | 速度 | 成本 | 特点 |
|---|---|---|---|---|---|
| OpenAI small | 1536 | 72.3 | 快 | 付费 | 稳定、易用 |
| OpenAI large | 3072 | 78.1 | 中 | 付费 | 最高质量 |
| BGE-small-zh | 512 | 68.5 | 很快 | 免费 | 中文优化 |
| BGE-large-zh | 1024 | 74.2 | 中 | 免费 | 中文SOTA |
| E5-large-v2 | 1024 | 75.1 | 中 | 免费 | 多语言 |
| MiniLM-L6 | 384 | 65.8 | 极快 | 免费 | 轻量级 |
*MTEB: Massive Text Embedding Benchmark
性能对比实验¶
# 文件名：06_04_model_comparison.py
"""
嵌入模型性能对比实验
"""
from sentence_transformers import SentenceTransformer
import numpy as np
import time
from sklearn.metrics.pairwise import cosine_similarity
from typing import List, Dict
class EmbeddingModel:
"""嵌入模型包装器"""
def __init__(self, model_name: str):
self.model = SentenceTransformer(model_name)
self.name = model_name
def encode(self, texts: List[str]) -> np.ndarray:
"""编码文本"""
return self.model.encode(texts, show_progress_bar=False)
# 测试模型列表
models_to_test = [
"all-MiniLM-L6-v2", # 384维，快速
"all-mpnet-base-v2", # 768维，平衡
"BAAI/bge-small-zh-v1.5", # 中文优化
]
# 测试数据
test_documents = [
"Python是一种高级编程语言",
"Java也是一门编程语言",
"JavaScript主要用于Web开发",
"今天天气很好",
"我喜欢吃苹果"
]
test_queries = [
"Python是什么？",
"什么语言用于Web开发？",
"今天天气怎么样？"
]
def evaluate_model(model: EmbeddingModel) -> Dict:
"""评估单个模型"""
print(f"\n评估模型: {model.name}")
print("-" * 60)
results = {}
# 1. 编码速度
print("测试1: 编码速度")
start_time = time.time()
doc_embeddings = model.encode(test_documents)
query_embeddings = model.encode(test_queries)
encode_time = time.time() - start_time
results['encode_time'] = encode_time
print(f" 编码时间: {encode_time:.3f}秒")
# 2. 嵌入维度
results['embedding_dim'] = doc_embeddings.shape[1]
print(f" 嵌入维度: {doc_embeddings.shape[1]}")
# 3. 检索质量
print("\n测试2: 检索质量")
hit_count = 0
for i, query_emb in enumerate(query_embeddings):
# 计算相似度
similarities = cosine_similarity(
[query_emb],
doc_embeddings
)[0]
# 找到最相关的文档
top_idx = np.argmax(similarities)
# 检查是否正确（假设前3个查询对应前3个文档）
if i < 3 and top_idx == i:
hit_count += 1
hit_rate = hit_count / 3 # 前3个查询
results['hit_rate'] = hit_rate
print(f" Hit Rate (前3查询): {hit_rate:.2%}")
# 4. 内存占用
print("\n测试3: 内存占用")
import sys
model_size = sum(p.numel() * p.element_size() for p in model.model.parameters())
model_size_mb = model_size / (1024 * 1024)
results['model_size_mb'] = model_size_mb
print(f" 模型大小: {model_size_mb:.2f}MB")
return results
# 运行对比
print("="*70)
print("嵌入模型性能对比")
print("="*70)
all_results = {}
for model_name in models_to_test:
try:
model = EmbeddingModel(model_name)
results = evaluate_model(model)
all_results[model_name] = results
except Exception as e:
print(f"错误: {e}")
continue
# 总结对比
print("\n" + "="*70)
print("性能对比总结")
print("="*70)
print(f"\n{'模型':<30} {'维度':<8} {'速度':<10} {'Hit Rate':<12} {'大小':<10}")
print("-" * 70)
for model_name, results in all_results.items():
short_name = model_name.split('/')[-1][:28]
print(f"{short_name:<30} "
f"{results['embedding_dim']:<8} "
f"{results['encode_time']:.3f}s {'':<2}"
f"{results['hit_rate']:<12.0%} "
f"{results['model_size_mb']:<10.1f}")
选择建议¶
根据应用场景选择¶
场景1：快速原型
场景2：生产环境（英文）
场景3：中文应用
场景4：多语言
决策树¶
开始
↓
需要中文优化？
├─ 是 → BGE-large-zh
└─ 否 ↓
预算充足？
├─ 是 → OpenAI large
└─ 否 ↓
需要速度？
├─ 是 → MiniLM-L6
└─ 否 → MPNet-base
6.3 嵌入模型微调¶
为什么微调？¶
问题：预训练模型可能不适合你的特定领域
解决方案：在领域数据上微调
微调数据准备¶
数据格式¶
# 文件名：06_05_finetuning_data.py
"""
准备微调数据
"""
# 标准微调数据格式
finetuning_data = [
{
"anchor": "Python是什么？",
"positive": "Python是一种编程语言",
"negative": "今天天气很好"
},
{
"anchor": "如何使用Pandas？",
"positive": "Pandas是Python数据分析库",
"negative": "Java也有类似的功能"
}
]
# 转换为训练格式
from torch.utils.data import Dataset
class FinetuningDataset(Dataset):
"""微调数据集"""
def __init__(self, data):
self.data = data
def __len__(self):
return len(self.data)
def __getitem__(self, idx):
item = self.data[idx]
return {
"anchor": item["anchor"],
"positive": item["positive"],
"negative": item["negative"]
}
# 创建数据集
dataset = FinetuningDataset(finetuning_data)
print(f"数据集大小: {len(dataset)}")
使用Sentence-Transformers微调¶
# 文件名：06_06_finetuning.py
"""
微调嵌入模型
"""
from sentence_transformers import SentenceTransformer, InputExample, losses, models
from torch.utils.data import DataLoader
def finetune_model(
base_model_name: str,
train_data: list,
output_path: str,
num_epochs: int = 3,
batch_size: int = 16
):
"""
微调嵌入模型
Args:
base_model_name: 基础模型名称
train_data: 训练数据列表
output_path: 输出路径
num_epochs: 训练轮数
batch_size: 批次大小
"""
print("="*60)
print("嵌入模型微调")
print("="*60 + "\n")
# 1. 加载基础模型
print("步骤1: 加载基础模型")
model = SentenceTransformer(base_model_name)
print(f"✅ 加载模型: {base_model_name}")
# 2. 准备训练数据
print("\n步骤2: 准备训练数据")
train_examples = [
InputExample(texts=[item["anchor"], item["positive"]], label=1.0)
for item in train_data
] + [
InputExample(texts=[item["anchor"], item["negative"]], label=0.0)
for item in train_data
]
train_dataloader = DataLoader(train_examples, shuffle=True, batch_size=batch_size)
print(f"✅ 训练样本: {len(train_examples)}")
# 3. 配置损失函数
print("\n步骤3: 配置损失函数")
train_loss = losses.CosineSimilarityLoss(model=model)
# 4. 配置优化器
print("步骤4: 配置优化器")
# 第一步：只训练最后一个变换层
warmup_steps = 100
# 5. 训练
print("\n步骤5: 开始训练")
print(f"Epochs: {num_epochs}")
print(f"Batch size: {batch_size}")
model.fit(
train_objectives=[(train_dataloader, train_loss)],
epochs=num_epochs,
warmup_steps=warmup_steps,
output_path=output_path,
show_progress_bar=True
)
print(f"\n✅ 模型已保存到: {output_path}")
# 6. 评估
print("\n步骤6: 评估微调效果")
# 这里可以添加评估代码
return model
# 使用示例
if __name__ == "__main__":
# 准备示例数据
train_data = [
{
"anchor": "Python有哪些特点？",
"positive": "Python的特点是语法简洁、易学易用",
"negative": "JavaScript是Web开发语言"
},
{
"anchor": "什么是RAG？",
"positive": "RAG是检索增强生成技术",
"negative": "今天天气晴朗"
}
# ... 更多数据
]
# 微调
finetune_model(
base_model_name="BAAI/bge-small-zh-v1.5",
train_data=train_data,
output_path="./models/finetuned_bge",
num_epochs=3
)
6.4 嵌入可视化与评估¶
嵌入空间可视化¶
使用t-SNE或UMAP降维可视化嵌入空间
# 文件名：06_07_embedding_visualization.py
"""
嵌入空间可视化
"""
import numpy as np
import matplotlib.pyplot as plt
from sklearn.manifold import TSNE
from sentence_transformers import SentenceTransformer
import pandas as pd
def visualize_embeddings(model_name: str, texts: list, labels: list):
"""
可视化嵌入空间
Args:
model_name: 模型名称
texts: 文本列表
labels: 标签列表
"""
print("="*60)
print("嵌入空间可视化")
print("="*60 + "\n")
# 1. 生成嵌入
print("步骤1: 生成嵌入向量")
model = SentenceTransformer(model_name)
embeddings = model.encode(texts)
print(f"✅ 嵌入形状: {embeddings.shape}")
# 2. 降维（t-SNE）
print("\n步骤2: t-SNE降维")
tsne = TSNE(n_components=2, random_state=42, perplexity=min(30, len(texts)-1))
embeddings_2d = tsne.fit_transform(embeddings)
print(f"✅ 降维后形状: {embeddings_2d.shape}")
# 3. 绘制散点图
print("\n步骤3: 绘制可视化")
plt.figure(figsize=(12, 8))
# 按标签分组着色
unique_labels = list(set(labels))
colors = plt.cm.tab10(np.linspace(0, 1, len(unique_labels)))
for i, label in enumerate(unique_labels):
mask = np.array(labels) == label
plt.scatter(
embeddings_2d[mask, 0],
embeddings_2d[mask, 1],
c=[colors[i]],
label=label,
alpha=0.7,
s=100
)
plt.xlabel("t-SNE 维度 1")
plt.ylabel("t-SNE 维度 2")
plt.title("嵌入空间可视化")
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
# 保存
output_path = "./outputs/embedding_visualization.png"
plt.savefig(output_path, dpi=300, bbox_inches='tight')
print(f"✅ 图片已保存: {output_path}")
plt.show()
# 使用示例
if __name__ == "__main__":
# 准备数据
texts = [
# 编程语言
"Python是一种编程语言",
"Java也是编程语言",
"JavaScript是Web语言",
"C++适合系统编程",
# 水果
"苹果是水果",
"香蕉也是水果",
"橙子富含维生素C",
# 动物
"猫是宠物",
"狗是宠物",
"老虎是野生动物"
]
labels = ["编程"] * 4 + ["水果"] * 3 + ["动物"] * 3
# 可视化
visualize_embeddings(
model_name="all-MiniLM-L6-v2",
texts=texts,
labels=labels
)
嵌入质量评估¶
# 文件名：06_08_embedding_evaluation.py
"""
评估嵌入质量
"""
import numpy as np
from typing import List, Dict
from sklearn.metrics.pairwise import cosine_similarity
class EmbeddingEvaluator:
"""嵌入评估器"""
def __init__(self, model):
self.model = model
def evaluate_retrieval(self, queries: list, documents: list, relevant_docs: list):
"""
评估检索质量
Args:
queries: 查询列表
documents: 文档列表
relevant_docs: 真实相关文档（列表的列表）
Returns:
评估指标
"""
# 生成嵌入
query_embeddings = self.model.encode(queries)
doc_embeddings = self.model.encode(documents)
# 计算每个查询的检索质量
hit_count = 0
mrr_sum = 0
for i, query_emb in enumerate(query_embeddings):
# 计算相似度
similarities = cosine_similarity([query_emb], doc_embeddings)[0]
# 排序
sorted_indices = np.argsort(similarities)[::-1]
# Hit Rate (top-5)
top5_indices = sorted_indices[:5]
if any(idx in relevant_docs[i] for idx in top5_indices):
hit_count += 1
# MRR
for rank, idx in enumerate(sorted_indices, 1):
if idx in relevant_docs[i]:
mrr_sum += 1 / rank
break
metrics = {
"hit_rate_at_5": hit_count / len(queries),
"mrr": mrr_sum / len(queries)
}
return metrics
def print_report(self, metrics: Dict):
"""打印评估报告"""
print("\n" + "="*60)
print("嵌入质量评估报告")
print("="*60)
for metric, value in metrics.items():
metric_name = metric.replace("_", " ").title()
if "rate" in metric.lower():
print(f"{metric_name:20s}: {value:.2%}")
else:
print(f"{metric_name:20s}: {value:.3f}")
print("="*60)
# 使用示例
if __name__ == "__main__":
from sentence_transformers import SentenceTransformer
model = SentenceTransformer('all-MiniLM-L6-v2')
evaluator = EmbeddingEvaluator(model)
# 测试数据
queries = [
"Python是什么？",
"什么语言用于Web开发？",
"C++有什么特点？"
]
documents = [
"Python是一种编程语言",
"Java也是编程语言",
"JavaScript主要用于Web开发",
"C++适合系统编程",
"今天天气很好"
]
relevant_docs = [
{0}, # 第一个查询相关第0个文档
{2}, # 第二个查询相关第2个文档
{3} # 第三个查询相关第3个文档
]
# 评估
metrics = evaluator.evaluate_retrieval(queries, documents, relevant_docs)
evaluator.print_report(metrics)
总结¶
本章要点回顾¶
- 嵌入模型原理
- Transformer架构回顾
- 双编码器网络
-
对比学习损失
-
模型选择
- 主流模型对比
- 根据场景选择
-
权衡性能、速度、成本
-
模型微调
- 数据准备
- 微调流程
-
效果评估
-
可视化评估
- t-SNE降维
- 空间可视化
- 质量评估指标
学习检查清单¶
- 理解Transformer嵌入原理
- 掌握主流嵌入模型对比
- 能够选择合适的模型
- 理解微调过程
- 能够可视化嵌入空间
- 掌握质量评估方法
下一步学习¶
- 下一章：第7章：高级分块策略
- 相关：优化嵌入模型是第一步，接下来优化数据质量
扩展资源¶
推荐阅读： 1. Sentence-BERT论文 2. BGE模型论文 3. MTEB Benchmark
实践项目： - 对比3种不同的嵌入模型 - 在自己的数据上微调模型 - 可视化并分析嵌入空间
返回目录 | 上一章：模块1总结 | 下一章：高级分块策略
本章结束
选择合适的嵌入模型是RAG优化的第一步。一个好的嵌入模型可以让检索质量提升10-15%，这是最简单也是最重要的优化之一！