diff --git a/api/document_routes.py b/api/document_routes.py index 0aae6e2..e3a5b9a 100644 --- a/api/document_routes.py +++ b/api/document_routes.py @@ -775,7 +775,22 @@ def delete_document(doc_path: str) -> Tuple[Any, int]: if kb_manager: kb_manager.delete_document(collection, filename) - # 2. 删除文件 + # 2. 缓存失效 + try: + from core.cache import get_cache_manager + _cm = get_cache_manager() + _cm.increment_kb_version(collection) + except Exception: + pass + try: + from core.semantic_cache import get_semantic_cache + _sc = get_semantic_cache() + if _sc: + _sc.clear() + except Exception: + pass + + # 3. 删除文件 os.remove(filepath) return jsonify({ diff --git a/api/kb_routes.py b/api/kb_routes.py index 9f5f886..5c488b1 100644 --- a/api/kb_routes.py +++ b/api/kb_routes.py @@ -260,6 +260,21 @@ def delete_collection(kb_name: str) -> Tuple[Any, int]: success, message = kb_manager.delete_collection(kb_name, delete_documents) if success: + # 缓存失效 + try: + from core.cache import get_cache_manager + _cm = get_cache_manager() + _cm.increment_kb_version(kb_name) + except Exception: + pass + try: + from core.semantic_cache import get_semantic_cache + _sc = get_semantic_cache() + if _sc: + _sc.clear() + except Exception: + pass + return jsonify({ "success": True, "message": message, diff --git a/core/agentic.py b/core/agentic.py index acc1249..f30b1f9 100644 --- a/core/agentic.py +++ b/core/agentic.py @@ -31,17 +31,11 @@ from core.llm_utils import call_llm, quick_yes_no, parse_json_from_response from .agentic_base import ( API_KEY, BASE_URL, MODEL, HAS_SERPER, - HAS_BUDGET, SEMANTIC_CACHE_ENABLED, + HAS_BUDGET, MAX_CONTEXT_TOKENS, MAX_CONTEXT_COUNT, RERANK_THRESHOLD, SOURCE_KB, SOURCE_WEB, ) -# 尝试导入语义缓存 -try: - from core.semantic_cache import SemanticCache -except ImportError: - SemanticCache = None - # 导入 Mixin 类 from .agentic_query import QueryRewriteMixin from .agentic_search import SearchMixin @@ -120,30 +114,6 @@ class AgenticRAG( except ImportError: self.loop_guard = None - # 初始化语义缓存 - self.semantic_cache = None - self.embedding_model = None - if SEMANTIC_CACHE_ENABLED and SemanticCache: - try: - engine = get_engine() - if engine and hasattr(engine, 'embedding_model'): - self.embedding_model = engine.embedding_model - emb_dim = 768 - # 优先使用新 API，兼容旧版本 - if hasattr(self.embedding_model, 'get_embedding_dimension'): - emb_dim = self.embedding_model.get_embedding_dimension() - elif hasattr(self.embedding_model, 'get_sentence_embedding_dimension'): - emb_dim = self.embedding_model.get_sentence_embedding_dimension() - self.semantic_cache = SemanticCache( - dim=emb_dim, - threshold=0.92, - max_size=5000 - ) - logger.info(f"语义缓存已启用，维度={emb_dim}") - except Exception as e: - logger.warning(f"语义缓存初始化失败: {e}") - self.semantic_cache = None - # Context Compression 配置 self.MAX_CONTEXT_TOKENS = MAX_CONTEXT_TOKENS self.MAX_CONTEXT_COUNT = MAX_CONTEXT_COUNT diff --git a/core/agentic_base.py b/core/agentic_base.py index 9558769..15eef54 100644 --- a/core/agentic_base.py +++ b/core/agentic_base.py @@ -25,22 +25,6 @@ except ImportError: HAS_BUDGET = False CallType = None -# 语义缓存 -try: - from config import SEMANTIC_CACHE_ENABLED, SEMANTIC_CACHE_THRESHOLD - HAS_SEMANTIC_CACHE_CONFIG = True -except ImportError: - SEMANTIC_CACHE_ENABLED = False - SEMANTIC_CACHE_THRESHOLD = 0.92 - HAS_SEMANTIC_CACHE_CONFIG = False - -try: - from core.semantic_cache import SemanticCache, get_semantic_cache - HAS_SEMANTIC_CACHE = True -except ImportError: - HAS_SEMANTIC_CACHE = False - SemanticCache = None - # LLM 配置 try: from config import API_KEY, BASE_URL, MODEL diff --git a/core/cache.py b/core/cache.py index e9c2beb..22fbd7a 100644 --- a/core/cache.py +++ b/core/cache.py @@ -189,6 +189,8 @@ class RAGCacheManager: # 失效旧版本缓存 self.query_cache.invalidate_by_version(old_version) self.embedding_cache.invalidate_by_version(old_version) + # Rerank cache 无 kb_version 字段，文档变更时全量清空以防过时分数 + self.rerank_cache.clear() logger.info(f"知识库 {kb_name} 版本更新: {old_version} -> {new_version}") return new_version @@ -196,87 +198,41 @@ class RAGCacheManager: # ==================== Query Cache 方法 ==================== @staticmethod - def _make_query_cache_key(query: str, kb_name: str, kb_version: int, doc_hash: str = "") -> str: + def _make_query_cache_key(query: str, kb_name: str, kb_version: int) -> str: """ - 生成查询缓存 key - - Args: - query: 查询文本 - kb_name: 知识库名称 - kb_version: 知识库版本号 - doc_hash: 相关文档版本哈希（细粒度失效） - - Returns: - 缓存 key + 生成查询缓存 key（基于 kb_version 的粗粒度失效） """ - if doc_hash: - # 细粒度：只失效相关文档的缓存 - return hashlib.md5( - f"query:{query}:{kb_name}:{doc_hash}".encode() - ).hexdigest() - else: - # 粗粒度：整个知识库版本变化时失效 - return hashlib.md5( - f"query:{query}:{kb_name}:{kb_version}".encode() - ).hexdigest() + return hashlib.md5( + f"query:{query}:{kb_name}:{kb_version}".encode() + ).hexdigest() - def get_query_result(self, query: str, kb_name: str, doc_ids: List[str] = None) -> Optional[Dict]: + def get_query_result(self, query: str, kb_name: str) -> Optional[Dict]: """ 获取查询缓存结果 - 始终使用粗粒度 key（基于 kb_version），确保 GET/SET key 一致。 - Args: query: 查询文本 kb_name: 知识库名称 - doc_ids: 保留参数以兼容调用方签名（当前未使用） """ kb_version = self.get_kb_version(kb_name) - - # 使用粗粒度 key，与 SET 保持一致 key = self._make_query_cache_key(query, kb_name, kb_version) return self.query_cache.get(key) - def set_query_result(self, query: str, kb_name: str, result: Dict, doc_ids: List[str] = None) -> None: + def set_query_result(self, query: str, kb_name: str, result: Dict) -> None: """ 设置查询缓存结果 - 始终使用粗粒度 key（基于 kb_version），确保 GET/SET key 一致。 kb_version 在文档变更时自增，触发整个知识库的缓存失效。 Args: query: 查询文本 kb_name: 知识库名称 result: 缓存结果 - doc_ids: 保留参数以兼容调用方签名（当前未使用） """ kb_version = self.get_kb_version(kb_name) - - # 使用与 GET 相同的粗粒度 key，确保缓存可命中 key = self._make_query_cache_key(query, kb_name, kb_version) self.query_cache.set(key, result, kb_version=kb_version) - def _compute_doc_hash(self, kb_name: str, doc_ids: List[str]) -> str: - """ - 计算文档版本哈希 - - 用于细粒度缓存失效：只失效相关文档变化时的缓存 - """ - if not doc_ids: - return "" - - # 从文档 ID 中提取 source（文件名） - sources = set() - for doc_id in doc_ids: - # doc_id 格式通常为 "filename_text_0" 或类似 - parts = doc_id.split('_') - if parts: - sources.add(parts[0]) - - # 生成哈希 - sources_str = ','.join(sorted(sources)) - return hashlib.md5(f"docs:{sources_str}".encode()).hexdigest() - # ==================== Embedding Cache 方法 ==================== @staticmethod diff --git a/core/engine.py b/core/engine.py index f4283fc..2749f30 100644 --- a/core/engine.py +++ b/core/engine.py @@ -560,7 +560,7 @@ class RAGEngine: top_score = 1.0 - top_dist if top_score >= CACHE_MIN_SCORE: doc_ids = result.get('ids', [[]])[0] if result.get('ids') else [] - cache.set_query_result(query, kb_name, result, doc_ids=doc_ids) + cache.set_query_result(query, kb_name, result) result['_debug'] = _debug return result @@ -582,7 +582,7 @@ class RAGEngine: top_score = 1.0 - top_dist if top_score >= CACHE_MIN_SCORE: doc_ids = result.get('ids', [[]])[0] if result.get('ids') else [] - cache.set_query_result(query, kb_name, result, doc_ids=doc_ids) + cache.set_query_result(query, kb_name, result) result['_debug'] = _debug return result except Exception as e: @@ -619,7 +619,7 @@ class RAGEngine: if top_score >= CACHE_MIN_SCORE: # 传递 doc_ids 实现细粒度缓存失效 doc_ids = result.get('ids', [[]])[0] if result.get('ids') else [] - cache.set_query_result(query, kb_name, result, doc_ids=doc_ids) + cache.set_query_result(query, kb_name, result) _debug['timing']['total_ms'] = int((time.time() - _overall_start) * 1000) result['_debug'] = _debug return result @@ -812,7 +812,7 @@ class RAGEngine: if top_score >= CACHE_MIN_SCORE: # 置信度阈值 # 传递 doc_ids 实现细粒度缓存失效 doc_ids = fused_results.get('ids', [[]])[0] if fused_results.get('ids') else [] - cache.set_query_result(query, kb_name, fused_results, doc_ids=doc_ids) + cache.set_query_result(query, kb_name, fused_results) fused_results['_debug'] = _debug _debug['timing']['total_ms'] = int((time.time() - _overall_start) * 1000) diff --git a/core/intent_analyzer.py b/core/intent_analyzer.py index 296f47d..4043dc8 100644 --- a/core/intent_analyzer.py +++ b/core/intent_analyzer.py @@ -316,8 +316,8 @@ class IntentAnalyzer: if query_emb is not None: cached = cache.get(query_emb) - # 确保缓存条目是意图分析结果（非 RAG 回答缓存） - if cached and cached.get("cache_type") != "rag_answer": + # 确保缓存条目是意图分析结果（包含式校验，避免新增缓存类型时误命中） + if cached and cached.get("cache_type") == "intent_analysis": # 二次验证：检查原始 query 文本相似度 cached_query = cached.get("_raw_query", "") if cached_query and self._query_text_similar(query, cached_query): diff --git a/knowledge/sync.py b/knowledge/sync.py index a669458..69f1d59 100644 --- a/knowledge/sync.py +++ b/knowledge/sync.py @@ -664,6 +664,17 @@ class KnowledgeSyncService: except Exception as e: logger.warning(f"递增缓存版本号失败: {e}") + # 语义缓存无版本号机制，文档变更后必须清空， + # 否则可能返回过时的 images/sources/citations（如已删除的图片 404） + try: + from core.semantic_cache import get_semantic_cache + _sc = get_semantic_cache() + if _sc: + _sc.clear() + logger.debug(f"已清空语义缓存（文档变更触发）: {kb_name}") + except Exception as e: + logger.warning(f"清空语义缓存失败: {e}") + return True except Exception as e: