CAG缓存检索生成 案例流程

用户提问
比如：用户问「黄金基金风险大吗？」

查询缓存层

首先检查缓存（Cache）里是否有与该问题高度相似的历史问题及答案。

如果相似度 > 阈值（比如 0.9），直接返回缓存答案，避免重新检索和生成。

命中缓存（Cache Hit）

如果命中，就直接返回缓存中的答案。

例如，缓存里有「黄金基金风险分析」的结果，就直接复用。

未命中缓存（Cache Miss）

如果没有命中缓存，就调用 检索引擎（RAG：Retrieval-Augmented Generation），从知识库/数据库中找到相关文档。

然后调用大模型（LLM）进行生成。

存入缓存

将新的问题 + 检索结果 + 生成的回答 存入缓存，供下次相似问题直接使用。

from sentence_transformers import SentenceTransformer
import faiss
import numpy as np

# 向量模型
model = SentenceTransformer("all-MiniLM-L6-v2")

# 模拟缓存数据库（用FAISS做语义检索）
dimension = 384  # embedding维度
index = faiss.IndexFlatL2(dimension)
cache_questions = []
cache_answers = []

def add_to_cache(question, answer):
    embedding = model.encode([question])
    index.add(np.array(embedding, dtype=np.float32))
    cache_questions.append(question)
    cache_answers.append(answer)

def search_cache(query, threshold=0.9):
    embedding = model.encode([query])
    D, I = index.search(np.array(embedding, dtype=np.float32), 1)
    if len(I) > 0 and I[0][0] != -1:
        similarity = 1 / (1 + D[0][0])  # 简化余弦相似度计算
        if similarity > threshold:
            return cache_answers[I[0][0]]
    return None

# 模拟使用
query = "黄金基金风险大吗？"
answer = search_cache(query)

if answer:
    print("命中缓存:", answer)
else:
    # 模拟RAG生成
    generated_answer = "黄金基金相对股票波动小，但仍有市场风险。"
    print("新生成:", generated_answer)
    add_to_cache(query, generated_answer)

📌 输出效果

第一次问：

新生成: 黄金基金相对股票波动小，但仍有市场风险。

第二次问（相似问题）：

命中缓存: 黄金基金相对股票波动小，但仍有市场风险。

要不要我帮你写一个 CAG缓存检索生成的完整架构图（流程图/时序图），更直观展示缓存 + 检索 + 生成的逻辑？