嵌入模型

嵌入模型（Embedding Model）负责将文本转换为向量表示，是 RAG 系统中连接文本和数学世界的桥梁。好的嵌入模型能准确捕捉语义信息，让相似的文本在向量空间中靠近，不同的文本远离。本章将深入介绍嵌入模型的原理、选择和使用方法。

什么是文本嵌入？

文本嵌入是将文本转换为高维向量的过程。这些向量在几何空间中的位置和距离反映了文本之间的语义关系。

直观理解

想象一个三维空间，我们根据三个维度给词语定位：

维度1: 正面/负面
维度2: 具体/抽象
维度3: 动词/名词

"开心" → [0.9, 0.3, 0.1]
"悲伤" → [-0.8, 0.3, 0.1]
"汽车" → [0.0, 0.9, 0.2]
"梦想" → [0.1, -0.7, 0.1]

"开心"和"悲伤"在维度1上相反，但在维度2、3上相似
"汽车"和"梦想"在维度2上差异很大

实际的嵌入向量有数百到数千个维度，每个维度代表一个抽象的语义特征。

数学表示

# 嵌入模型的输出示例
text = "机器学习是人工智能的一个分支"
embedding = [0.023, -0.156, 0.089, ..., 0.041]  # 1536维向量

# 向量的属性
dimension = len(embedding)      # 维度：1536
magnitude = np.linalg.norm(embedding)  # 模长：约 1.0（归一化后）

相似度计算

向量之间的相似度通常用余弦相似度衡量：

$\text{similarity}(\vec{A}, \vec{B}) = \cos(\theta) = \frac{\vec{A} \cdot \vec{B}}{|\vec{A}| \cdot |\vec{B}|}$

import numpy as np

def cosine_similarity(a, b):
    """计算余弦相似度"""
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# 示例
embedding_hello = embed("你好")
embedding_hi = embed("嗨")
embedding_weather = embed("天气")

similarity_hello_hi = cosine_similarity(embedding_hello, embedding_hi)      # 0.92
similarity_hello_weather = cosine_similarity(embedding_hello, embedding_weather)  # 0.15

# "你好"和"嗨"相似度高，"你好"和"天气"相似度低

嵌入模型的类型

密集向量（Dense Embeddings）

最常用的类型，每个维度都是连续值。

from openai import OpenAI

client = OpenAI()

response = client.embeddings.create(
    model="text-embedding-3-small",
    input="这是一段文本"
)

dense_embedding = response.data[0].embedding
# [0.0023, -0.0156, 0.0089, ..., 0.0041]
# 维度: 1536

优点：语义表达能力强、压缩效率高 缺点：难以解释每个维度的含义

稀疏向量（Sparse Embeddings）

大多数维度为 0，只有少数维度有值，通常与词项对应。

from transformers import AutoTokenizer, AutoModel

# SPLADE 是典型的稀疏嵌入模型
sparse_embedding = {
    "机器": 2.34,
    "学习": 3.12,
    "人工智能": 2.89,
    "技术": 1.45,
    # ... 其他维度为 0
}

优点：可解释、支持精确匹配 缺点：语义表达能力较弱、维度高

多向量嵌入（Multi-Vector Embeddings）

为一段文本生成多个向量，捕捉不同层面的语义。

# ColBERT 风格的多向量
text = "机器学习是人工智能的分支"
tokens = ["机器", "学习", "是", "人工智能", "的", "分支"]

multi_vectors = {
    "机器": [0.12, 0.34, ...],
    "学习": [0.23, 0.45, ...],
    # 每个词一个向量
}

优点：细粒度语义匹配 缺点：存储和计算开销大

主流嵌入模型对比

商业模型

模型	提供商	维度	特点	定价（$/1M tokens）
text-embedding-3-small	OpenAI	1536	性价比高、速度快	$0.02
text-embedding-3-large	OpenAI	3072	性能最佳	$0.13
embed-english-v3.0	Cohere	1024	英文优化、支持压缩	$0.10
embed-multilingual-v3.0	Cohere	1024	多语言支持	$0.10
voyage-large-2	Voyage	1536	检索性能优秀	$0.12

开源模型

模型	提供商	维度	特点
BGE-large-zh-v1.5	BAAI	1024	中文最佳、开源免费
BGE-m3	BAAI	1024	多语言、长文本支持
E5-large-v2	Microsoft	1024	英文优秀、需加前缀
GTE-large	Alibaba	1024	多语言、性能均衡
MTEB top models	各机构	-	参考排行榜

MTEB 排行榜

MTEB（Massive Text Embedding Benchmark）是评估嵌入模型的权威榜单：

排名参考（2024）:
1. Voyage-large-2-instruct
2. OpenAI text-embedding-3-large
3. BGE-large-en-v1.5
4. E5-mistral-7b
...

访问 MTEB Leaderboard 查看最新排名。

嵌入模型选择指南

按语言选择

# 中文场景 - 推荐使用 BGE
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('BAAI/bge-large-zh-v1.5')
embeddings = model.encode(["这是一段中文文本"])

# 英文场景 - 推荐 E5 或 OpenAI
model = SentenceTransformer('intfloat/e5-large-v2')
# E5 需要添加前缀
embeddings = model.encode(["query: What is machine learning?"])

# 多语言场景 - 推荐 BGE-m3 或 Cohere
model = SentenceTransformer('BAAI/bge-m3')

按场景选择

场景	推荐模型	原因
企业内部知识库	BGE-large-zh + 本地部署	中文优化、数据安全
全球化产品	Cohere multilingual 或 OpenAI	多语言支持
高精度检索	Voyage-large-2	检索性能顶尖
成本敏感	text-embedding-3-small	最低成本
隐私敏感	BGE 本地部署	数据不出服务器

性能 vs 成本权衡

性能/成本矩阵:

高性能 ↑
       │  text-embedding-3-large    Voyage-large-2
       │
       │  BGE-large-zh-v1.5         text-embedding-3-small
       │
低性能 │  (开源本地运行)
       └───────────────────────────────────────→ 高成本
                低成本

嵌入模型的使用

OpenAI 嵌入模型

from openai import OpenAI
from langchain_openai import OpenAIEmbeddings

# 直接使用 OpenAI API
client = OpenAI()

def get_embeddings(texts):
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=texts,
        dimensions=512  # 可选：减少维度
    )
    return [item.embedding for item in response.data]

# 使用 LangChain
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    dimensions=512  # 支持维度缩减
)

# 单个文本
vector = embeddings.embed_query("这是用户问题")

# 批量文本
vectors = embeddings.embed_documents(["文档1", "文档2", "文档3"])

开源模型（Sentence Transformers）

from sentence_transformers import SentenceTransformer

# 加载模型
model = SentenceTransformer('BAAI/bge-large-zh-v1.5')

# 编码
sentences = [
    "机器学习是人工智能的一个分支",
    "深度学习使用多层神经网络",
    "今天天气真好"
]

embeddings = model.encode(sentences)

# embeddings.shape: (3, 1024)
# 每句话一个 1024 维向量

E5 模型的使用注意事项

E5 模型需要在文本前添加前缀来区分查询和文档：

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('intfloat/e5-large-v2')

# 查询需要加 "query: " 前缀
query = "query: 什么是机器学习？"
query_embedding = model.encode([query])

# 文档需要加 "passage: " 前缀
documents = [
    "passage: 机器学习是人工智能的一个分支...",
    "passage: 深度学习是机器学习的子集..."
]
doc_embeddings = model.encode(documents)

BGE 模型的使用

BGE 模型同样建议添加指令：

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('BAAI/bge-large-zh-v1.5')

# 查询添加指令
query = "为这个句子生成表示以用于检索相关文章：如何学习Python？"
query_embedding = model.encode([query], normalize_embeddings=True)

# 文档直接编码
documents = ["Python是一种编程语言...", "学习Python的建议..."]
doc_embeddings = model.encode(documents, normalize_embeddings=True)

# BGE 建议归一化

维度缩减

高维向量占用更多存储空间和计算资源。某些模型支持缩减维度：

from langchain_openai import OpenAIEmbeddings

# 原始 1536 维
full_embeddings = OpenAIEmbeddings(model="text-embedding-3-large")

# 缩减到 256 维（仍保持较高性能）
reduced_embeddings = OpenAIEmbeddings(
    model="text-embedding-3-large",
    dimensions=256
)

# 存储节省
# 1536 维: 1536 * 4 bytes = 6 KB / 向量
# 256 维:  256 * 4 bytes = 1 KB / 向量
# 节省 83% 存储空间

Matryoshka 嵌入

某些模型支持"俄罗斯套娃"式嵌入，可以截断到任意维度：

# 支持 Matryoshka 的模型
model = SentenceTransformer('nomic-ai/nomic-embed-text-v1')

# 生成 768 维向量
embedding_768 = model.encode(["text"])

# 截断到 256 维（性能略有下降但可用）
embedding_256 = embedding_768[:256]

# 截断到 128 维
embedding_128 = embedding_768[:128]

批量处理与性能优化

批量编码

处理大量文档时，使用批量编码显著提高效率：

from sentence_transformers import SentenceTransformer

model = SentenceTransformer('BAAI/bge-large-zh-v1.5')

# 不好的做法：逐个编码
# for doc in documents:
#     embedding = model.encode(doc)  # 非常慢

# 好的做法：批量编码
batch_size = 32
all_embeddings = model.encode(
    documents,
    batch_size=batch_size,
    show_progress_bar=True,
    convert_to_numpy=True
)

# 更高效：使用 GPU
model = SentenceTransformer('BAAI/bge-large-zh-v1.5', device='cuda')

异步处理

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def get_embedding(text):
    response = await client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

async def batch_embed(texts, batch_size=100):
    results = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i+batch_size]
        tasks = [get_embedding(text) for text in batch]
        embeddings = await asyncio.gather(*tasks)
        results.extend(embeddings)
    return results

# 使用
embeddings = asyncio.run(batch_embed(documents))

嵌入模型评估

内在评估

使用标准数据集评估模型质量：

from sentence_transformers import SentenceTransformer
from sentence_transformers.evaluation import InformationRetrievalEvaluator

# 准备评估数据
evaluator = InformationRetrievalEvaluator(
    queries={"q1": "机器学习是什么", "q2": "Python教程"},
    corpus={"d1": "机器学习是AI的分支", "d2": "Python是编程语言"},
    relevant_docs={"q1": ["d1"], "q2": ["d2"]}
)

# 评估
model = SentenceTransformer('BAAI/bge-large-zh-v1.5')
results = evaluator(model)

print(f"NDCG@10: {results['ndcg@10']}")
print(f"MRR@10: {results['mrr@10']}")

实际场景评估

用业务数据评估检索效果：

def evaluate_retrieval_quality(test_cases, model, vectorstore):
    """评估检索质量"""
    total = len(test_cases)
    hits = 0
    
    for case in test_cases:
        query = case["query"]
        expected_chunks = case["relevant_chunks"]
        
        # 编码查询
        query_embedding = model.encode([query])
        
        # 检索
        results = vectorstore.similarity_search_by_vector(
            query_embedding[0], k=5
        )
        
        # 检查是否命中
        retrieved_ids = [r.metadata["chunk_id"] for r in results]
        if any(eid in retrieved_ids for eid in expected_chunks):
            hits += 1
    
    return hits / total

常见问题与解决方案

问题 1：不同长度文本的嵌入质量差异

短文本（如标题）和长文本（如段落）的嵌入可能有系统性差异。

解决方案：使用针对检索优化的模型，或在长文本中提取关键句。

def embed_with_fallback(text, model, min_length=50):
    """带回退的嵌入"""
    if len(text) < min_length:
        # 短文本：补充上下文
        text = f"文档内容：{text}"
    
    return model.encode([text])

问题 2：中英文混合文本

中英文混合的文本处理不当会影响检索质量。

解决方案：使用多语言模型或统一处理。

# 使用多语言模型
model = SentenceTransformer('BAAI/bge-m3')  # 支持多语言

# 或预处理：统一语言
from langdetect import detect

def detect_and_translate(text):
    lang = detect(text)
    if lang != 'zh':
        text = translate_to_chinese(text)
    return text

问题 3：嵌入模型更新

模型更新后，旧向量与新模型不兼容。

解决方案：版本管理 + 重新索引。

# 存储模型版本信息
def index_documents(docs, model_name):
    model = SentenceTransformer(model_name)
    embeddings = model.encode(docs)
    
    for i, (doc, emb) in enumerate(zip(docs, embeddings)):
        vectorstore.add(
            id=f"doc_{i}",
            vector=emb,
            metadata={
                "text": doc,
                "model": model_name,
                "indexed_at": datetime.now().isoformat()
            }
        )

# 检查版本一致性
def search(query, model_name):
    current_model = get_config("embedding_model")
    if current_model != model_name:
        raise ModelMismatchError("需要重新索引")

最佳实践总结

1. 中文场景优先 BGE：性能优秀、开源免费
2. 商业场景考虑 OpenAI：稳定可靠、易用
3. 添加适当前缀：某些模型需要区分查询和文档
4. 批量处理：提高编码效率
5. 归一化向量：某些模型需要，提高检索稳定性
6. 版本管理：记录使用的模型版本

小结

嵌入模型是 RAG 系统的"翻译器"，将文本转换为计算机可处理的向量。关键要点：

• 选择适合语言和场景的模型
• 注意模型的使用规范（前缀、归一化）
• 批量处理提高效率
• 持续评估和优化

下一步

• 向量数据库：学习如何存储和检索嵌入向量
• 检索策略：深入理解向量检索的原理和优化

参考资料

• MTEB Leaderboard
• OpenAI Embeddings Guide
• BGE Models
• Sentence Transformers Documentation