RAG已经过时了？试试CAG，缓存增强生成技术实战大揭秘！

import numpy as npfrom typing import List, Dictimport timefrom datetime import datetime# 模拟向量数据库class SimpleVectorDB:    """简单的向量数据库实现"""        def __init__(self):        self.documents = []        self.embeddings = []        self.metadata = []            def add_document(self, text: str, metadata: Dict = None):        """添加文档到数据库"""        # 这里用简单的词频向量模拟embedding        embedding = self._text_to_vector(text)        self.documents.append(text)        self.embeddings.append(embedding)        self.metadata.append(metadata or {})            def _text_to_vector(self, text: str) -> np.ndarray:        """将文本转换为向量（简化版）"""        # 实际应该用OpenAI/HuggingFace的embedding模型        # 这里简化处理：基于字符出现频率        vector = np.zeros(100)        for i, char in enumerate(text[:100]):            vector[i] = ord(char) / 1000        return vector        def search(self, query: str, top_k: int = 3) -> List[Dict]:        """检索最相关的文档"""        query_vector = self._text_to_vector(query)                # 计算余弦相似度        similarities = []        for i, doc_vector in enumerate(self.embeddings):            similarity = np.dot(query_vector, doc_vector) / (                np.linalg.norm(query_vector) * np.linalg.norm(doc_vector) + 1e-10            )            similarities.append({                'index': i,                'score': similarity,                'text': self.documents[i],                'metadata': self.metadata[i]            })                # 返回top_k结果        similarities.sort(key=lambda x: x['score'], reverse=True)        return similarities[:top_k]class TraditionalRAG:    """传统RAG系统"""        def __init__(self):        self.vector_db = SimpleVectorDB()        self.search_count = 0  # 统计检索次数        self.search_times = []  # 记录每次检索耗时            def add_knowledge(self, text: str, metadata: Dict = None):        """添加知识到系统"""        self.vector_db.add_document(text, metadata)            def query(self, question: str) -> Dict:        """处理查询"""        start_time = time.time()                # 每次都要检索        search_results = self.vector_db.search(question, top_k=2)                search_time = time.time() - start_time        self.search_count += 1        self.search_times.append(search_time)                # 组装上下文        context = "\n\n".join([r['text'] for r in search_results])                # 模拟LLM生成答案（实际应调用GPT/Claude API）        answer = self._generate_answer(question, context)                return {            'question': question,            'answer': answer,            'context': context,            'search_time': search_time,            'total_searches': self.search_count        }        def _generate_answer(self, question: str, context: str) -> str:        """模拟LLM生成答案"""        # 实际应该调用OpenAI API或其他LLM        return f"基于知识库：{context[:100]}... 回答：[模拟答案]"        def get_statistics(self) -> Dict:        """获取性能统计"""        return {            'total_searches': self.search_count,            'avg_search_time': np.mean(self.search_times) if self.search_times else 0,            'total_time': sum(self.search_times)        }# 使用示例def demo_traditional_rag():    """演示传统RAG的问题"""    print("=" * 60)    print("传统RAG系统演示")    print("=" * 60)        # 创建RAG系统    rag = TraditionalRAG()        # 添加企业知识（这些都是稳定的制度文档）    knowledge_base = [        {            "text": "公司年假政策：入职满1年员工享有5天年假，满3年享有10天，满5年享有15天。年假必须在当年使用，不可跨年累积。",            "metadata": {"category": "HR政策", "update_date": "2024-01-01"}        },        {            "text": "差旅费报销标准：国内出差每天补贴200元，住宿费实报实销上限500元/天。需提供发票和出差申请单。",            "metadata": {"category": "财务制度", "update_date": "2024-01-01"}        },        {            "text": "公积金缴纳比例：公司和个人各缴纳12%，基数为上年度月平均工资。每年7月调整一次。",            "metadata": {"category": "薪酬福利", "update_date": "2024-01-01"}        },        {            "text": "病假规定：员工因病需请假，需提供医院证明。病假工资按基本工资的80%发放，每年累计不超过30天。",            "metadata": {"category": "HR政策", "update_date": "2024-01-01"}        }    ]        for kb in knowledge_base:        rag.add_knowledge(kb['text'], kb['metadata'])        print(f"\n已加载 {len(knowledge_base)} 条企业知识\n")        # 模拟重复查询（这是关键问题所在）    repeated_questions = [        "年假怎么算？",        "年假政策是什么？",        "我能休几天年假？",        "差旅费怎么报销？",        "出差补贴标准是多少？",        "年假能累积吗？",  # 又问年假        "公积金比例是多少？",        "年假政策详细说明",  # 再问年假    ]        print("开始处理查询...\n")    for i, question in enumerate(repeated_questions, 1):        result = rag.query(question)        print(f"查询 {i}: {question}")        print(f"  检索耗时: {result['search_time']*1000:.2f}ms")        print(f"  累计检索次数: {result['total_searches']}")        print()        # 显示统计信息    stats = rag.get_statistics()    print("=" * 60)    print("性能统计")    print("=" * 60)    print(f"总检索次数: {stats['total_searches']}")    print(f"平均检索耗时: {stats['avg_search_time']*1000:.2f}ms")    print(f"总耗时: {stats['total_time']*1000:.2f}ms")    print()    print("⚠️  问题分析：")    print("  - 关于'年假'的问题被问了4次，但每次都重新检索")    print("  - 这些制度文档几个月都不会变，却要反复访问数据库")    print("  - 随着查询量增加，成本和延迟线性上升")    print()# 运行演示demo_traditional_rag()

import hashlibfrom typing import Optional, Tupleimport jsonclass KnowledgeCache:    """知识缓存管理器"""        def __init__(self, max_size: int = 100):        self.cache = {}  # 缓存存储        self.max_size = max_size        self.hit_count = 0  # 命中次数        self.miss_count = 0  # 未命中次数        self.access_log = []  # 访问日志            def _generate_key(self, query: str) -> str:        """生成查询的缓存键"""        # 使用语义哈希（这里简化为文本哈希）        # 实际应该用embedding的相似度匹配        normalized = query.lower().strip()        return hashlib.md5(normalized.encode()).hexdigest()[:16]        def get(self, query: str, similarity_threshold: float = 0.85) -> Optional[Dict]:        """从缓存获取答案"""        # 简化版：精确匹配        # 实际应该用语义相似度匹配        query_key = query.lower().strip()                # 查找语义相似的缓存项        for cached_query, cached_data in self.cache.items():            if self._is_similar(query_key, cached_query):                self.hit_count += 1                self.access_log.append({                    'query': query,                    'result': 'HIT',                    'timestamp': datetime.now().isoformat()                })                return cached_data                self.miss_count += 1        self.access_log.append({            'query': query,            'result': 'MISS',            'timestamp': datetime.now().isoformat()        })        return None        def _is_similar(self, query1: str, query2: str) -> bool:        """判断两个查询是否相似"""        # 简化版：包含关键词就算相似        # 实际应该用向量相似度        keywords1 = set(query1.split())        keywords2 = set(query2.split())                if not keywords1 or not keywords2:            return False                    intersection = keywords1 & keywords2        union = keywords1 | keywords2        similarity = len(intersection) / len(union)                return similarity > 0.5        def set(self, query: str, context: str, answer: str, metadata: Dict = None):        """设置缓存"""        query_key = query.lower().strip()                # 检查容量限制        if len(self.cache) >= self.max_size:            # 简单的LRU：删除最旧的项            oldest_key = next(iter(self.cache))            del self.cache[oldest_key]                self.cache[query_key] = {            'query': query,            'context': context,            'answer': answer,            'metadata': metadata or {},            'cached_at': datetime.now().isoformat()        }        def get_statistics(self) -> Dict:        """获取缓存统计"""        total_access = self.hit_count + self.miss_count        hit_rate = self.hit_count / total_access if total_access > 0 else 0                return {            'hit_count': self.hit_count,            'miss_count': self.miss_count,            'total_access': total_access,            'hit_rate': hit_rate,            'cache_size': len(self.cache)        }class CAGSystem:    """CAG（缓存增强生成）系统"""        def __init__(self, cache_size: int = 100):        self.vector_db = SimpleVectorDB()        self.cache = KnowledgeCache(max_size=cache_size)        self.search_count = 0        self.search_times = []        self.cache_hit_times = []            def add_knowledge(self, text: str, metadata: Dict = None, cacheable: bool = False):        """添加知识"""        self.vector_db.add_document(text, metadata)                # 如果标记为可缓存，预先生成常见问题的缓存        if cacheable and metadata and 'common_questions' in metadata:            for question in metadata['common_questions']:                # 预先缓存答案                answer = f"基于缓存：{text[:100]}..."                self.cache.set(question, text, answer, metadata)        def query(self, question: str) -> Dict:        """处理查询（带缓存）"""        start_time = time.time()                # 先查缓存        cached_result = self.cache.get(question)                if cached_result:            # 缓存命中！            cache_time = time.time() - start_time            self.cache_hit_times.append(cache_time)                        return {                'question': question,                'answer': cached_result['answer'],                'context': cached_result['context'],                'source': 'CACHE',                'response_time': cache_time,                'total_searches': self.search_count            }                # 缓存未命中，执行检索        search_results = self.vector_db.search(question, top_k=2)        search_time = time.time() - start_time        self.search_count += 1        self.search_times.append(search_time)                # 组装上下文        context = "\n\n".join([r['text'] for r in search_results])        answer = self._generate_answer(question, context)                # 存入缓存（如果是静态知识）        if search_results and self._is_cacheable(search_results[0]):            self.cache.set(question, context, answer,                           search_results[0].get('metadata', {}))                return {            'question': question,            'answer': answer,            'context': context,            'source': 'RETRIEVAL',            'response_time': search_time,            'total_searches': self.search_count        }        def _is_cacheable(self, search_result: Dict) -> bool:        """判断检索结果是否应该缓存"""        metadata = search_result.get('metadata', {})        # 如果有更新日期且超过30天未更新，认为是静态知识        update_date = metadata.get('update_date')        if update_date:            # 简化判断：只要有update_date就认为是静态的            return True        return False        def _generate_answer(self, question: str, context: str) -> str:        """模拟LLM生成答案"""        return f"基于知识库：{context[:100]}... 回答：[模拟答案]"        def get_statistics(self) -> Dict:        """获取完整统计信息"""        cache_stats = self.cache.get_statistics()                return {            'retrieval': {                'total_searches': self.search_count,                'avg_search_time': np.mean(self.search_times) if self.search_times else 0,                'total_time': sum(self.search_times)            },            'cache': {                'hit_count': cache_stats['hit_count'],                'miss_count': cache_stats['miss_count'],                'hit_rate': cache_stats['hit_rate'],                'avg_hit_time': np.mean(self.cache_hit_times) if self.cache_hit_times else 0,                'cache_size': cache_stats['cache_size']            },            'overall': {                'total_queries': cache_stats['total_access'],                'searches_saved': cache_stats['hit_count'],                'cost_reduction': f"{cache_stats['hit_rate']*100:.1f}%"            }        }# 使用示例def demo_cag_system():    """演示CAG系统的优势"""    print("=" * 60)    print("CAG系统演示（带缓存优化）")    print("=" * 60)        # 创建CAG系统    cag = CAGSystem(cache_size=50)        # 添加知识（标记静态知识为可缓存）    knowledge_base = [        {            "text": "公司年假政策：入职满1年员工享有5天年假，满3年享有10天，满5年享有15天。年假必须在当年使用，不可跨年累积。",            "metadata": {                "category": "HR政策",                "update_date": "2024-01-01",                "common_questions": [                    "年假怎么算",                    "年假政策是什么",                    "我能休几天年假",                    "年假能累积吗"                ]            },            "cacheable": True        },        {            "text": "差旅费报销标准：国内出差每天补贴200元，住宿费实报实销上限500元/天。需提供发票和出差申请单。",            "metadata": {                "category": "财务制度",                "update_date": "2024-01-01",                "common_questions": [                    "差旅费怎么报销",                    "出差补贴标准是多少",                    "出差住宿费报销"                ]            },            "cacheable": True        },        {            "text": "公积金缴纳比例：公司和个人各缴纳12%，基数为上年度月平均工资。每年7月调整一次。",            "metadata": {                "category": "薪酬福利",                "update_date": "2024-01-01",                "common_questions": [                    "公积金比例是多少",                    "公积金怎么缴纳"                ]            },            "cacheable": True        }    ]        for kb in knowledge_base:        cag.add_knowledge(kb['text'], kb['metadata'], kb['cacheable'])        print(f"\n已加载 {len(knowledge_base)} 条企业知识（已预缓存常见问题）\n")        # 模拟重复查询    test_questions = [        "年假怎么算？",          # 第1次：缓存命中        "年假政策是什么？",      # 第2次：缓存命中        "我能休几天年假？",      # 第3次：缓存命中        "差旅费怎么报销？",      # 第1次：缓存命中        "出差补贴标准是多少？",  # 第2次：缓存命中        "年假能累积吗？",        # 第4次：缓存命中        "公积金比例是多少？",    # 第1次：缓存命中        "年假政策详细说明",      # 第5次：缓存命中    ]        print("开始处理查询...\n")    for i, question in enumerate(test_questions, 1):        result = cag.query(question)                # 显示结果        source_icon = "⚡ [缓存]" if result['source'] == 'CACHE' else "🔍 [检索]"        print(f"查询 {i}: {question}")        print(f"  数据源: {source_icon}")        print(f"  响应时间: {result['response_time']*1000:.2f}ms")        print(f"  累计检索次数: {result['total_searches']}")        print()        # 显示详细统计    stats = cag.get_statistics()    print("=" * 60)    print("性能统计对比")    print("=" * 60)        print("\n【检索统计】")    print(f"  实际检索次数: {stats['retrieval']['total_searches']}")    print(f"  平均检索耗时: {stats['retrieval']['avg_search_time']*1000:.2f}ms")        print("\n【缓存统计】")    print(f"  缓存命中次数: {stats['cache']['hit_count']}")    print(f"  缓存未命中: {stats['cache']['miss_count']}")    print(f"  缓存命中率: {stats['cache']['hit_rate']*100:.1f}%")    print(f"  平均缓存响应: {stats['cache']['avg_hit_time']*1000:.2f}ms")    print(f"  当前缓存大小: {stats['cache']['cache_size']}")        print("\n【整体优化】")    print(f"  总查询次数: {stats['overall']['total_queries']}")    print(f"  节省检索次数: {stats['overall']['searches_saved']}")    print(f"  成本降低: {stats['overall']['cost_reduction']}")        print("\n✅ 优势总结：")    print("  - 重复问题直接从缓存返回，无需检索")    print("  - 响应时间从 50-200ms 降低到 <5ms")    print("  - 数据库访问次数大幅减少，成本显著降低")    print()# 运行CAG演示demo_cag_system()

def compare_rag_vs_cag():    """直接对比RAG和CAG的性能"""    print("=" * 60)    print("RAG vs CAG 性能对比实验")    print("=" * 60)        # 准备测试数据    knowledge = {        "text": "公司年假政策：入职满1年员工享有5天年假，满3年享有10天，满5年享有15天。",        "metadata": {            "category": "HR政策",            "common_questions": ["年假怎么算", "年假政策", "休几天年假"]        }    }        # 重复查询100次    questions = ["年假怎么算？"] * 100        # 测试传统RAG    print("\n【测试1：传统RAG】")    rag = TraditionalRAG()    rag.add_knowledge(knowledge['text'], knowledge['metadata'])        rag_start = time.time()    for q in questions:        rag.query(q)    rag_total_time = time.time() - rag_start    rag_stats = rag.get_statistics()        print(f"总耗时: {rag_total_time*1000:.2f}ms")    print(f"检索次数: {rag_stats['total_searches']}")    print(f"平均延迟: {rag_stats['avg_search_time']*1000:.2f}ms")        # 测试CAG    print("\n【测试2：CAG系统】")    cag = CAGSystem()    cag.add_knowledge(knowledge['text'], knowledge['metadata'], cacheable=True)        cag_start = time.time()    for q in questions:        cag.query(q)    cag_total_time = time.time() - cag_start    cag_stats = cag.get_statistics()        print(f"总耗时: {cag_total_time*1000:.2f}ms")    print(f"检索次数: {cag_stats['retrieval']['total_searches']}")    print(f"缓存命中率: {cag_stats['cache']['hit_rate']*100:.1f}%")    print(f"平均延迟: {cag_stats['cache']['avg_hit_time']*1000:.2f}ms")        # 性能提升计算    print("\n" + "=" * 60)    print("性能提升")    print("=" * 60)    speedup = rag_total_time / cag_total_time    search_reduction = (rag_stats['total_searches'] - cag_stats['retrieval']['total_searches']) / rag_stats['total_searches']        print(f"速度提升: {speedup:.1f}x")    print(f"检索次数减少: {search_reduction*100:.1f}%")    print(f"成本节约: ~{search_reduction*100:.1f}%")    # 运行对比测试compare_rag_vs_cag()

class HybridRAGCAG:    """混合RAG+CAG系统"""        def __init__(self, cache_size: int = 100):        # 静态知识缓存        self.static_cache = KnowledgeCache(max_size=cache_size)                # 动态知识向量库        self.dynamic_db = SimpleVectorDB()                # 静态知识向量库（用于缓存未命中时的后备）        self.static_db = SimpleVectorDB()                # 统计信息        self.stats = {            'static_cache_hits': 0,            'static_db_queries': 0,            'dynamic_db_queries': 0,            'total_queries': 0        }            def add_static_knowledge(self, text: str, metadata: Dict = None,                             common_questions: List[str] = None):        """添加静态知识（长期不变）"""        # 添加到静态数据库        self.static_db.add_document(text, metadata)                # 预缓存常见问题        if common_questions:            for question in common_questions:                answer = f"[静态知识] {text}"                self.static_cache.set(question, text, answer, metadata)        def add_dynamic_knowledge(self, text: str, metadata: Dict = None):        """添加动态知识（经常更新）"""        # 只添加到动态数据库，不缓存        self.dynamic_db.add_document(text, metadata)        def query(self, question: str, require_realtime: bool = False) -> Dict:        """智能查询：自动判断用缓存还是检索                Args:            question: 用户问题            require_realtime: 是否强制要求实时数据        """        self.stats['total_queries'] += 1        start_time = time.time()                # 如果不要求实时数据，先查静态缓存        if not require_realtime:            cached_result = self.static_cache.get(question)            if cached_result:                self.stats['static_cache_hits'] += 1                return {                    'question': question,                    'answer': cached_result['answer'],                    'source': 'STATIC_CACHE',                    'response_time': time.time() - start_time,                    'confidence': 'high'                }                # 判断问题类型：需要动态数据还是静态数据？        question_type = self._classify_question(question)                if question_type == 'dynamic' or require_realtime:            # 查询动态数据库            results = self.dynamic_db.search(question, top_k=2)            self.stats['dynamic_db_queries'] += 1            source = 'DYNAMIC_RETRIEVAL'        else:            # 查询静态数据库            results = self.static_db.search(question, top_k=2)            self.stats['static_db_queries'] += 1            source = 'STATIC_RETRIEVAL'                        # 将结果缓存起来，下次直接用            if results:                context = results[0]['text']                answer = f"[静态知识] {context}"                self.static_cache.set(question, context, answer,                                      results[0].get('metadata', {}))                # 生成答案        context = "\n".join([r['text'] for r in results]) if results else ""        answer = self._generate_answer(question, context, source)                return {            'question': question,            'answer': answer,            'source': source,            'response_time': time.time() - start_time,            'confidence': 'high' if results else 'low'        }        def _classify_question(self, question: str) -> str:        """判断问题需要动态数据还是静态数据"""        # 简化版：通过关键词判断        dynamic_keywords = ['今天', '最新', '现在', '当前', '实时', '昨天', '最近']        static_keywords = ['政策', '制度', '规定', '标准', '流程', '怎么', '如何']                question_lower = question.lower()                # 包含动态关键词，返回dynamic        for keyword in dynamic_keywords:            if keyword in question_lower:                return 'dynamic'                # 包含静态关键词，返回static        for keyword in static_keywords:            if keyword in question_lower:                return 'static'                # 默认当作静态        return 'static'        def _generate_answer(self, question: str, context: str, source: str) -> str:        """生成答案"""        if not context:            return "抱歉，没有找到相关信息。"        return f"基于{source}：{context[:150]}..."        def get_statistics(self) -> Dict:        """获取详细统计"""        cache_stats = self.static_cache.get_statistics()        total_queries = self.stats['total_queries']                return {            'total_queries': total_queries,            'static_cache_hits': self.stats['static_cache_hits'],            'static_db_queries': self.stats['static_db_queries'],            'dynamic_db_queries': self.stats['dynamic_db_queries'],            'cache_hit_rate': cache_stats['hit_rate'],            'db_access_rate': (self.stats['static_db_queries'] + self.stats['dynamic_db_queries']) / total_queries if total_queries > 0 else 0        }def demo_hybrid_system():    """演示混合系统"""    print("=" * 60)    print("混合RAG+CAG系统演示")    print("=" * 60)        # 创建混合系统    hybrid = HybridRAGCAG(cache_size=50)        # 添加静态知识（制度文档）    print("\n【加载静态知识】")    static_knowledge = [        {            "text": "公司年假政策：入职满1年员工享有5天年假，满3年享有10天，满5年享有15天。年假必须在当年使用，不可跨年累积。",            "common_questions": ["年假怎么算", "年假政策", "休几天年假", "年假能累积吗"]        },        {            "text": "报销流程：提交申请单→部门主管审批→财务审核→财务打款。处理时间约3-5个工作日。",            "common_questions": ["怎么报销", "报销流程", "报销要多久"]        }    ]        for kb in static_knowledge:        hybrid.add_static_knowledge(            kb['text'],             {'type': 'static', 'category': 'policy'},            kb['common_questions']        )    print(f"已加载 {len(static_knowledge)} 条静态知识（已预缓存）")        # 添加动态知识（实时数据）    print("\n【加载动态知识】")    dynamic_knowledge = [        {            "text": "今天公司食堂菜单：午餐有红烧肉、清蒸鱼、麻婆豆腐。晚餐有宫保鸡丁、酸菜鱼、素炒时蔬。",            "metadata": {'type': 'dynamic', 'date': '2025-11-05'}        },        {            "text": "本周会议通知：周三下午3点全体会议，周五上午10点部门例会。请提前准备材料。",            "metadata": {'type': 'dynamic', 'date': '2025-11-05'}        }    ]        for kb in dynamic_knowledge:        hybrid.add_dynamic_knowledge(kb['text'], kb['metadata'])    print(f"已加载 {len(dynamic_knowledge)} 条动态知识")        # 测试不同类型的查询    print("\n" + "=" * 60)    print("开始测试查询")    print("=" * 60)        test_cases = [        {"q": "年假怎么算？", "type": "静态问题（应命中缓存）"},        {"q": "年假政策是什么？", "type": "静态问题（应命中缓存）"},        {"q": "报销流程是什么？", "type": "静态问题（应命中缓存）"},        {"q": "今天食堂吃什么？", "type": "动态问题（应查询动态库）"},        {"q": "本周有什么会议？", "type": "动态问题（应查询动态库）"},        {"q": "年假能累积吗？", "type": "静态问题（应命中缓存）"},        {"q": "今天食堂有什么菜？", "type": "动态问题（应查询动态库）"},    ]        for i, test in enumerate(test_cases, 1):        result = hybrid.query(test['q'])                # 根据source显示不同图标        if result['source'] == 'STATIC_CACHE':            icon = "⚡"            color = "缓存"        elif result['source'] == 'STATIC_RETRIEVAL':            icon = "📚"            color = "静态库"        else:            icon = "🔄"            color = "动态库"                print(f"\n查询 {i}: {test['q']}")        print(f"  类型: {test['type']}")        print(f"  数据源: {icon} {color}")        print(f"  响应时间: {result['response_time']*1000:.2f}ms")        # 显示统计    print("\n" + "=" * 60)    print("系统统计")    print("=" * 60)    stats = hybrid.get_statistics()        print(f"\n总查询次数: {stats['total_queries']}")    print(f"  ├─ 静态缓存命中: {stats['static_cache_hits']} ({stats['cache_hit_rate']*100:.1f}%)")    print(f"  ├─ 静态库查询: {stats['static_db_queries']}")    print(f"  └─ 动态库查询: {stats['dynamic_db_queries']}")    print(f"\n数据库访问率: {stats['db_access_rate']*100:.1f}%")    print(f"成本节约: ~{(1-stats['db_access_rate'])*100:.1f}%")        print("\n✅ 混合架构优势：")    print("  - 静态知识走缓存，响应极快")    print("  - 动态知识走检索，保证实时性")    print("  - 自动判断问题类型，智能路由")    print("  - 兼顾速度、成本和准确性")# 运行混合系统演示demo_hybrid_system()

class SmartCache:    """智能缓存系统（基于LFU+LRU）"""        def __init__(self, max_size: int = 100, min_access_count: int = 3):        self.cache = {}        self.access_count = {}  # 访问计数        self.last_access = {}   # 最后访问时间        self.max_size = max_size        self.min_access_count = min_access_count  # 最小访问次数才缓存                # 候选池：访问次数不够的暂存这里        self.candidate_pool = {}        self.candidate_access = {}            def should_cache(self, key: str) -> bool:        """判断是否应该缓存"""        # 如果已经在候选池        if key in self.candidate_access:            self.candidate_access[key] += 1                        # 访问次数达到阈值，提升到正式缓存            if self.candidate_access[key] >= self.min_access_count:                return True        else:            # 首次访问，加入候选池            self.candidate_access[key] = 1                return False        def set(self, key: str, value: Dict):        """设置缓存（只缓存热数据）"""        if not self.should_cache(key):            # 暂存到候选池            self.candidate_pool[key] = value            return False                # 达到缓存条件，正式缓存        if len(self.cache) >= self.max_size:            # 淘汰策略：LFU + LRU            self._evict()                self.cache[key] = value        self.access_count[key] = self.candidate_access.get(key, 1)        self.last_access[key] = time.time()                # 从候选池移除        if key in self.candidate_pool:            del self.candidate_pool[key]                return True        def get(self, key: str) -> Optional[Dict]:        """获取缓存"""        if key in self.cache:            # 更新访问统计            self.access_count[key] += 1            self.last_access[key] = time.time()            return self.cache[key]                # 检查候选池        if key in self.candidate_pool:            self.candidate_access[key] += 1            # 如果访问够多了，提升到正式缓存            if self.candidate_access[key] >= self.min_access_count:                self.set(key, self.candidate_pool[key])                return self.cache[key]            return self.candidate_pool[key]                return None        def _evict(self):        """淘汰缓存项（LFU + LRU组合）"""        if not self.cache:            return                # 找出访问次数最少的项        min_count = min(self.access_count.values())        candidates = [k for k, v in self.access_count.items() if v == min_count]                # 如果有多个，选最久未访问的        if len(candidates) > 1:            evict_key = min(candidates, key=lambda k: self.last_access[k])        else:            evict_key = candidates[0]                # 删除        del self.cache[evict_key]        del self.access_count[evict_key]        del self.last_access[evict_key]        def get_statistics(self) -> Dict:        """获取统计信息"""        return {            'cache_size': len(self.cache),            'candidate_size': len(self.candidate_pool),            'total_size': len(self.cache) + len(self.candidate_pool),            'avg_access_count': np.mean(list(self.access_count.values())) if self.access_count else 0,            'hot_items': sorted(                [(k, v) for k, v in self.access_count.items()],                key=lambda x: x[1],                reverse=True            )[:5]  # 前5个热门项        }class SmartCachingSystem:    """带智能缓存的完整系统"""        def __init__(self, cache_size: int = 50, min_access: int = 3):        self.vector_db = SimpleVectorDB()        self.smart_cache = SmartCache(max_size=cache_size, min_access_count=min_access)                self.stats = {            'total_queries': 0,            'cache_hits': 0,            'db_queries': 0,            'promoted_to_cache': 0  # 从候选池提升到正式缓存的次数        }        def add_knowledge(self, text: str, metadata: Dict = None):        """添加知识"""        self.vector_db.add_document(text, metadata)        def query(self, question: str) -> Dict:        """查询"""        self.stats['total_queries'] += 1        start_time = time.time()                # 查缓存        cached = self.smart_cache.get(question)        if cached and question in self.smart_cache.cache:  # 正式缓存命中            self.stats['cache_hits'] += 1            return {                'question': question,                'answer': cached['answer'],                'source': 'CACHE',                'response_time': time.time() - start_time            }                # 检索        results = self.vector_db.search(question, top_k=2)        self.stats['db_queries'] += 1                context = "\n".join([r['text'] for r in results]) if results else ""        answer = f"基于检索: {context[:100]}..."                # 尝试缓存（智能判断）        cached_result = self.smart_cache.set(question, {            'answer': answer,            'context': context,            'metadata': results[0].get('metadata', {}) if results else {}        })                if cached_result:            self.stats['promoted_to_cache'] += 1                return {            'question': question,            'answer': answer,            'source': 'RETRIEVAL',            'response_time': time.time() - start_time,            'will_cache': cached_result        }        def get_statistics(self) -> Dict:        """获取统计"""        cache_stats = self.smart_cache.get_statistics()                return {            'queries': self.stats,            'cache': cache_stats,            'cache_hit_rate': self.stats['cache_hits'] / self.stats['total_queries'] if self.stats['total_queries'] > 0 else 0        }def demo_smart_caching():    """演示智能缓存"""    print("=" * 60)    print("智能缓存系统演示")    print("=" * 60)        # 创建系统    system = SmartCachingSystem(cache_size=10, min_access=3)        # 添加知识    knowledge = [        "年假政策：入职满1年5天，满3年10天，满5年15天",        "报销流程：提交申请→审批→财务审核→打款",        "公积金比例：公司和个人各12%",        "加班政策：工作日1.5倍，周末2倍，节假日3倍",        "社保缴纳：养老8%医疗2%失业0.5%"    ]        for kb in knowledge:        system.add_knowledge(kb)        print(f"\n已加载 {len(knowledge)} 条知识\n")        # 模拟真实查询分布（符合二八定律）    print("模拟真实查询场景（80%查询集中在20%的问题）\n")        # 热门问题（会被频繁查询）    hot_questions = [        "年假怎么算",        "怎么报销",        "公积金比例"    ]        # 冷门问题（偶尔查一次）    cold_questions = [        "加班怎么算",        "社保比例",        "病假政策",        "迟到扣款",        "离职流程"    ]        # 生成查询序列（80/20分布）    query_sequence = []    for _ in range(50):        if np.random.random() < 0.8:  # 80%概率查热门问题            query_sequence.append(np.random.choice(hot_questions))        else:  # 20%概率查冷门问题            query_sequence.append(np.random.choice(cold_questions))        # 执行查询    print("开始处理50次查询...\n")    cache_hits_timeline = []        for i, question in enumerate(query_sequence, 1):        result = system.query(question)                if i <= 10 or i % 10 == 0:  # 只显示部分结果            source_icon = "⚡" if result['source'] == 'CACHE' else "🔍"            cached_tag = " [已提升到缓存]" if result.get('will_cache') else ""            print(f"查询{i:2d}: {question:15s} {source_icon} {result['source']}{cached_tag}")                # 记录命中率变化        stats = system.get_statistics()        cache_hits_timeline.append(stats['cache_hit_rate'])        # 最终统计    print("\n" + "=" * 60)    print("最终统计")    print("=" * 60)        final_stats = system.get_statistics()        print(f"\n【查询统计】")    print(f"  总查询次数: {final_stats['queries']['total_queries']}")    print(f"  缓存命中: {final_stats['queries']['cache_hits']}")    print(f"  数据库查询: {final_stats['queries']['db_queries']}")    print(f"  提升到缓存: {final_stats['queries']['promoted_to_cache']}")        print(f"\n【缓存统计】")    print(f"  正式缓存: {final_stats['cache']['cache_size']} 项")    print(f"  候选池: {final_stats['cache']['candidate_size']} 项")    print(f"  总存储: {final_stats['cache']['total_size']} 项")    print(f"  缓存命中率: {final_stats['cache_hit_rate']*100:.1f}%")    print(f"  平均访问次数: {final_stats['cache']['avg_access_count']:.1f}")        print(f"\n【热门问题Top5】")    for i, (question, count) in enumerate(final_stats['cache']['hot_items'], 1):        print(f"  {i}. {question} - 访问{count}次")        print("\n✅ 智能缓存特点：")    print("  - 只缓存被多次访问的热门问题（访问≥3次）")    print("  - 冷门问题不占用宝贵的缓存空间")    print("  - 自动淘汰不常用的缓存项")    print("  - 符合真实业务场景的访问分布")        # 显示命中率趋势    print(f"\n【命中率趋势】前20次查询：")    for i in range(0, min(20, len(cache_hits_timeline)), 5):        rate = cache_hits_timeline[i]        bar = "█" * int(rate * 50)        print(f"  查询{i+1:2d}: {bar} {rate*100:.1f}%")# 运行智能缓存演示demo_smart_caching()

from datetime import datetime, timedeltaclass CacheWithTTL:    """带过期时间的缓存"""        def __init__(self, max_size: int = 100, default_ttl: int = 86400):        """        Args:            max_size: 最大缓存数量            default_ttl: 默认过期时间（秒），默认24小时        """        self.cache = {}        self.max_size = max_size        self.default_ttl = default_ttl                self.stats = {            'hits': 0,            'misses': 0,            'expires': 0,            'invalidations': 0        }        def set(self, key: str, value: Dict, ttl: Optional[int] = None):        """设置缓存项                Args:            key: 缓存键            value: 缓存值            ttl: 过期时间（秒），None则使用默认值        """        if len(self.cache) >= self.max_size:            self._evict_oldest()                expire_at = time.time() + (ttl if ttl is not None else self.default_ttl)                self.cache[key] = {            'value': value,            'expire_at': expire_at,            'created_at': time.time(),            'version': value.get('metadata', {}).get('version', 1)        }        def get(self, key: str) -> Optional[Dict]:        """获取缓存项"""        if key not in self.cache:            self.stats['misses'] += 1            return None                item = self.cache[key]                # 检查是否过期        if time.time() > item['expire_at']:            self.stats['expires'] += 1            del self.cache[key]            return None                self.stats['hits'] += 1        return item['value']        def invalidate(self, key: str):        """主动失效某个缓存"""        if key in self.cache:            del self.cache[key]            self.stats['invalidations'] += 1        def invalidate_by_pattern(self, pattern: str):        """按模式批量失效"""        keys_to_delete = [k for k in self.cache.keys() if pattern in k]        for key in keys_to_delete:            self.invalidate(key)        def update_version(self, key: str, new_version: int):        """更新版本号（触发重新缓存）"""        if key in self.cache:            current_version = self.cache[key]['version']            if new_version > current_version:                # 版本更新，失效旧缓存                self.invalidate(key)        def _evict_oldest(self):        """淘汰最旧的项"""        if not self.cache:            return        oldest_key = min(self.cache.keys(),                         key=lambda k: self.cache[k]['created_at'])        del self.cache[oldest_key]        def get_statistics(self) -> Dict:        """获取统计"""        total = self.stats['hits'] + self.stats['misses']        return {            **self.stats,            'hit_rate': self.stats['hits'] / total if total > 0 else 0,            'cache_size': len(self.cache)        }class VersionedKnowledgeBase:    """带版本控制的知识库"""        def __init__(self):        self.documents = {}  # key: doc_id, value: {content, version, metadata}        self.cache = CacheWithTTL(max_size=50, default_ttl=3600)  # 1小时TTL        self.vector_db = SimpleVectorDB()            def add_or_update_document(self, doc_id: str, content: str,                                 metadata: Dict = None, version: int = 1):        """添加或更新文档"""        # 检查是否是更新        is_update = doc_id in self.documents        old_version = self.documents[doc_id]['version'] if is_update else 0                # 保存文档        self.documents[doc_id] = {            'content': content,            'version': version,            'metadata': metadata or {},            'updated_at': datetime.now().isoformat()        }                # 更新向量数据库        self.vector_db.add_document(content, {            'doc_id': doc_id,            'version': version,            **(metadata or {})        })                # 如果是更新，失效相关缓存        if is_update and version > old_version:            print(f"📝 文档 {doc_id} 更新: v{old_version} -> v{version}")            self.cache.invalidate_by_pattern(doc_id)            return True                return False        def query(self, question: str, doc_id: Optional[str] = None) -> Dict:        """查询（支持版本检查）"""        # 构建缓存键        cache_key = f"{doc_id}:{question}" if doc_id else question                # 查缓存        cached = self.cache.get(cache_key)        if cached:            return {                'question': question,                'answer': cached['answer'],                'source': 'CACHE',                'version': cached.get('version', 'unknown')            }                # 检索        results = self.vector_db.search(question, top_k=2)        if not results:            return {'question': question, 'answer': '未找到相关信息', 'source': 'NONE'}                # 生成答案        context = results[0]['text']        result_doc_id = results[0]['metadata'].get('doc_id', 'unknown')        result_version = results[0]['metadata'].get('version', 1)        answer = f"[v{result_version}] {context}"                # 缓存结果        self.cache.set(cache_key, {            'answer': answer,            'context': context,            'metadata': {                'doc_id': result_doc_id,                'version': result_version            }        }, ttl=3600)  # 1小时过期                return {            'question': question,            'answer': answer,            'source': 'RETRIEVAL',            'version': result_version,            'doc_id': result_doc_id        }def demo_cache_update():    """演示缓存更新机制"""    print("=" * 60)    print("缓存更新与版本控制演示")    print("=" * 60)        kb = VersionedKnowledgeBase()        # 场景1：初始知识    print("\n【场景1：初始加载知识】")    kb.add_or_update_document(        doc_id="policy_annual_leave",        content="年假政策v1：入职满1年5天，满3年10天，满5年15天",        metadata={'category': 'HR'},        version=1    )    print("✅ 已添加：年假政策 v1")        # 第一次查询    print("\n第1次查询：年假怎么算？")    result1 = kb.query("年假怎么算")    print(f"  来源: {result1['source']}")    print(f"  版本: {result1['version']}")    print(f"  答案: {result1['answer'][:50]}...")        # 第二次查询（应该命中缓存）    print("\n第2次查询：年假怎么算？")    result2 = kb.query("年假怎么算")    print(f"  来源: {result2['source']} ⚡")    print(f"  版本: {result2['version']}")        # 场景2：政策更新    print("\n" + "=" * 60)    print("【场景2：政策更新】")    print("公司调整年假政策...")        kb.add_or_update_document(        doc_id="policy_annual_leave",        content="年假政策v2：入职满1年7天，满3年12天，满5年20天。新增：满10年25天",        metadata={'category': 'HR'},        version=2    )        # 再次查询（缓存已失效，应该返回新版本）    print("\n第3次查询：年假怎么算？")    result3 = kb.query("年假怎么算")    print(f"  来源: {result3['source']}")    print(f"  版本: {result3['version']}")    print(f"  答案: {result3['answer'][:60]}...")        # 场景3：缓存重建    print("\n第4次查询：年假怎么算？")    result4 = kb.query("年假怎么算")    print(f"  来源: {result4['source']} ⚡ (新版本已缓存)")    print(f"  版本: {result4['version']}")        # 统计    print("\n" + "=" * 60)    print("缓存统计")    print("=" * 60)    stats = kb.cache.get_statistics()    print(f"缓存命中: {stats['hits']}")    print(f"缓存未命中: {stats['misses']}")    print(f"缓存失效: {stats['invalidations']}")    print(f"命中率: {stats['hit_rate']*100:.1f}%")        print("\n✅ 更新机制总结：")    print("  - 文档更新时自动失效相关缓存")    print("  - 版本号控制确保数据一致性")    print("  - 支持TTL自动过期")    print("  - 下次查询会获取最新版本并重新缓存")# 运行缓存更新演示demo_cache_update()

def compare_invalidation_strategies():    """对比不同的缓存失效策略"""    print("=" * 60)    print("三种缓存失效策略对比")    print("=" * 60)        print("\n【策略1：固定TTL（Time To Live）】")    print("特点：设置固定过期时间")    print("优点：实现简单，自动清理")    print("缺点：可能返回过期数据")    print("适用：可以容忍短期延迟的场景")    print("\n示例代码：")    print("""    cache.set('question', answer, ttl=3600)  # 1小时后过期    """)        print("\n【策略2：版本号控制】")    print("特点：每次更新增加版本号")    print("优点：精确控制，不会返回旧数据")    print("缺点：需要维护版本号系统")    print("适用：数据一致性要求高的场景")    print("\n示例代码：")    print("""    # 更新文档时    doc.version += 1    cache.invalidate_by_version(doc.id, doc.version)    """)        print("\n【策略3：主动推送失效】")    print("特点：内容更新时主动通知缓存失效")    print("优点：实时性最好")    print("缺点：需要额外的通知机制")    print("适用：分布式系统、多节点部署")    print("\n示例代码：")    print("""    # 发布更新事件    event_bus.publish('document_updated', doc_id='policy_123')        # 监听器失效缓存    @event_bus.subscribe('document_updated')    def on_document_updated(doc_id):        cache.invalidate_by_pattern(doc_id)    """)        # 实际测试对比    print("\n" + "=" * 60)    print("实际场景测试")    print("=" * 60)        # 模拟：文档每小时更新一次，查询每分钟一次    ttl_configs = [        {'name': 'TTL=10分钟', 'ttl': 600, 'update_interval': 3600},        {'name': 'TTL=30分钟', 'ttl': 1800, 'update_interval': 3600},        {'name': 'TTL=60分钟', 'ttl': 3600, 'update_interval': 3600},    ]        print("\n假设：文档每小时更新，查询每分钟一次（共120次查询）")    print("\n不同TTL配置的效果：\n")        for config in ttl_configs:        ttl = config['ttl']        update_interval = config['update_interval']                # 计算可能返回过期数据的次数        stale_responses = max(0, (update_interval - ttl) / 60)  # 分钟        freshness_rate = (60 - stale_responses) / 60 * 100                print(f"{config['name']}:")        print(f"  可能过期的响应: ~{int(stale_responses)}次")        print(f"  数据新鲜度: {freshness_rate:.1f}%")        print()        print("💡 建议：")    print("  - 制度类文档：TTL = 24小时 + 版本控制")    print("  - 产品信息：TTL = 1小时 + 版本控制")    print("  - 实时数据：不缓存或TTL < 5分钟")# 运行对比compare_invalidation_strategies()