jhzhang/rag
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 缓存系统全面加固 — 版本失效 + 死代码清理 + 防御性改进

Browse Source 
基于缓存审查报告 + Claude Code 交叉审查的反馈：

P0+ 语义缓存版本失效:
- sync.py: 文档变更后清空语义缓存，防止返回过时的 images/sources
- document_routes.py: 删除文档时同步清理缓存
- kb_routes.py: 删除向量库时同步清理缓存

P1.5 cache_type 包含式校验:
- intent_analyzer.py: != "rag_answer" 改为 == "intent_analysis"

P2+ Rerank Cache 版本失效:
- cache.py: increment_kb_version 时同时清空 rerank_cache

P1+P6 死代码清理:
- agentic.py: 移除未使用的 SemanticCache 实例和相关导入
- agentic_base.py: 移除未使用的语义缓存配置导入
- cache.py: 移除 _compute_doc_hash 和未使用的 doc_ids 参数
- engine.py: 移除 set_query_result 调用中的 doc_ids 参数

P7 缓存质量门槛:
- config.py: CACHE_MIN_SCORE 从 0.0 改为 0.3
This commit is contained in:
lacerate551
2026-06-21 16:08:58 +08:00
parent 21470ed28c
commit 0f339a0998
8 changed files with 58 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
api/document_routes.py
View File

@@ -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({

15
api/kb_routes.py
View File

@@ -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,

32
core/agentic.py
View File

@@ -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

16
core/agentic_base.py
View File

@@ -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

56
core/cache.py
View File

@@ -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()
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

8
core/engine.py
View File

@@ -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)

4
core/intent_analyzer.py
View File

@@ -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):

11
knowledge/sync.py
View File

@@ -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: