从OpenAI到国产模型：RAG系统中文本嵌入模型的替换实践与选型指南

1. 项目背景与国产模型替换的动机

最近在折腾一个RAG（检索增强生成）项目，核心的文本嵌入（Embedding）环节一直用的是OpenAI的text-embedding-ada-002。模型效果确实稳定，但每次调用都得走API，延迟、费用和潜在的稳定性问题，在项目后期越来越让人头疼。尤其是当你想把项目部署到内网，或者对数据隐私有更高要求时，依赖海外服务总感觉不那么踏实。正好，国产大模型在近一年里突飞猛进，不仅在通用大语言模型（LLM）上表现亮眼，在文本嵌入这个“幕后英雄”领域也涌现出不少优秀选手。于是，我决定动手，把项目里的嵌入模型从“洋枪”换成“国产炮”。

这个决定背后有几个很实际的考量。首先是成本与控制权。使用国产开源模型，无论是按量计费还是私有化部署，长期来看成本更可控，也避免了因国际网络或服务政策变动带来的风险。其次是数据安全与合规。对于涉及企业内部知识、个人隐私数据的应用，将嵌入计算留在本地或国内云环境，是很多项目的硬性要求。最后是技术自主与定制化。开源模型允许我们深入其内部，针对特定领域语料进行微调（Fine-tuning），从而获得比通用嵌入模型更精准的向量表示，这对于垂直领域的RAG应用效果提升至关重要。

这次替换，我瞄准了几个在中文社区口碑不错且有官方详细文档的国产模型，比如智谱的text2vec系列、百度的ERNIE-Embedding，以及一些通用的双语模型如BGE-M3。目标很明确：在不显著损失（甚至提升）中文语义表征能力的前提下，实现嵌入模型的平滑替换，并确保整个RAG流水线（切片、向量化、检索、重排序）依然能高效协同工作。

2. 主流国产文本嵌入模型选型与对比

替换不是盲目地换，首先要搞清楚我们有哪些“国产炮”可用，以及各自的特点。文本嵌入模型的核心任务是将一段文本（无论长短）映射为一个固定维度的稠密向量（比如1024维）。这个向量应当能够很好地捕捉文本的语义信息，使得语义相似的文本在向量空间中的距离（通常用余弦相似度衡量）也更近。

目前，有几类国产或在国内生态中表现优异的嵌入模型值得关注：

2.1 智谱AI的 text2vec 系列

这是智谱开源的中文文本嵌入模型家族。例如text2vec-base-chinese就是一个基于BERT架构、在大量中文语料上训练的基础模型。它的特点是专为中文优化，对中文词语、句子的语义理解比较到位，开箱即用效果不错，而且模型大小相对适中（约300M参数），便于部署。

2.2 百度的 ERNIE-Embedding

百度基于其文心大模型（ERNIE）技术推出的嵌入模型。它不仅仅考虑了词语的共现，还融入了知识图谱等信息，旨在更好地理解真实世界中的实体和概念关系。对于包含大量实体、专业术语的文本（如科技文章、医疗报告），ERNIE-Embedding 可能具有优势。它通常通过百度智能云API提供，也有部分轻量版开源。

2.3 北京智源研究院的 BGE (BAAI General Embedding) 系列

尤其是BGE-M3，是近期的一个明星模型。它由北京智源人工智能研究院开源，号称是“大规模多语言、多功能、多粒度的下一代嵌入模型”。BGE-M3的强大之处在于其多功能性：它同时支持稠密向量检索、稀疏向量检索（即Lexical Search，基于词频）和多向量检索（ColBERT-like），并且对多语言（特别是中英文）支持很好。如果你的RAG系统需要混合检索（既看语义相似，也看关键词匹配），BGE-M3提供了一个“全家桶”式的解决方案。

2.4 其他优秀候选

• M3E (Moka Massive Mixed Embedding)：由 MokaAI 开源，在中文文本匹配和检索任务上表现强劲，同样在中文社区有广泛应用。
• 腾讯的 Embedding 模型：腾讯混元大模型也提供了相应的文本嵌入能力，可通过API调用，深度集成在腾讯云生态中。
• 阿里云的灵积模型服务平台：也提供了多种文本嵌入模型，方便阿里云用户直接集成。

为了更直观地对比，我们可以从几个关键维度进行考量：

注意：模型选择没有绝对的最好，只有最合适。如果你的数据全是中文，text2vec或M3E可能是最直接高效的选择。如果数据中英文混杂，或者你需要更先进的检索功能，BGE-M3值得投入时间研究。如果追求与企业现有云服务（百度、腾讯、阿里）深度集成，那么选择对应的云服务模型会更方便。

3. 实战：将 OpenAI Embedding 替换为 text2vec

理论说完，我们来点实际的。我以最经典的text2vec-base-chinese为例，展示如何在一个使用 LangChain 框架的简易 RAG 应用中，替换掉原本的 OpenAI Embedding。

假设我们原有的核心嵌入代码是这样的（使用 LangChain 和 OpenAI）：

from langchain.embeddings.openai import OpenAIEmbeddings from langchain.vectorstores import Chroma from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.document_loaders import TextLoader # 1. 加载文档并分割 loader = TextLoader("./state_of_the_union.txt") documents = loader.load() text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) texts = text_splitter.split_documents(documents) # 2. 使用 OpenAI Embedding 生成向量并存入向量库 embeddings = OpenAIEmbeddings(openai_api_key="your-api-key") # 依赖外部API vectorstore = Chroma.from_documents(documents=texts, embedding=embeddings, persist_directory="./chroma_db")

替换步骤：

3.1 安装依赖首先，需要安装text2vec和sentence-transformers（一个常用的嵌入模型调用库）。

pip install sentence-transformers # 或者直接从 Hugging Face 使用 transformers 库，但 sentence-transformers 接口更友好

3.2 创建自定义 Embeddings 类LangChain 提供了Embeddings基类，我们需要为text2vec实现一个适配器。幸运的是，sentence-transformers本身就有SentenceTransformerEmbeddings的封装，但为了更清晰，我们也可以自己写一个简单的版本。

from langchain.embeddings.base import Embeddings from sentence_transformers import SentenceTransformer from typing import List import numpy as np class Text2VecEmbeddings(Embeddings): """自定义 text2vec 嵌入类""" def __init__(self, model_name: str = "shibing624/text2vec-base-chinese", device: str = None): """ 初始化模型。 Args: model_name: Hugging Face 上的模型ID，默认为一个中文模型。 device: 指定运行设备，如 'cuda', 'cpu'。为None则自动选择。 """ self.model = SentenceTransformer(model_name, device=device) # 获取模型输出的向量维度，便于后续知晓 self.embedding_dimension = self.model.get_sentence_embedding_dimension() print(f"Loaded model '{model_name}', embedding dimension: {self.embedding_dimension}") def embed_documents(self, texts: List[str]) -> List[List[float]]: """将一组文档嵌入为向量。""" # sentence-transformers 的 encode 方法直接返回 numpy array embeddings = self.model.encode(texts, normalize_embeddings=True, # 归一化，方便计算余弦相似度 show_progress_bar=False) # 转换为 List[List[float]] 格式 return embeddings.tolist() def embed_query(self, text: str) -> List[float]: """将单个查询文本嵌入为向量。""" embedding = self.model.encode([text], normalize_embeddings=True, show_progress_bar=False)[0] return embedding.tolist() # 使用示例 my_embeddings = Text2VecEmbeddings(model_name="shibing624/text2vec-base-chinese", device="cpu")

3.3 集成到原有RAG流程现在，只需将原来初始化OpenAIEmbeddings的那一行替换成我们自定义的类即可。

# 替换前：embeddings = OpenAIEmbeddings(openai_api_key="your-api-key") # 替换后： embeddings = Text2VecEmbeddings(model_name="shibing624/text2vec-base-chinese", device="cpu") # 使用CPU运行 # 后续创建向量库的代码完全不变！ vectorstore = Chroma.from_documents(documents=texts, embedding=embeddings, persist_directory="./chroma_db_chinese")

3.4 进行检索测试替换后，务必进行检索测试，验证效果。

# 假设我们有一个问题 query = "总统在国情咨文中主要谈了哪些经济政策？" # 从向量库中检索相似文档 docs = vectorstore.similarity_search(query, k=3) print(f"检索到 {len(docs)} 个相关文档片段：") for i, doc in enumerate(docs): print(f"\n--- 片段 {i+1} ---") print(doc.page_content[:200] + "...") # 打印前200字符

实操心得：第一次运行Text2VecEmbeddings时，它会从 Hugging Face 下载模型（约300MB），需要一点时间。下载后模型会缓存到本地~/.cache/huggingface/hub目录，后续加载就很快了。另外，normalize_embeddings=True这个参数非常重要，它会对生成的向量进行L2归一化，使得点积（dot product）就等于余弦相似度，这是大多数向量数据库进行相似度计算时的默认假设。

4. 进阶：集成多功能模型 BGE-M3 与混合检索

如果你对检索效果有更高要求，特别是面对复杂查询或中英文混合内容时，BGE-M3提供了更强大的能力。它不仅生成稠密向量，还能同时生成用于稀疏检索的词汇权重（Lexical Weights）和用于多向量检索的令牌级向量（Token Vectors）。这里我们演示如何利用其稠密和稀疏检索能力，实现一个简单的混合检索。

4.1 安装与初始化 BGE-M3

pip install -U FlagEmbedding

from FlagEmbedding import BGEM3FlagModel from typing import List, Dict, Tuple import numpy as np class BGEM3Embeddings(Embeddings): """自定义 BGE-M3 嵌入类（主要使用稠密向量）""" def __init__(self, model_name: str = "BAAI/bge-m3", use_fp16: bool = False, device: str = None): self.model = BGEM3FlagModel(model_name, use_fp16=use_fp16, device=device) # BGE-M3的稠密向量维度是1024 self.embedding_dimension = 1024 def embed_documents(self, texts: List[str]) -> List[List[float]]: # 这里我们只取稠密向量（dense_vecs） embeddings = self.model.encode(texts, return_dense=True, return_sparse=False, return_colbert_vecs=False) dense_embeddings = embeddings['dense_vecs'] # 归一化 dense_embeddings = dense_embeddings / np.linalg.norm(dense_embeddings, axis=1, keepdims=True) return dense_embeddings.tolist() def embed_query(self, text: str) -> List[float]: embeddings = self.model.encode([text], return_dense=True, return_sparse=False, return_colbert_vecs=False) dense_embedding = embeddings['dense_vecs'][0] dense_embedding = dense_embedding / np.linalg.norm(dense_embedding) return dense_embedding.tolist() def encode_for_hybrid(self, texts: List[str]) -> Dict: """为混合检索编码，返回包含稠密向量和稀疏权重的字典""" return self.model.encode(texts, return_dense=True, return_sparse=True, return_colbert_vecs=False)

4.2 实现简易混合检索逻辑混合检索的核心思想是：同时进行语义检索（用稠密向量）和关键词检索（用稀疏向量/词权重），然后将两者的结果按照某种规则融合（如加权分数、RRF等）。

def hybrid_retrieval(query: str, vectorstore, bge_model: BGEM3Embeddings, dense_weight: float = 0.7, sparse_weight: float = 0.3, top_k: int = 5): """ 简易混合检索。 Args: query: 查询文本。 vectorstore: 已用BGE-M3稠密向量构建的向量库（如Chroma）。 bge_model: 初始化好的BGEM3Embeddings模型实例。 dense_weight, sparse_weight: 稠密和稀疏检索分数的权重。 top_k: 最终返回的文档数量。 """ # 1. 稠密检索（语义检索） dense_docs = vectorstore.similarity_search_with_score(query, k=top_k*2) # 多取一些，方便后续融合 dense_dict = {doc.metadata.get("id", i): (doc, score) for i, (doc, score) in enumerate(dense_docs)} # 2. 稀疏检索（关键词检索）- 这里需要自己实现一个简单的基于词权重的检索 # 首先，获取查询和所有文档的稀疏表示（在实际中，文档的稀疏表示应预先计算并存储） # 为简化，我们假设有一个函数能根据文档ID获取其预计算的稀疏向量（词权重字典） # sparse_vectors = get_precomputed_sparse_vectors(doc_ids) # 然后计算查询与每个文档的稀疏相似度（如BM25、TF-IDF等） # 这里我们用模型实时编码查询，并假设有一个简单的内存索引进行演示（生产环境需用Elasticsearch等） # 模拟：假设我们有一个简单的文档列表和对应的稀疏向量（实际应从数据库获取） all_doc_texts = ["文档1内容...", "文档2内容...", "..."] # 你的所有文档文本列表 # 预计算所有文档的稀疏表示（在生产中，这应该是一次性离线完成的） # encoded_docs = bge_model.encode_for_hybrid(all_doc_texts) # doc_sparse_weights = encoded_docs['lexical_weights'] # 这是一个列表，每个元素是字典{词:权重} # 编码查询的稀疏表示 encoded_query = bge_model.model.encode([query], return_dense=False, return_sparse=True, return_colbert_vecs=False) query_sparse_weights = encoded_query['lexical_weights'][0] # 查询词的权重字典 # 计算稀疏分数（简化版：计算查询词与文档词的权重点积） sparse_scores = [] for i, doc_text in enumerate(all_doc_texts): # 这里需要获取文档i的预计算稀疏权重 doc_weights # score = sum(query_sparse_weights.get(word, 0) * doc_weights.get(word, 0) for word in query_sparse_weights) # sparse_scores.append((i, score)) pass # 实际实现需要完整的索引和计算 # 3. 分数融合 (Score Fusion) # 假设我们得到了 sparse_scores: [(doc_id, sparse_score), ...] # 将稠密检索和稀疏检索的分数归一化到同一尺度，然后加权求和 # normalized_dense_score = (dense_score - min_dense) / (max_dense - min_dense) # normalized_sparse_score = (sparse_score - min_sparse) / (max_sparse - min_sparse) # final_score = dense_weight * normalized_dense_score + sparse_weight * normalized_sparse_score # 4. 按最终分数排序，返回 top_k 个文档 # sorted_docs = sorted(combined_results, key=lambda x: x['final_score'], reverse=True)[:top_k] # 由于稀疏检索实现较复杂，此处仅提供融合思路。实际应用中，可以使用现成的库（如Elasticsearch的hybrid search）或更成熟的框架。 print("混合检索逻辑框架已说明，具体实现需结合你的数据存储和索引方式。") # 作为 fallback，先返回稠密检索的结果 return [doc for doc, _ in dense_docs[:top_k]] # 使用示例（需完善稀疏检索部分） bge_embeddings = BGEM3Embeddings(device='cpu') # 假设 vectorstore_bge 是用 bge_embeddings.embed_documents 构建的 # results = hybrid_retrieval("你的查询", vectorstore_bge, bge_embeddings)

踩坑提醒：实现生产级的混合检索（Hybrid Search）是一个系统工程。BGE-M3虽然提供了稀疏向量，但你需要一个能同时高效处理稠密向量和稀疏倒排索引的数据库。Chroma目前对稀疏检索的支持还在发展中。更成熟的选择是Weaviate(支持混合检索)、Elasticsearch(通过插件或自定义脚本)、Qdrant(支持稀疏向量) 或Milvus(需要结合其他组件)。在原型阶段，可以先用稠密检索，待效果评估稳定后再引入混合检索。

5. 效果评估与调优：如何判断替换是否成功？

模型换完了，代码跑通了，但效果到底怎么样？不能凭感觉，需要有量化的评估。对于RAG系统，嵌入模型的好坏直接影响检索质量，进而影响最终答案的准确性。

5.1 构建评估数据集首先，你需要一个小的测试集。这个测试集应该来自你的实际业务数据或相近的领域。至少包含：

• 一组查询（Queries）：用户可能提出的问题，例如20-50个。
• 对应的标准答案/相关文档片段（Ground Truth）：每个查询，人工标注出知识库中哪些文档片段是真正相关的。

5.2 核心评估指标

1. 检索阶段指标：评估嵌入模型找到相关文档的能力。

   • 命中率（Hit Rate @ k）：对于每个查询，在前k个检索结果中，至少出现一个相关文档的比例。例如，Hit@3=0.9，表示90%的查询，其前3个结果里至少有一个是相关的。
   • 平均倒数排名（Mean Reciprocal Rank, MRR）：衡量相关文档出现位置的指标。对于每个查询，第一个相关文档出现的位置的倒数（如第一个相关文档排第2位，得分1/2=0.5），然后对所有查询取平均。MRR越高，说明相关文档排得越靠前。

2. 端到端RAG指标：将检索到的文档送给LLM生成答案后，评估最终答案的质量。

   • 答案相关性（Answer Relevance）：生成的答案与查询问题的匹配程度。
   • 事实一致性（Faithfulness）：生成的答案是否严格基于检索到的文档，没有“胡编乱造”。
   • 这些指标通常需要人工评估，或者使用像RAGAS、TruLens这样的评估框架，利用LLM本身作为裁判进行自动化评估。

5.3 执行A/B测试将原来的OpenAI Embedding系统（A组）和新的国产模型系统（B组）在同一个测试集上跑一遍，对比上述指标。

# 伪代码：评估 Hit Rate @ 3 def evaluate_hit_rate(vectorstore, queries, ground_truth, k=3): """ vectorstore: 向量数据库实例 queries: 查询列表 ground_truth: 字典，key为查询索引，value为相关文档的ID列表 """ hit_count = 0 for idx, query in enumerate(queries): retrieved_docs = vectorstore.similarity_search(query, k=k) retrieved_ids = [doc.metadata.get("doc_id") for doc in retrieved_docs] # 检查是否有检索到的ID在标准答案ID列表中 if any(True for rid in retrieved_ids if rid in ground_truth[idx]): hit_count += 1 hit_rate = hit_count / len(queries) return hit_rate # 分别对 OpenAI 和 text2vec 构建的向量库进行评估 # hit_rate_openai = evaluate_hit_rate(vectorstore_openai, test_queries, ground_truth) # hit_rate_text2vec = evaluate_hit_rate(vectorstore_text2vec, test_queries, ground_truth) # print(f"OpenAI Hit@{k}: {hit_rate_openai:.4f}") # print(f"Text2Vec Hit@{k}: {hit_rate_text2vec:.4f}")

5.4 针对性调优如果评估发现国产模型在某些查询上表现不佳，可以考虑以下调优手段：

• 文本分块策略：嵌入模型对输入长度敏感。尝试调整chunk_size和chunk_overlap。对于中文，可能更适合按句号、换行符分割，而不是单纯按字符数。
• 提示工程（Prompt Engineering）：在将检索到的上下文送给LLM生成答案时，优化你的提示词（Prompt），明确指示LLM基于给定的上下文回答。
• 模型微调（Fine-tuning）：如果领域数据非常特殊（如大量专业术语、行业黑话），且你有足够的标注数据（文本对及其相似度分数），可以考虑对开源的国产嵌入模型进行微调，让它更适应你的领域。SentenceTransformers库提供了完善的微调接口。
• 重排序（Re-ranking）：在初步检索出较多文档（如20个）后，使用一个更精细的、专门用于句子对排序的模型（Cross-Encoder）对结果进行重新排序，再将Top结果送给LLM。这能显著提升最终答案的质量。BGE-M3本身也具备重排序能力。

经验之谈：不要期望国产模型在第一个版本就全面超越OpenAI。评估时，重点关注在你的数据和你的场景下的表现。有时，一个在通用基准上分数稍低的模型，因为对中文语言习惯或你的领域术语捕捉更好，实际效果反而更佳。替换是一个迭代过程，评估-调整-再评估是关键。

6. 工程化部署与性能考量

当原型验证通过，准备将使用国产嵌入模型的RAG应用投入生产时，就需要考虑工程化部署的问题了。

6.1 部署模式选择

• 本地部署：将模型文件（.bin或.pth）下载到服务器，使用sentence-transformers或transformers库加载。这是数据最安全、网络延迟最低的方式，但需要消耗本地计算资源（CPU/GPU）。
  • CPU推理：对于text2vec-base这类轻量模型，在现代化的CPU上推理速度也是可接受的（单条文本几十到几百毫秒）。适用于并发不高的场景。
  • GPU推理：如果文档数量多、需要实时处理高并发请求，必须使用GPU。一张消费级的RTX 4090甚至RTX 3090就能让BGE-M3这类模型的推理速度提升一个数量级。
• 模型服务化（Model as a Service）：使用Triton Inference Server、TensorFlow Serving或FastAPI + Uvicorn将模型封装成HTTP/gRPC服务。这样，你的RAG应用后端通过网络调用这个嵌入服务，实现了解耦和弹性伸缩。你可以独立扩缩容嵌入服务，而不影响应用其他部分。
• 使用云服务商提供的模型API：如果不想管理基础设施，可以直接使用百度、阿里、腾讯等云厂商提供的嵌入模型API。这相当于把OpenAI的模式平替到了国内云，优点是省心，缺点是有调用费用和网络延迟（虽然国内网络快很多）。

6.2 性能优化技巧

1. 批处理（Batch Inference）：这是提升吞吐量最有效的手段。不要一次只编码一条文本，而是积累一定数量（如32、64条）后一次性送入模型。model.encode()方法天然支持批处理。

   # 低效 vectors = [] for text in text_list: vec = embeddings.embed_documents([text])[0] vectors.append(vec) # 高效 vectors = embeddings.embed_documents(text_list) # 一次性编码整个列表

2. 量化（Quantization）：将模型从FP32精度转换为INT8甚至INT4精度，可以大幅减少模型体积和内存占用，提升推理速度，而精度损失通常很小。可以使用bitsandbytes或onnxruntime进行量化。

   # 使用 optimum 和 onnxruntime 进行量化示例（步骤较多，需参考官方文档） pip install optimum onnxruntime-gpu # 然后可以将模型导出为ONNX格式并进行量化

3. 缓存（Caching）：对于不变的文档库，其向量化结果可以永久缓存。对于频繁出现的相似查询，也可以缓存其查询向量和检索结果。可以使用Redis或Memcached。

4. 向量数据库优化：选择高性能的向量数据库（如Milvus、Weaviate、Qdrant），并合理创建索引（如HNSW、IVF_FLAT）。索引的构建参数（如M、ef_construction对于HNSW）需要根据数据规模和查询延迟要求进行调优。

6.3 监控与日志在生产环境中，必须对嵌入服务进行监控。

• 延迟监控：记录每个嵌入请求的耗时（P50, P95, P99）。
• 吞吐量监控：统计每秒处理的请求数（QPS）。
• 准确性监控：定期（如每周）在测试集上跑一遍评估指标，确保模型效果没有因为数据分布漂移而下降。
• 错误日志：详细记录模型加载失败、推理错误、输入过长被截断等情况。

将嵌入模型替换为国产模型，并成功上线，只是第一步。接下来，你可以探索更高级的特性，比如根据查询动态选择不同的检索策略（Dense, Sparse, Hybrid），或者实现一个多阶段的检索-重排序管道，持续优化你的RAG系统，使其在成本、性能和效果上找到最佳平衡点。