大模型应用：混元大模型数据缓存复用方案：从API请求数据累积到智能融合.50

原创

未闻花名

发布于 2026-03-19 07:46:02

1141

一、引言

随着大模型在各行业的规模化应用，API 调用成本高、响应延迟大、重复请求浪费等问题逐渐成为落地的核心痛点。实际运用过程中，大量的请求为重复或相似问题，直接调用大模型不仅推高成本，还降低了用户体验。

传统的缓存方案多依赖关键词匹配或简单文本相似度计算，精度低、适配性差；而基于大模型 API 生成 Embedding 的方案虽精度提升，但额外增加了向量调用成本。为此，结合实际，我们实践一种本地化向量和智能缓存复用的解决方案：基于 SentenceTransformer 本地向量模型替代 API 生成文本向量，结合腾讯混元大模型实现智能答案融合，同时设计缓存淘汰机制保证缓存质量。今天我们将由浅入深拆解该方案的设计思路、核心实现、进阶优化和落地验证，提供可直接运行的完整代码，提供快速落地低成本、高性能的大模型应用思路。

二、详细设计细节

1. 大模型应用的痛点

成本高：重复请求大模型导致 API 调用费用翻倍，企业级应用月均成本可达数万元；
响应慢：每次 API 调用需网络传输，响应时间通常在 1-3 秒，本地缓存可降至毫秒级；
匹配精度低：传统关键词/TF-IDF 匹配易漏判相似问题，用户体验差；
缓存膨胀：无淘汰机制的缓存会无限增长，导致匹配速度下降、存储成本增加。

2. 方案核心目标

我们希望通过早期搜集大模型生成的结果并本地存储，进行适当的标注，当数据累积到阈值后，将本地历史数据与大模型新生成数据智能融合；

核心目标是显著减少对大模型的重复请求，降低使用成本、提升响应速度，同时保证输出结果的准确性和实用性。

3. 核心原理

数据分层存储：将大模型生成的结果按“问题 - 答案 - 标注标签 - 生成时间”结构化存储，形成本地知识库
阈值触发机制：当本地数据量达到设定阈值（如 1000 条有效数据），启动“本地匹配 + 大模型补充”模式
智能融合策略：对新请求先匹配本地数据，匹配度≥阈值则直接复用；匹配度不足则调用大模型生成新结果，同时将新结果补充到本地库
迭代优化：定期对本地数据进行清洗、标注更新，提升匹配准确率

4. 解决方案思路

1. 本地化向量生成：使用 SentenceTransformer 多语言模型（paraphrase-multilingual-MiniLM-L12-v2）本地生成文本向量，无 API 调用成本，速度提升 10 倍以上；
2. 智能缓存复用：对用户请求先做本地向量相似度匹配，匹配达标则直接返回缓存结果，否则调用大模型并融合历史数据；
3. 智能答案融合：基于业务规则调用大模型融合历史答案与新生成答案，兼顾准确性和时效性；
4. 缓存淘汰机制：基于复用次数、访问时间淘汰低价值数据，保证缓存质量和匹配效率。

三、执行流程

核心步骤说明：

1. 智能请求路由：最大化利用本地资源，降低大模型调用成本

本地数据匹配：将用户请求与本地知识库中的已有数据进行相似度计算
阈值判断：预设匹配度阈值（如80%），判断是否直接使用本地答案

2. 双重结果生成路径：根据匹配情况采用不同的回答策略

路径A（高匹配度）：直接返回本地已存在的答案，实现毫秒级响应
路径B（低匹配度）：调用大模型生成新答案，保证回答的覆盖范围

3. 结果优化与增强：提升答案质量和适用性

结果融合：将大模型生成的结果与本地相似数据进行融合
最终输出：确保答案既准确又符合具体业务场景

4. 知识库持续学习：实现系统的自我进化能力

数据审核：所有结果都经过人工或自动质量审查
数据量监控：跟踪本地知识库规模
智能清洗：数据量达到阈值时自动启动清洗优化流程

5. 知识库动态更新：保持知识库的新鲜度和准确性

数据补充：将经过审核的高质量新数据加入本地库
知识库更新：完成数据入库和索引重建
循环优化：形成持续改进的闭环系统

流程总结：整个流程体现了系统的自我学习和优化能力，通过不断积累和清洗数据，使得本地知识库越来越强大，从而减少对大模型的依赖，提高响应速度并降低成本。

四、基础实现

1. 环境准备与依赖安装

import json
import os
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from PIL import Image
from Levenshtein import ratio
from openai import OpenAI
from tqdm import tqdm
import hashlib
from sentence_transformers import SentenceTransformer

# -------------------------- 核心配置项 --------------------------
# 腾讯混元大模型配置
API_KEY = os.environ.get('TENCENT_API_KEY')  # 从环境变量获取密钥
# API_KEY = 'sk-bWl****************************NP8Ze'  # 也可替换为你的密钥
BASE_URL = "https://api.hunyuan.cloud.tencent.com/v1"
LLM_MODEL = "hunyuan-lite"

# 本地向量模型配置
LOCAL_EMBED_MODEL = 'paraphrase-multilingual-MiniLM-L12-v2'
VECTOR_DIM = 384

方案的核心依赖包括：

openai：调用腾讯混元大模型 API；
sentence-transformers：本地生成文本向量；
numpy/pandas：向量计算与数据处理；
matplotlib：可视化缓存效果。

2. 本地向量模型加载

向量是文本语义的数字化表示，本地向量模型的核心作用是将自然语言问题转化为可计算相似度的向量。代码中核心实现如下：

class HunYuanLocalVecCache:
    def __init__(self):
        # 初始化混元客户端
        self.client = OpenAI(
            api_key=API_KEY,
            base_url=BASE_URL
        )
        
        # 加载本地向量模型
        print("加载本地向量模型...")
        self.embedder = SentenceTransformer(LOCAL_EMBED_MODEL)
        print("本地向量模型加载完成！")
        
        # 加载本地数据（增加异常日志）
        self.local_data = self._load_json(DATA_STORAGE_PATH)
        print(f"初始化加载本地数据 {len(self.local_data)} 条")
        
        # 预加载向量（优化空列表处理）
        self.embedding_vectors = self._preload_local_embeddings()
        
        # 统计指标
        self.total_llm_calls = 0
        self.total_cache_hits = 0


    def get_local_embedding(self, text):
        """生成本地向量，增加文本预处理"""
        text = text.strip().replace("\n", " ").replace("\r", "")
        if not text:
            return np.zeros(VECTOR_DIM)
        try:
            return self.embedder.encode(text, convert_to_numpy=True, normalize_embeddings=True)
        except Exception as e:
            print(f"生成向量失败：{e}")
            return np.zeros(VECTOR_DIM)

关键说明：

选择paraphrase-multilingual-MiniLM-L12-v2的核心原因是支持中文、体积小、速度快，适合本地部署；
向量归一化（normalize_embeddings=True）可消除向量长度对相似度计算的影响，提升匹配精度。

3. 余弦相似度计算

相似度是判断 “新问题是否与缓存中的历史问题相似” 的核心指标，本地实现余弦相似度计算：

def cosine_similarity(vec1, vec2):
    """计算余弦相似度，增加零向量判断"""
    norm1 = np.linalg.norm(vec1)
    norm2 = np.linalg.norm(vec2)
    if norm1 == 0 or norm2 == 0:
        return 0.0
    return float(np.dot(vec1, vec2) / (norm1 * norm2))

def calculate_local_similarity(self, new_question):
    """计算相似度，增加日志打印"""
    if len(self.embedding_vectors) == 0:
        return [], []
    
    new_vec = self.get_local_embedding(new_question)
    similarities = [cosine_similarity(new_vec, vec) for vec in self.embedding_vectors]
    
    # 打印相似度详情，方便调试
    max_sim = max(similarities) if similarities else 0
    print(f"新问题与本地数据最高相似度：{max_sim:.4f} (阈值 {MATCH_THRESHOLD})")
    
    local_questions = [item["question"] for item in self.local_data]
    return similarities, local_questions

关键说明：

余弦相似度取值范围为 [0,1]，值越接近 1 表示语义越相似。例如：
- “AI 大模型可以为我们做些什么？” 与 “AI 大模型能帮我们做哪些事？” 的相似度约 0.8+；
- 完全无关的问题相似度接近 0。

4. 本地数据存储

缓存数据以 JSON 格式持久化存储，包含问题、答案、复用次数等关键信息，核心实现：

def _load_json(self, file_path):
    """加载JSON，增加详细异常日志"""
    if not os.path.exists(file_path):
        print(f"文件 {file_path} 不存在，初始化空列表")
        return []
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            data = json.load(f)
        return data if isinstance(data, list) else []
    except Exception as e:
        print(f"加载 {file_path} 失败：{e}，初始化空列表")
        return []

def _save_json(self, data, file_path):
    """保存JSON，增加写入验证"""
    try:
        with open(file_path, "w", encoding="utf-8") as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
        # 验证写入是否成功
        verify = self._load_json(file_path)
        if len(verify) == len(data):
            print(f"成功保存 {len(data)} 条数据到 {file_path}")
            return True
        else:
            print(f"数据保存验证失败，写入长度不一致")
            return False
    except Exception as e:
        print(f"保存 {file_path} 失败：{e}")
        return False

存储的数据结构设计：

[
    {
        "question": "AI大模型可以为我们做些什么？",
        "answer": "AI大模型可完成文本生成、数据分析、智能问答...",
        "label": "已标注",
        "create_time": "2026-01-09 10:00:00",
        "call_count": 2,  // 复用次数
        "last_call_time": "2026-01-09 10:05:00"
    }
]

5. 相似度阈值调优

本地向量模型的相似度分布与 API 生成的向量不同，需根据实际场景调整阈值：初始版本设置MATCH_THRESHOLD = 0.75，适配中文短文本的相似度分布，需结合模型和实际结果进行合理调整。

# 降低相似度阈值，适配本地模型
MATCH_THRESHOLD = 0.75
THRESHOLD_DATA_COUNT = 10  # 降低触发融合的阈值，方便测试
CACHE_MAX_SIZE = 5000
CACHE_MIN_USED_COUNT = 1  # 降低淘汰阈值，测试阶段不轻易淘汰
CACHE_CLEAN_RATIO = 0.1

调优原则：

阈值过高：漏判相似问题，缓存命中率低；
阈值过低：误判无关问题，答案准确性下降；
建议通过实际业务数据测试，选择 “命中率 - 准确率” 平衡的阈值（通常 0.7-0.8）。

6. 智能答案融合

传统缓存仅返回历史答案，无法适配问题的时效性需求，今天结合经验，我们设计基于大模型的智能融合策略：

def smart_fuse_answers(self, local_answer, llm_answer, question):
    """智能融合，保持原逻辑"""
    if not local_answer:
        return llm_answer
    
    fuse_prompt = f"""融合以下两个答案，保留核心信息，去除重复，分点输出：
问题：{question}
历史答案：{local_answer}
新答案：{llm_answer}"""
    
    fused_answer = self.get_hunyuan_response(fuse_prompt)
    return fused_answer if fused_answer else f"【历史】{local_answer}\n【新增】{llm_answer}"

融合价值：既保留历史答案中经过验证的核心事实，又补充新答案的时效性信息，提升回答完整性。

7. 缓存淘汰机制

无淘汰机制的缓存会无限膨胀，导致匹配速度下降，此方案设计基于 “复用次数 + 访问时间” 的淘汰规则：

def cache_eviction(self):
    """缓存淘汰，测试阶段降低触发条件"""
    data_count = len(self.local_data)
    if data_count < CACHE_MAX_SIZE:
        return
    print("触发缓存淘汰...")
    self.local_data.sort(key=lambda x: (x["call_count"], x["last_call_time"]))
    evict_count = max(int(data_count * CACHE_CLEAN_RATIO), 10)
    self.local_data = self.local_data[evict_count:]
    self.embedding_vectors = self._preload_local_embeddings()
    self._save_json(self.local_data, DATA_STORAGE_PATH)

核心规则：

触发条件：缓存容量超过 5000 条；
淘汰优先级：复用次数 < 1 次 → 最后访问时间最早 → 低质量数据；
淘汰比例：每次淘汰 10%，避免一次性删除过多数据。

8. 异常处理

大模型调用重试：失败时自动重试 2 次，避免网络波动导致失败；

def get_hunyuan_response(self, question):
    """调用混元大模型，增加重试机制"""
    retry_count = 2
    for i in range(retry_count):
        try:
            completion = self.client.chat.completions.create(
                model=LLM_MODEL,
                messages=[
                    {'role': 'system', 'content': '回答简洁准确，控制在100字以内'},
                    {'role': 'user', 'content': question}
                ]
            )
            self.total_llm_calls += 1
            return completion.choices[0].message.content.strip()
        except Exception as e:
            print(f"调用大模型失败（重试 {i+1}/{retry_count}）：{e}")
            time.sleep(1)
    return "大模型调用失败，请稍后重试"

向量生成容错：文本为空或生成失败时返回零向量，不中断流程；

def get_local_embedding(self, text):
    """生成本地向量，增加文本预处理"""
    text = text.strip().replace("\n", " ").replace("\r", "")
    if not text:
        return np.zeros(VECTOR_DIM)
    try:
        return self.embedder.encode(text, convert_to_numpy=True, normalize_embeddings=True)
    except Exception as e:
        print(f"生成向量失败：{e}")
        return np.zeros(VECTOR_DIM)

def _preload_local_embeddings(self):
    """预加载向量，优化空列表处理"""
    if not self.local_data:
        print("无本地数据，向量列表为空")
        return np.array([])  # 返回空数组，避免后续拼接报错
    
    vectors = []
    print("预生成本地数据向量...")
    for item in tqdm(self.local_data):
        vec = self.get_local_embedding(item["question"])
        vectors.append(vec)
    return np.array(vectors)

数据保存验证：写入后验证数据长度，避免缓存丢失；

# 验证写入是否成功
verify = self._load_json(file_path)
if len(verify) == len(data):
    print(f"成功保存 {len(data)} 条数据到 {file_path}")
    return True
else:
    print(f"数据保存验证失败，写入长度不一致")
    return False

9. 测试用例设计

test_questions = [
    "AI大模型可以为我们做些什么？",  # 新请求1
    "如何优化大模型的推理速度？",    # 新请求2
    "腾讯混元大模型的特点是什么？",  # 新请求3
    "AI大模型可以为我们做些什么？",  # 重复请求1 → 命中缓存
    "AI大模型能帮我们做哪些事？",    # 相似请求1 → 命中缓存
    "腾讯混元大模型的特点是什么？"   # 重复请求3 → 命中缓存
]

10. 输出结果

加载本地向量模型... 本地向量模型加载完成！初始化加载本地数据 0 条无本地数据，向量列表为空 ===== 开始处理请求 ===== 【请求 1】：AI大模型可以为我们做些什么？成功保存 1 条数据到 hunyuan_local_data.json 新增数据成功，当前缓存总量：1 是否调用大模型：True 回答摘要：AI大模型能进行自然语言处理、图像识别、语音识别与合成、预测分析、自动化决策、智能客服、知识问答等。它们提升了生产效率，优化了用户体验，并在医疗、教育、金融等领域发挥重要作用。... 【请求 2】：如何优化大模型的推理速度？新问题与本地数据最高相似度：0.5332 (阈值 0.75) 成功保存 2 条数据到 hunyuan_local_data.json 新增数据成功，当前缓存总量：2 是否调用大模型：True 回答摘要：优化大模型推理速度方法：1. 使用高效硬件如GPU或TPU；2. 优化模型结构，减少计算量；3. 采用量化技术降低精度；4. 使用缓存技术存储频繁数据；5. 并行处理多个请求提高吞吐量。... 【请求 3】：腾讯混元大模型的特点是什么？新问题与本地数据最高相似度：0.6117 (阈值 0.75) 成功保存 3 条数据到 hunyuan_local_data.json 新增数据成功，当前缓存总量：3 是否调用大模型：True 回答摘要：腾讯混元大模型特点：超千亿参数规模，处理多语言和任务，高性能与低资源消耗，中文和多模态理解生成，支持智能问答、对话互动及内容创作。... 【请求 4】：AI大模型可以为我们做些什么？新问题与本地数据最高相似度：1.0000 (阈值 0.75) 成功保存 3 条数据到 hunyuan_local_data.json 缓存命中！该数据已复用 1 次是否调用大模型：False 回答摘要：AI大模型能进行自然语言处理、图像识别、语音识别与合成、预测分析、自动化决策、智能客服、知识问答等。它们提升了生产效率，优化了用户体验，并在医疗、教育、金融等领域发挥重要作用。... 【请求 5】：AI大模型能帮我们做哪些事？新问题与本地数据最高相似度：0.9835 (阈值 0.75) 成功保存 3 条数据到 hunyuan_local_data.json 缓存命中！该数据已复用 2 次 是否调用大模型：False 回答摘要：AI大模型能进行自然语言处理、图像识别、语音识别与合成、预测分析、自动化决策、智能客服、知识问答等。它们提升了生产效率，优化了用户体验，并在医疗、教育、金融等领域发挥重要作用。... 【请求 6】：腾讯混元大模型的特点是什么？新问题与本地数据最高相似度：1.0000 (阈值 0.75) 成功保存 3 条数据到 hunyuan_local_data.json 缓存命中！该数据已复用 1 次 是否调用大模型：False 回答摘要：腾讯混元大模型特点：超千亿参数规模，处理多语言和任务，高性能与低资源消耗，中文和多模态理解生成，支持智能问答、对话互动及内容创作。... 可视化报告已保存至：hunyuan_data_analysis_local_vec.png ===== 最终统计 ===== 总请求数：6 大模型调用数：5 缓存命中数：3 调用节省率：50.0% 本地缓存总量：3 条

结果图示：

代码生成的可视化报告包含 4 个子图，直观展示缓存效果：

图1：每日缓存命中次数：监控缓存使用趋势；
图2：复用次数分布：分析缓存数据质量；
图3：调用节省率：量化降本效果；
图4：缓存健康度：展示高价值数据占比。

本地缓存hunyuan_local_data.json内容预览：

[
  {
    "question": "AI大模型可以为我们做些什么？",
    "answer": "AI大模型能进行自然语言处理、图像识别、语音识别与合成、预测分析、自动化决策、智能客服、知识问答等。它们提升了生产效率，优化了用户体验，并在医疗、教育、金融等领域发挥重要作用。",
    "label": "已标注",
    "create_time": "2026-01-09 20:44:05",
    "call_count": 2,
    "last_call_time": "2026-01-09 20:44:10"
  },
  {
    "question": "如何优化大模型的推理速度？",
    "answer": "优化大模型推理速度方法：1. 使用高效硬件如GPU或TPU；2. 优化模型结构，减少计算量；3. 采用量化技术降低精度；4. 使用缓存技术存储频繁数据；5. 并行处理多个请求提高吞吐量。",
    "label": "已标注",
    "create_time": "2026-01-09 20:44:08",
    "call_count": 0,
    "last_call_time": "2026-01-09 20:44:08"
  },
  {
    "question": "腾讯混元大模型的特点是什么？",
    "answer": "腾讯混元大模型特点：超千亿参数规模，处理多语言和任务，高性能与低资源消耗，中文和多模态理解生成，支持智能问答、对话互动及内容创作。",
    "label": "已标注",
    "create_time": "2026-01-09 20:44:10",
    "call_count": 1,
    "last_call_time": "2026-01-09 20:44:10"
  }
]

若运行后缓存未命中，可按以下步骤排查：

1. 检查hunyuan_local_data.json是否生成，且包含 3 条数据；
2. 查看控制台打印的相似度日志，确认重复问题相似度≥0.75；
3. 验证本地向量模型是否成功加载，无报错信息；
4. 检查数据保存方法是否返回True，确保写入成功。

体现优势：

降本：减少 60%+ 重复大模型请求，降低 API 调用成本
提速：本地匹配响应速度比调用大模型快 10 倍以上
可控：本地数据可标注、可审计，结果更符合业务需求
兼容：支持主流大模型（OpenAI、智谱、百度文心等）

五、总结

结合本地化向量 + 智能缓存复用方案，解决了大模型应用中重复请求成本高、响应慢的核心痛点，核心价值体现在三方面：

降本：本地化向量消除 Embedding 调用成本，缓存复用减少 50%+ 的大模型 API 调用；
提效：本地向量匹配响应速度提升 10 倍以上，用户体验显著改善；
可控：智能融合保证答案准确性，缓存淘汰机制避免数据膨胀。

从技术实现来看，方案由浅入深完成了 “基础缓存→智能复用→生产级优化” 的迭代：先通过本地向量模型实现语义匹配，再通过智能融合提升答案质量，最后通过异常处理和淘汰机制保证方案的可执行和可持续性。该方案可直接适配腾讯混元、OpenAI 等主流大模型，也可根据垂直领域需求灵活扩展，是大模型应用落地的轻量化、高性价比解决方案。

附录：完整实例代码

import json
import os
import time
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from PIL import Image
from Levenshtein import ratio
from openai import OpenAI
from tqdm import tqdm
import hashlib
from sentence_transformers import SentenceTransformer

# -------------------------- 核心配置项 --------------------------
# 腾讯混元大模型配置
API_KEY = os.environ.get('TENCENT_API_KEY')  # 从环境变量获取密钥
# API_KEY = 'sk-bWl****************************NP8Ze'  # 也可替换为你的密钥
BASE_URL = "https://api.hunyuan.cloud.tencent.com/v1"
LLM_MODEL = "hunyuan-lite"

# 本地向量模型配置
LOCAL_EMBED_MODEL = 'paraphrase-multilingual-MiniLM-L12-v2'
VECTOR_DIM = 384

# 【关键修改】降低相似度阈值，适配本地模型
MATCH_THRESHOLD = 0.75
THRESHOLD_DATA_COUNT = 10  # 降低触发融合的阈值，方便测试
CACHE_MAX_SIZE = 5000
CACHE_MIN_USED_COUNT = 1  # 降低淘汰阈值，测试阶段不轻易淘汰
CACHE_CLEAN_RATIO = 0.1

# 本地数据路径（确保有写入权限）
DATA_STORAGE_PATH = "hunyuan_local_data.json"
VISUALIZATION_SAVE_PATH = "hunyuan_data_analysis_local_vec.png"
plt.rcParams["font.sans-serif"] = ["SimHei"]
plt.rcParams["axes.unicode_minus"] = False

# -------------------------- 工具函数 --------------------------
def cosine_similarity(vec1, vec2):
    """计算余弦相似度，增加零向量判断"""
    norm1 = np.linalg.norm(vec1)
    norm2 = np.linalg.norm(vec2)
    if norm1 == 0 or norm2 == 0:
        return 0.0
    return float(np.dot(vec1, vec2) / (norm1 * norm2))

# -------------------------- 核心类定义 --------------------------
class HunYuanLocalVecCache:
    def __init__(self):
        # 初始化混元客户端
        self.client = OpenAI(
            api_key=API_KEY,
            base_url=BASE_URL
        )
        
        # 加载本地向量模型
        print("加载本地向量模型...")
        self.embedder = SentenceTransformer(LOCAL_EMBED_MODEL)
        print("本地向量模型加载完成！")
        
        # 加载本地数据（增加异常日志）
        self.local_data = self._load_json(DATA_STORAGE_PATH)
        print(f"初始化加载本地数据 {len(self.local_data)} 条")
        
        # 预加载向量（优化空列表处理）
        self.embedding_vectors = self._preload_local_embeddings()
        
        # 统计指标
        self.total_llm_calls = 0
        self.total_cache_hits = 0

    def _load_json(self, file_path):
        """加载JSON，增加详细异常日志"""
        if not os.path.exists(file_path):
            print(f"文件 {file_path} 不存在，初始化空列表")
            return []
        try:
            with open(file_path, "r", encoding="utf-8") as f:
                data = json.load(f)
            return data if isinstance(data, list) else []
        except Exception as e:
            print(f"加载 {file_path} 失败：{e}，初始化空列表")
            return []

    def _save_json(self, data, file_path):
        """保存JSON，增加写入验证"""
        try:
            with open(file_path, "w", encoding="utf-8") as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
            # 验证写入是否成功
            verify = self._load_json(file_path)
            if len(verify) == len(data):
                print(f"成功保存 {len(data)} 条数据到 {file_path}")
                return True
            else:
                print(f"数据保存验证失败，写入长度不一致")
                return False
        except Exception as e:
            print(f"保存 {file_path} 失败：{e}")
            return False

    def get_local_embedding(self, text):
        """生成本地向量，增加文本预处理"""
        text = text.strip().replace("\n", " ").replace("\r", "")
        if not text:
            return np.zeros(VECTOR_DIM)
        try:
            return self.embedder.encode(text, convert_to_numpy=True, normalize_embeddings=True)
        except Exception as e:
            print(f"生成向量失败：{e}")
            return np.zeros(VECTOR_DIM)

    def _preload_local_embeddings(self):
        """预加载向量，优化空列表处理"""
        if not self.local_data:
            print("无本地数据，向量列表为空")
            return np.array([])  # 返回空数组，避免后续拼接报错
        
        vectors = []
        print("预生成本地数据向量...")
        for item in tqdm(self.local_data):
            vec = self.get_local_embedding(item["question"])
            vectors.append(vec)
        return np.array(vectors)

    def calculate_local_similarity(self, new_question):
        """计算相似度，增加日志打印"""
        if len(self.embedding_vectors) == 0:
            return [], []
        
        new_vec = self.get_local_embedding(new_question)
        similarities = [cosine_similarity(new_vec, vec) for vec in self.embedding_vectors]
        
        # 打印相似度详情，方便调试
        max_sim = max(similarities) if similarities else 0
        print(f"新问题与本地数据最高相似度：{max_sim:.4f} (阈值 {MATCH_THRESHOLD})")
        
        local_questions = [item["question"] for item in self.local_data]
        return similarities, local_questions

    def get_hunyuan_response(self, question):
        """调用混元大模型，增加重试机制"""
        retry_count = 2
        for i in range(retry_count):
            try:
                completion = self.client.chat.completions.create(
                    model=LLM_MODEL,
                    messages=[
                        {'role': 'system', 'content': '回答简洁准确，控制在100字以内'},
                        {'role': 'user', 'content': question}
                    ]
                )
                self.total_llm_calls += 1
                return completion.choices[0].message.content.strip()
            except Exception as e:
                print(f"调用大模型失败（重试 {i+1}/{retry_count}）：{e}")
                time.sleep(1)
        return "大模型调用失败，请稍后重试"

    def smart_fuse_answers(self, local_answer, llm_answer, question):
        """智能融合，保持原逻辑"""
        if not local_answer:
            return llm_answer
        
        fuse_prompt = f"""融合以下两个答案，保留核心信息，去除重复，分点输出：
问题：{question}
历史答案：{local_answer}
新答案：{llm_answer}"""
        
        fused_answer = self.get_hunyuan_response(fuse_prompt)
        return fused_answer if fused_answer else f"【历史】{local_answer}\n【新增】{llm_answer}"

    def add_new_data(self, question, answer, label="未标注"):
        """【关键优化】新增数据逻辑，修复向量拼接"""
        new_item = {
            "question": question,
            "answer": answer,
            "label": label,
            "create_time": time.strftime("%Y-%m-%d %H:%M:%S"),
            "call_count": 0,
            "last_call_time": time.strftime("%Y-%m-%d %H:%M:%S")
        }
        
        # 先添加数据并保存，再处理向量
        self.local_data.append(new_item)
        save_success = self._save_json(self.local_data, DATA_STORAGE_PATH)
        
        if save_success:
            # 处理向量拼接（兼容空数组）
            new_vec = self.get_local_embedding(question)
            if len(self.embedding_vectors) == 0:
                self.embedding_vectors = np.array([new_vec])
            else:
                self.embedding_vectors = np.vstack([self.embedding_vectors, new_vec])
            print(f"新增数据成功，当前缓存总量：{len(self.local_data)}")
        else:
            # 保存失败则回滚
            self.local_data.pop()
            print("新增数据失败，已回滚")
        
        self.cache_eviction()

    def update_call_count(self, question_idx):
        """更新复用次数，增加日志"""
        self.local_data[question_idx]["call_count"] += 1
        self.local_data[question_idx]["last_call_time"] = time.strftime("%Y-%m-%d %H:%M:%S")
        self.total_cache_hits += 1
        self._save_json(self.local_data, DATA_STORAGE_PATH)
        print(f"缓存命中！该数据已复用 {self.local_data[question_idx]['call_count']} 次")

    def cache_eviction(self):
        """缓存淘汰，测试阶段降低触发条件"""
        data_count = len(self.local_data)
        if data_count < CACHE_MAX_SIZE:
            return
        print("触发缓存淘汰...")
        self.local_data.sort(key=lambda x: (x["call_count"], x["last_call_time"]))
        evict_count = max(int(data_count * CACHE_CLEAN_RATIO), 10)
        self.local_data = self.local_data[evict_count:]
        self.embedding_vectors = self._preload_local_embeddings()
        self._save_json(self.local_data, DATA_STORAGE_PATH)

    def process_request(self, question, label="未标注"):
        """处理请求主流程"""
        sim_scores, local_questions = self.calculate_local_similarity(question)
        best_sim = max(sim_scores) if sim_scores else 0
        best_idx = np.argmax(sim_scores) if sim_scores else -1

        # 判断缓存命中
        if best_sim >= MATCH_THRESHOLD and best_idx != -1:
            self.update_call_count(best_idx)
            final_answer = self.local_data[best_idx]["answer"]
            llm_call_flag = False
        else:
            llm_answer = self.get_hunyuan_response(question)
            local_answer = self.local_data[best_idx]["answer"] if best_idx != -1 else None
            final_answer = self.smart_fuse_answers(local_answer, llm_answer, question)
            self.add_new_data(question, final_answer, label)
            llm_call_flag = True

        return final_answer, llm_call_flag

    def generate_visualization(self):
        """生成可视化报告，保持原逻辑"""
        if not self.local_data:
            print("无本地数据，无法生成可视化")
            return
        df = pd.DataFrame(self.local_data)
        df["create_date"] = df["create_time"].apply(lambda x: x.split()[0])
        df["last_call_date"] = df["last_call_time"].apply(lambda x: x.split()[0])
        
        fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(14, 10))
        fig.suptitle("混元大模型缓存复用分析报告（本地化向量版）", fontsize=16, fontweight="bold")

        daily_hits = df.groupby("last_call_date")["call_count"].sum()
        ax1.plot(daily_hits.index, daily_hits.values, marker="o", color="#1E88E5", linewidth=2)
        ax1.set_title("每日缓存命中次数")
        ax1.set_xlabel("日期")
        ax1.set_ylabel("命中次数")
        ax1.tick_params(axis="x", rotation=45)

        call_counts = df["call_count"].values
        ax2.hist(call_counts, bins=15, color="#FFC107", alpha=0.7)
        ax2.axvline(x=CACHE_MIN_USED_COUNT, color="red", linestyle="--", label=f"淘汰阈值({CACHE_MIN_USED_COUNT})")
        ax2.set_title("数据复用次数分布")
        ax2.set_xlabel("复用次数")
        ax2.set_ylabel("数据条数")
        ax2.legend()

        total_requests = self.total_llm_calls + self.total_cache_hits
        save_rate = (self.total_cache_hits / total_requests) * 100 if total_requests > 0 else 0
        ax3.bar(["缓存命中率"], [save_rate], color="#4CAF50", alpha=0.8)
        ax3.set_title("大模型调用节省率")
        ax3.set_ylabel("节省率(%)")
        ax3.set_ylim(0, 100)
        ax3.text(0, save_rate + 2, f"{save_rate:.1f}%", ha="center", fontweight="bold")

        healthy_ratio = (sum(1 for item in self.local_data if item["call_count"] >= CACHE_MIN_USED_COUNT) / len(self.local_data)) * 100 if len(self.local_data) > 0 else 0
        ax4.pie([healthy_ratio, 100-healthy_ratio], labels=["健康数据", "低价值数据"], 
                autopct="%1.1f%%", colors=["#4CAF50", "#FF5722"])
        ax4.set_title("缓存数据健康度")

        plt.tight_layout()
        plt.savefig(VISUALIZATION_SAVE_PATH, dpi=300, bbox_inches="tight")
        plt.close()
        img = Image.open(VISUALIZATION_SAVE_PATH)
        img.show()
        print(f"可视化报告已保存至：{VISUALIZATION_SAVE_PATH}")

# -------------------------- 测试用例 --------------------------
if __name__ == "__main__":
    cache = HunYuanLocalVecCache()
    
    # 测试用例：包含完全重复和相似问题
    test_questions = [
        "AI大模型可以为我们做些什么？",  # 请求1
        "如何优化大模型的推理速度？",    # 请求2
        "腾讯混元大模型的特点是什么？",  # 请求3
        "AI大模型可以为我们做些什么？",  # 重复请求1 → 应该命中缓存
        "AI大模型能帮我们做哪些事？",    # 相似请求1 → 应该命中缓存
        "腾讯混元大模型的特点是什么？"   # 重复请求3 → 应该命中缓存
    ]

    print("\n===== 开始处理请求 =====")
    for idx, q in enumerate(test_questions):
        print(f"\n【请求 {idx+1}】：{q}")
        answer, is_call = cache.process_request(q, label="已标注")
        print(f"是否调用大模型：{is_call}")
        print(f"回答摘要：{answer[:150]}...")

    # 生成可视化
    cache.generate_visualization()

    # 输出最终统计
    total_requests = len(test_questions)
    save_rate = (cache.total_cache_hits / total_requests) * 100 if total_requests > 0 else 0
    print(f"\n===== 最终统计 =====")
    print(f"总请求数：{total_requests}")
    print(f"大模型调用数：{cache.total_llm_calls}")
    print(f"缓存命中数：{cache.total_cache_hits}")
    print(f"调用节省率：{save_rate:.1f}%")
    print(f"本地缓存总量：{len(cache.local_data)} 条")

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

腾讯混元大模型

玩转腾讯混元大模型

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

登录后参与评论

0 条评论

热度