介绍 AI 大模型中嵌入模型的性能优化方案。针对嵌入计算成本高、重复计算浪费、API 限制及响应速度瓶颈等痛点，提出使用缓存机制的核心价值。详细解析 LangChain 提供的 CacheBackedEmbeddings 缓存装饰器，涵盖技术架构、核心参数、支持的存储类型（本地文件、Redis 等）。通过智能客服知识库案例，对比无缓存与有缓存方案的代码实现及性能差异，展示耗时降低 98.7% 的效果。最后给出适用场景、存储选择策略及进阶优化技巧，帮助开发者降低 API 成本并提升系统响应速度。
在 RAG 系统中，嵌入模型的作用是将文本（文档/查询）转换为高维向量，为后续的相似度检索提供基础。但在实际应用中，嵌入计算往往会成为系统的性能瓶颈，这也是面试中经常被问到的'RAG 系统优化痛点'之一。
先看一组真实场景数据：某智能客服知识库包含 10 万条 QA 对，使用 OpenAI Embeddings 计算嵌入时，单次全量计算需要消耗约 200 元 API 费用，单条查询响应时间约 800ms，且每天因重复计算浪费 30% 的资源——这正是大多数开发者在落地 RAG 时会遇到的问题。
在 RAG 系统中，嵌入模型的作用是将文本（文档/查询）转换为高维向量，为后续的相似度检索提供基础。但在实际应用中，嵌入计算往往会成为系统的性能瓶颈，这也是面试中经常被问到的'RAG 系统优化痛点'之一。
针对上述痛点，缓存（Cache） 是最直接有效的优化方案。其核心逻辑是：将首次计算的嵌入结果存储起来，后续遇到相同文本时直接读取缓存，无需重复调用模型。具体优势如下：
LangChain 作为大模型开发的'瑞士军刀'，提供了专门的缓存装饰器——CacheBackedEmbeddings，可无缝集成各类嵌入模型和存储介质，无需手动实现缓存逻辑。
注：CacheBackedEmbeddings采用文本哈希生成唯一键（默认使用 SHA-256），确保相同文本对应唯一缓存键，避免冲突。
# 导入核心组件
from langchain.embeddings import CacheBackedEmbeddings, OpenAIEmbeddings
from langchain.storage import LocalFileStore
# 1. 初始化原始嵌入模型（如 OpenAI Embeddings）
embedding_model = OpenAIEmbeddings(openai_api_key="sk-xxx")
# 2. 初始化缓存存储（以本地文件存储为例）
storage = LocalFileStore("./embedding_cache/") # 缓存文件存储目录
# 3. 组合缓存与嵌入模型（装饰器模式）
cached_embedder = CacheBackedEmbeddings(
underlying_embeddings=embedding_model, # 原始嵌入模型
document_embedding_store=storage, # 缓存存储对象
namespace="openai-v3" # 可选：命名空间（隔离不同项目/模型版本）
)
| 参数名 | 类型 | 作用说明 |
|---|---|---|
| underlying_embeddings | Embeddings | 原始嵌入模型实例（如 OpenAIEmbeddings、DashScopeEmbeddings、本地模型等） |
| document_embedding_store | BaseStore | 缓存存储实现类（LangChain 提供多种开箱即用的存储方案） |
| namespace | str | 缓存命名空间，用于隔离不同项目或模型版本（如'openai-v3'和'openai-v2'分开存储） |
LangChain 的 langchain.storage 模块提供多种存储方案，适配不同场景：
# langchain.storage 支持的存储类型
__all__ = [
"InMemoryStore", # 内存存储（最快，重启丢失）
"LocalFileStore", # 本地文件存储（零配置，易调试）
"RedisStore", # Redis 存储（分布式，高并发）
"UpstashRedisStore", # Upstash Redis（Serverless，无需运维）
"EncoderBackedStore" # 自定义编码存储（支持复杂数据类型）
]
以'10 万条 QA 对的智能客服知识库'为例，对比无缓存和有缓存方案的差异。
from langchain.embeddings import OpenAIEmbeddings
# 初始化模型
embedder = OpenAIEmbeddings(openai_api_key="sk-xxx")
# 每次请求都重新计算嵌入（即使文本重复）
def get_embedding(text):
return embedder.embed_documents([text]) # 批量接口，单文本需用列表包裹
# 第一次调用：计算嵌入（800ms 左右）
vector1 = get_embedding("如何重置密码？")
# 第二次调用：重复计算（同样 800ms 左右，浪费资源）
vector2 = get_embedding("如何重置密码？")
问题：10 万条 QA 对每次全量更新需计算 10 万次，API 费用高且耗时久；用户重复提问时，响应速度无优化。
from langchain.embeddings import CacheBackedEmbeddings, OpenAIEmbeddings
from langchain.storage import LocalFileStore
# 1. 初始化组件
embedder = OpenAIEmbeddings(openai_api_key="sk-xxx")
storage = LocalFileStore("./kb_embedding_cache/") # 知识库缓存目录
cached_embedder = CacheBackedEmbeddings(
underlying_embeddings=embedder,
document_embedding_store=storage,
namespace="customer-service-kb" # 命名空间：隔离客服知识库缓存
)
# 2. 缓存优化的嵌入获取函数
def get_cached_embedding(text):
return cached_embedder.embed_documents([text])
# 第一次调用：未命中缓存，计算并存储（800ms 左右）
vector1 = get_cached_embedding("如何重置密码？")
print(f"首次调用嵌入维度：{len(vector1[0])}") # 输出：1536（OpenAI Embeddings 维度）
# 第二次调用：命中缓存，直接读取（10ms 左右）
vector2 = get_cached_embedding("如何重置密码？")
print(f"结果一致性：{vector1 == vector2}") # 输出：True（向量完全一致）
优化效果：首次全量预计算后，后续查询响应时间从 800ms 降至 10ms，API 调用次数减少 99%，成本降低 90% 以上。
对于集群部署的生产环境，本地文件存储无法共享缓存，推荐使用 Redis 存储（支持高并发、分布式共享、TTL 过期策略）：
# 安装依赖：pip install redis langchain
from redis import Redis
from langchain.embeddings import CacheBackedEmbeddings, OpenAIEmbeddings
from langchain.storage import RedisStore
# 1. 连接 Redis（本地或远程集群）
redis_client = Redis(
host="localhost",
port=6379,
password="xxx",
db=0
)
# 2. 初始化 Redis 缓存（设置 24 小时过期，避免缓存膨胀）
redis_store = RedisStore(redis_client, ttl=86400) # ttl：缓存过期时间（秒）
# 3. 初始化带 Redis 缓存的嵌入模型
embedder = OpenAIEmbeddings(openai_api_key="sk-xxx")
cached_embedder = CacheBackedEmbeddings(
underlying_embeddings=embedder,
document_embedding_store=redis_store,
namespace="prod-rag-kb" # 生产环境命名空间
)
# 调用方式与本地缓存一致，支持多节点共享缓存
vector = cached_embedder.embed_documents(["如何查询订单物流？"])
CacheBackedEmbeddings对两个核心接口的缓存策略不同，需根据场景选择：
| 接口名 | 作用场景 | 缓存策略 | 设计考量 |
|---|---|---|---|
| embed_documents | 批量处理文档（如知识库构建、预计算） | 默认开启缓存 | 文档重复率高，缓存收益大；批量处理可分摊缓存读写开销 |
| embed_query | 处理用户实时查询（如'如何重置密码？'） | 默认不缓存 | 用户查询多样性高，缓存命中率低，反而增加存储开销和延迟 |
以阿里云 DashScope Embeddings 为例，对比缓存前后的调用耗时：
from langchain.embeddings import CacheBackedEmbeddings, DashScopeEmbeddings
from langchain.storage import LocalFileStore
import time
# 1. 初始化组件
embedding_model = DashScopeEmbeddings(
model="text-embedding-v2",
dashscope_api_key="sk-xxx",
max_retries=3
)
storage = LocalFileStore("./dashscope_cache/")
cached_embeddings = CacheBackedEmbeddings.from_bytes_store(
underlying_embeddings=embedding_model,
document_embedding_store=storage,
namespace="dashscope-v2"
)
# 2. 测试文本（故意重复，模拟知识库重复内容）
texts = ["AI 大模型开发实战", "AI 大模型开发实战"]
# 3. 首次调用（未命中缓存）
start_time = time.time()
emb1 = cached_embeddings.embed_documents(texts)
first_cost = time.time() - start_time
print(f"首次调用：嵌入维度={len(emb1[0])}，耗时={first_cost:.2f}s")
# 4. 二次调用（命中缓存）
start_time = time.time()
emb2 = cached_embeddings.embed_documents(texts)
second_cost = time.time() - start_time
print(f"二次调用：结果一致={emb1 == emb2}，耗时={second_cost:.2f}s")
首次调用：嵌入维度=768，耗时=0.78s
二次调用：结果一致=True，耗时=0.01s
结论：缓存后耗时降低 98.7%，效果显著！
| 存储类型 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| LocalFileStore | 零配置、易调试、无需额外依赖 | 不支持分布式、并发性能差 | 本地开发、单节点测试 |
| RedisStore | 高并发、分布式共享、支持 TTL | 需要部署 Redis、运维成本高 | 生产环境、集群部署 |
| InMemoryStore | 速度最快（内存读写） | 重启丢失、不支持分布式 | 临时测试、短期缓存 |
| UpstashRedisStore | Serverless、无需运维 | 云服务收费、依赖网络 | 中小规模生产环境、快速部署 |
嵌入模型的缓存优化是 RAG 系统落地的关键步骤，通过 CacheBackedEmbeddings 可快速实现'一次计算、多次复用'，显著降低成本、提升响应速度。面试中遇到'RAG 系统性能优化'问题时，可从'缓存机制 + 存储选型 + 预计算策略'三个维度展开，结合本文案例能体现你的实战经验～
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
使用加密算法（如AES、TripleDES、Rabbit或RC4）加密和解密文本明文。 在线工具，加密/解密文本在线工具，online
生成新的随机RSA私钥和公钥pem证书。 在线工具，RSA密钥对生成器在线工具，online
基于 Mermaid.js 实时预览流程图、时序图等图表，支持源码编辑与即时渲染。 在线工具，Mermaid 预览与可视化编辑在线工具，online
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online