diff --git a/api/__init__.py b/api/__init__.py index 864a92f..85ccc20 100644 --- a/api/__init__.py +++ b/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 diff --git a/api/chat_routes.py b/api/chat_routes.py index 7a389bd..d0fcb08 100644 --- a/api/chat_routes.py +++ b/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 分词精准匹配） diff --git a/config.example.py b/config.example.py index 5d0a711..1d18fcc 100644 --- a/config.example.py +++ b/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 diff --git a/core/__init__.py b/core/__init__.py index 5ecf564..894f2e7 100644 --- a/core/__init__.py +++ b/core/__init__.py @@ -3,7 +3,6 @@ RAG 核心引擎模块 包含： - engine: RAGEngine 单例类，管理模型和共享资源 -- agentic: AgenticRAG 智能问答 - bm25_index: BM25 关键词检索索引 - chunker: 文本分块器 """ diff --git a/core/agentic.py b/core/agentic.py deleted file mode 100644 index acc1249..0000000 --- a/core/agentic.py +++ /dev/null @@ -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}") diff --git a/core/agentic_answer.py b/core/agentic_answer.py deleted file mode 100644 index 4a57db8..0000000 --- a/core/agentic_answer.py +++ /dev/null @@ -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)}" diff --git a/core/agentic_base.py b/core/agentic_base.py deleted file mode 100644 index 9558769..0000000 --- a/core/agentic_base.py +++ /dev/null @@ -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 # 幻觉修正最多重试次数 diff --git a/core/agentic_citation.py b/core/agentic_citation.py deleted file mode 100644 index 3859cb1..0000000 --- a/core/agentic_citation.py +++ /dev/null @@ -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) - } diff --git a/core/agentic_context.py b/core/agentic_context.py deleted file mode 100644 index baeb569..0000000 --- a/core/agentic_context.py +++ /dev/null @@ -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 diff --git a/core/agentic_media.py b/core/agentic_media.py deleted file mode 100644 index 48c0ce1..0000000 --- a/core/agentic_media.py +++ /dev/null @@ -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] - } diff --git a/core/agentic_meta.py b/core/agentic_meta.py deleted file mode 100644 index 1ae18a3..0000000 --- a/core/agentic_meta.py +++ /dev/null @@ -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您可以直接提问，我会尝试从知识库中检索相关信息。" diff --git a/core/agentic_quality.py b/core/agentic_quality.py deleted file mode 100644 index b1868b2..0000000 --- a/core/agentic_quality.py +++ /dev/null @@ -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": "默认重写"} diff --git a/core/agentic_query.py b/core/agentic_query.py deleted file mode 100644 index 4bbc1b4..0000000 --- a/core/agentic_query.py +++ /dev/null @@ -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 diff --git a/core/agentic_search.py b/core/agentic_search.py deleted file mode 100644 index bcb5c45..0000000 --- a/core/agentic_search.py +++ /dev/null @@ -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 diff --git a/core/cache.py b/core/cache.py index aaf78d1..e9c2beb 100644 --- a/core/cache.py +++ b/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） - doc_hash = "" - 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) + # 使用粗粒度 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: """ 设置查询缓存结果 + 始终使用粗粒度 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) - # 计算相关文档的版本哈希（细粒度失效） - doc_hash = "" - 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) + # 使用与 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: diff --git a/core/engine.py b/core/engine.py index b4468b6..7eb43aa 100644 --- a/core/engine.py +++ b/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 - # 获取查询向量 - query_emb = np.array(self.embedding_model.encode(query)) + # 获取查询向量（使用 embedding 缓存） + 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): """