一、简介

嵌入模型（Embedding Model）是 RAG 流程的核心中枢—— 它将文本片段（Chunk）转换为高维向量（Embedding Vector），让计算机能通过「向量相似度」而非关键词匹配来检索相关信息。好的嵌入模型直接决定 RAG 的检索精度，甚至能弥补文本分割、加载环节的小瑕疵。

1.1 核心概念

嵌入（Embedding）是将文本（单词、句子、段落）映射到高维向量空间的技术。在这个空间中：

• 语义相似的文本在向量空间中也距离相近
• 不同含义的文本则相互远离

例如："猫"和"狗"的向量距离比"猫"和"汽车"更近。

1.2 核心作用

在RAG中，嵌入模型起到桥梁作用：

1. 索引阶段：将文档块转化为向量，存入向量数据库。
2. 查询阶段：将用户问题转化为向量，在向量空间中搜索最相似的文档块。

没有嵌入模型，RAG就无法实现语义级别的检索，只能依赖关键词匹配。

1.3 核心指标

1. 向量维度（Dimension）

   • 输出向量的长度，常见：768 / 1024 / 1536 / 3072
   • 维度越高精度可能越高，但存储 / 计算成本也越高，RAG 常用 768-1536 维。

2. 上下文窗口

   • 模型能处理的最大文本长度。
   • 例如：
     • text-embedding-3-small：8191
     • bge-small-zh：512
   • 超过会被截断，直接废掉检索效果。

3. 语义召回率

   • 能检索到相关文本的比例。
   • 核心指标，越高越好（优先选经过中文评测的模型）

4. 支持语言

   • 英文模型 → 中文拉胯
   • 中文专用模型 → 中文效果强很多
   • 多语言模型 = 通用

5. 速度 & 成本

   • 开源本地：免费、慢、隐私强
   • 闭源 API：快、准、收费

二、LangChain中的嵌入模型接口

LangChain提供了一个统一的接口Embeddings，所有嵌入模型都遵循这个接口：

class Embeddings(ABC):
    @abstractmethod
    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        """嵌入多个文档（批量处理）"""
        pass
    
    @abstractmethod
    def embed_query(self, text: str) -> List[float]:
        """嵌入单个查询（与文档使用相同的模型，但可能不同的处理逻辑）"""
        pass

基本使用：

from langchain_openai import OpenAIEmbeddings

# 初始化嵌入模型
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    openai_api_key="your-api-key"
)

# 嵌入文档（批量）
doc_texts = ["这是第一段文本", "这是第二段文本", "这是第三段文本"]
doc_vectors = embeddings.embed_documents(doc_texts)
print(f"生成了 {len(doc_vectors)} 个向量，每个维度 {len(doc_vectors[0])}")

# 嵌入查询
query = "用户的问题"
query_vector = embeddings.embed_query(query)
print(f"查询向量维度：{len(query_vector)}")

异步支持：

import asyncio

async def embed_async():
    embeddings = OpenAIEmbeddings()
    
    # 异步嵌入文档
    doc_vectors = await embeddings.aembed_documents(doc_texts)
    
    # 异步嵌入查询
    query_vector = await embeddings.aembed_query(query)
    
    return doc_vectors, query_vector

# 执行
vectors = asyncio.run(embed_async())

三、主流嵌入模型

3.1 开源模型（本地部署，企业首选）

3.1.1 中文最优：BGE 系列（上海人工智能实验室）

• 代表模型：bge-m3（最强）、bge-large-zh-v1.5（经典）、bge-small-zh-v1.5（轻量）

• 优势：中文语义理解顶尖、支持多粒度检索、可本地部署、免费商用

• 适用场景：所有中文 RAG 场景（文档问答、知识库、智能客服）

• 使用示例：

  # 安装依赖
  pip install sentence-transformers  # 加载BGE模型
  pip install langchain-huggingface   # LangChain集成
  

  # LangChain 集成 BGE-m3（最优）
  from langchain_huggingface import HuggingFaceEmbeddings
  
  # 初始化模型（自动下载权重，首次运行需等待）
  embeddings = HuggingFaceEmbeddings(
     model_name="BAAI/bge-m3",
     model_kwargs={
         "device": "cuda"  # 用GPU加速（无GPU则设为"cpu"）
     },
     encode_kwargs={
         "normalize_embeddings": True  # 归一化，提升相似度计算精度
     }
  )
  
  # 生成文本向量（单文本）
  vector = embeddings.embed_query("产品价格是99元")
  print(f"向量维度：{len(vector)}")  # bge-m3 默认 1024 维
  
  # 批量生成向量（推荐，速度更快）
  texts = ["产品价格是99元", "99元可以购买该产品", "今天天气很好"]
  vectors = embeddings.embed_documents(texts)
  print(f"批量向量数量：{len(vectors)}，单个维度：{len(vectors[0])}")
  

3.1.2 轻量快模型：m3e 系列

• 代表模型：m3e-base、m3e-small

• 优势：体积小、推理快（CPU 也能跑）、中文适配好

• 适用场景：低算力环境（如边缘设备、低配服务器）

• 使用示例：

  embeddings = HuggingFaceEmbeddings(
      model_name="moka-ai/m3e-base",
      model_kwargs={"device": "cpu"},  # CPU 部署
      encode_kwargs={"normalize_embeddings": True}
  )
  

3.1.3 多语言模型：all-MiniLM-L6-v2

• 优势：支持多语言、体积极小、速度极快
• 缺点：中文精度略低于 BGE/m3e
• 适用场景：多语言 RAG 场景

3.2 闭源 API 模型（快速上线，无需部署）

3.2.1 OpenAI Embeddings

• 代表模型：text-embedding-3-small（轻量）、text-embedding-3-large（高精度）

• 优势：通用语义理解强、集成简单

• 缺点：付费、数据出海、中文精度略逊于 BGE

• 适用场景：快速验证原型、英文为主的 RAG

• 使用示例：

  pip install langchain-openai
  

  from langchain_openai import OpenAIEmbeddings
  
  embeddings = OpenAIEmbeddings(
      model="text-embedding-3-large",
      api_key="YOUR_OPENAI_KEY",
      # 可选：指定国内代理
      # base_url="https://api.openai-proxy.com/v1"
  )
  
  vector = embeddings.embed_query("产品价格是99元")
  print(f"向量维度：{len(vector)}")  # text-embedding-3-large 默认为 3072 维
  

3.2.2 国内大厂 API

• 代表模型：阿里云通义千问：DashScopeEmbeddings；百度文心一言：ErnieEmbeddings；腾讯混元：HunyuanEmbeddings

• 优势：国内访问快、中文优化、合规性高

• 适用场景：企业级合规要求高的 RAG

• 使用示例：

  # 阿里云通义千问示例
  from langchain_community.embeddings import DashScopeEmbeddings
  
  embeddings = DashScopeEmbeddings(
      model="text-embedding-v1",
      dashscope_api_key="YOUR_DASHSCOPE_KEY"
  )
  

四、嵌入模型的核心参数

4.1 通用参数

参数	说明	示例
model	模型名称	“text-embedding-3-small”
dimensions	输出向量维度	256（仅部分模型支持）
normalize_embeddings	是否归一化（余弦相似度需要）	True
show_progress_bar	是否显示进度条	True（批量处理时）
batch_size	批量处理大小	32
max_retries	失败重试次数	3

4.2 模型特定参数

# OpenAI
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    dimensions=512,           # 降维
    openai_api_key="...",
    request_timeout=60        # 超时时间
)

# HuggingFace
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-base-zh",
    model_kwargs={
        'device': 'cuda',      # GPU
        'trust_remote_code': True
    },
    encode_kwargs={
        'normalize_embeddings': True,
        'batch_size': 32
    }
)

五、嵌入模型的评估与选择

5.1 评估指标

指标	说明	理想值
检索准确率	检索出的top-k文档中相关文档的比例	越高越好
召回率	所有相关文档中被检索出的比例	越高越好
MTEB分数	大规模文本嵌入基准测试分数	查看排行榜
延迟	生成嵌入的时间	越快越好
维度	向量维度	权衡精度和存储

5.2 选择指南

场景	推荐模型	理由
快速原型、通用场景	OpenAI text-embedding-3-small	简单易用，性能优秀
中文文档为主	BAAI/bge-base-zh 或 M3E	针对中文优化，免费
对精度要求极高	OpenAI text-embedding-3-large	目前顶尖水平
多语言混合	BAAI/bge-m3 或 Cohere embed-multilingual	支持100+语言
本地部署、数据隐私	HuggingFace模型（BGE、E5等）	完全本地运行
资源受限（移动端）	all-MiniLM-L6-v2	轻量（384维）
需要降维节省存储	OpenAI text-embedding-3系列	支持自定义维度

六、高级技巧与最佳实践

6.1 批量处理优化

# 对于大量文档，使用批次处理
embeddings = OpenAIEmbeddings()

# 自动分批（默认batch_size=500）
vectors = embeddings.embed_documents(huge_document_list)

# 手动控制批次大小和并发
from langchain_core.embeddings import Embeddings

class BatchedEmbeddings(Embeddings):
    def __init__(self, embeddings_model, batch_size=100):
        self.model = embeddings_model
        self.batch_size = batch_size
    
    def embed_documents(self, texts):
        all_vectors = []
        for i in range(0, len(texts), self.batch_size):
            batch = texts[i:i+self.batch_size]
            batch_vectors = self.model.embed_documents(batch)
            all_vectors.extend(batch_vectors)
            print(f"已处理 {i+len(batch)}/{len(texts)}")
        return all_vectors

6.2 缓存嵌入结果

对于重复的文本，可以缓存嵌入结果节省时间和费用：

import hashlib
import json
import redis

class CachedEmbeddings:
    def __init__(self, embeddings_model, cache_client):
        self.model = embeddings_model
        self.cache = cache_client  # 可以是redis、内存字典等
    
    def _get_key(self, text):
        return f"embedding:{hashlib.md5(text.encode()).hexdigest()}"
    
    def embed_query(self, text):
        key = self._get_key(text)
        cached = self.cache.get(key)
        if cached:
            return json.loads(cached)
        
        vector = self.model.embed_query(text)
        self.cache.set(key, json.dumps(vector))
        return vector
    
    def embed_documents(self, texts):
        # 批量检查缓存
        results = []
        uncached_texts = []
        uncached_indices = []
        
        for i, text in enumerate(texts):
            key = self._get_key(text)
            cached = self.cache.get(key)
            if cached:
                results.append((i, json.loads(cached)))
            else:
                uncached_texts.append(text)
                uncached_indices.append(i)
        
        if uncached_texts:
            new_vectors = self.model.embed_documents(uncached_texts)
            for idx, text, vector in zip(uncached_indices, uncached_texts, new_vectors):
                key = self._get_key(text)
                self.cache.set(key, json.dumps(vector))
                results.append((idx, vector))
        
        # 按原顺序返回
        results.sort(key=lambda x: x[0])
        return [v for _, v in results]

6.3 向量归一化

大多数相似度计算（如余弦相似度）假设向量是归一化的。确保嵌入模型输出归一化向量或在计算前归一化：

import numpy as np

def normalize(vector):
    norm = np.linalg.norm(vector)
    return vector / norm if norm > 0 else vector

# 在HuggingFace模型中设置
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-base-zh",
    encode_kwargs={'normalize_embeddings': True}
)

# 手动归一化
vector = embeddings.embed_query("text")
vector = normalize(vector)

6.4 混合检索（稀疏+稠密）

有时结合传统的关键词检索（BM25）和向量检索可以取得更好效果：

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import Chroma

# 向量检索器
vectorstore = Chroma.from_documents(chunks, embeddings)
vector_retriever = vectorstore.as_retriever()

# 关键词检索器
keyword_retriever = BM25Retriever.from_documents(chunks)

# 组合检索器
ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, keyword_retriever],
    weights=[0.7, 0.3]  # 权重分配
)

docs = ensemble_retriever.get_relevant_documents("查询")

6.5 维度约简（降维）

对于存储空间敏感的场景，可以对嵌入向量进行降维：

from sklearn.decomposition import PCA
import numpy as np

# 假设已经有很多文档向量
vectors = np.array(all_vectors)  # shape: (n_docs, original_dim)

# 训练PCA
pca = PCA(n_components=256)
reduced_vectors = pca.fit_transform(vectors)

# 存储降维后的向量
# 查询时也需要先嵌入再降维
query_vector = embeddings.embed_query(query)
reduced_query = pca.transform([query_vector])[0]

或者使用OpenAI的原生降维：

embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    dimensions=256  # 直接在API层面降维
)

6.6 适配长文本（超出模型窗口）

当文本片段长度超过模型上下文窗口时，需先分段再嵌入，最后合并：

from langchain_huggingface import HuggingFaceEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 初始化模型（bge-m3 窗口约 8192 字符）
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-m3")

# 长文本（10000 字符）
long_text = "产品介绍：" * 2000  # 模拟长文本

# 先分割成模型能处理的小块
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=5000,
    chunk_overlap=100
)
chunks = text_splitter.split_text(long_text)

# 嵌入所有小块，然后取平均值作为长文本的向量
chunk_vectors = embeddings.embed_documents(chunks)
long_text_vector = [sum(col) / len(col) for col in zip(*chunk_vectors)]

print(f"长文本向量维度：{len(long_text_vector)}")  # 1024 维

七、完整RAG嵌入实战

import os
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# 1. 初始化嵌入模型（选择适合你的模型）
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    dimensions=512,  # 降维节省存储
)

# 2. 加载和分割文档
loader = TextLoader("knowledge_base.txt")
documents = loader.load()

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
chunks = text_splitter.split_documents(documents)

# 3. 嵌入并存储
print(f"正在嵌入 {len(chunks)} 个文档块...")
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 4. 测试检索
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

test_queries = [
    "公司的年假政策是什么？",
    "如何申请报销？"
]

for query in test_queries:
    print(f"\n查询：{query}")
    docs = retriever.get_relevant_documents(query)
    print(f"检索到 {len(docs)} 个相关文档")
    for i, doc in enumerate(docs):
        print(f"  文档{i+1}（相似度评分：{doc.metadata.get('score', 'N/A')}）")
        print(f"  内容预览：{doc.page_content[:100]}...")

# 5. 完整RAG问答
prompt = ChatPromptTemplate.from_template("""
基于以下上下文回答问题：
{context}

问题：{input}
回答：""")

model = ChatOpenAI()
document_chain = create_stuff_documents_chain(model, prompt)
rag_chain = create_retrieval_chain(retriever, document_chain)

response = rag_chain.invoke({"input": "年假有多少天？"})
print(f"\n答案：{response['answer']}")

八、常见问题与解决方案

Q1: 中文语义检索精度低

• 现象：检索结果和问题无关（如问 “产品价格”，返回 “产品功能”）
• 根因：用了英文优先的模型（如 all-MiniLM）、未归一化向量、chunk 太小
• 解决方案：
  • 替换为 BGE-m3/m3e 等中文优化模型；
  • 开启 normalize_embeddings=True（关键！）；
  • 调整 chunk_size 到 800-1200 字符。

Q2: 嵌入速度极慢（CPU 环境）

• 现象：处理 1000 条文本需要几分钟
• 根因：用了大模型（如 BGE-m3）、单条嵌入、未量化
• 解决方案：
  • 换轻量模型（m3e-small/BGE-small-zh）；
  • 批量嵌入（每次 100-500 条）；
  • 开启 CPU 量化（如 torch.float16）。

Q3: 显存溢出（GPU 环境）

• 现象：加载模型时抛出 CUDA out of memory
• 根因：模型体积大、未量化、批量太大
• 解决方案：
  • 启用 4bit/8bit 量化；
  • 减小批量大小（如每次 50 条）；
  • 用更小的模型（如 BGE-small-zh）。

Q4: 向量存储占用过大

• 现象：100 万条文本的向量占用几十 GB 磁盘
• 根因：用了高维度模型（如 text-embedding-3-large 3072 维）
• 解决方案：
  • 换低维度模型（BGE-m3 1024 维 /m3e-base 768 维）；
  • 向量量化（如 FAISS 的 IVF_PQ 量化）；
  • 只保留核心文本的向量，过滤冗余内容。

Q5: API 调用成本高

• 现象：大量文本嵌入导致 API 费用飙升
• 根因：未缓存、重复嵌入、用了高精度高维度模型
• 解决方案：
  • 启用缓存（本地 / Redis）；
  • 优先用开源模型私有化部署；
  • 对低频更新的文本预生成向量并存储。