perf(cache): 修复缓存失效 Bug + 集成语义缓存 + 删除 AgenticRAG 死代码

- fix: Query Cache GET/SET Key 不匹配导致命中率始终为 0%
- fix: CACHE_MIN_SCORE 阈值 0.3 对 ChromaDB cosine distance 过于严格
- feat: 在 /rag 端点集成语义缓存（命中时跳过检索+生成，92x 加速）
- refactor: 删除 AgenticRAG 死代码路径（10 个 agentic_*.py，约 1950 行）
- cleanup: 移除 engine.py 死方法、路由死函数、初始化死代码

api/__init__.py

@@ -3,7 +3,7 @@ API 路由层 — Flask 应用工厂
 
 本模块实现 Flask 应用工厂模式，负责：
 - 创建和配置 Flask 应用实例
-- 初始化核心服务（AgenticRAG、同步服务）
+- 初始化核心服务（同步服务）
 - 注册所有 API Blueprint
 - 配置前端静态文件路由
 
@@ -47,7 +47,7 @@ def create_app() -> 'Flask':
 
     1. 创建 Flask 应用，配置 CORS
     2. 初始化 Repository（会话存储）
-    3. 初始化核心服务（AgenticRAG、同步服务）
+    3. 初始化核心服务（同步服务）
     4. 注册所有 API Blueprint
     5. 配置前端静态文件路由
     6. 执行生产环境配置校验
@@ -93,19 +93,6 @@ def create_app() -> 'Flask':
 
     # ==================== 核心服务初始化 ====================
 
-    # Agentic RAG 引擎
-    try:
-        from core.agentic import AgenticRAG
-        from config import ENABLE_WEB_SEARCH
-        agentic_rag = AgenticRAG(
-            enable_web_search=ENABLE_WEB_SEARCH,
-        )
-        app.config['AGENTIC_RAG'] = agentic_rag
-        logger.info(f"Agentic RAG 引擎已初始化（网络搜索={'启用' if ENABLE_WEB_SEARCH else '关闭'}）")
-    except Exception as e:
-        app.config['AGENTIC_RAG'] = None
-        logger.warning(f"Agentic RAG 初始化失败: {e}")
-
     # 同步服务
     try:
         from knowledge.sync import KnowledgeSyncService

api/chat_routes.py

@@ -381,16 +381,6 @@ def _build_context_with_budget(contexts: List[Dict], max_chars: int, soft_limit:
     return "\n\n".join(parts)
 
 
-def _get_agentic_rag() -> 'AgenticRAG':
-    """
-    获取 AgenticRAG 实例
-
-    Returns:
-        AgenticRAG: 当前应用中的 AgenticRAG 实例
-    """
-    return current_app.config['AGENTIC_RAG']
-
-
 def _attach_citations(answer: str, contexts: List[Dict]) -> Dict[str, Any]:
     """
     自动为回答添加引用标记（按段落级别匹配，jieba 分词精准匹配）
@@ -1464,6 +1454,47 @@ def rag():
             except Exception as e:
                 logger.warning(f"意图分析失败: {e}，继续执行检索流程")
 
+            # 1.5 语义缓存检查（跳过检索+生成全流程）
+            _semantic_cache_emb = None
+            try:
+                from core.semantic_cache import get_semantic_cache
+                from core.engine import get_engine as _get_eng
+                _sc = get_semantic_cache()
+                _eng = _get_eng()
+                if _sc and _eng and hasattr(_eng, 'embedding_model'):
+                    _semantic_cache_emb = _eng.embedding_model.encode(message)
+                    cached = _sc.get(_semantic_cache_emb)
+                    if cached is not None:
+                        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
+
             # 1. 发送开始事件
             yield f"data: {json.dumps({'type': 'start', 'message': '正在检索知识库...'}, ensure_ascii=False)}\n\n"
 
@@ -1951,6 +1982,28 @@ def rag():
                     "rerank_cached": rerank_cached,
                     "total_ms": duration_ms
                 }
+            # 10.5 写入语义缓存
+            try:
+                from core.semantic_cache import get_semantic_cache
+                _sc = get_semantic_cache()
+                if _sc and filtered_answer:
+                    if _semantic_cache_emb is None:
+                        from core.engine import get_engine as _get_eng
+                        _eng = _get_eng()
+                        if _eng and hasattr(_eng, 'embedding_model'):
+                            _semantic_cache_emb = _eng.embedding_model.encode(message)
+                    if _semantic_cache_emb is not None:
+                        _sc.set(_semantic_cache_emb, {
+                            "answer": filtered_answer,
+                            "sources": sources,
+                            "citations": citation_result.get("citations", []),
+                            "images": rich_media.get("images", []),
+                            "tables": rich_media.get("tables", []),
+                        })
+                        logger.debug(f"[语义缓存] 写入成功: {message[:50]}...")
+            except Exception as e:
+                logger.info(f"[语义缓存] 写入失败: {e}")
+
             yield f"data: {json.dumps(finish_event, ensure_ascii=False)}\n\n"
 
         except Exception as e:
@@ -2019,3 +2072,68 @@ def search():
         'metadatas': results['metadatas'][0],
         'scores': results['scores'][0]
     })
+
+
+# ==================== 缓存调试接口（临时） ====================
+
+@chat_bp.route('/cache/stats', methods=['GET'])
+def cache_stats():
+    """缓存统计接口（调试用）"""
+    stats = {}
+    try:
+        from core.cache import get_cache_manager
+        cache = get_cache_manager()
+        all_stats = cache.get_all_stats()
+        for name, s in all_stats.items():
+            stats[name] = {
+                'total_entries': s.total_entries,
+                'hits': s.hits,
+                'misses': s.misses,
+                'hit_rate': f"{s.hit_rate:.2%}",
+                'evictions': s.evictions,
+            }
+    except Exception as e:
+        stats['_error'] = str(e)
+
+    # 语义缓存统计（全局单例 + 意图分析器实例）
+    try:
+        from core.semantic_cache import get_semantic_cache
+        sc = get_semantic_cache()
+        if sc:
+            stats['semantic_cache'] = sc.get_stats()
+    except Exception:
+        pass
+
+    try:
+        from core.intent_analyzer import IntentAnalyzer
+        if hasattr(IntentAnalyzer, '_instance'):
+            ia = IntentAnalyzer._instance
+            if hasattr(ia, 'semantic_cache') and ia.semantic_cache:
+                stats['semantic_cache_intent'] = ia.semantic_cache.get_stats()
+    except Exception:
+        pass
+
+    return jsonify(stats)
+
+
+@chat_bp.route('/cache/clear', methods=['POST'])
+def cache_clear():
+    """缓存清空接口（调试用）"""
+    result = {}
+    try:
+        from core.cache import get_cache_manager
+        cache = get_cache_manager()
+        cache.clear_all()
+        result['lru_cache'] = 'cleared'
+    except Exception as e:
+        result['lru_cache'] = f'error: {e}'
+
+    try:
+        from core.semantic_cache import get_semantic_cache
+        sc = get_semantic_cache()
+        sc.clear()
+        result['semantic_cache'] = 'cleared'
+    except Exception:
+        result['semantic_cache'] = 'not_available'
+
+    return jsonify({'status': 'ok', 'cleared': result})

config.example.py

@@ -181,7 +181,10 @@ SEMANTIC_CACHE_ENABLED = True
 SEMANTIC_CACHE_THRESHOLD = 0.92  # 相似度阈值
 
 # 缓存写入最低置信度
-CACHE_MIN_SCORE = 0.3
+# 注意：ChromaDB cosine distance 范围 [0,2]，score = 1 - dist
+# 当前 embedding 模型的 cosine similarity 普遍在 0.03-0.06 之间
+# 搜索管线已通过 rerank 过滤低质量结果，此处不再额外限制
+CACHE_MIN_SCORE = 0.0
 
 # LLM 调用预算
 MAX_LLM_CALLS_PER_QUERY = 2

core/__init__.py

@@ -3,7 +3,6 @@ RAG 核心引擎模块
 
 包含：
 - engine: RAGEngine 单例类，管理模型和共享资源
-- agentic: AgenticRAG 智能问答
 - bm25_index: BM25 关键词检索索引
 - chunker: 文本分块器
 """

core/agentic.py

@@ -1,390 +0,0 @@
-"""
-Agentic RAG - 知识库智能问答系统
-
-核心能力：
-1. 知识库检索 - 向量检索 + BM25 + Rerank
-2. 网络搜索 - 当知识库不足时自动搜索（需配置SERPER_API_KEY）
-3. 图谱检索 - 实体关系推理（需配置Neo4j）
-4. 多源融合 - 智能处理知识库和网络内容
-5. Agent决策 - 动态决定检索、改写、分解等操作
-
-使用方式：
-    from core.agentic import AgenticRAG
-
-    rag = AgenticRAG()
-    result = rag.process("你的问题")
-    print(result["answer"])
-"""
-
-import json
-import logging
-from openai import OpenAI
-
-# 配置日志
-logger = logging.getLogger(__name__)
-
-# 导入基础模块
-from core.engine import get_engine
-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,
-    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
-from .agentic_answer import AnswerMixin
-from .agentic_citation import CitationMixin
-from .agentic_media import RichMediaMixin
-from .agentic_quality import QualityMixin
-from .agentic_context import ContextMixin
-from .agentic_meta import MetaQuestionMixin
-
-
-class AgenticRAG(
-    QueryRewriteMixin,
-    SearchMixin,
-    AnswerMixin,
-    CitationMixin,
-    RichMediaMixin,
-    QualityMixin,
-    ContextMixin,
-    MetaQuestionMixin
-):
-    """
-    Agentic RAG - 知识库智能问答
-
-    通过 Mixin 模式组合功能：
-    - QueryRewriteMixin: 查询重写
-    - SearchMixin: 检索功能
-    - AnswerMixin: 答案生成
-    - CitationMixin: 引用处理
-    - RichMediaMixin: 富媒体处理
-    - QualityMixin: 质量评估
-    - ContextMixin: 上下文处理
-    - MetaQuestionMixin: 元问题处理
-    """
-
-    def __init__(
-        self,
-        max_iterations: int = 3,
-        enable_web_search: bool = True,
-        **kwargs
-    ):
-        """初始化"""
-        self.max_iterations = max_iterations
-        self.enable_web_search = enable_web_search and HAS_SERPER
-        self.client = OpenAI(api_key=API_KEY, base_url=BASE_URL)
-
-        # 信息来源标记
-        self.SOURCE_KB = SOURCE_KB
-        self.SOURCE_WEB = SOURCE_WEB
-
-        # 初始化置信度门控
-        try:
-            from core.confidence_gate import create_gate
-            self.confidence_gate = create_gate()
-        except ImportError:
-            self.confidence_gate = None
-
-        # 初始化多维质量评估器
-        try:
-            from core.quality_assessor import create_assessor
-            self.quality_assessor = create_assessor()
-        except ImportError:
-            self.quality_assessor = None
-
-        # 初始化推理反思器
-        try:
-            from core.reasoning_reflector import create_reflector
-            self.reasoning_reflector = create_reflector()
-        except ImportError:
-            self.reasoning_reflector = None
-
-        # 初始化循环防护器
-        try:
-            from core.loop_guard import create_guard
-            self.loop_guard = create_guard(max_iterations=max_iterations)
-        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
-        self.RERANK_THRESHOLD = RERANK_THRESHOLD
-
-        # Answer Grounding 配置
-        self.MAX_GROUNDING_RETRY = 1
-        self.grounding_retry_count = 0
-
-    def should_rewrite(self, query: str, history: list = None) -> bool:
-        """判断是否需要重写查询"""
-        # 口语化表达模式
-        colloquial_patterns = [
-            "这个", "那个", "它", "这", "那",
-            "上面", "下面", "刚才", "之前",
-            "能不能", "可以吗", "行不行",
-            "怎么办", "怎么弄", "咋整"
-        ]
-
-        for pattern in colloquial_patterns:
-            if pattern in query:
-                return True
-
-        # 查询太短
-        if len(query) < 5:
-            return True
-
-        # 有对话历史时，可能需要实体补全
-        if history:
-            for msg in reversed(history[-3:]):
-                if msg.get("role") == "user":
-                    prev_query = msg.get("content", "")
-                    # 如果当前查询缺少主语，可能需要补全
-                    if any(kw in query for kw in ["标准", "规定", "流程", "制度"]):
-                        if not any(kw in query for kw in ["报销", "出差", "请假", "工资", "合同"]):
-                            return True
-
-        return False
-
-    def process(self, query: str, verbose: bool = True, history: list = None,
-                allowed_levels: list = None, role: str = None, department: str = None,
-                emit_log=None) -> dict:
-        """
-        主流程：智能问答
-
-        Args:
-            query: 用户问题
-            verbose: 是否打印详细过程
-            history: 对话历史
-            allowed_levels: 允许访问的安全级别
-            role: 用户角色
-            department: 用户部门
-            emit_log: 日志发射函数（流式输出）
-
-        Returns:
-            {
-                "answer": str,
-                "sources": list,
-                "images": list,
-                "tables": list,
-                "citations": list,
-                "log_trace": list
-            }
-        """
-        log_trace = []
-
-        # 1. 检查元问题
-        if self._is_meta_question(query):
-            answer = self._answer_meta_question(query, allowed_levels, role, department)
-            return {
-                "answer": answer,
-                "sources": [],
-                "images": [],
-                "tables": [],
-                "citations": [],
-                "log_trace": [{"phase": "meta_question", "query": query}]
-            }
-
-        # 2. 查询重写
-        current_query = query
-        if self.should_rewrite(query, history):
-            current_query = self._rewrite_query(query, history)
-            log_trace.append({"phase": "rewrite", "original": query, "rewritten": current_query})
-            if emit_log:
-                emit_log(f"📝 查询重写: {query} → {current_query}")
-
-        # 3. 知识库检索
-        contexts = []
-        try:
-            engine = get_engine()
-            if not engine._initialized:
-                engine.initialize()
-
-            # 获取用户可访问的向量库
-            from knowledge.manager import get_kb_manager
-            kb_mgr = get_kb_manager()
-            accessible = kb_mgr.get_accessible_collections(role or 'user', department or '', 'read')
-
-            # 统一使用 search_knowledge() — 生产路径的同一 API
-            # search_knowledge() 返回 dict: {ids, documents, metadatas, distances}，每项为 list[list]
-            # top_k 与生产路径对齐（30），确保 Rerank 后仍有足够结果
-            results = engine.search_knowledge(
-                query=current_query,
-                top_k=30,
-                collections=accessible if accessible else None,
-            )
-
-            docs = results.get('documents', [[]])[0]
-            metas = results.get('metadatas', [[]])[0]
-            dists = results.get('distances', [[]])[0]
-
-            for doc, meta, score in zip(docs, metas, dists):
-                contexts.append({
-                    'doc': doc,
-                    'meta': meta,
-                    'score': 1 - score if score <= 1 else 1 / (1 + score),  # 距离→相似度
-                    'source_type': self.SOURCE_KB,
-                    'query': current_query
-                })
-
-            log_trace.append({"phase": "kb_search", "query": current_query, "results": len(contexts)})
-            if emit_log:
-                emit_log(f"🔍 知识库检索: {len(contexts)} 条结果")
-
-        except Exception as e:
-            logger.error(f"知识库检索失败: {e}")
-            log_trace.append({"phase": "kb_search", "error": str(e)})
-
-        # 3.5 图片独立检索 + 打分选择（与生产路径对齐）
-        selected_images = []
-        try:
-            from api.chat_routes import select_images
-            selected_images = select_images(contexts, current_query)
-            if selected_images and emit_log:
-                emit_log(f"🖼️ 图片选择: {len(selected_images)} 张相关图片")
-            log_trace.append({"phase": "image_selection", "count": len(selected_images)})
-        except Exception as e:
-            logger.debug(f"图片选择失败: {e}")
-
-        # 4. 上下文压缩
-        contexts = self._compress_contexts(current_query, contexts)
-
-        # 5. 网络搜索（如果需要）
-        web_contexts = []
-        if self.enable_web_search and (
-            not contexts or
-            not self._is_kb_result_sufficient(current_query, [c['doc'] for c in contexts]) or
-            self._should_web_search(current_query)
-        ):
-            web_contexts = self._web_search_flow(current_query, log_trace, emit_log, verbose, allowed_levels)
-            contexts.extend(web_contexts)
-
-        # 7. 生成答案（注入图片描述到上下文）
-        if contexts:
-            # 将选中图片的描述注入上下文，让 LLM 能"看到"图片内容
-            if selected_images:
-                image_contexts = []
-                for i, img in enumerate(selected_images, 1):
-                    full_desc = img.get('full_description', '') or img.get('description', '')
-                    if full_desc:
-                        img_source = img.get('source', '')
-                        img_page = img.get('page', '')
-                        source_info = f"（来源：{img_source} 第{img_page}页）" if img_source and img_page else ""
-                        image_contexts.append({
-                            'doc': f"【图片{i}】{full_desc}{source_info}",
-                            'meta': {'source': img_source, 'page': img_page, 'chunk_type': img.get('type', 'image')},
-                            'score': img.get('score', 0),
-                            'source_type': self.SOURCE_KB,
-                            'query': current_query
-                        })
-                # 图片上下文追加到知识库上下文前面（让 LLM 优先看到图片信息）
-                contexts = image_contexts + contexts
-
-            answer = self._generate_fused_answer(current_query, contexts, allowed_levels)
-
-            # 答案验证（防止幻觉）
-            if self.grounding_retry_count < self.MAX_GROUNDING_RETRY:
-                answer = self._verify_and_refine_answer(current_query, answer, contexts)
-        else:
-            answer = self._generate_no_context_answer(current_query, allowed_levels)
-
-        # 8. 构建引用
-        citations = self._attach_citations(answer, contexts)
-
-        # 9. 图片结果：优先使用 select_images 的结构化结果（含 URL + 打分）
-        #    回退到 _extract_rich_media（从 metadata 提取）
-        if selected_images:
-            images_result = selected_images
-        else:
-            rich_media = self._extract_rich_media(contexts)
-            images_result = rich_media.get("images", [])
-
-        return {
-            "answer": answer,
-            "sources": citations.get("sources", []),
-            "images": images_result,
-            "tables": citations.get("tables", []) if isinstance(citations, dict) else [],
-            "citations": citations.get("citations", []),
-            "log_trace": log_trace
-        }
-
-    def chat_search(self, query: str, history: list = None, role: str = None,
-                    department: str = None, allowed_levels: list = None) -> dict:
-        """聊天式检索接口"""
-        return self.process(
-            query,
-            verbose=False,
-            history=history,
-            role=role,
-            department=department,
-            allowed_levels=allowed_levels
-        )
-
-    def chat(self):
-        """命令行交互模式"""
-        print("🤖 Agentic RAG 已启动，输入 'quit' 退出")
-        print("-" * 50)
-
-        history = []
-        while True:
-            try:
-                query = input("\n👤 你: ").strip()
-                if not query:
-                    continue
-                if query.lower() in ['quit', 'exit', 'q']:
-                    print("👋 再见！")
-                    break
-
-                result = self.process(query, verbose=True, history=history)
-                print(f"\n🤖 AI: {result['answer']}")
-
-                if result['sources']:
-                    print("\n📚 来源:")
-                    for src in result['sources'][:3]:
-                        print(f"  - {src['source']}")
-
-                history.append({"role": "user", "content": query})
-                history.append({"role": "assistant", "content": result['answer']})
-
-            except KeyboardInterrupt:
-                print("\n👋 再见！")
-                break
-            except Exception as e:
-                print(f"❌ 错误: {e}")

core/agentic_answer.py

@@ -1,214 +0,0 @@
-"""
-Agentic RAG - 答案生成 Mixin
-
-包含答案生成、上下文构建、融合回答等方法
-"""
-
-import logging
-
-from .agentic_base import logger, MODEL, SOURCE_KB, SOURCE_WEB
-from core.llm_utils import call_llm
-
-logger = logging.getLogger(__name__)
-
-
-class AnswerMixin:
-    """答案生成方法"""
-
-    def _generate_fused_answer(self, query: str, contexts: list, allowed_levels: list = None) -> str:
-        """生成融合答案 - 智能处理多源信息"""
-        # 分离不同来源
-        kb_contexts = [c for c in contexts if c.get('source_type') == self.SOURCE_KB]
-        web_contexts = [c for c in contexts if c.get('source_type') == self.SOURCE_WEB]
-
-        # 如果没有任何上下文，检测是否因权限限制
-        if not contexts:
-            return self._generate_no_context_answer(query, allowed_levels)
-
-        # 正常生成答案
-        context_str = self._build_context_string(kb_contexts, web_contexts)
-        prompt = self._build_normal_answer_prompt(query, context_str, kb_contexts, web_contexts)
-
-        result = call_llm(
-            self.client, prompt, MODEL,
-            temperature=0.7,
-            max_tokens=2000
-        )
-        return result or f"生成答案失败"
-
-    def _build_context_string(self, kb_contexts, web_contexts):
-        """构建上下文字符串 - FAQ 优先策略，按分数排序"""
-        # 分离 FAQ 和普通知识库内容
-        faq_contexts = [c for c in kb_contexts if c.get('meta', {}).get('chunk_type') == 'faq']
-        regular_contexts = [c for c in kb_contexts if c.get('meta', {}).get('chunk_type') != 'faq']
-
-        # 按分数降序排列，确保最相关的内容优先展示
-        regular_contexts.sort(key=lambda c: c.get('score', 0), reverse=True)
-
-        # FAQ 部分（优先展示）
-        faq_parts = []
-        for i, c in enumerate(faq_contexts[:3], 1):
-            meta = c['meta']
-            answer = meta.get('faq_answer', c['doc'])
-            faq_parts.append(f"[FAQ-{i}] 常见问题\n问题：{c['doc']}\n标准答案：{answer}")
-
-        # 普通知识库部分（用 12 条，提升覆盖率）
-        kb_parts = []
-        for i, c in enumerate(regular_contexts[:12], 1):
-            meta = c['meta']
-            source_str = meta.get('source', '未知')
-            section = meta.get('section', '')
-            source_info = f"{source_str}"
-            if section:
-                source_info += f" > {section[:60]}"
-            kb_parts.append(f"[知识库-{i}] {source_info}\n{c['doc']}")
-
-        web_parts = []
-        for i, c in enumerate(web_contexts[:5], 1):
-            meta = c['meta']
-            web_parts.append(f"[网络-{i}] {meta.get('title', '')}\n来源:{meta.get('source', '')}\n{c['doc']}")
-
-        return "\n\n".join(faq_parts + kb_parts + web_parts)
-
-    def _build_normal_answer_prompt(self, query, context_str, kb_contexts, web_contexts):
-        """构建正常回答的提示词（与生产路径 generate_answer_stream 对齐）"""
-        # 检测是否有图片上下文
-        has_images = any(c.get('meta', {}).get('chunk_type') in ('image', 'chart', 'table')
-                        for c in kb_contexts)
-
-        image_instruction = ""
-        if has_images:
-            image_instruction = "\n5. 如果参考资料中包含【图片N】信息，请在回答中简要介绍每张图片的内容和用途"
-
-        return f"""你是一个严谨的知识库问答助手。你必须且只能根据用户提供的【参考资料】回答问题。
-
-【参考资料】
-{context_str}
-
-【用户问题】
-{query}
-
-【回答要求】
-1. 如果参考资料中有答案，必须引用对应内容回答，并在回答末尾标注引用编号（如[1]、[2]）
-2. 如果参考资料中确实没有相关信息，简短说明"参考资料中没有相关信息"即可，不要编造或补充资料外的内容
-3. 禁止使用参考资料以外的知识进行补充或推测
-4. 分点列举，条理清晰，语言简洁{image_instruction}
-
-请仔细阅读以上全部参考资料后回答："""
-
-    def _build_answer_prompt_with_permission(self, query, context_str, levels_str, sources_str, kb_contexts, web_contexts):
-        """构建带权限提示的回答提示词"""
-        return f"""你是一个严谨的智能助手。
-
-【用户问题】
-{query}
-
-【重要提示】
-检测到与用户问题更相关的信息可能存在于「{levels_str}」级别的文档中，但用户当前的权限级别无法访问。
-
-【可访问的信息来源】
-{context_str}
-
-【回答要求】
-1. 首先明确告知用户：当前回答基于您有权限访问的文档，可能不完整
-2. 基于现有信息如实回答
-3. 建议用户如需完整信息，请联系管理员申请相应权限
-
-请回答："""
-
-    def _generate_no_context_answer(self, query: str, allowed_levels: list = None) -> str:
-        """无上下文时的回答 — 诚实告知，不编造"""
-        return "参考资料中没有找到与该问题相关的信息，无法根据现有知识库内容回答您的问题。"
-
-    def _verify_and_refine_answer(self, query: str, answer: str, contexts: list) -> str:
-        """验证并精炼答案 - 防止幻觉
-
-        返回值始终是干净的答案文本，不包含验证推理过程。
-        """
-        prompt = f"""请检查以下回答是否存在"幻觉"（与参考信息不符的内容）。
-
-【用户问题】
-{query}
-
-【参考信息】
-{chr(10).join([f"[{i+1}] {c['doc'][:200]}" for i, c in enumerate(contexts[:8])])}
-
-【AI回答】
-{answer}
-
-【检查规则】
-1. 逐条核对回答中的事实是否能在参考信息中找到依据
-2. 如果没有幻觉，只回复一个英文单词：PASS
-3. 如果有幻觉，只输出修正后的完整回答（不要输出检查过程、不要加标题、不要加"检查结果"等前缀）
-
-修正后的回答："""
-
-        try:
-            result = call_llm(
-                self.client, prompt, MODEL,
-                temperature=0.1,
-                max_tokens=2000
-            )
-            if not result:
-                return answer
-            # 如果返回 PASS 或很短的确认，说明无幻觉
-            cleaned = result.strip()
-            if cleaned.upper() == "PASS" or len(cleaned) < 10:
-                return answer
-            # 有幻觉时，返回修正后的干净答案（去掉可能的前缀）
-            for prefix in ["修正后的回答：", "修正后回答：", "修正回答：", "修正后："]:
-                if cleaned.startswith(prefix):
-                    cleaned = cleaned[len(prefix):].strip()
-            return cleaned
-        except Exception as e:
-            logger.warning(f"答案验证失败: {e}")
-            return answer
-
-    def _generate_uncertain_answer(self, query: str, contexts: list) -> str:
-        """生成不确定性回答"""
-        context_str = "\n".join([c['doc'][:200] for c in contexts[:3]])
-
-        prompt = f"""用户问题：{query}
-
-找到的信息可能不够完整或相关性不高：
-{context_str}
-
-请基于这些信息给出一个谨慎的回答，明确说明哪些部分是有依据的，哪些部分可能需要更多验证。
-
-回答："""
-
-        try:
-            result = call_llm(
-                self.client, prompt, MODEL,
-                temperature=0.7,
-                max_tokens=1000
-            )
-            return result or "根据现有信息无法确定答案。"
-        except Exception as e:
-            logger.error(f"生成不确定性回答失败: {e}")
-            return "根据现有信息无法确定答案。"
-
-    def _direct_answer(self, query: str, history: list = None) -> str:
-        """直接使用 LLM 回答（无知识库检索）"""
-        messages = [
-            {"role": "system", "content": "你是一个专业的助手，请用中文回答用户的问题。"}
-        ]
-
-        if history:
-            for h in history[-4:]:
-                if h.get("role") in ["user", "assistant"]:
-                    messages.append({"role": h["role"], "content": h.get("content", "")})
-
-        messages.append({"role": "user", "content": query})
-
-        try:
-            result = call_llm(
-                self.client, "", MODEL,
-                temperature=0.7,
-                max_tokens=1500,
-                messages=messages
-            )
-            return result or "抱歉，我无法回答这个问题。"
-        except Exception as e:
-            logger.error(f"直接回答失败: {e}")
-            return f"回答生成失败：{str(e)}"

core/agentic_base.py

@@ -1,62 +0,0 @@
-"""
-Agentic RAG - 基础模块
-
-包含常量、导入和共享配置
-"""
-
-import logging
-
-# 配置日志
-logger = logging.getLogger(__name__)
-
-# 尝试导入搜索API配置
-try:
-    from config import SERPER_API_KEY
-    HAS_SERPER = True
-except ImportError:
-    HAS_SERPER = False
-    SERPER_API_KEY = None
-
-# LLM 预算控制
-try:
-    from core.llm_budget import get_budget_controller, should_use_agent, CallType
-    HAS_BUDGET = True
-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
-except ImportError:
-    API_KEY = None
-    BASE_URL = None
-    MODEL = None
-
-# 来源标记
-SOURCE_KB = "知识库"
-SOURCE_WEB = "网络搜索"
-
-# Context Compression 配置
-MAX_CONTEXT_TOKENS = 8000  # 最大上下文 token 数（与生产路径对齐）
-MAX_CONTEXT_COUNT = 20     # 最大上下文数量
-RERANK_THRESHOLD = 0.3    # Rerank 过滤阈值
-
-# Answer Grounding 配置
-MAX_GROUNDING_RETRY = 1   # 幻觉修正最多重试次数

core/agentic_citation.py

@@ -1,237 +0,0 @@
-"""
-Agentic RAG - 引用处理 Mixin
-
-包含来源提取、引用构建、引用附加等方法
-"""
-
-import json
-import logging
-
-from .agentic_base import logger, SOURCE_KB
-
-logger = logging.getLogger(__name__)
-
-
-class CitationMixin:
-    """引用处理方法"""
-
-    def _extract_sources(self, contexts: list) -> list:
-        """提取来源列表，返回结构化定位信息"""
-        source_map = {}
-
-        for c in contexts:
-            meta = c.get('meta', {})
-            source_type = c.get('source_type', '未知')
-
-            if source_type == self.SOURCE_KB:
-                source_key = meta.get('source', '未知')
-                page = meta.get('page')
-                page_end = meta.get('page_end', page)
-                section = meta.get('section', '')
-                doc_type = meta.get('doc_type', 'other')
-                preview = meta.get('preview', '')
-                section_chunk_id = meta.get('section_chunk_id')
-            else:
-                source_key = meta.get('title', meta.get('source', '未知'))
-                page = None
-                page_end = None
-                section = ''
-                doc_type = 'other'
-                preview = ''
-                section_chunk_id = None
-
-            if source_key not in source_map:
-                source_map[source_key] = {
-                    "source": source_key,
-                    "type": source_type,
-                    "doc_type": doc_type,
-                    "count": 0,
-                    "pages": [],
-                    "sections": set(),
-                    "previews": [],
-                    "section_chunk_ids": set()
-                }
-
-            source_map[source_key]["count"] += 1
-
-            if page:
-                page_range = (page, page_end if page_end else page)
-                if page_range not in source_map[source_key]["pages"]:
-                    source_map[source_key]["pages"].append(page_range)
-
-            if section:
-                source_map[source_key]["sections"].add(section)
-
-            if preview and len(source_map[source_key]["previews"]) < 3:
-                if preview not in source_map[source_key]["previews"]:
-                    source_map[source_key]["previews"].append(preview)
-
-            if section_chunk_id:
-                source_map[source_key]["section_chunk_ids"].add(section_chunk_id)
-
-        sources = []
-        for key, info in source_map.items():
-            source_str = info["source"]
-            doc_type = info.get("doc_type", "other")
-            location_parts = []
-
-            if doc_type == 'pdf':
-                if info["pages"]:
-                    valid_pages = [(s, e) for s, e in info["pages"] if s > 1 or e > 1]
-                    if valid_pages or not info["sections"]:
-                        page_strs = []
-                        for start, end in sorted(info["pages"], key=lambda x: x[0]):
-                            if start == end:
-                                page_strs.append(f"第{start}页")
-                            else:
-                                page_strs.append(f"第{start}-{end}页")
-                        location_parts.append(", ".join(page_strs))
-
-                if info["sections"]:
-                    sections_list = sorted(info["sections"])[:3]
-                    sections_str = "、".join(sections_list)
-                    if len(info["sections"]) > 3:
-                        sections_str += f"等{len(info['sections'])}个章节"
-                    location_parts.append(sections_str)
-
-            elif doc_type == 'word':
-                if info["sections"]:
-                    sections_list = sorted(info["sections"])[:3]
-                    sections_str = "、".join(sections_list)
-                    if len(info["sections"]) > 3:
-                        sections_str += f"等{len(info['sections'])}个章节"
-                    location_parts.append(sections_str)
-
-                if info.get("section_chunk_ids"):
-                    chunk_ids = sorted(info["section_chunk_ids"])[:5]
-                    if chunk_ids:
-                        chunk_str = f"第{chunk_ids[0]}"
-                        if len(chunk_ids) > 1:
-                            chunk_str = f"第{chunk_ids[0]}-{chunk_ids[-1]}段"
-                        location_parts.append(chunk_str)
-
-            elif doc_type == 'excel':
-                if info["sections"]:
-                    sections_list = sorted(info["sections"])[:3]
-                    sections_str = "、".join(sections_list)
-                    location_parts.append(sections_str)
-
-            else:
-                if info["pages"]:
-                    valid_pages = [(s, e) for s, e in info["pages"] if s > 1 or e > 1]
-                    if valid_pages or not info["sections"]:
-                        page_strs = []
-                        for start, end in sorted(info["pages"], key=lambda x: x[0]):
-                            if start == end:
-                                page_strs.append(f"第{start}页")
-                            else:
-                                page_strs.append(f"第{start}-{end}页")
-                        location_parts.append(", ".join(page_strs))
-
-                if info["sections"]:
-                    sections_list = sorted(info["sections"])[:3]
-                    sections_str = "、".join(sections_list)
-                    if len(info["sections"]) > 3:
-                        sections_str += f"等{len(info['sections'])}个章节"
-                    location_parts.append(sections_str)
-
-            if location_parts:
-                source_str = f"{source_str} ({' | '.join(location_parts)})"
-
-            sources.append({
-                "source": source_str,
-                "type": info["type"],
-                "count": info["count"],
-                "doc_type": doc_type,
-                "previews": info.get("previews", []),
-                "section_chunk_ids": sorted(info.get("section_chunk_ids", []))[:5]
-            })
-
-        return sources
-
-    def _build_citation(self, meta: dict) -> dict:
-        """根据文档类型构建定位信息"""
-        # 从 chunk_id 中提取全局切片序号（格式: "filename_N"）
-        chunk_id_raw = meta.get('chunk_id', '')
-        chunk_index = None
-        if chunk_id_raw and '_' in str(chunk_id_raw):
-            try:
-                chunk_index = int(str(chunk_id_raw).rsplit('_', 1)[-1])
-            except (ValueError, IndexError):
-                chunk_index = meta.get('chunk_index')
-        else:
-            chunk_index = meta.get('chunk_index')
-
-        citation = {
-            "chunk_id": chunk_id_raw,
-            "chunk_index": chunk_index,  # 全局切片序号，用于前端文档预览跳转
-            "source": meta.get('source', ''),
-            "collection": meta.get('_collection', ''),  # 所属向量库，用于前端文档预览跳转
-            "doc_type": meta.get('doc_type', 'other'),
-            "section": meta.get('section', ''),
-            "preview": meta.get('preview', ''),
-            "content": meta.get('preview', ''),  # 初始用 preview，_attach_citations 中会用完整内容覆盖
-            "chunk_type": meta.get('chunk_type', 'text'),
-        }
-
-        doc_type = meta.get('doc_type', 'other')
-
-        if doc_type == 'pdf':
-            bbox_raw = meta.get('bbox')
-            bbox = None
-            if bbox_raw:
-                try:
-                    bbox = json.loads(bbox_raw) if isinstance(bbox_raw, str) else bbox_raw
-                except (json.JSONDecodeError, TypeError):
-                    bbox = bbox_raw
-
-            citation.update({
-                "page": meta.get('page'),
-                "page_end": meta.get('page_end'),
-                "bbox": bbox,
-                "bbox_mode": meta.get('bbox_mode'),
-            })
-        elif doc_type == 'word':
-            citation.update({
-                "section_chunk_id": meta.get('section_chunk_id'),
-            })
-        elif doc_type == 'excel':
-            citation.update({
-                "page": meta.get('page'),
-            })
-        else:
-            bbox_raw = meta.get('bbox')
-            bbox = None
-            if bbox_raw:
-                try:
-                    bbox = json.loads(bbox_raw) if isinstance(bbox_raw, str) else bbox_raw
-                except (json.JSONDecodeError, TypeError):
-                    bbox = bbox_raw
-
-            citation.update({
-                "page": meta.get('page'),
-                "page_end": meta.get('page_end'),
-                "bbox": bbox,
-                "bbox_mode": meta.get('bbox_mode'),
-            })
-
-        return citation
-
-    def _attach_citations(self, answer: str, contexts: list) -> dict:
-        """将引用信息附加到答案"""
-        citations = []
-
-        for c in contexts:
-            meta = c.get('meta', {})
-            full_content = c.get('doc', '')
-            citation = self._build_citation(meta)
-            # 用上下文中的完整文档内容覆盖 content 字段
-            if full_content:
-                citation['content'] = full_content[:300]
-            citations.append(citation)
-
-        return {
-            "answer": answer,
-            "citations": citations,
-            "sources": self._extract_sources(contexts)
-        }

core/agentic_context.py

@@ -1,111 +0,0 @@
-"""
-Agentic RAG - 上下文处理 Mixin
-
-包含上下文压缩、去重、Token 控制等方法
-"""
-
-import logging
-
-from .agentic_base import logger, MAX_CONTEXT_TOKENS, RERANK_THRESHOLD
-
-logger = logging.getLogger(__name__)
-
-
-class ContextMixin:
-    """上下文处理方法"""
-
-    def _compress_contexts(self, query: str, contexts: list) -> list:
-        """上下文压缩三步走：Rerank 过滤 → 去重 → Token 控制"""
-        if not contexts:
-            return contexts
-
-        # Step 1: Rerank 过滤
-        filtered = self._rerank_filter(contexts)
-
-        # Step 2: 去重
-        deduped = self._deduplicate_contexts(filtered)
-
-        # Step 3: Token 控制
-        result = self._truncate_to_tokens(deduped, self.MAX_CONTEXT_TOKENS)
-
-        return result
-
-    def _rerank_filter(self, contexts: list) -> list:
-        """Rerank 过滤 - 保留相关性分数 >= 阈值的上下文"""
-        scored_contexts = [c for c in contexts if c.get('score') is not None]
-
-        if scored_contexts:
-            filtered = [c for c in contexts if c.get('score', 0) >= self.RERANK_THRESHOLD]
-            return filtered if filtered else contexts
-
-        return contexts
-
-    def _deduplicate_contexts(self, contexts: list, threshold: float = 0.9) -> list:
-        """去重 - 基于内容相似度去重"""
-        if len(contexts) <= 1:
-            return contexts
-
-        result = []
-        seen_keys = set()
-
-        for c in contexts:
-            doc = c.get('doc', '')
-            key = doc[:100] if doc else ''
-
-            meta = c.get('meta', {})
-            source = meta.get('source', '')
-            page = meta.get('page', '')
-
-            composite_key = f"{source}|{page}|{key}"
-
-            if composite_key not in seen_keys:
-                seen_keys.add(composite_key)
-                result.append(c)
-
-        return result
-
-    def _truncate_to_tokens(self, contexts: list, max_tokens: int) -> list:
-        """Token 控制 - 截断到最大 Token 数"""
-        result = []
-        total_tokens = 0
-
-        for c in contexts:
-            doc = c.get('doc', '')
-            # 简单估算：1 token ≈ 1.5 中文字符
-            tokens = len(doc) // 1.5
-
-            if total_tokens + tokens <= max_tokens:
-                result.append(c)
-                total_tokens += tokens
-            else:
-                break
-
-        return result
-
-    def _merge_and_deduplicate(self, old_contexts: list, new_contexts: list) -> list:
-        """合并并去重两个上下文列表"""
-        result = list(old_contexts)
-        seen_keys = set()
-
-        # 记录已有上下文的 key
-        for c in old_contexts:
-            doc = c.get('doc', '')
-            key = doc[:100] if doc else ''
-            meta = c.get('meta', {})
-            source = meta.get('source', '')
-            composite_key = f"{source}|{key}"
-            seen_keys.add(composite_key)
-
-        # 添加新上下文（去重）
-        for c in new_contexts:
-            doc = c.get('doc', '')
-            key = doc[:100] if doc else ''
-            meta = c.get('meta', {})
-            source = meta.get('source', '')
-            composite_key = f"{source}|{key}"
-
-            if composite_key not in seen_keys:
-                seen_keys.add(composite_key)
-                result.append(c)
-
-        return result

core/agentic_media.py

@@ -1,200 +0,0 @@
-"""
-Agentic RAG - 富媒体处理 Mixin
-
-包含图表查找、图片提取、富媒体附加等方法
-"""
-
-import re
-import json
-import logging
-
-from .agentic_base import logger
-
-logger = logging.getLogger(__name__)
-
-
-class RichMediaMixin:
-    """富媒体处理方法"""
-
-    def _find_figure(self, query: str, contexts: list, source: str = None) -> dict:
-        """精确查找图表，带 fallback"""
-        patterns = [
-            r'图\s*(\d+[\.\-]\d+)',
-            r'Fig\.?\s*(\d+[\.\-]\d+)',
-            r'Figure\s*(\d+[\.\-]\d+)',
-        ]
-
-        target_figure = None
-        for pattern in patterns:
-            match = re.search(pattern, query, re.IGNORECASE)
-            if match:
-                target_figure = match.group(1).replace('-', '.')
-                break
-
-        if not target_figure:
-            return {"found": False}
-
-        # 从 contexts 中查找
-        for ctx in contexts:
-            meta = ctx.get('meta', {})
-            fig_num = meta.get('figure_number', '')
-            if fig_num == target_figure:
-                if not source or meta.get('source') == source:
-                    return {
-                        "found": True,
-                        "chunk_id": meta.get('chunk_id'),
-                        "source": meta.get('source'),
-                        "page": meta.get('page'),
-                        "caption": meta.get('caption'),
-                        "image_path": meta.get('image_path'),
-                    }
-
-        # Fallback: 直接查向量库
-        try:
-            from knowledge.manager import get_kb_manager
-            kb_mgr = get_kb_manager()
-            coll = kb_mgr.get_collection('public_kb')
-
-            if coll:
-                where_conditions = [{'chunk_type': {'$in': ['image', 'chart']}}]
-                if source:
-                    where_conditions.append({'source': source})
-
-                result = coll.get(
-                    where={'$and': where_conditions} if len(where_conditions) > 1 else where_conditions[0],
-                    include=['metadatas', 'documents']
-                )
-
-                for meta, doc in zip(result.get('metadatas', []), result.get('documents', [])):
-                    if meta.get('figure_number') == target_figure:
-                        return {
-                            "found": True,
-                            "chunk_id": meta.get('chunk_id'),
-                            "source": meta.get('source'),
-                            "page": meta.get('page'),
-                            "caption": meta.get('caption'),
-                            "image_path": meta.get('image_path'),
-                        }
-                    caption = meta.get('caption', '') or (doc if doc else '')
-                    if f"图{target_figure}" in caption or f"图 {target_figure}" in caption:
-                        return {
-                            "found": True,
-                            "chunk_id": meta.get('chunk_id'),
-                            "source": meta.get('source'),
-                            "page": meta.get('page'),
-                            "caption": meta.get('caption'),
-                            "image_path": meta.get('image_path'),
-                        }
-        except Exception as e:
-            logger.warning(f"_find_figure fallback 查询失败: {e}")
-
-        return {"found": False}
-
-    def _get_images_for_source(self, source: str, collections: list = None) -> list:
-        """直接从向量库获取指定文件的所有图片"""
-        try:
-            from knowledge.manager import get_kb_manager
-            kb_mgr = get_kb_manager()
-        except ImportError:
-            return []
-
-        images = []
-        seen_ids = set()
-
-        target_collections = collections or ['public_kb']
-
-        for kb_name in target_collections:
-            try:
-                coll = kb_mgr.get_collection(kb_name)
-                if not coll:
-                    continue
-
-                result = coll.get(
-                    where={'source': source},
-                    include=['metadatas']
-                )
-
-                for meta in result.get('metadatas', []):
-                    images_json = meta.get('images_json')
-                    if images_json:
-                        try:
-                            imgs = json.loads(images_json)
-                            for img in imgs:
-                                img_id = img.get('id')
-                                if img_id and img_id not in seen_ids:
-                                    seen_ids.add(img_id)
-                                    images.append({
-                                        "id": img_id,
-                                        "caption": img.get("caption", ""),
-                                        "url": f"/images/{img_id}",
-                                        "page": img.get("page") or meta.get("page"),
-                                        "source": source,
-                                        "width": img.get("width"),
-                                        "height": img.get("height")
-                                    })
-                        except (json.JSONDecodeError, TypeError):
-                            pass
-            except Exception as e:
-                logger.warning(f"从 {kb_name} 获取图片失败: {e}")
-                continue
-
-        return images
-
-    def _extract_rich_media(self, contexts: list, sources_filter: list = None, max_images: int = 10,
-                           max_tables: int = 5) -> dict:
-        """从检索结果中提取富媒体（图片、表格）"""
-        images = []
-        tables = []
-        seen_image_ids = set()
-        seen_table_ids = set()
-
-        for ctx in contexts:
-            meta = ctx.get('meta', {})
-            source = meta.get('source', '')
-
-            # 过滤来源
-            if sources_filter and source not in sources_filter:
-                continue
-
-            # 提取图片
-            images_json = meta.get('images_json')
-            if images_json:
-                try:
-                    imgs = json.loads(images_json)
-                    for img in imgs:
-                        img_id = img.get('id')
-                        if img_id and img_id not in seen_image_ids:
-                            seen_image_ids.add(img_id)
-                            images.append({
-                                "id": img_id,
-                                "caption": img.get("caption", ""),
-                                "url": f"/images/{img_id}",
-                                "page": img.get("page") or meta.get("page"),
-                                "source": source,
-                                "type": img.get("type", "image")
-                            })
-                except (json.JSONDecodeError, TypeError):
-                    pass
-
-            # 提取表格
-            table_json = meta.get('table_json')
-            if table_json:
-                try:
-                    tbl = json.loads(table_json)
-                    tbl_id = tbl.get('id') or meta.get('chunk_id')
-                    if tbl_id and tbl_id not in seen_table_ids:
-                        seen_table_ids.add(tbl_id)
-                        tables.append({
-                            "id": tbl_id,
-                            "caption": tbl.get("caption", ""),
-                            "markdown": tbl.get("markdown", ""),
-                            "page": meta.get("page"),
-                            "source": source
-                        })
-                except (json.JSONDecodeError, TypeError):
-                    pass
-
-        return {
-            "images": images[:max_images],
-            "tables": tables[:max_tables]
-        }

core/agentic_meta.py

@@ -1,133 +0,0 @@
-"""
-Agentic RAG - 元问题处理 Mixin
-
-包含元问题判断和知识库元数据回答方法
-"""
-
-import logging
-
-from .agentic_base import logger
-
-logger = logging.getLogger(__name__)
-
-
-class MetaQuestionMixin:
-    """元问题处理方法"""
-
-    def _is_meta_question(self, query: str) -> bool:
-        """判断是否为元问题（关于知识库本身的问题）"""
-        meta_patterns = [
-            "有哪些文件", "什么文件", "哪些文件", "文件列表", "文件目录",
-            "可以查看", "能查看", "有权限查看", "权限查看",
-            "能访问", "可以访问", "有权限访问",
-            "我的权限", "用户权限", "查看权限", "访问权限",
-            "权限能", "权限可以", "有什么权限", "有哪些权限",
-            "我能看", "我可以看", "我能查", "我可以查",
-            "能看到什么", "能查到什么", "可以看什么", "可以查什么",
-            "知识库有哪些", "库里有", "文档有哪些", "有哪些文档",
-            "有什么文档", "有什么文件", "包含什么", "包含哪些",
-            "你知道什么", "你都知道", "你能回答什么",
-            "系统里有什么", "库里有什么",
-            "public_kb", "dept_tech", "dept_hr", "dept_finance", "dept_operation",
-            "kb里", "向量库", "有哪些库", "库列表", "kb有哪些"
-        ]
-        query_lower = query.lower()
-        return any(kw in query_lower for kw in meta_patterns)
-
-    def _answer_meta_question(self, query: str, allowed_levels: list = None,
-                              role: str = None, department: str = None) -> str:
-        """回答元问题（关于知识库本身的问题）"""
-        try:
-            source_map = {}
-
-            try:
-                from knowledge.manager import get_kb_manager
-                from auth.gateway import get_accessible_collections as _get_accessible
-
-                kb_mgr = get_kb_manager()
-                accessible = _get_accessible(role or 'user', department or '', 'read')
-
-                for kb_name in accessible:
-                    coll = kb_mgr.get_collection(kb_name)
-                    if not coll:
-                        continue
-                    try:
-                        result = coll.get(include=['metadatas'])
-                    except Exception as e:
-                        logger.debug(f"获取{kb_name}元数据失败: {e}")
-                        continue
-
-                    for meta in result.get('metadatas', []):
-                        source = meta.get('source', '未知')
-                        level = meta.get('security_level', 'public')
-                        page = meta.get('page')
-
-                        if source not in source_map:
-                            source_map[source] = {
-                                'count': 0, 'levels': set(),
-                                'pages': set(), 'collections': set()
-                            }
-
-                        source_map[source]['count'] += 1
-                        source_map[source]['levels'].add(level)
-                        source_map[source]['collections'].add(kb_name)
-                        if page:
-                            source_map[source]['pages'].add(page)
-
-            except ImportError:
-                from core.engine import get_engine
-                all_docs = get_engine().collection.get(include=['metadatas'])
-                for meta in all_docs.get('metadatas', []):
-                    source = meta.get('source', '未知')
-                    level = meta.get('security_level', 'public')
-                    page = meta.get('page')
-
-                    if source not in source_map:
-                        source_map[source] = {
-                            'count': 0, 'levels': set(),
-                            'pages': set(), 'collections': set()
-                        }
-
-                    source_map[source]['count'] += 1
-                    source_map[source]['levels'].add(level)
-                    if page:
-                        source_map[source]['pages'].add(page)
-
-            # 根据安全级别过滤
-            if allowed_levels:
-                allowed_set = set(allowed_levels)
-                filtered_sources = {}
-                for source, info in source_map.items():
-                    if info['levels'] & allowed_set:
-                        filtered_sources[source] = info
-                source_map = filtered_sources
-
-            if not source_map:
-                return "抱歉，您当前没有权限查看任何文档，或者知识库为空。"
-
-            sorted_sources = sorted(source_map.items(), key=lambda x: x[1]['count'], reverse=True)
-
-            answer_parts = [f"📚 **知识库文档列表**（共 {len(sorted_sources)} 个文档）\n"]
-
-            for i, (source, info) in enumerate(sorted_sources, 1):
-                colls = info.get('collections', set())
-                coll_str = f"，所属: {', '.join(sorted(colls))}" if colls else ""
-                pages_str = ''
-                if info['pages']:
-                    pages_list = sorted(info['pages'])
-                    if len(pages_list) <= 5:
-                        pages_str = f"，页码: {', '.join(map(str, pages_list))}"
-                    else:
-                        pages_str = f"，共 {len(info['pages'])} 页"
-
-                answer_parts.append(f"{i}. **{source}** ({info['count']} 条片段{coll_str}{pages_str})")
-
-            answer_parts.append(f"\n**总计**: {sum(s[1]['count'] for s in sorted_sources)} 条知识片段")
-            answer_parts.append(f"\n**您的权限级别**: {', '.join(allowed_levels) if allowed_levels else '全部'}")
-
-            answer_parts.append("\n\n💡 **提示**: 您可以直接提问关于这些文档内容的问题。")
-
-            return '\n'.join(answer_parts)
-
-        except Exception as e:
-            return f"获取文档列表时出错: {str(e)}\n\n您可以直接提问，我会尝试从知识库中检索相关信息。"

core/agentic_quality.py

@@ -1,137 +0,0 @@
-"""
-Agentic RAG - 质量评估 Mixin
-
-包含置信度门控、质量评估、推理反思等方法
-"""
-
-import logging
-
-from .agentic_base import logger
-
-logger = logging.getLogger(__name__)
-
-
-class QualityMixin:
-    """质量评估方法"""
-
-    def _check_confidence_gate(self, query: str, docs: list, verbose: bool = True,
-                               precomputed_scores: list = None):
-        """检查置信度门控
-
-        Args:
-            query: 用户查询
-            docs: 文档列表
-            verbose: 是否详细输出
-            precomputed_scores: 预计算的 Rerank 分数（可选，避免重复推理）
-        """
-        if not self.confidence_gate:
-            return {"passed": True, "reason": "no_gate"}
-
-        try:
-            result = self.confidence_gate.evaluate(query, docs,
-                                                    precomputed_scores=precomputed_scores)
-            return result
-        except Exception as e:
-            logger.warning(f"置信度门控检查失败: {e}")
-            return {"passed": True, "reason": "error"}
-
-    def _assess_quality(self, query: str, docs: list, metas: list = None,
-                       verbose: bool = True) -> dict:
-        """多维质量评估"""
-        if not self.quality_assessor:
-            return {"overall_score": 0.5, "dimensions": {}}
-
-        try:
-            result = self.quality_assessor.assess(query, docs, metas)
-            return result
-        except Exception as e:
-            logger.warning(f"质量评估失败: {e}")
-            return {"overall_score": 0.5, "dimensions": {}}
-
-    def _reflect_on_answer(self, query: str, answer: str, contexts: list,
-                          verbose: bool = True) -> dict:
-        """推理反思"""
-        if not self.reasoning_reflector:
-            return {"needs_reflection": False, "issues": []}
-
-        try:
-            result = self.reasoning_reflector.reflect(query, answer, contexts)
-            return result
-        except Exception as e:
-            logger.warning(f"推理反思失败: {e}")
-            return {"needs_reflection": False, "issues": []}
-
-    def _think(self, original_query: str, current_query: str,
-               iteration: int, contexts: list, verbose: bool = True) -> dict:
-        """
-        Agent 思考：决定下一步行动
-
-        Returns:
-            {
-                "action": "answer" | "rewrite" | "search_web" | "decompose",
-                "reason": "...",
-                "rewrite_query": "..."  # 如果 action == "rewrite"
-            }
-        """
-        from core.llm_utils import call_llm, parse_json_from_response
-        from .agentic_base import MODEL
-
-        # 构建思考提示
-        context_summary = ""
-        if contexts:
-            for i, ctx in enumerate(contexts[:3], 1):
-                meta = ctx.get('meta', {})
-                source = meta.get('source', '未知')
-                doc_preview = ctx.get('doc', '')[:100]
-                context_summary += f"{i}. [{source}] {doc_preview}...\n"
-
-        prompt = f"""你是一个 RAG 系统的决策 Agent，需要判断下一步行动。
-
-【原始问题】
-{original_query}
-
-【当前问题】
-{current_query}
-
-【迭代轮次】
-{iteration} / {self.max_iterations}
-
-【已检索到的上下文】
-{context_summary if context_summary else "（无）"}
-
-【可选行动】
-1. answer - 已有足够信息，可以回答
-2. rewrite - 查询不够清晰，需要重写
-3. search_web - 知识库信息不足，需要网络搜索
-4. decompose - 问题太复杂，需要分解
-
-【决策要求】
-- 如果上下文足够回答问题，选择 answer
-- 如果上下文不足且迭代未超限，选择 search_web 或 rewrite
-- 返回 JSON 格式
-
-请决策："""
-
-        try:
-            result = call_llm(
-                self.client, prompt, MODEL,
-                temperature=0.3,
-                max_tokens=200
-            )
-
-            decision = parse_json_from_response(result) if result else {}
-
-            # 默认决策
-            if not decision or "action" not in decision:
-                if contexts and len(contexts) >= 2:
-                    decision = {"action": "answer", "reason": "有足够上下文"}
-                else:
-                    decision = {"action": "rewrite", "reason": "上下文不足"}
-
-            return decision
-
-        except Exception as e:
-            logger.warning(f"Agent 思考失败: {e}")
-            if contexts:
-                return {"action": "answer", "reason": "默认回答"}
-            return {"action": "rewrite", "reason": "默认重写"}

core/agentic_query.py

@@ -1,271 +0,0 @@
-"""
-Agentic RAG - 查询重写 Mixin
-
-包含查询改写、实体补全、专业术语映射等方法
-"""
-
-import re
-import logging
-
-from .agentic_base import logger, MODEL
-
-logger = logging.getLogger(__name__)
-
-
-class QueryRewriteMixin:
-    """查询重写方法"""
-
-    def _rewrite_query(self, query: str, history: list = None,
-                       strategy: str = "professional") -> str:
-        """
-        增强版查询重写：将口语化表达转为专业术语
-
-        Args:
-            query: 原始查询
-            history: 对话历史（用于实体补全）
-            strategy: 重写策略
-                - professional: 口语化→专业术语
-                - expand: 扩展关键词
-                - clarify: 消歧义
-                - entity: 实体补全
-
-        Returns:
-            str: 重写后的查询
-        """
-        # 尝试多种策略组合
-        rewritten = query
-
-        # 策略1: 口语化→专业术语映射
-        if strategy in ["professional", "all"]:
-            rewritten = self._apply_professional_mapping(rewritten)
-
-        # 策略2: 实体补全（利用对话历史）
-        if strategy in ["entity", "all"] and history:
-            rewritten = self._complete_entities(rewritten, history)
-
-        # 策略3: LLM 深度重写（仅在需要时调用）
-        if strategy in ["professional", "all"]:
-            llm_rewritten = self._llm_rewrite(rewritten)
-            if llm_rewritten and len(llm_rewritten) > len(rewritten) * 0.5:
-                rewritten = llm_rewritten
-
-        return rewritten
-
-    def _apply_professional_mapping(self, query: str) -> str:
-        """应用口语化→专业术语映射"""
-        TERM_MAPPING = {
-            "报销": "差旅报销 费用报销 报销审批",
-            "请假": "休假申请 请假审批 考勤管理",
-            "加班": "加班申请 工时管理 加班审批",
-            "工资": "薪酬管理 工资发放 薪资结构",
-            "合同": "合同管理 合同签署 合同审批",
-            "流程": "审批流程 业务流程 工作流",
-            "制度": "管理制度 规章制度 企业规范",
-            "规定": "管理规定 制度规定 政策要求",
-            "几天": "时限 审批时限 办理时限",
-            "多久": "处理时效 审批周期 办理周期",
-            "多少": "标准 额度 限额 标准",
-            "能不能": "是否允许 是否可以 权限",
-            "人事": "人力资源 HR 人力部门",
-            "财务": "财务部 财务部门 财务管理",
-            "技术": "技术部 研发部 IT部门",
-        }
-
-        result = query
-        for colloquial, professional in TERM_MAPPING.items():
-            if colloquial in query:
-                result = result.replace(colloquial, f"{colloquial} {professional.split()[0]}")
-
-        return result
-
-    def _complete_entities(self, query: str, history: list) -> str:
-        """实体补全：利用对话历史补充缺失的实体"""
-        if not history:
-            return query
-
-        # 图片指代识别
-        image_reference = self._detect_image_reference(query, history)
-        if image_reference:
-            return image_reference
-
-        # 获取最近用户消息
-        last_user_msg = None
-        for msg in reversed(history):
-            if msg.get("role") == "user":
-                last_user_msg = msg.get("content", "")
-                break
-
-        if not last_user_msg:
-            return query
-
-        # 检查当前查询是否缺少主语
-        BUSINESS_KEYWORDS = ["报销", "出差", "请假", "工资", "合同", "审批", "流程",
-                           "制度", "规定", "标准", "金额", "时间"]
-
-        has_subject = any(kw in query for kw in BUSINESS_KEYWORDS)
-
-        if not has_subject:
-            try:
-                import jieba
-                entities = []
-                for word in jieba.cut(last_user_msg):
-                    word = word.strip()
-                    if len(word) >= 2 and any(kw in word for kw in BUSINESS_KEYWORDS):
-                        entities.append(word)
-
-                if entities:
-                    return f"{entities[0]} {query}"
-            except ImportError:
-                pass
-
-        return query
-
-    def _detect_image_reference(self, query: str, history: list) -> str:
-        """检测图片指代查询并重写"""
-        IMAGE_REFERENCE_PATTERNS = [
-            r'这[张些]图片', r'那[张些]图片', r'上面的图片', r'刚才的图片',
-            r'这[张些]图', r'那[张些]图', r'上面的图', r'刚才的图',
-            r'解释一下这[张些]图', r'说明一下这[张些]图',
-            r'这[张些]是什么图', r'图[里内]是什么', r'图片[里内]是什么',
-        ]
-
-        is_image_reference = False
-        for pattern in IMAGE_REFERENCE_PATTERNS:
-            if re.search(pattern, query):
-                is_image_reference = True
-                break
-
-        if not is_image_reference:
-            return ""
-
-        last_images = []
-        for msg in reversed(history):
-            if msg.get("role") == "assistant":
-                metadata = msg.get("metadata", {})
-                if isinstance(metadata, dict):
-                    images = metadata.get("images", [])
-                    if images:
-                        for img in images[:5]:
-                            if isinstance(img, dict):
-                                desc = img.get("description", "")
-                                img_type = img.get("type", "图片")
-                                if desc:
-                                    last_images.append(f"{img_type}：{desc}")
-                            elif isinstance(img, str):
-                                last_images.append(f"图片：{img}")
-
-                if not last_images:
-                    content = msg.get("content", "")
-                    if "图片" in content or "图表" in content or "图" in content:
-                        sentences = content.split("。")
-                        for sentence in sentences:
-                            if "图片" in sentence or "图表" in sentence:
-                                last_images.append(sentence.strip())
-
-                if last_images:
-                    break
-
-        if last_images:
-            image_context = " ".join(last_images[:3])
-            question_intent = re.sub(
-                r'这[张些]图片?|那[张些]图片?|上面的图片?|刚才的图片?|解释一下|说明一下',
-                '', query
-            ).strip()
-
-            if question_intent:
-                return f"{image_context} {question_intent}"
-            else:
-                return f"详细解释：{image_context}"
-
-        return query
-
-    def _extract_image_context_from_history(self, history: list) -> str:
-        """从对话历史中提取图片上下文"""
-        if not history:
-            return ""
-
-        for msg in reversed(history):
-            if msg.get("role") == "assistant":
-                metadata = msg.get("metadata", {})
-                images = metadata.get("images", [])
-                content = msg.get("content", "")
-
-                image_descriptions = []
-
-                if images:
-                    for i, img in enumerate(images[:5], 1):
-                        if isinstance(img, dict):
-                            desc = img.get("description", "")
-                            img_type = img.get("type", "图片")
-                            source = img.get("source", "")
-                            page = img.get("page", "")
-
-                            img_info = f"图片{i}：{img_type}"
-                            if desc:
-                                img_info += f"，描述：{desc}"
-                            if source:
-                                img_info += f"，来源：{source}"
-                            if page:
-                                img_info += f"，第{page}页"
-                            image_descriptions.append(img_info)
-
-                if not image_descriptions:
-                    if "图片" in content or "图表" in content:
-                        sentences = content.split("。")
-                        for sentence in sentences:
-                            if "图片" in sentence or "图表" in sentence:
-                                image_descriptions.append(sentence.strip())
-                                if len(image_descriptions) >= 3:
-                                    break
-
-                if image_descriptions:
-                    return "\n".join(image_descriptions)
-
-        return ""
-
-    def _answer_image_reference(self, enhanced_query: str, history: list) -> str:
-        """回答图片引用问题"""
-        from core.llm_utils import call_llm
-
-        messages = [
-            {"role": "system", "content": "你是一个专业的助手，请根据提供的图片信息回答用户的问题。"}
-        ]
-
-        for h in history[-4:]:
-            if h.get("role") in ["user", "assistant"]:
-                messages.append({"role": h["role"], "content": h.get("content", "")})
-
-        messages.append({"role": "user", "content": enhanced_query})
-
-        try:
-            result = call_llm(
-                self.client, "", MODEL,
-                temperature=0.3,
-                max_tokens=1000,
-                messages=messages
-            )
-            return result or ""
-        except Exception as e:
-            logger.error(f"图片引用回答失败: {e}")
-            return f"抱歉，回答图片问题时出现错误：{str(e)}"
-
-    def _llm_rewrite(self, query: str) -> str:
-        """LLM 深度重写查询"""
-        from core.llm_utils import call_llm
-
-        prompt = f"""请将以下用户问题改写为更专业、更清晰的表达，保持原意不变。
-
-原问题：{query}
-
-改写后的问题："""
-
-        try:
-            rewritten = call_llm(
-                self.client, prompt, MODEL,
-                temperature=0.3,
-                max_tokens=100
-            )
-            return rewritten.strip() if rewritten else query
-        except Exception as e:
-            logger.warning(f"LLM 重写失败: {e}")
-            return query

core/agentic_search.py

@@ -1,152 +0,0 @@
-"""
-Agentic RAG - 检索 Mixin
-
-包含知识库检索、网络搜索等方法
-"""
-
-import json
-import logging
-import requests
-
-from .agentic_base import (
-    logger, HAS_SERPER, SERPER_API_KEY,
-    SOURCE_KB, SOURCE_WEB
-)
-
-logger = logging.getLogger(__name__)
-
-
-class SearchMixin:
-    """检索功能方法"""
-
-    def _web_search(self, query: str, top_k: int = 5) -> list:
-        """网络搜索（使用Serper API）"""
-        if not HAS_SERPER:
-            return []
-
-        try:
-            url = "https://google.serper.dev/search"
-            payload = json.dumps({
-                "q": query,
-                "gl": "cn",
-                "hl": "zh-cn",
-                "num": top_k
-            })
-            headers = {
-                'X-API-KEY': SERPER_API_KEY,
-                'Content-Type': 'application/json'
-            }
-
-            response = requests.post(url, headers=headers, data=payload, timeout=10)
-            response.raise_for_status()
-            data = response.json()
-
-            results = []
-            for item in data.get('organic', [])[:top_k]:
-                results.append({
-                    'title': item.get('title', ''),
-                    'link': item.get('link', ''),
-                    'snippet': item.get('snippet', ''),
-                    'date': item.get('date', '')
-                })
-
-            return results
-
-        except Exception as e:
-            logger.warning(f"网络搜索失败: {e}")
-            return []
-
-    def _should_web_search(self, query: str) -> bool:
-        """判断是否需要网络搜索"""
-        realtime_keywords = [
-            "今天", "最新", "今日", "当前", "现在",
-            "天气", "新闻", "股价", "行情", "汇率",
-            "最近", "近期", "这周", "本月", "今年",
-            "实时", "动态", "热点", "发生"
-        ]
-
-        query_lower = query.lower()
-        return any(kw in query_lower for kw in realtime_keywords)
-
-    def _web_search_flow(self, query: str, log_trace: list, emit_log, verbose: bool,
-                         allowed_levels: list = None) -> list:
-        """
-        网络搜索流程
-
-        Args:
-            query: 查询
-            log_trace: 日志追踪列表
-            emit_log: 日志发射函数
-            verbose: 是否详细输出
-            allowed_levels: 允许的安全级别
-
-        Returns:
-            网络搜索结果列表
-        """
-        if not self.enable_web_search or not HAS_SERPER:
-            return []
-
-        if emit_log:
-            emit_log("🌐 触发网络搜索...")
-
-        web_results = self._web_search(query, top_k=5)
-
-        if not web_results:
-            if emit_log:
-                emit_log("⚠️ 网络搜索未返回结果")
-            return []
-
-        # 转换为统一上下文格式
-        web_contexts = []
-        for item in web_results:
-            web_contexts.append({
-                'doc': f"{item.get('title', '')}\n{item.get('snippet', '')}",
-                'meta': {
-                    'source': self.SOURCE_WEB,
-                    'link': item.get('link', ''),
-                    'date': item.get('date', '')
-                },
-                'source_type': self.SOURCE_WEB,
-                'query': query
-            })
-
-        log_trace.append({
-            'phase': 'web_search',
-            'query': query,
-            'results_count': len(web_contexts)
-        })
-
-        if emit_log:
-            emit_log(f"✅ 网络搜索返回 {len(web_contexts)} 条结果")
-
-        return web_contexts
-
-    def _is_kb_result_sufficient(self, query: str, docs: list) -> bool:
-        """判断知识库检索结果是否充分"""
-        if not docs:
-            return False
-
-        # 结果数量检查
-        if len(docs) >= 3:
-            # 至少3条结果，检查相关性
-            high_rel_count = 0
-            for doc in docs:
-                score = doc.get('score', 0) or doc.get('distance', 1)
-                # cosine 距离转相似度
-                if isinstance(score, (int, float)):
-                    sim = 1 - score if score <= 1 else score
-                    if sim >= 0.6:
-                        high_rel_count += 1
-
-            if high_rel_count >= 2:
-                return True
-
-        # 有高质量结果
-        for doc in docs[:2]:
-            score = doc.get('score', 0) or doc.get('distance', 1)
-            if isinstance(score, (int, float)):
-                sim = 1 - score if score <= 1 else score
-                if sim >= 0.8:
-                    return True
-
-        return False

core/cache.py

@@ -224,39 +224,36 @@ class RAGCacheManager:
         """
         获取查询缓存结果
 
+        始终使用粗粒度 key（基于 kb_version），确保 GET/SET key 一致。
+
         Args:
             query: 查询文本
             kb_name: 知识库名称
-            doc_ids: 相关文档 ID 列表（用于细粒度缓存 key）
+            doc_ids: 保留参数以兼容调用方签名（当前未使用）
         """
         kb_version = self.get_kb_version(kb_name)
 
-        # 计算文档哈希（如果提供了 doc_ids）
+        # 使用粗粒度 key，与 SET 保持一致
-        doc_hash = ""
+        key = self._make_query_cache_key(query, kb_name, kb_version)
-        if doc_ids:
-            doc_hash = self._compute_doc_hash(kb_name, doc_ids)
-
-        key = self._make_query_cache_key(query, kb_name, kb_version, doc_hash)
         return self.query_cache.get(key)
 
     def set_query_result(self, query: str, kb_name: str, result: Dict, doc_ids: List[str] = None) -> None:
         """
         设置查询缓存结果
 
+        始终使用粗粒度 key（基于 kb_version），确保 GET/SET key 一致。
+        kb_version 在文档变更时自增，触发整个知识库的缓存失效。
+
         Args:
             query: 查询文本
             kb_name: 知识库名称
             result: 缓存结果
-            doc_ids: 相关文档 ID 列表（用于细粒度失效）
+            doc_ids: 保留参数以兼容调用方签名（当前未使用）
         """
         kb_version = self.get_kb_version(kb_name)
 
-        # 计算相关文档的版本哈希（细粒度失效）
+        # 使用与 GET 相同的粗粒度 key，确保缓存可命中
-        doc_hash = ""
+        key = self._make_query_cache_key(query, kb_name, kb_version)
-        if doc_ids:
-            doc_hash = self._compute_doc_hash(kb_name, doc_ids)
-
-        key = self._make_query_cache_key(query, kb_name, kb_version, doc_hash)
         self.query_cache.set(key, result, kb_version=kb_version)
 
     def _compute_doc_hash(self, kb_name: str, doc_ids: List[str]) -> str:

core/engine.py

@@ -625,7 +625,7 @@ class RAGEngine:
         elif len(conditions) > 1:
             where_filter = {"$and": conditions}
 
-        query_vector = self.embedding_model.encode(query).tolist()
+        query_vector = self._encode_cached(query).tolist()
         recall_k = RERANK_CANDIDATES if (USE_RERANK or USE_HYBRID_SEARCH) else top_k
         recall_k = max(recall_k, top_k * RECALL_MULTIPLIER)
 
@@ -811,6 +811,76 @@ class RAGEngine:
             logger.warning(f"FAQ 集合查询失败: {e}")
             return get_empty_result()
 
+    def _encode_cached(self, text):
+        """
+        带缓存的 embedding 编码
+
+        优先从 Embedding Cache（LRU）读取，未命中再调用模型编码并写入缓存。
+        支持单文本和批量文本输入。
+
+        Args:
+            text: 单个文本字符串 或 文本列表
+
+        Returns:
+            numpy 数组（单文本为一维，批量为二维）
+        """
+        import numpy as _np
+
+        # 检查 embedding 缓存是否启用（缓存配置查询结果，避免每次重复导入）
+        if not hasattr(self, '_emb_cache_enabled'):
+            self._emb_cache_enabled = True  # 默认启用
+            if CACHE_AVAILABLE:
+                try:
+                    from config import EMBEDDING_CACHE_ENABLED
+                    self._emb_cache_enabled = EMBEDDING_CACHE_ENABLED
+                except ImportError:
+                    pass
+
+        if not self._emb_cache_enabled:
+            return self.embedding_model.encode(text)
+
+        try:
+            _cache = get_cache_manager()
+        except Exception:
+            return self.embedding_model.encode(text)
+
+        # 批量输入
+        if isinstance(text, list):
+            try:
+                cached_embs, missed_indices = _cache.get_embeddings_batch(text)
+                if missed_indices:
+                    missed_texts = [text[i] for i in missed_indices]
+                    # encode(list) 始终返回 2D ndarray，直接按行索引即可
+                    new_embs = self.embedding_model.encode(missed_texts)
+                    if len(missed_indices) == 1:
+                        # 单条时 encode 可能返回 1D，需统一处理
+                        if new_embs.ndim == 1:
+                            new_embs = new_embs.reshape(1, -1)
+                    for idx, mi in enumerate(missed_indices):
+                        emb_list = new_embs[idx].tolist()
+                        cached_embs[mi] = emb_list
+                        try:
+                            _cache.set_embedding(text[mi], emb_list)
+                        except Exception:
+                            pass
+                return _np.array(cached_embs)
+            except Exception:
+                # 缓存故障时优雅降级为直接编码
+                return self.embedding_model.encode(text)
+
+        # 单文本输入
+        cached = _cache.get_embedding(text)
+        if cached is not None:
+            return _np.array(cached)
+
+        embedding = self.embedding_model.encode(text)
+        try:
+            emb_list = embedding.tolist() if hasattr(embedding, 'tolist') else list(embedding)
+            _cache.set_embedding(text, emb_list)
+        except Exception:
+            pass
+        return embedding
+
     def _search_image_chunks(self, query_vector: list, top_k: int = 5, where_filter: dict = None) -> dict:
         """
         独立检索图片切片（P0：图片独立召回通道）
@@ -1387,7 +1457,7 @@ class RAGEngine:
         if not target_collections:
             return get_empty_result()
 
-        query_vector = self.embedding_model.encode(query).tolist()
+        query_vector = self._encode_cached(query).tolist()
         # 扩大召回数量，以便过滤废止切片后仍有足够结果
         recall_k = RERANK_CANDIDATES if (USE_RERANK or USE_HYBRID_SEARCH) else top_k
         recall_k = max(recall_k, top_k * RECALL_MULTIPLIER)
@@ -1739,12 +1809,12 @@ class RAGEngine:
                 # === 高精度版：基于语义向量 ===
                 from core.mmr import mmr_rerank
 
-                # 获取查询向量
+                # 获取查询向量（使用 embedding 缓存）
-                query_emb = np.array(self.embedding_model.encode(query))
+                query_emb = np.array(self._encode_cached(query))
 
-                # 批量编码所有文档
+                # 批量编码所有文档（使用 embedding 缓存）
                 docs_list = results['documents'][0]
-                all_embeddings = self.embedding_model.encode(docs_list)
+                all_embeddings = self._encode_cached(docs_list)
 
                 # 构建候选列表
                 candidates = []
@@ -1940,104 +2010,7 @@ class RAGEngine:
                 reranked[key] = results[key]
         return reranked
 
-    # ---------------- 安全与工具 ----------------
+    # ---------------- 流式生成 ----------------
-
-    def check_restricted_documents(self, query, allowed_levels, top_k=3, role=None, department=None):
-        if not self._initialized:
-            self.initialize()
-
-        if USE_MULTI_KB and self.kb_manager and role and department:
-            from auth.gateway import get_accessible_collections
-            all_colls = [c.name for c in self.kb_manager.list_collections()]
-            accessible = set(get_accessible_collections(role, department, 'read'))
-            restricted = set(all_colls) - accessible
-
-            if not restricted:
-                return {"has_restricted": False, "restricted_levels": [], "restricted_sources": []}
-
-            query_vector = self.embedding_model.encode(query).tolist()
-            found_sources = set()
-            top_score = 0.0
-
-            for coll_name in restricted:
-                try:
-                    coll = self.kb_manager.get_collection(coll_name)
-                    if not coll: continue
-                    res = coll.query(query_embeddings=[query_vector], n_results=top_k)
-                    if res['metadatas'] and res['metadatas'][0]:
-                        for meta in res['metadatas'][0]:
-                            found_sources.add(meta.get('source', '未知'))
-                        for dist in (res.get('distances', [[]])[0] or []):
-                            if dist > top_score: top_score = dist
-                except Exception as e:
-                    logger.debug(f"权限检查遍历失败: {e}")
-
-            return {
-                "has_restricted": len(found_sources) > 0,
-                "restricted_levels": [c.replace('dept_', '') for c in restricted if True][:3],
-                "restricted_sources": list(found_sources)[:3],
-                "top_restricted_score": top_score
-            }
-
-        if not allowed_levels:
-            return {"has_restricted": False, "restricted_levels": [], "restricted_sources": [], "top_restricted_score": 0.0}
-
-        restricted_levels = {"public", "internal", "confidential", "secret"} - set(allowed_levels)
-        if not restricted_levels:
-            return {"has_restricted": False, "restricted_levels": [], "restricted_sources": [], "top_restricted_score": 0.0}
-
-        query_vector = self.embedding_model.encode(query).tolist()
-        try:
-            res = self.collection.query(
-                query_embeddings=[query_vector],
-                n_results=top_k,
-                where={"security_level": {"$in": list(restricted_levels)}}
-            )
-            docs = res.get('documents', [[]])[0]
-            if not docs:
-                return {"has_restricted": False, "restricted_levels": [], "restricted_sources": [], "top_restricted_score": 0.0}
-
-            metas = res.get('metadatas', [[]])[0]
-            dists = res.get('distances', [[]])[0]
-            found_levels, found_sources, top_score = set(), set(), 0.0
-
-            for meta, dist in zip(metas, dists):
-                found_levels.add(meta.get('security_level', 'public'))
-                found_sources.add(meta.get('source', '未知'))
-                if dist > top_score: top_score = dist
-
-            return {
-                "has_restricted": True,
-                "restricted_levels": list(found_levels),
-                "restricted_sources": list(found_sources)[:3],
-                "top_restricted_score": top_score
-            }
-        except Exception as e:
-            logger.warning(f"受限内容检查失败: {e}")
-            return {"has_restricted": False, "restricted_levels": [], "restricted_sources": [], "top_restricted_score": 0.0}
-
-    def generate_answer(self, query, context):
-        """底层生成答复能力"""
-        prompt = f"""你是一个严谨的智能助手，请根据以下参考资料回答用户的问题。
-...
-参考资料：
-{context}
-
-用户问题：{query}
-
-请回答："""
-        try:
-            from core.llm_utils import call_llm
-            result = call_llm(
-                self.llm_client,
-                prompt,
-                MODEL,
-                temperature=LLM_TEMPERATURE,
-                max_tokens=LLM_MAX_TOKENS
-            )
-            return result or f"调用大模型失败: 返回结果为空"
-        except Exception as e:
-            return f"调用大模型失败: {str(e)}"
 
     def generate_answer_stream(self, query, context, history=None):
         """