零基础学AI大模型之嵌入模型性能优化

# 导入核心组件
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"                   # 可选：命名空间（隔离不同项目/模型版本）
)

# langchain.storage 支持的存储类型
__all__ = [
    "InMemoryStore",        # 内存存储（最快，重启丢失）
    "LocalFileStore",       # 本地文件存储（零配置，易调试）
    "RedisStore",           # Redis存储（分布式，高并发）
    "UpstashRedisStore",    # Upstash Redis（Serverless，无需运维）
    "EncoderBackedStore",   # 自定义编码存储（支持复杂数据类型）
]

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("如何重置密码？")

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（向量完全一致）

# 安装依赖：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",  #  Redis地址
    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(["如何查询订单物流？"])

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",  # 替换为你的API密钥
    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