langchain.embeddings.CacheBackedEmbeddings 类是 LangChain 中用于缓存嵌入向量（embeddings）的工具类，通过存储已计算的文本嵌入结果来避免重复计算，从而提高效率和降低成本。

本文基于 LangChain 0.3.x，详细介绍 CacheBackedEmbeddings 的定义、参数、方法和典型场景，并提供一个独立示例，展示如何使用 CacheBackedEmbeddings 结合 ChatOpenAI 和向量存储处理人工智能主题的问答任务，示例加载 AI 相关文档，使用缓存嵌入优化 RAG 链的性能。

langchain.embeddings.CacheBackedEmbeddings 简介

CacheBackedEmbeddings 是 LangChain 的嵌入包装器类，通过将文本的嵌入向量存储在缓存（支持多种后端，如内存、文件或数据库）中，复用已计算的结果，避免重复调用嵌入模型（如 OpenAIEmbeddings）。它特别适合处理大规模文档或重复文本，降低 API 调用成本和计算时间。

核心功能：

  • 缓存文本的嵌入向量，复用现有结果。
  • 支持多种缓存后端（如 InMemoryStore、LocalFileStore、RedisStore）。
  • 包装现有嵌入模型，保持接口一致。
  • 提供同步和异步嵌入方法。

适用场景：

  • 优化 RAG 系统，减少嵌入计算开销。
  • 处理大规模文档集，复用重复文本的嵌入。
  • 降低商业嵌入 API（如 OpenAI）的调用成本。
  • 在开发或测试中加速嵌入生成。

与普通嵌入模型对比：

  • 普通嵌入模型（如 OpenAIEmbeddings）：每次计算嵌入。
  • CacheBackedEmbeddings：缓存结果，仅首次计算。

类定义和初始化

以下是 CacheBackedEmbeddings 的定义，基于 LangChain 源码（langchain/embeddings/cache.py）和官方文档（CacheBackedEmbeddings)。

类签名
class CacheBackedEmbeddings(Embeddings):
    def __init__(
        self,
        underlying_embeddings: Embeddings,
        document_embedding_cache: BaseStore[str, List[float]],
        namespace: str = "",
        **kwargs: Any
    ) -> None
  • 参数：
    • underlying_embeddings（Embeddings）：底层嵌入模型（如 OpenAIEmbeddings）。
    • document_embedding_cache（BaseStore[str, List[float]]）：缓存存储，键为文本，值为嵌入向量。
    • namespace（str，默认 ""）：缓存键的命名空间，用于区分不同任务。
    • **kwargs：传递给 Embeddings 基类的其他参数。
  • 功能：
    • 检查缓存中是否存在文本的嵌入。
    • 若存在，从缓存获取；若不存在，调用 underlying_embeddings 计算并存储。
初始化示例
from langchain.embeddings import CacheBackedEmbeddings
from langchain_openai import OpenAIEmbeddings
from langchain.storage import InMemoryStore
underlying_embeddings = OpenAIEmbeddings()
cache = InMemoryStore()
cached_embeddings = CacheBackedEmbeddings(
    underlying_embeddings=underlying_embeddings,
    document_embedding_cache=cache,
    namespace="ai_docs"
)

常用方法

CacheBackedEmbeddings 继承自 langchain_core.embeddings.Embeddings，提供以下核心方法。

1. embed_documents
def embed_documents(self, texts: List[str]) -> List[List[float]]
  • 功能：同步嵌入文档列表，返回嵌入向量列表。
  • 输入：texts（List[str]），待嵌入的文本列表。
  • 输出：List[List[float]]，每个文本的嵌入向量。
  • 行为：
    • 检查缓存，若命中则返回缓存结果。
    • 否则调用 underlying_embeddings.embed_documents 并缓存结果。
  • 示例： 
    texts = ["人工智能是计算机科学的一个分支。", "AI 在医疗中应用广泛。"]
embeddings = cached_embeddings.embed_documents(texts)
print(len(embeddings))  # 2
2. embed_query
def embed_query(self, text: str) -> List[float]
  • 功能：同步嵌入单个查询文本，返回嵌入向量。
  • 输入：text（str），查询文本。
  • 输出：List[float]，文本的嵌入向量。
  • 示例： 
    embedding = cached_embeddings.embed_query("人工智能是什么？")
print(len(embedding))  # 嵌入维度（如 1536）
3. aembed_documents
async def aembed_documents(self, texts: List[str]) -> List[List[float]]
  • 功能：异步嵌入文档列表。
  • 使用场景：高并发场景。
  • 示例： 
    import asyncio
embeddings = await cached_embeddings.aembed_documents(texts)
4. aembed_query
async def aembed_query(self, text: str) -> List[float]
  • 功能：异步嵌入单个查询文本。
  • 示例： 
    embedding = await cached_embeddings.aembed_query("人工智能是什么？")

使用方式

以下是使用 CacheBackedEmbeddings 的步骤。

1. 安装依赖
pip install --upgrade langchain langchain-openai faiss-cpu
2. 设置 OpenAI API 密钥
export OPENAI_API_KEY="your-api-key"

或在代码中：

import os
os.environ["OPENAI_API_KEY"] = "your-api-key"
3. 准备文档
from langchain_core.documents import Document
docs = [
    Document(page_content="人工智能是计算机科学的一个分支。"),
    Document(page_content="AI 在医疗中应用广泛。")
]
4. 初始化 CacheBackedEmbeddings
from langchain.embeddings import CacheBackedEmbeddings
from langchain_openai import OpenAIEmbeddings
from langchain.storage import InMemoryStore
underlying_embeddings = OpenAIEmbeddings()
cache = InMemoryStore()
cached_embeddings = CacheBackedEmbeddings(
    underlying_embeddings=underlying_embeddings,
    document_embedding_cache=cache,
    namespace="ai_docs"
)
5. 集成到向量存储
from langchain.vectorstores import FAISS
vectorstore = FAISS.from_documents(docs, cached_embeddings)
6. 构建 RAG 链
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
retriever = vectorstore.as_retriever()
llm = ChatOpenAI(model="gpt-3.5-turbo")
prompt = ChatPromptTemplate.from_template("根据以下上下文回答问题：\n{context}\n问题：{question}")
chain = {"context": retriever, "question": lambda x: x["question"]} | prompt | llm
7. 调用链
result = chain.invoke({"question": "人工智能是什么？"})
print(result.content)

使用 CacheBackedEmbeddings 的示例

以下是一个独立示例，展示如何使用 CacheBackedEmbeddings 加载 AI 相关文档，结合 FAISS 向量存储和 ChatOpenAI 构建 RAG 链，回答用户关于人工智能的查询。示例使用 LocalFileStore 持久化缓存，优化嵌入性能。

准备环境：

  • 获取 OpenAI API 密钥：OpenAI Platform ，并设置环境变量： 
    export OPENAI_API_KEY="your-api-key"
  • 安装依赖： 
    pip install --upgrade langchain langchain-openai faiss-cpu
  • 创建目录 ai_docs 并添加文件：
    • ai_docs/doc1.txt： 
      人工智能（AI）是计算机科学的一个分支，旨在模拟人类智能，如学习和推理。
    • ai_docs/doc2.txt： 
      AI 在医疗中的应用包括诊断、药物研发和个性化治疗。

代码：

from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain.embeddings import CacheBackedEmbeddings
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.storage import LocalFileStore
from langchain.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel

# 初始化 DirectoryLoader
loader = DirectoryLoader(
    path="./ai_docs",
    glob="*.txt",
    loader_cls=TextLoader,
    loader_kwargs={"encoding": "utf-8"},
    show_progress=True
)

# 加载文档
docs = loader.load()

# 初始化 CacheBackedEmbeddings
underlying_embeddings = OpenAIEmbeddings()
cache = LocalFileStore("./cache/ai_embeddings")
cached_embeddings = CacheBackedEmbeddings(
    underlying_embeddings=underlying_embeddings,
    document_embedding_cache=cache,
    namespace="ai_docs"
)

# 初始化向量存储
vectorstore = FAISS.from_documents(docs, cached_embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# 初始化 ChatOpenAI
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7)

# 定义提示模板
prompt = ChatPromptTemplate.from_template(
    "根据以下上下文回答问题：\n{context}\n问题：{question}"
)

# 定义输出解析器
parser = StrOutputParser()

# 创建 RAG 链
chain = RunnableParallel({
    "context": retriever,
    "question": lambda x: x["question"]
}) | prompt | llm | parser

# 测试 CacheBackedEmbeddings 和 RAG 链
print("测试 CacheBackedEmbeddings 和 RAG 链：")
try:
    question = "人工智能是什么？"
    result = chain.invoke({"question": question})
    
    print(f"输入问题: {question}")
    print(f"回答: {result}")
    
    # 显示加载的文档
    print("\n加载的文档：")
    for i, doc in enumerate(docs):
        print(f"文档 {i+1}:")
        print(f"内容: {doc.page_content[:100]}...")  # 截取前 100 字符
        print(f"元数据: {doc.metadata}")
except Exception as e:
    print(f"错误: {e}")

输出示例（实际输出取决于模型和 API 响应）：

测试 CacheBackedEmbeddings 和 RAG 链：
输入问题: 人工智能是什么？
回答: 人工智能（AI）是计算机科学的一个分支，旨在模拟人类智能，如学习、推理和问题解决。

加载的文档：
文档 1:
内容: 人工智能（AI）是计算机科学的一个分支，旨在模拟人类智能，如学习和推理。...
元数据: {'source': 'ai_docs/doc1.txt'}
文档 2:
内容: AI 在医疗中的应用包括诊断、药物研发和个性化治疗。...
元数据: {'source': 'ai_docs/doc2.txt'}
代码说明
  1. DirectoryLoader：
    • 加载 ai_docs 目录中的 *.txt 文件，使用 TextLoader。
    • 启用 show_progress 显示加载进度。
  2. CacheBackedEmbeddings：
    • 使用 OpenAIEmbeddings 作为底层模型。
    • 使用 LocalFileStore 持久化缓存到 ./cache/ai_embeddings。
    • 设置 namespace="ai_docs" 区分任务。
  3. 向量存储：
    • 使用 FAISS 和 cached_embeddings 创建向量存储。
    • 配置 retriever 检索前 2 个最相似文档。
  4. LLM 初始化：
    • 使用 ChatOpenAI 调用 gpt-3.5-turbo，设置 temperature=0.7。
  5. 提示模板：
    • ChatPromptTemplate 组合检索上下文和用户问题。
  6. RAG 链：
    • 使用 RunnableParallel 传递 context（检索结果）和 question。
    • LCEL 链连接 prompt、llm 和 parser，生成回答。
  7. 测试：
    • 测试 RAG 链，回答 AI 定义问题。
    • 显示加载的文档，展示 page_content 和 metadata。
  8. 错误处理：
    • 使用 try-except 捕获文件、API 或嵌入错误。

运行要求：

  • 有效的 OpenAI API 密钥： 
    export OPENAI_API_KEY="your-api-key"
  • 安装依赖： 
    pip install --upgrade langchain langchain-openai faiss-cpu
  • 准备 ai_docs 目录和文本文件。
  • 网络连接：访问 https://api.openai.com.

注意事项

  1. API 密钥：
    • 确保 OPENAI_API_KEY 已设置： 
      echo $OPENAI_API_KEY
    • 或在代码中设置： 
      llm = ChatOpenAI(api_key="your-api-key")
  2. 缓存后端：
    • 内存缓存（InMemoryStore）：适合临时任务： 
      cache = InMemoryStore()
    • 文件缓存（LocalFileStore）：持久化结果： 
      cache = LocalFileStore("./cache")
    • 清理缓存： 
      rm -rf ./cache
  3. 嵌入模型：
    • 使用高效模型： 
      underlying_embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    • 开源替代： 
      from langchain_huggingface import HuggingFaceEmbeddings
underlying_embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
  4. 性能优化：
    • 异步嵌入：使用 aembed_documents： 
      embeddings = await cached_embeddings.aembed_documents(texts)
    • 批量处理：分批嵌入大文档集： 
      batch_size = 100
for i in range(0, len(docs), batch_size):
    vectorstore.add_documents(docs[i:i+batch_size])
    • 缓存检查：验证缓存命中： 
      print(cache.mget(["text1", "text2"]))
  5. 错误调试：
    • 缓存错误：
      • 检查缓存路径： 
        import os
print(os.path.exists("./cache"))
      • 验证缓存内容： 
        print(list(cache.yield_keys()))
    • API 错误：
      • 检查密钥： 
        print(os.environ.get("OPENAI_API_KEY"))
      • 增加超时： 
        llm = ChatOpenAI(timeout=30)
    • 检索失败：
      • 检查文档： 
        print(len(docs))
      • 调整 k： 
        retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

常见问题

Q1：如何切换缓存后端？
A：使用不同 BaseStore：

from langchain.storage import RedisStore
cache = RedisStore(redis_url="redis://localhost:6379")
cached_embeddings = CacheBackedEmbeddings(underlying_embeddings, cache)

Q2：如何清除缓存？
A：删除缓存内容：

cache.mdelete(list(cache.yield_keys()))
# 或删除文件缓存
import shutil
shutil.rmtree("./cache")

Q3：如何与少样本提示结合？
A：使用缓存嵌入生成示例：

from langchain_core.prompts import FewShotPromptTemplate
examples = [{"content": doc.page_content} for doc in docs]
embeddings = cached_embeddings.embed_documents([ex["content"] for ex in examples])
prompt = FewShotPromptTemplate(
    examples=examples,
    example_prompt=PromptTemplate.from_template("文档：{content}"),
    prefix="根据文档回答：",
    suffix="问题：{question}",
    input_variables=["question"]
)

Q4：如何支持开源模型？
A：使用 ChatOllama 和开源嵌入：

from langchain_ollama import ChatOllama
from langchain_huggingface import HuggingFaceEmbeddings
underlying_embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
cached_embeddings = CacheBackedEmbeddings(underlying_embeddings, cache)
llm = ChatOllama(model="llama3")
chain = {"context": retriever, "question": lambda x: x["question"]} | prompt | llm | parser

总结

langchain.embeddings.CacheBackedEmbeddings 是 LangChain 中优化嵌入计算的核心工具，核心功能包括：

  • 定义：缓存嵌入向量，复用结果。
  • 初始化：配置 underlying_embeddings 和 document_embedding_cache。
  • 常用方法：embed_documents（同步）、embed_query（查询）、aembed_documents（异步）。
  • 适用场景：RAG 优化、成本降低、大规模文档处理。
# 缓存
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

【免费下载】 index-tts-vllm：高性能文本转语音解决方案

`index-tts-vllm` 是一款开源的文本转语音（TTS）项目，基于 `index-tts` 进行了优化，通过引入 `vllm` 库重新实现了 GPT 模型的推理过程，大幅提升了推理速度和并发处理能力。项目旨在为开发者提供一个高效、稳定、易于集成的 TTS 解决方案。## 项目技术分析`index-tts-vllm` 采用了以下核心技术：- **vllm 库**：用于加速 GP

avatar AI Agent技术社区
cover

2025年8月SCI-炭黑四角蜂优化算法 Tetragonula carbonaria Optimization Algorithm-附Matlab免费代码

avatar AI Agent技术社区

AgentScope示例应用：多智能体对话

在人工智能快速发展的今天，单一智能体已经难以满足复杂任务的需求。想象一下这样的场景：一个项目团队需要讨论技术方案，不同专业背景的成员各抒己见，最终达成共识。这正是多智能体对话要解决的问题——让多个AI智能体像真实团队成员一样进行有意义的交流。AgentScope作为面向开发者的多智能体编程框架，提供了强大的多智能体对话能力。本文将深入探讨如何使用AgentScope构建高效的多智能体对话系统。..

avatar AI Agent技术社区