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

fix: 语义缓存误命中修复 — 二次 Jaccard 验证 + query-only embedding

Browse Source 
问题: 同会话中语义相近但不同的问题(如"总体要求"vs"工作流程和规范要求")
会错误命中缓存，返回上一个问题的答案。

修复:
- intent_analyzer: 缓存 embedding 改用 query-only(不含历史)，避免上下文污染
- intent_analyzer + chat_routes: 缓存读取增加字符级 Jaccard 二次验证(阈值0.5)
- 缓存写入存储 _raw_query 供二次验证比对
- 兼容旧缓存条目(无 _raw_query 时放行)
This commit is contained in:
lacerate551
2026-06-21 15:37:37 +08:00
parent 258be54df7
commit 21470ed28c
2 changed files with 89 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
api/chat_routes.py
View File

@@ -2298,32 +2298,45 @@ def rag():
cached = _sc.get(_semantic_cache_emb)
# 防御性校验：必须是 RAG 回答缓存（非 intent_analyzer 缓存），且回答非空
if cached is not None and cached.get("cache_type") == "rag_answer" and cached.get("answer"):
logger.info(f"[语义缓存] 命中: {message[:50]}...")
cached_answer = cached.get("answer", "")
# 流式返回缓存的答案
yield f"data: {json.dumps({'type': 'start', 'message': '正在检索知识库...'}, ensure_ascii=False)}\n\n"
# 分块发送缓存答案
chunk_size = 20
for i in range(0, len(cached_answer), chunk_size):
chunk = cached_answer[i:i+chunk_size]
full_answer.append(chunk)
yield f"data: {json.dumps({'type': 'chunk', 'content': chunk}, ensure_ascii=False)}\n\n"
finish_event = {
"type": "finish",
"answer": cached_answer,
"mode": "rag",
"session_id": session_id,
"sources": cached.get("sources", []),
"citations": cached.get("citations", []),
"images": cached.get("images", []),
"tables": cached.get("tables", []),
"sections": [],
"duration_ms": int((_time.time() - start_time) * 1000),
"confidence_score": 1.0,
"semantic_cache_hit": True
}
yield f"data: {json.dumps(finish_event, ensure_ascii=False)}\n\n"
return
# 二次验证：检查原始查询文本相似度，防止语义相近但不同问题误命中
_cached_query = cached.get("_raw_query", "")
if _cached_query:
_qa, _qb = set(retrieval_query), set(_cached_query)
_jaccard = len(_qa & _qb) / len(_qa | _qb) if (_qa and _qb) else 0
if _jaccard < 0.5:
logger.info(
f"[语义缓存] 二次验证拒绝 (Jaccard={_jaccard:.2f}): "
f"query='{retrieval_query[:30]}' vs cached='{_cached_query[:30]}'"
)
cached = None # 拒绝命中，继续正常流程
# 通过验证（或旧缓存无 _raw_query 字段兼容放行）
if cached is not None and cached.get("cache_type") == "rag_answer" and cached.get("answer"):
logger.info(f"[语义缓存] 命中: {message[:50]}...")
cached_answer = cached.get("answer", "")
# 流式返回缓存的答案
yield f"data: {json.dumps({'type': 'start', 'message': '正在检索知识库...'}, ensure_ascii=False)}\n\n"
# 分块发送缓存答案
chunk_size = 20
for i in range(0, len(cached_answer), chunk_size):
chunk = cached_answer[i:i+chunk_size]
full_answer.append(chunk)
yield f"data: {json.dumps({'type': 'chunk', 'content': chunk}, ensure_ascii=False)}\n\n"
finish_event = {
"type": "finish",
"answer": cached_answer,
"mode": "rag",
"session_id": session_id,
"sources": cached.get("sources", []),
"citations": cached.get("citations", []),
"images": cached.get("images", []),
"tables": cached.get("tables", []),
"sections": [],
"duration_ms": int((_time.time() - start_time) * 1000),
"confidence_score": 1.0,
"semantic_cache_hit": True
}
yield f"data: {json.dumps(finish_event, ensure_ascii=False)}\n\n"
return
except Exception as e:
logger.warning(f"[语义缓存] 检查失败: {e}")
_semantic_cache_emb = None
@@ -2886,6 +2899,7 @@ def rag():
"citations": citation_result.get("citations", []),
"images": rich_media.get("images", []),
"tables": rich_media.get("tables", []),
"_raw_query": retrieval_query,
})
logger.debug(f"[语义缓存] 写入成功: {message[:50]}...")
except Exception as e:

58
core/intent_analyzer.py
View File

@@ -308,18 +308,30 @@ class IntentAnalyzer:
return self._exact_cache[exact_key]
# 2. 尝试从语义缓存获取
# 关键：语义缓存只用原始 query 做 embedding（不含历史），
# 避免同会话中不同问题因历史上下文污染导致误命中
cache = self._get_cache()
if cache:
# 使用 query + 历史关键信息作为缓存键
cache_key = self._build_cache_key(query, history)
cache_emb = self._get_embedding(cache_key)
query_emb = self._get_embedding(query)
if cache_emb is not None:
cached = cache.get(cache_emb)
if query_emb is not None:
cached = cache.get(query_emb)
# 确保缓存条目是意图分析结果（非 RAG 回答缓存）
if cached and cached.get("cache_type") != "rag_answer":
logger.info(f"意图分析缓存命中: {cached.get('reason', '')[:50]}")
return IntentAnalysis.from_dict(cached)
# 二次验证：检查原始 query 文本相似度
cached_query = cached.get("_raw_query", "")
if cached_query and self._query_text_similar(query, cached_query):
logger.info(f"意图分析缓存命中: {cached.get('reason', '')[:50]}")
return IntentAnalysis.from_dict(cached)
elif cached_query:
logger.info(
f"意图分析缓存二次验证拒绝: "
f"query='{query[:30]}' vs cached='{cached_query[:30]}'"
)
else:
# 旧缓存无 _raw_query 字段，兼容放行
logger.info(f"意图分析缓存命中(无验证): {cached.get('reason', '')[:50]}")
return IntentAnalysis.from_dict(cached)
else:
logger.debug(f"意图分析缓存未命中，缓存状态: {cache.get_stats()}")
else:
@@ -388,10 +400,12 @@ class IntentAnalyzer:
)
# 存入语义缓存（标记类型，避免与 RAG 回答缓存混淆）
if cache and cache_emb is not None:
# 使用仅含 query 的 embedding，不含历史，防止同会话误命中
if cache and query_emb is not None:
cache_data = analysis.to_dict()
cache_data["cache_type"] = "intent_analysis"
cache.set(cache_emb, cache_data)
cache_data["_raw_query"] = query # 供二次验证使用
cache.set(query_emb, cache_data)
# 存入精确匹配缓存
if len(self._exact_cache) < self._exact_cache_max:
@@ -431,6 +445,32 @@ class IntentAnalyzer:
return " | ".join(parts)
@staticmethod
def _query_text_similar(query: str, cached_query: str, threshold: float = 0.5) -> bool:
"""
判断两个 query 文本是否足够相似（字符级 Jaccard）。
用于语义缓存命中后的二次验证，防止语义相近但实际意图不同的问题误命中。
Args:
query: 当前查询
cached_query: 缓存中的原始查询
threshold: 相似度阈值，默认 0.5
Returns:
True 表示足够相似，可以命中缓存
"""
# 精确匹配快速路径
if query.strip() == cached_query.strip():
return True
# 字符级 Jaccard 相似度
set_a = set(query)
set_b = set(cached_query)
if not set_a or not set_b:
return False
intersection = len(set_a & set_b)
union = len(set_a | set_b)
return (intersection / union) >= threshold
def _build_history_summary(
self,
history: List[dict],