开源模型应用落地-LangChain实用小技巧-使用CacheBackedEmbeddings组件（九）

一、前言

在 LangChain 中使用 CacheBackedEmbeddings组件具有诸多好处。一方面，它提供了强大的缓存功能，当处理大量文本数据进行嵌入操作时，可避免重复计算已经处理过的文本嵌入向量，极大地提升了系统运行效率，尤其对于频繁访问的文本内容效果显著。另一方面，它还具备备份作用，增加了数据的可靠性和稳定性，在系统出现故障或计算中断时，已缓存的嵌入向量可作为备份数据，便于任务中断后的恢复和继续计算，减少了因意外情况导致的重新计算成本。

上一篇：开源模型应用落地-LangChain实用小技巧-使用Embedding嵌入模型（八）

二、术语

**2.1.**CacheBackedEmbeddings组件作用

缓存功能

1. 提升效率

• 当处理大量文本数据进行嵌入操作时，这个组件可以缓存已经计算过的嵌入向量。如果后续再次遇到相同的文本内容，就可以直接从缓存中获取对应的嵌入向量，而无需再次进行复杂的计算过程。这大大减少了计算时间和资源消耗，提高了整个系统的运行效率。

2.减少重复计算

• 在一些场景下，可能会对相同的文本集合进行多次不同的分析或处理流程。确保了相同文本的嵌入向量只计算一次，避免了不必要的重复计算。这对于需要进行多次迭代或不同角度分析的任务非常有帮助。

备份功能

1. 数据可靠性

• 除了缓存嵌入向量以提高性能外，该组件还可以起到一定的备份作用。如果在计算嵌入向量的过程中出现意外情况，比如系统故障或计算中断，已经缓存的嵌入向量可以作为备份数据使用，确保不会因为意外而丢失所有的计算结果。

2.恢复和继续计算

• 如果任务被中断后需要恢复，CacheBackedEmbeddings可以利用缓存的嵌入向量快速恢复到中断前的状态，并继续进行后续的处理。这使得在复杂的计算流程中，即使遇到问题也能够相对容易地恢复和继续执行任务。

2.2.LocalFileStore

是一个用于本地文件存储的工具。它允许你在本地磁盘上创建一个存储区域，用于保存各种数据，尤其是在与缓存相关的操作中非常有用。LocalFileStore 可以作为一个可靠的本地存储解决方案，确保数据在需要时能够被快速访问和检索。

2.3.InMemoryStore

是一个将数据存储在内存中工具。这意味着数据可以被快速地读写和访问，因为不需要进行磁盘 I/O 操作。它非常适合存储临时数据或者在需要快速响应的场景中使用。

2.4.RedisStore

是一种利用 Redis 数据库进行数据存储的方式。RedisStore 允许将数据持久化地存储在 Redis 中，相比仅在内存中存储（如 InMemoryStore），它提供了更好的数据可靠性和可恢复性。它可以用于存储各种类型的数据，特别是在需要高效缓存和快速检索的场景中表现出色。

**三、前提条件 **

3.1. 基础环境

1. 操作系统：不限

3.2. 安装依赖库

conda create --name langchain python=3.10
conda activate langchain
pip install langchain langchain-openai

四、技术实现

4.1.集成LocalFileStore

import os
import time

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

os.environ["OPENAI_API_KEY"] = 'sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'  # 你的Open AI Key

def cosine_similarity(vector1: list, vector2: list) -> float:
    # 1.计算内积/点职
    dot_product = np.dot(vector1, vector2)
    # 2.计算向量的范数/长度
    vec1 = norm(vector1)
    vec2 = norm(vector2)
    # 3.计算余弦相似度
    return dot_product / (vec1 * vec2)

# 嵌入单个文档
def embed_query(str):
    query_vector = embeddings_cache.embed_query(str)
    return query_vector

# 嵌入多个个文档
def embed_documents(strs):
    documents_vector = embeddings_cache.embed_documents(strs)
    return documents_vector

if __name__ == '__main__':
    start = time.time()
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    store = LocalFileStore("E:\pycharm\data")

    embeddings_cache = CacheBackedEmbeddings.from_bytes_store(embeddings,store)

    str1 = '我叫张无忌，我会九阳神功'
    str2 = '张无忌会九阳神功'
    str3 = '我是蜘蛛侠，我会爬墙'

    documents_vector = embed_documents([str1,str2,str3])
    query_vector1 = documents_vector[0]
    query_vector2 = documents_vector[1]
    query_vector3 = documents_vector[2]

    print(f'query_vector1与query_vector2的余弦相似度为：{cosine_similarity(query_vector1,query_vector2)}')
    print(f'query_vector1与query_vector3的余弦相似度为：{cosine_similarity(query_vector1, query_vector3)}')
    end = time.time()
    print(f"执行时间为：{end - start} 秒")

第一次调用结果：

第二次调用结果：

第三次调用结果：

说明：

1.第一次调用完成后，生成的向量会被缓存在本地文件中

2.第二/三次调用时，直接从本地缓存获取数据，所以执行时间比第一次调用要显著减少

4.2.集成InMemoryStore

import os
import time

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

os.environ["OPENAI_API_KEY"] = 'sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx'  # 你的Open AI Key

def cosine_similarity(vector1: list, vector2: list) -> float:
    # 1.计算内积/点职
    dot_product = np.dot(vector1, vector2)
    # 2.计算向量的范数/长度
    vec1 = norm(vector1)
    vec2 = norm(vector2)
    # 3.计算余弦相似度
    return dot_product / (vec1 * vec2)

# 嵌入单个文档
def embed_query(str):
    query_vector = embeddings_cache.embed_query(str)
    return query_vector

# 嵌入多个个文档
def embed_documents(strs):
    documents_vector = embeddings_cache.embed_documents(strs)
    return documents_vector

if __name__ == '__main__':
    str1 = '我叫张无忌，我会九阳神功'
    str2 = '张无忌会九阳神功'
    str3 = '我是蜘蛛侠，我会爬墙'

    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    store = InMemoryStore()

    embeddings_cache = CacheBackedEmbeddings.from_bytes_store(embeddings, store)

    start = time.time()
    documents_vector = embed_documents([str1,str2,str3])
    query_vector1 = documents_vector[0]
    query_vector2 = documents_vector[1]
    query_vector3 = documents_vector[2]

    print(f'query_vector1与query_vector2的余弦相似度为：{cosine_similarity(query_vector1,query_vector2)}')
    print(f'query_vector1与query_vector3的余弦相似度为：{cosine_similarity(query_vector1, query_vector3)}')
    end = time.time()
    print(f"第一次执行时间为：{end - start} 秒")

    print('-----------------------------------------------------')
    start = time.time()
    documents_vector = embed_documents([str1, str2, str3])
    query_vector1 = documents_vector[0]
    query_vector2 = documents_vector[1]
    query_vector3 = documents_vector[2]

    print(f'query_vector1与query_vector2的余弦相似度为：{cosine_similarity(query_vector1, query_vector2)}')
    print(f'query_vector1与query_vector3的余弦相似度为：{cosine_similarity(query_vector1, query_vector3)}')
    end = time.time()
    print(f"第二次执行时间为：{end - start} 秒")

    print('-----------------------------------------------------')
    start = time.time()
    documents_vector = embed_documents([str1, str2, str3])
    query_vector1 = documents_vector[0]
    query_vector2 = documents_vector[1]
    query_vector3 = documents_vector[2]

    print(f'query_vector1与query_vector2的余弦相似度为：{cosine_similarity(query_vector1, query_vector2)}')
    print(f'query_vector1与query_vector3的余弦相似度为：{cosine_similarity(query_vector1, query_vector3)}')
    end = time.time()
    print(f"第三次执行时间为：{end - start} 秒")

调用结果：

五、附带说明

5.1.除了上面示例中的Store以外，还有很多其他的Store组件，例如：RedisStore、PostgresStore，大家可以动手实践一下哈。