LangChain嵌入模型组件介绍！

月伴飞鱼2025-10-142025-10-14

OpenAI Embedding 嵌入模型

OpenAI 服务提供商提供了线上的 Embeddings 服务。

OpenAI Embeddings 官网文档：https://platform.openai.com/docs/guides/embeddings

OpenAI 的 Embeddings 嵌入模型虽然是市面上效果最好的嵌入模型。

价格非常低廉，不过由于没有本地版本，而且接口响应速度相对较慢，在需要对大量文档进行嵌入时，效率非常低。

所以国内使用得并不多（国内一般会使用对应的本地或者本地服务提供商的嵌入模型）。

Embedding 组件使用

在 LangChain 中，Embeddings 类是一个专为与文本嵌入模型交互而设计的类。

这个类为许多嵌入模型提供商（如 OpenAI、Cohere、HuggingFace 等）提供一个标准的接口。

LangChain 中 Embeddings 类提供了两种方法：

embed_documents：

用于嵌入文档列表，传入一个文档列表，得到这个文档列表对应的向量列表。

embed_query：

用于嵌入单个查询，传入一个字符串，得到这个字符串对应的向量。

并且 Embeddings 类并不是一个 Runnable 组件，所以并不能直接接入到 Runnable 序列链中。

需要额外的 RunnableLambda 函数进行转换。

class Embeddings(ABC):
    """Interface for embedding models."""

    @abstractmethod
    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        """Embed search docs."""

    @abstractmethod
    def embed_query(self, text: str) -> List[float]:
        """Embed query text."""

    async def aembed_documents(self, texts: List[str]) -> List[List[float]]:
        """Asynchronous Embed search docs."""
        return await run_in_executor(None, self.embed_documents, texts)

    async def aembed_query(self, text: str) -> List[float]:
        """Asynchronous Embed query text."""
        return await run_in_executor(None, self.embed_query, text)

如果想对接自定义的 Embedding 嵌入模型，只需实现 embed_documents 和 embed_query 这两个方法即可。

LangChain 使用 OpenAI Embeddings 嵌入模型示例：

import dotenv
import numpy as np
from langchain_openai import OpenAIEmbeddings
from numpy.linalg import norm

dotenv.load_dotenv()


def cosine_similarity(vector1: list, vector2: list) -> float:
    """计算传入两个向量的余弦相似度"""
    # 1.计算内积/点积
    dot_product = np.dot(vector1, vector2)

    # 2.计算向量的范数/长度
    norm_vec1 = norm(vector1)
    norm_vec2 = norm(vector2)

    # 3.计算余弦相似度
    return dot_product / (norm_vec1 * norm_vec2)


embeddings = OpenAIEmbeddings()

query_vector = embeddings.embed_query("你好，我是openai，我喜欢打篮球")
documents_vector = embeddings.embed_documents([
    "你好，我是openai，我喜欢打篮球",
    "这个喜欢打篮球的人叫openai",
    "求知若渴，虚心若愚"
])

print(query_vector)
print(len(query_vector))

print("============")

print(len(documents_vector))
print("vector1与vector2的余弦相似度:", cosine_similarity(documents_vector[0], documents_vector[1]))
print("vector2与vector3的余弦相似度:", cosine_similarity(documents_vector[0], documents_vector[2]))

CacheBackedEmbedding 使用与场景

通过嵌入模型计算传递数据的向量需要昂贵的算力，对于重复的内容，Embeddings 计算的结果肯定是一致的。

如果数据重复仍然二次计算，会导致效率非常低，而且增加无用功。

在 LangChain 中提供了一个叫CacheBackedEmbedding的包装类。

一般通过类方法 from_bytes_store 进行实例化，它接受以下参数：

underlying_embedder：

用于嵌入的嵌入模型。

document_embedding_cache：

用于缓存文档嵌入的任何存储库（ByteStore）。

batch_size：

可选参数，默认为 None，在存储更新之间要嵌入的文档数量。

namespace：

可选参数，默认为””，用于文档缓存的命名空间。

此命名空间用于避免与其他缓存发生冲突。

例如，将其设置为所使用的嵌入模型的名称。

query_embedding_cache：

可选默认为 None 或者不缓存，用于缓存查询/文本嵌入的 ByteStore。

或这是为 True 以使用与 document_embedding_cache 相同的存储。

注意事项：

CacheBackEmbedding 默认不缓存 embed_query 生成的向量。

如果要缓存，需要设置 query_embedding_cache 的值。

另外请尽可能设置 namespace，以避免使用不同嵌入模型嵌入的相同文本发生冲突。

import dotenv
import numpy as np
from langchain.embeddings import CacheBackedEmbeddings
from langchain.storage import LocalFileStore
from langchain_openai import OpenAIEmbeddings
from numpy.linalg import norm

dotenv.load_dotenv()


def cosine_similarity(vector1: list, vector2: list) -> float:
    """计算传入两个向量的余弦相似度"""
    # 1.计算内积/点积
    dot_product = np.dot(vector1, vector2)

    # 2.计算向量的范数/长度
    norm_vec1 = norm(vector1)
    norm_vec2 = norm(vector2)

    # 3.计算余弦相似度
    return dot_product / (norm_vec1 * norm_vec2)


embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
embeddings_with_cache = CacheBackedEmbeddings.from_bytes_store(
    embeddings,
    LocalFileStore("./cache/"),
    namespace=embeddings.model,
    query_embedding_cache=True,
)

query_vector = embeddings_with_cache.embed_query("你好，我是openai，我喜欢打篮球")
documents_vector = embeddings_with_cache.embed_documents([
    "你好，我是openai，我喜欢打篮球",
    "这个喜欢打篮球的人叫openai",
    "求知若渴，虚心若愚"
])

print(query_vector)
print(len(query_vector))

print("============")

print(len(documents_vector))
print("vector1与vector2的余弦相似度:", cosine_similarity(documents_vector[0], documents_vector[1]))
print("vector2与vector3的余弦相似度:", cosine_similarity(documents_vector[0], documents_vector[2]))

CacheBackedEmbedding 运行流程

CacheBackedEmbedding 底层本质上就是封装了一个持久化存储的数据存储仓库。

在每次进行数据嵌入前，会从前数据存储仓库中检索对应的向量，然后逐个匹配对应的数据是否相等。

找到缓存中没有的文本，然后将这些文本调用嵌入生成向量。

最后将生成的新向量存储到数据仓库中，从而完成对数据的存储。

Hugging Face 文本嵌入模型

Hugging Face 本地模型：

在某些对数据保密要求极高的场合下，数据不允许传递到外网，这个时候就可以考虑使用本地的文本嵌入模型。

安装 langchain-huggingface 与 sentence-transformers 包，命令如下：

1	pip install -U langchain-huggingface sentence-transformers

其中 langchain-huggingface 是 LangChain 团队基于 HuggingFace 封装的第三方社区包。

sentence-transformers 是一个用于生成和使用预训练的文本嵌入，基于 transformer 架构。

也是目前使用量最大的本地文本嵌入模型。

配置好后，就可以像正常的文本嵌入模型一样使用了，示例：

from langchain_huggingface import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings()

query_vector = embeddings.embed_query("你好，我是xxxx，我喜欢打篮球")

print(query_vector)
print(len(query_vector))

sentence-transformers 除此加载的时候，会将模型从本地应用加载到内容中。

首次加载相对会慢，后续调用速度即可恢复正常。

除此之外，使用 Hugging Face 文本嵌入还可以加载任意本地的文本嵌入模型，传递 model_name 与 cache_folder 参数即可。

from langchain_huggingface import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(
    model_name="neuml/pubmedbert-base-embeddings",
    cache_folder="./embeddings/"
)

query_vector = embeddings.embed_query("你好，我是xxxx，我喜欢打篮球")

print(query_vector)
print(len(query_vector))

HuggingFace远程嵌入模型

部分模型的文件比较大，如果只是短期内调试，可以考虑使用 HuggingFace 提供的远程嵌入模型。

1	pip install huggingface_hub

然后在 Hugging Face 官网（https://huggingface.co/）的 setting 中添加对应的访问秘钥。

并配置到 .env 文件中：HUGGINGFACEHUB_API_TOKEN=xxx

接下来就可以使用 Hugging Face 提供的推理服务，这样在本地服务器上就无需配置对应的文本嵌入模型了。

HuggingFace 嵌入文档：

https://python.langchain.com/v0.2/docs/integrations/text_embedding/sentence_transformers/

HuggingFace 嵌入翻译文档：

http://imooc-langchain.shortvar.com/docs/integrations/text_embedding/sentence_transformers/

import dotenv
from langchain_huggingface import HuggingFaceEndpointEmbeddings

dotenv.load_dotenv()

embeddings = HuggingFaceEndpointEmbeddings(model="sentence-transformers/all-MiniLM-L12-v2")

query_vector = embeddings.embed_query("你好，我是openai，我喜欢打篮球")

print(query_vector)