""" 核心聊天与检索 API 本模块提供 RAG 系统的核心问答接口，包括： - 普通聊天模式（/chat） - 知识库问答模式（/rag，支持 SSE 流式返回） - 混合检索接口（/search，供外部系统调用） 路由列表: POST /chat : 普通聊天模式（JSON 响应） POST /rag : 知识库问答模式（SSE 流式返回） POST /search : 混合检索接口（供 Dify 调用） 架构说明: - 会话管理由后端服务负责，RAG 服务不存储对话历史 - 权限验证由后端网关完成（通过 request.current_user 获取用户信息） - /rag 接口已升级为 SSE 流式返回，支持实时输出 Example: curl -X POST http://localhost:5001/rag \\ -H "Content-Type: application/json" \\ -H "Authorization: Bearer mock-token-admin" \\ -d '{"query": "公司报销制度是什么？"}' """ import json import os import queue import re import threading import time as _time from typing import List, Dict, Any, Optional, Tuple from pathlib import Path import numpy as np from flask import Blueprint, request, jsonify, Response, current_app import logging logger = logging.getLogger(__name__) from auth.gateway import require_gateway_auth from core.status_codes import SUCCESS, BAD_REQUEST, NOT_FOUND, INTERNAL_ERROR from api.response_utils import success_response, error_response from auth.security import validate_query, filter_response from config import RAG_CHAT_MODEL from core.llm_utils import call_llm, call_llm_stream from core.prompt_guard import detect_injection, sanitize_user_input def _get_vlm_cache(image_path: str) -> Optional[str]: """ 从缓存获取图片的 VLM 描述 Args: image_path: 图片文件名（不含路径），如 '0cd5e156ff0a.png' Returns: VLM 描述文本，无缓存返回 None """ import hashlib try: images_dir = Path(".data/images") full_path = images_dir / image_path if not full_path.exists(): return None img_hash = hashlib.md5(full_path.read_bytes()).hexdigest() cache_file = Path(f".data/cache/vlm/{img_hash}.txt") if cache_file.exists(): return cache_file.read_text(encoding='utf-8') except Exception as e: logger.debug(f"读取 VLM 缓存失败: {e}") return None def _check_vlm_relevance(query: str, vlm_desc: str) -> float: """ 使用 LLM 评估图片描述与查询的相关性 Args: query: 用户查询 vlm_desc: 图片的 VLM 描述 Returns: 相关性分数 (0-1)，0=不相关，1=高度相关 """ if not vlm_desc or len(vlm_desc) < 20: return 0.5 # 无有效描述，中性评分 # 使用 jieba 分词提取关键词 import jieba stop_words = {'图', '表', '图片', '图表', '如图', '所示', '的', '是', '在', '有', '了', '和', '与', '或', '及', '等', '中', '对'} query_keywords = [w for w in jieba.lcut(query) if len(w) >= 2 and w not in stop_words] if not query_keywords: return 0.5 # 检查关键词是否在 VLM 描述中出现 matches = sum(1 for kw in query_keywords if kw in vlm_desc) if matches == 0: return 0.0 # 完全不相关 # 按匹配比例评分 ratio = matches / len(query_keywords) return min(ratio, 1.0) chat_bp = Blueprint('chat', __name__) def _safe_int(value: Any, default: int = 10**9) -> int: """ 安全的整数转换 Args: value: 待转换的值 default: 转换失败时的默认值 Returns: 转换后的整数值，失败返回默认值 """ try: if value is None or value == "": return default return int(value) except (TypeError, ValueError): return default def _is_enum_query(query: str) -> bool: """ 判断是否为枚举型查询 枚举型查询通常包含"哪些"、"列出"等关键词， 需要特殊的上下文排序策略。 Args: query: 用户查询文本 Returns: 是枚举型查询返回 True，否则返回 False """ try: from core.query_classifier import is_enumeration_query return is_enumeration_query(query) except Exception as e: logger.debug(f"枚举查询检测失败: {e}") markers = ("哪些", "有哪些", "列出", "严禁", "禁止", "不得", "包括", "要求", "情形", "场景") return bool(query) and any(marker in query for marker in markers) def _get_full_table_from_docstore(chunk_id: str) -> Optional[str]: """ 从 DocStore 获取表格的完整 Markdown 内容 Args: chunk_id: 切片 ID，如 '组织架构.xlsx_0' 或 'test_report.pdf_3' Returns: 完整表格 Markdown，未找到返回 None """ docstore_dir = Path(".data/docstore") if not docstore_dir.exists(): return None # chunk_id 格式: {filename}_{index} # DocStore 格式: {filename}_table_{index}.json 或 {filename}_{index}.json possible_paths = [ docstore_dir / f"{chunk_id}.json", # 直接匹配 docstore_dir / f"{chunk_id.replace('_', '_table_', 1)}.json", # xxx_0 -> xxx_table_0 ] # 如果 chunk_id 是 {filename}_{num} 格式，尝试 {filename}_table_{num} parts = chunk_id.rsplit('_', 1) if len(parts) == 2: filename, idx = parts possible_paths.append(docstore_dir / f"{filename}_table_{idx}.json") for doc_path in possible_paths: if doc_path.exists(): try: with open(doc_path, 'r', encoding='utf-8') as f: record = json.load(f) return record.get('markdown', '') except Exception as e: logger.debug(f"读取 DocStore 失败: {doc_path}, {e}") return None def _strip_semantic_prefix(doc: str, chunk_type: str) -> str: """ 去除切片 doc 中的冗余语义前缀，保留关键标识信息 表格切片的 doc 由 _build_semantic_content_for_table 生成，格式为： 主题：section_path（保留，用于关联章节） 字段：A, B, C（去除，冗余） 描述：该表包含N行数据（去除，冗余） 示例：字段=值（去除，冗余） 表格内容： | A | B | C | |---|---|---| | 1 | 2 | 3 | 优化策略： - 保留"主题："行（表格所属章节标识） - 去除"字段："/"描述："/"示例："行（冗余信息） - 添加"【表格】"标记，让 LLM 明确识别表格类型 - 保留 Markdown 表格内容（| 开头）和 HTML 表格内容（ List[Dict]: """ 对文本上下文排序，优化提示词构建 对于列举型查询，保持同一文档章节的切片连续排列， 便于 LLM 理解完整的语义上下文。 Args: contexts: 检索到的上下文列表 query: 用户查询 max_chunks: 最大切片数量 Returns: 排序后的上下文列表 Example: >>> ordered = _order_text_contexts_for_prompt(contexts, "有哪些禁止情形？", 10) """ # 文本切片 + 表格切片（从 DocStore 获取完整内容） text_contexts = [] for ctx in contexts: chunk_type = ctx.get('meta', {}).get('chunk_type', '') if chunk_type in ('image', 'chart'): continue # 图片/图表单独处理 if chunk_type == 'table': # 表格：从 DocStore 获取完整 Markdown chunk_id = ctx.get('meta', {}).get('chunk_id', '') full_table = _get_full_table_from_docstore(chunk_id) if chunk_id else None if full_table: # 替换为完整内容，保留 score text_contexts.append({ 'doc': full_table, 'meta': {**ctx.get('meta', {}), '_from_docstore': True}, 'score': ctx.get('score', 0) }) else: # DocStore 中没有，使用原始摘要 text_contexts.append(ctx) else: # 普通文本切片 text_contexts.append(ctx) # 图片/图表切片：提取其 doc 内容（包含前文/后文上下文） # 这些切片虽然 chunk_type 是 image/chart，但其 doc 字段包含有价值的上下文信息 chart_contexts = [] for ctx in contexts: chunk_type = ctx.get('meta', {}).get('chunk_type', '') if chunk_type in ('image', 'chart'): doc = ctx.get('doc', '') if doc and len(doc) > 20: # 有实际内容 # 标记来源，避免重复 chart_contexts.append({ 'doc': doc, 'meta': {**ctx.get('meta', {}), '_is_chart_context': True}, 'score': ctx.get('score', 0) }) # Phase 1：按 Rerank 分数过滤低分切片 # 保护策略：同一 section 内如有切片通过阈值，则同 section 的 table 切片也保留 # 原因：表格切片的 rerank 分数往往偏低（尤其是元问题如"有表格吗？"）， # 但它们与同 section 的 text 切片属于同一语义单元，不应割裂 # 安全下限：被保护的 table 切片自身 score 不得低于 min_score * 0.3， # 防止 section 粒度较粗时完全不相关的表格被无条件保护 if min_score > 0: _table_floor = min_score * 0.3 # table 保护最低分数下限 # 先找出所有通过阈值的 section passing_sections = set() for c in text_contexts: if c.get('score', 0) >= min_score: meta = c.get('meta', {}) section_key = (meta.get('source', ''), meta.get('section', '') or meta.get('section_path', '')) if section_key[1]: # 有 section 信息的才保护 passing_sections.add(section_key) text_contexts = [ c for c in text_contexts if c.get('score', 0) >= min_score or ( c.get('meta', {}).get('chunk_type') == 'table' and c.get('score', 0) >= _table_floor and (c.get('meta', {}).get('source', ''), c.get('meta', {}).get('section', '') or c.get('meta', {}).get('section_path', '')) in passing_sections ) ] chart_contexts = [c for c in chart_contexts if c.get('score', 0) >= min_score] # 合并：文本切片优先，图表切片补充 # 限制图表切片数量，避免过多 max_chart_contexts = 3 combined_contexts = text_contexts + chart_contexts[:max_chart_contexts] if not _is_enum_query(query): # 表格不受 max_chunks 限制（CrossEncoder 对表格评分偏低， # 但表格是结构化关键内容，不应因分数低而被截断） table_ctx = [c for c in combined_contexts if c.get('meta', {}).get('chunk_type') == 'table'] non_table_ctx = [c for c in combined_contexts if c.get('meta', {}).get('chunk_type') != 'table'] return non_table_ctx[:max_chunks] + table_ctx def sort_key(ctx): meta = ctx.get('meta', {}) source = meta.get('source', '') section = meta.get('section', '') or meta.get('section_path', '') rank = _safe_int(meta.get('_retrieval_rank'), 10**6) return ( source != primary_source, section != primary_section, source, section, _safe_int(meta.get('chunk_index')), _safe_int(meta.get('section_chunk_id')), rank, ) primary = text_contexts[0].get('meta', {}) if text_contexts else {} primary_source = primary.get('source', '') primary_section = primary.get('section', '') or primary.get('section_path', '') primary_index = _safe_int(primary.get('chunk_index'), None) try: from config import CONTEXT_EXPANSION_BEFORE, CONTEXT_EXPANSION_AFTER except Exception as e: logger.debug(f"读取上下文扩展配置失败: {e}") CONTEXT_EXPANSION_BEFORE, CONTEXT_EXPANSION_AFTER = 1, 5 if primary_source and primary_index is not None: window_start = primary_index - CONTEXT_EXPANSION_BEFORE window_end = primary_index + CONTEXT_EXPANSION_AFTER + 1 window = [] for ctx in text_contexts: meta = ctx.get('meta', {}) idx = _safe_int(meta.get('chunk_index'), None) if meta.get('source') == primary_source and idx is not None and window_start <= idx <= window_end: window.append(ctx) window_ids = { ctx.get('meta', {}).get('chunk_id') or (ctx.get('meta', {}).get('source'), ctx.get('meta', {}).get('chunk_index')) for ctx in window } window = sorted(window, key=lambda ctx: _safe_int(ctx.get('meta', {}).get('chunk_index'))) remainder = [ ctx for ctx in text_contexts if (ctx.get('meta', {}).get('chunk_id') or (ctx.get('meta', {}).get('source'), ctx.get('meta', {}).get('chunk_index'))) not in window_ids ] # 枚举查询不截断 max_chunks：需要跨 section 信息， # window 优先保证主命中文档连续，remainder 按相关性补充 return window + sorted(remainder, key=sort_key) ordered = sorted(text_contexts, key=sort_key) return ordered def _process_table_doc(doc: str, meta: Dict) -> str: """ 处理单个切片的 doc：精简表格语义前缀 + 注入嵌入图片 URL 集中处理两处共用逻辑（正常路径和预算截断路径），避免重复代码。 Args: doc: 切片原始 doc meta: 切片 metadata Returns: 处理后的 doc """ doc = _strip_semantic_prefix(doc, meta.get('chunk_type', '')) if meta.get('chunk_type') == 'table' and meta.get('images_json'): try: img_list = json.loads(meta['images_json']) if img_list: img_urls = [ f"/images/{img.get('id', '')}" for img in img_list if isinstance(img, dict) and img.get('id') ] if img_urls: doc += "\n\n[该表格包含以下图片，可在回答中引用]: " + ", ".join(img_urls) except (json.JSONDecodeError, TypeError): pass return doc def _section_similarity(section_a: str, section_b: str) -> float: """ 计算两个章节路径的层级相似度（数据驱动，无需硬编码格式假设）。 策略： 1. 如果两个路径都有数值编号（如 "2.3"），优先用精确数值匹配 2. 否则按 " > " 层级拆分，计算 Jaccard 相似度 3. 无数值编号时优雅降级，适用于 "第七章 附则" 或无结构文档 Returns: 0.0 ~ 1.0 """ import re as _re if not section_a or not section_b: return 0.0 # 快速路径：完全相同 if section_a == section_b: return 1.0 # 优先精确匹配：数值编号（如 "2.3"） num_a = _re.search(r'(\d+\.\d+)', section_a) num_b = _re.search(r'(\d+\.\d+)', section_b) if num_a and num_b: return 1.0 if num_a.group(1) == num_b.group(1) else 0.0 # 层级文本匹配：拆分路径为各级标题，计算 Jaccard 系数 def _split_levels(path): parts = [p.strip() for p in path.split('>') if p.strip()] # 去掉常见序号前缀（如 "1.1 "、"第1章 "），保留语义部分 cleaned = set() for p in parts: cleaned.add(_re.sub(r'^(?:\d+[\.\d]*\s*|第\s*\d+\s*章\s*|[一二三四五六七八九十]+、\s*)', '', p).strip()) return cleaned levels_a = _split_levels(section_a) levels_b = _split_levels(section_b) if not levels_a or not levels_b: return 0.0 overlap = len(levels_a & levels_b) union = len(levels_a | levels_b) return overlap / union if union > 0 else 0.0 def _rescue_bm25_divergence(contexts: List[Dict], search_result: dict, min_score: float) -> List[Dict]: """ BM25-CrossEncoder 分歧检测救援：当 BM25 排名靠前（top-3）的切片 被 CrossEncoder rerank 压制（分数低于 min_score 或被截断）时， 将其分数提升至保底值，使其通过后续的 min_score 过滤。 适用场景： - BM25 top-1/top-2 的切片（精确关键词匹配强信号）被 rerank 评分极低 - 正确切片在 rerank 后被截断，根本不在 contexts 中 Args: contexts: 全部上下文切片 search_result: engine.search_hybrid() 返回的原始结果（含 _bm25_top3） min_score: 最低分数阈值 Returns: 修改后的 contexts """ if not contexts: return contexts from config import ( BM25_DIVERGENCE_RESCUE_ENABLED, BM25_DIVERGENCE_MAX_RANK, CLUSTER_RESCUE_FLOOR, ) if not BM25_DIVERGENCE_RESCUE_ENABLED: return contexts bm25_top3 = search_result.get('_bm25_top3', []) if not bm25_top3: return contexts # 构建 contexts 中已有切片的 ID 索引，用于快速查找 existing_ids = {} for i, ctx in enumerate(contexts): chunk_id = ctx.get('meta', {}).get('chunk_id') or ctx.get('id') if chunk_id: existing_ids[chunk_id] = i # 计算 contexts 中的多数 source（用于情况 B 注入校验） # 防止跨文档注入无关切片（如 q013 场景：2.docx 的吸烟场所切片被注入到 1.docx 的查询中） source_counter: Dict[str, int] = {} for ctx in contexts: src = ctx.get('meta', {}).get('source', '') if src: source_counter[src] = source_counter.get(src, 0) + 1 majority_source = max(source_counter, key=source_counter.get) if source_counter else None rescued_count = 0 for bm25_item in bm25_top3: rank = bm25_item.get('rank', 99) if rank > BM25_DIVERGENCE_MAX_RANK: continue bm25_id = bm25_item.get('id') bm25_meta = bm25_item.get('meta', {}) bm25_doc = bm25_item.get('doc', '') # 情况 A：切片在 contexts 中但 score < min_score if bm25_id and bm25_id in existing_ids: ctx = contexts[existing_ids[bm25_id]] if ctx.get('score', 0) < min_score: ctx['score'] = CLUSTER_RESCUE_FLOOR rescued_count += 1 logger.debug( f"BM25 分歧救援 (情况A): rank={rank}, " f"source={bm25_meta.get('source', '')}, " f"section={bm25_meta.get('section', '')}, " f"原score→{CLUSTER_RESCUE_FLOOR}" ) continue # 情况 B：切片不在 contexts 中（被 rerank 截断或已被过滤） # 从 _bm25_top3 备份中注入，但需校验 source 一致性 if bm25_doc and bm25_meta: bm25_source = bm25_meta.get('source', '') # source 一致性校验：只注入与 contexts 多数 source 一致的切片 # 避免跨文档注入无关内容（如 2.docx 的吸烟场所切片混入 1.docx 的投放查询） if majority_source and bm25_source and bm25_source != majority_source: logger.debug( f"BM25 分歧救援 (情况B-跳过): rank={rank}, " f"source={bm25_source} != majority={majority_source}, " f"section={bm25_meta.get('section', '')}" ) continue injected_ctx = { 'doc': bm25_doc, 'meta': bm25_meta, 'score': CLUSTER_RESCUE_FLOOR, } contexts.append(injected_ctx) rescued_count += 1 logger.debug( f"BM25 分歧救援 (情况B-注入): rank={rank}, " f"source={bm25_meta.get('source', '')}, " f"section={bm25_meta.get('section', '')}" ) if rescued_count > 0: logger.info(f"BM25 分歧救援: 共救援 {rescued_count} 个切片") return contexts def _rescue_lexical_match(contexts: List[Dict], retrieval_query: str, min_score: float) -> List[Dict]: """ 词法匹配救援：当切片文本精确包含查询的核心关键词但 CrossEncoder 评分很低时， 将其分数提升至保底值，并同时救援同 source 下 chunk_index 相邻的切片。 适用场景： 1. 某个 section 只有一个正确切片（无法触发聚类救援） 2. 枚举类问题的 header 切片被词法匹配救援后，其后续子条目也应被一并保留 Args: contexts: 全部上下文切片 retrieval_query: 检索查询 min_score: 最低分数阈值 Returns: 修改后的 contexts """ if not contexts or not retrieval_query: return contexts import re from config import CLUSTER_RESCUE_FLOOR # 清理查询：去除 markdown 格式和标点 clean_query = re.sub(r'\*+|#+|`', '', retrieval_query) clean_query = re.sub(r'[？?！!。，,、；;：:"""\'\s]+', ' ', clean_query).strip() if len(clean_query) < 2: return contexts # 提取查询中的有意义 bigram（连续两字组） query_bigrams = set() for i in range(len(clean_query) - 1): w = clean_query[i:i+2].strip() if len(w) == 2: query_bigrams.add(w) if not query_bigrams: return contexts # Phase 1: 词法匹配救援——找到高分匹配的切片 rescued_seeds = [] # [(source, chunk_index)] for ctx in contexts: if ctx.get('score', 0) >= min_score: continue doc = ctx.get('doc', '') or '' meta = ctx.get('meta', {}) section = meta.get('section', '') or meta.get('section_path', '') combined = doc + ' ' + section matched = sum(1 for w in query_bigrams if w in combined) match_ratio = matched / len(query_bigrams) if match_ratio > 0.35: ctx['score'] = max(ctx.get('score', 0), CLUSTER_RESCUE_FLOOR) source = meta.get('source', '') chunk_index = meta.get('chunk_index') if source and chunk_index is not None: try: rescued_seeds.append((source, int(chunk_index))) except (ValueError, TypeError): pass # Phase 2: 邻居救援——对每个被词法匹配救援的种子， # 同时救援同 source 下 chunk_index 后续相邻的切片（枚举子条目） if rescued_seeds: for source, seed_idx in rescued_seeds: for ctx in contexts: if ctx.get('score', 0) >= min_score: continue meta = ctx.get('meta', {}) if meta.get('source') != source: continue try: n_idx = int(meta.get('chunk_index', -1)) except (ValueError, TypeError): continue # 救援种子后续 8 个相邻切片（覆盖大多数枚举/条款模式） if seed_idx < n_idx <= seed_idx + 8: ctx['score'] = max(ctx.get('score', 0), CLUSTER_RESCUE_FLOOR) return contexts def _normalize_section_for_rescue(section_path: str, levels: int = 2) -> str: """归一化 section_path：取前 N 级路径用于分组""" if not section_path: return '' parts = [p.strip() for p in section_path.split('>')] return ' > '.join(parts[:levels]) def _rescue_section_cluster(contexts: List[Dict], retrieval_query: str, min_score: float) -> List[Dict]: """ 路由层章节聚类救援：当某个 section 有多个候选切片但全部低于 min_score 时， 给该 section 的切片分配保底分数，使其通过后续的 min_score 过滤。 作为引擎层 _section_cluster_boost 的二次安全网，防止极端情况下所有正确切片被过滤。 Args: contexts: engine 返回的全部上下文切片（每个含 doc, meta, score） retrieval_query: 改写后的检索查询 min_score: 当前的最低分数阈值 Returns: 修改后的 contexts（部分切片 score 被提升至保底分数） """ if not contexts: return contexts from config import ( CLUSTER_MIN_MEMBERS, CLUSTER_MIN_TYPES, CLUSTER_RESCUE_FLOOR, CLUSTER_MAX_SECTIONS, CLUSTER_MAX_RESCUE_PER_SECTION, CLUSTER_SECTION_PREFIX_LEVELS, ) # 1. 按 (source, normalized_section) 分组 from collections import defaultdict section_groups = defaultdict(list) # key → [index_in_contexts] for i, ctx in enumerate(contexts): meta = ctx.get('meta', {}) source = meta.get('source', '') section_path = meta.get('section', '') or meta.get('section_path', '') norm_section = _normalize_section_for_rescue(section_path, CLUSTER_SECTION_PREFIX_LEVELS) if not source or not norm_section: continue key = (source, norm_section) section_groups[key].append(i) # 2. 检测"全灭 section"并计算聚类强度 rescue_candidates = [] # (strength, key, member_indices) for key, indices in section_groups.items(): if len(indices) < CLUSTER_MIN_MEMBERS: continue # 检查是否所有成员都低于 min_score（"全灭"） scores = [contexts[i].get('score', 0) for i in indices] if any(s >= min_score for s in scores): continue # 已有成员通过阈值，无需救援 # 计算聚类强度 chunk_types = set(contexts[i].get('meta', {}).get('chunk_type', 'text') for i in indices) type_diversity = len(chunk_types) if type_diversity < CLUSTER_MIN_TYPES: continue # 类型不够多样，可能是噪音 # 查询与 section_path 的字符重叠率 source, norm_section = key query_chars = set(retrieval_query) section_chars = set(norm_section) overlap = len(query_chars & section_chars) / max(len(query_chars), 1) # 聚类强度 = 成员数 × 类型多样性 × (1 + 查询匹配度) strength = len(indices) * type_diversity * (1.0 + overlap) rescue_candidates.append((strength, key, indices)) # 3. 按强度降序，救援 top-N section rescue_candidates.sort(key=lambda x: x[0], reverse=True) rescued_sections = [] total_rescued = 0 for strength, key, indices in rescue_candidates[:CLUSTER_MAX_SECTIONS]: # 对组内切片分配保底分数（仅低于 min_score 的） rescue_count = 0 for idx in indices: if rescue_count >= CLUSTER_MAX_RESCUE_PER_SECTION: break ctx = contexts[idx] if ctx.get('score', 0) < min_score: ctx['score'] = CLUSTER_RESCUE_FLOOR rescue_count += 1 total_rescued += 1 if rescue_count > 0: source, norm_section = key rescued_sections.append({ 'source': source, 'section': norm_section, 'members': len(indices), 'rescued': rescue_count, 'strength': round(strength, 2) }) return contexts def _rescue_table_chunks(contexts: List[Dict], context_text: str, retrieval_query: str, max_rescue_chars: int = 3000) -> str: """ 表格救援：当查询涉及表格但上下文预算截断了表格内容时，将最相关的表格补回。 根因：CrossEncoder 对 Markdown 表格格式的评分普遍偏低， 导致 _build_context_with_budget 按分数排序时表格被排到最后并被截断。 此函数作为安全网，确保与查询最相关的表格始终出现在上下文中。 Args: contexts: 经过 _order_text_contexts_for_prompt 处理的全部文本切片 context_text: _build_context_with_budget 的输出 retrieval_query: 改写后的检索查询 max_rescue_chars: 救援表格的最大字符数 Returns: 可能追加了表格内容的上下文文本 """ # 1. 数据驱动检测：检索结果中是否有表格类型切片（无需硬编码关键词） has_table_in_contexts = any( ctx.get('meta', {}).get('chunk_type') == 'table' for ctx in contexts ) if not has_table_in_contexts: return context_text # 2. 检测现有上下文是否已包含表格数据（Markdown 表格至少 2 列） if '|' in context_text and context_text.count('|') > 4: return context_text # 3. 从 contexts 中找被截断的表格切片 # 使用 _in_budget_context 标记精确判断（由 _build_context_with_budget 设置） table_candidates = [] for ctx in contexts: if ctx.get('meta', {}).get('chunk_type') != 'table': continue # 精确判断：如果 _build_context_with_budget 已标记该 chunk 为已纳入，跳过 if ctx.get('_in_budget_context'): continue table_candidates.append(ctx) if not table_candidates: return context_text # 4. 按分数降序取最佳表格，追加到上下文 table_candidates.sort(key=lambda c: c.get('score', 0), reverse=True) rescue_parts = [] rescue_chars = 0 for ctx in table_candidates[:3]: # 最多救援 3 个表格 meta = ctx.get('meta', {}) doc = _process_table_doc(ctx.get('doc', ''), meta) section = meta.get('section', '') or meta.get('section_path', '') part_text = "" if section: part_text += f"━ {section} ━\n" part_text += doc if rescue_chars + len(part_text) > max_rescue_chars: break rescue_parts.append(part_text) rescue_chars += len(part_text) if rescue_parts: separator = "\n\n--- 以下为与查询最相关的表格（CrossEncoder 评分偏低，自动补入）---\n\n" context_text += separator + "\n\n".join(rescue_parts) return context_text def _build_context_with_budget(contexts: List[Dict], max_chars: int, soft_limit: int) -> str: """ Phase 2：按字符预算构建上下文文本。 1. 按 (source, section) 分组，组内按 chunk_index 保持连续 2. 组间按组内最高 Rerank 分数降序排列 3. 逐组加入直到达到 max_chars 预算 4. 超过 soft_limit 后仅接受组内最高分 >= soft_min_score 的组 对于列举类查询（_is_enum_query），保持原始顺序直接拼接， 因为 _order_text_contexts_for_prompt 已经优化了排序。 Args: contexts: 已排序的上下文列表 max_chars: 硬性字符上限 soft_limit: 软限制，超过后收紧准入 Returns: 拼接好的上下文文本 """ if not contexts: return "" # 按 (source, section) 分组 groups = {} group_order = [] for ctx in contexts: meta = ctx.get('meta', {}) key = (meta.get('source', ''), meta.get('section', '') or meta.get('section_path', '')) if key not in groups: groups[key] = [] group_order.append(key) groups[key].append(ctx) # 组内按 chunk_index 排序 for key in groups: groups[key].sort(key=lambda c: _safe_int(c.get('meta', {}).get('chunk_index'))) # 组间按组内最高 score 降序 def group_max_score(key): return max((c.get('score', 0) for c in groups[key]), default=0) group_order.sort(key=group_max_score, reverse=True) # 贪心加入直到预算满 parts = [] total_chars = 0 for key in group_order: group = groups[key] source, section = key # 判断是否包含表格切片（表格是结构化关键内容，不受预算截断） has_table = any(c.get('meta', {}).get('chunk_type') == 'table' for c in group) # 构建组文本，表格切片附加图片 URL 供 LLM 引用 doc_parts = [] for ctx in group: doc = _process_table_doc(ctx.get('doc', ''), ctx.get('meta', {})) doc_parts.append(doc) group_text = "\n\n".join(doc_parts) # 在组首插入章节路径标题行，帮助 LLM 区分不同章节 section_header = '' if section: section_header = f"━ {section} ━" group_text = section_header + "\n" + group_text # 超过软限制后，只接受高分组（但表格组始终保留，不因分数低被跳过） if total_chars > soft_limit and group_max_score(key) < 0.1 and not has_table: continue if total_chars + len(group_text) > max_chars: # 尝试逐条加入该组，直到预算满 # 先加入章节标题（如果有） if section_header and total_chars + len(section_header) + 2 <= max_chars: parts.append(section_header) total_chars += len(section_header) + 2 for ctx in group: doc = _process_table_doc(ctx.get('doc', ''), ctx.get('meta', {})) if total_chars + len(doc) + 2 > max_chars: # +2 for "\n\n" break ctx['_in_budget_context'] = True # 标记已纳入上下文 parts.append(doc) total_chars += len(doc) + 2 # 超预算后一律 break，被截断的表格由 _rescue_table_chunks 补回 break for ctx in group: ctx['_in_budget_context'] = True # 标记已纳入上下文 parts.append(group_text) total_chars += len(group_text) + 2 # +2 for "\n\n" return "\n\n".join(parts) def _attach_citations(answer: str, contexts: List[Dict]) -> Dict[str, Any]: """ 自动为回答添加引用标记（按段落级别匹配，jieba 分词精准匹配） 流程: 1. 将 answer 按自然段落分割（\n\n） 2. 对每个段落用 jieba 分词后计算词级重叠度 3. Phase 6：动态阈值（短段落 0.55 / 长段落 0.45），每段最多 2 个引用 4. 前端负责对引用进行重新编号 Args: answer: LLM 生成的回答 contexts: 检索到的上下文列表 Returns: { "answer_with_refs": "回答文本（含 [ref:chunk_id] 标记）", "citations": [引用列表] } """ import re if not contexts: return {"answer_with_refs": answer, "citations": []} # 按 (collection, chunk_id) 复合键组织 contexts，防止跨库同名文件覆盖 ctx_by_chunk = {} for ctx in contexts: meta = ctx.get('meta', {}) chunk_id = meta.get('chunk_id') or f"{meta.get('source')}_{meta.get('chunk_index', 0)}" coll = meta.get('_collection') or meta.get('collection') or '' composite_key = f"{coll}/{chunk_id}" if coll else chunk_id # 保存原始 chunk_id，用于对外输出（ref tag / citation） ctx['_raw_chunk_id'] = chunk_id ctx_by_chunk[composite_key] = ctx # jieba 分词函数（fallback 到字符级） try: import jieba def _tokenize(text: str) -> set: return set(w for w in jieba.lcut(text) if len(w) >= 2) except ImportError: def _tokenize(text: str) -> set: return set(text) # 按自然段落分割（\n\n 为分隔符，保留分隔符用于重组） parts = re.split(r'(\n\n+)', answer) cited_chunks_ordered: List[str] = [] # 保序去重的 chunk_id 列表 cited_set: set = set() result_parts = [] for i in range(0, len(parts), 2): para = parts[i] sep = parts[i + 1] if i + 1 < len(parts) else '' stripped = para.strip() # 跳过过短段落或 markdown 表格/标题行 is_table = stripped.startswith('|') or '|---' in stripped is_short = len(stripped) < 15 if is_table or is_short: result_parts.append(para + sep) continue # 词级重叠匹配：找最相关的 chunk para_words = _tokenize(stripped[:300]) # Phase 6：动态阈值 — 短段落用更高阈值避免误匹配 overlap_threshold = 0.55 if len(stripped) < 50 else 0.45 # 收集所有超过阈值的候选 chunk，按分数降序 candidates = [] for chunk_id, ctx in ctx_by_chunk.items(): ctx_doc = ctx.get('doc', '') if not ctx_doc: continue ctx_words = _tokenize(ctx_doc[:400]) if not para_words: continue overlap = len(para_words & ctx_words) score = overlap / len(para_words) if score >= overlap_threshold: candidates.append((chunk_id, score)) candidates.sort(key=lambda x: x[1], reverse=True) # Phase 6：允许最多 2 个引用（分数差距 < 0.1 时附加第二引用） selected_ids = [] if candidates: selected_ids.append(candidates[0][0]) if len(candidates) > 1 and (candidates[0][1] - candidates[1][1]) < 0.1: selected_ids.append(candidates[1][0]) # 按 _raw_chunk_id 去重：不同 composite_key 可能指向同一个底层 chunk # 避免同一 chunk 产生重复引用标记（如 [3][3]） seen_raw_ids = set() deduped_ids = [] for cid in selected_ids: raw_id = ctx_by_chunk[cid].get('_raw_chunk_id', cid) if raw_id not in seen_raw_ids: seen_raw_ids.add(raw_id) deduped_ids.append(cid) selected_ids = deduped_ids if selected_ids: for cid in selected_ids: if cid not in cited_set: cited_set.add(cid) cited_chunks_ordered.append(cid) # 在段落末尾插入引用标记（使用原始 chunk_id，不暴露复合键） ref_tags = "".join( f"[ref:{ctx_by_chunk[cid].get('_raw_chunk_id', cid)}]" for cid in selected_ids ) result_parts.append(f"{para}{ref_tags}{sep}") else: result_parts.append(para + sep) # 构建引用列表（按出现顺序），使用原始 chunk_id 构建 citation # 按 _raw_chunk_id 去重，避免同一 chunk 产生重复引用条目 citations = [] seen_citation_raw_ids = set() for composite_key in cited_chunks_ordered: ctx = ctx_by_chunk.get(composite_key) if ctx: raw_id = ctx.get('_raw_chunk_id') or composite_key if raw_id in seen_citation_raw_ids: continue # 同一 chunk 已在引用列表中，跳过 seen_citation_raw_ids.add(raw_id) meta = ctx.get('meta', {}) full_content = ctx.get('doc', '') citation = _build_citation(meta, full_content) citation['chunk_id'] = raw_id citations.append(citation) return { "answer_with_refs": "".join(result_parts), "citations": citations } def _clean_section(raw: str) -> str: """清洗 section 字段：过滤掉像正文内容而非章节路径的值""" if not raw: return '' import re as _re def _is_heading(part: str) -> bool: """判断一个字符串是否像章节标题（而非正文内容）""" p = part.strip() if not p: return False # 以句号/问号/感叹号结尾 → 正文句子 if p.endswith(('。', '！', '？', '.', '!', '?')): return False # 含冒号且偏长 → 更像是内容摘要而非标题 if '：' in p and len(p) > 25: return False # 长度超过 30 字 → 不像是标题 if len(p) > 30: return False return True # 去掉 Markdown 加粗标记 cleaned = raw.replace('**', '').strip() # 如果含 ' > ' 分隔符，验证每一级都是合法标题 if ' > ' in cleaned: parts = [p.strip() for p in cleaned.split('>') if p.strip()] valid = [p for p in parts if _is_heading(p)] if valid: return ' > '.join(valid[:3]) # 所有部分都不像标题 → 清空 return '' # 匹配 【xxx】 或 第X篇/章/节 格式 if _re.match(r'^(【[^】]+】|第[一二三四五六七八九十\d]+[篇章节部])', cleaned): return cleaned[:40] # 短字符串（< 30字）且不像句子（无句号/逗号），保留 if len(cleaned) < 30 and not any(c in cleaned for c in '。，；：'): return cleaned # 其余视为正文内容，清空 return '' def _build_citation(meta: Dict, full_content: str = '') -> Dict[str, Any]: """ 根据文档类型构建定位信息 不同文档类型使用不同的定位策略： - PDF: 坐标定位（page + bbox） - Word: 语义定位（section + section_chunk_id + preview） - Excel: 表格定位（sheet + preview） Args: meta: 切片元数据 full_content: 完整切片内容（可选） Returns: 引用信息字典，包含定位信息 Example: >>> citation = _build_citation(meta, "完整内容...") >>> print(citation["page"]) # PDF: 页码 >>> print(citation["section"]) # Word: 章节 """ # 从 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') or meta.get('collection', ''), # 所属向量库，用于前端文档预览跳转 "doc_type": meta.get('doc_type', 'other'), "section": _clean_section(meta.get('section', '')), "preview": meta.get('preview', ''), "content": (full_content or meta.get('preview', ''))[:300], # 截断至 300 字避免冒返大量数据 "chunk_type": meta.get('chunk_type', 'text'), } doc_type = meta.get('doc_type', 'other') if doc_type == 'pdf': # 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': # Word: 语义定位 citation.update({ "section_chunk_id": meta.get('section_chunk_id'), # 章节内段落序号 }) elif doc_type == 'excel': # 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 score_image_relevance(query: str, meta: Dict, doc: str = '') -> float: """ 图片相关性打分（语义增强版） 通过多维度特征评估图片与查询的相关性： 1. 图片编号精确匹配（如 "图2.1"） 2. 关键词匹配（年份、数值+单位、中文词组） 3. 整体文本相似度 4. 章节匹配 5. 图片类型加分 Args: query: 用户查询 meta: 切片元数据 doc: document 字段（包含图片描述和上下文） Returns: 相关性分数（>= 3.0 推荐展示） Example: >>> score = score_image_relevance("图2.1是什么？", meta, doc) >>> if score >= 3.0: ... # 推荐展示该图片 """ import re score = 0.0 # 优先使用 doc 字段（包含完整描述和上下文） search_text = doc or meta.get('caption', '') section = meta.get('section', '') or meta.get('section_path', '') source = meta.get('source', '') # 1. 图片编号精确匹配（最高优先级） figure_pattern = r'图\s*(\d+\.?\d*)' figure_matches = re.findall(figure_pattern, query) if figure_matches: for fig_num in figure_matches: # 在所有文本中查找图号 all_text = f"{search_text} {section} {source}" if f"图{fig_num}" in all_text or f"图 {fig_num}" in all_text or f"见图{fig_num}" in all_text: score += 10.0 # 精确匹配，直接返回 return score # 1.5. 表格编号精确匹配（新增：支持表格图片） table_pattern = r'表\s*(\d+\.?\d*)' table_matches = re.findall(table_pattern, query) if table_matches: for table_num in table_matches: # 在所有文本中查找表号 all_text = f"{search_text} {section} {source}" if f"表{table_num}" in all_text or f"表 {table_num}" in all_text or f"见表{table_num}" in all_text: score += 10.0 # 精确匹配，直接返回 return score # 2. 查询词匹配（通用方式，不硬编码关键词） # 从查询中提取有意义的词：中文词组、数字+单位、年份等 # 使用 jieba 分词（如果可用）或简单的正则提取 query_keywords = [] # 提取年份（如 "2003年"） year_matches = re.findall(r'(\d{4})\s*年', query) query_keywords.extend(year_matches) # 提取数值+单位（如 "100亿"、"50万千瓦时"） num_unit_matches = re.findall(r'(\d+\.?\d*\s*[亿万万千百吨米秒])', query) query_keywords.extend(num_unit_matches) # 使用 jieba 分词提取中文关键词 import jieba jieba_words = [w for w in jieba.lcut(query) if len(w) >= 2] query_keywords.extend(jieba_words) # 过滤掉泛词（图、表、图片等） stop_words = {'图', '表', '图片', '图表', '如图', '所示', '如下', '如下表', '如下图', '的', '是', '在', '有', '了', '和', '与', '或', '及', '等', '中', '对'} query_keywords = [kw for kw in query_keywords if kw not in stop_words] # 在图片描述中匹配关键词 keyword_match_score = 0.0 for kw in query_keywords: if kw in search_text or kw in section: keyword_match_score += 2.0 score += min(keyword_match_score, 8.0) # 最多加 8 分 # 3. 整体文本相似度（字符级别） if search_text: # 复用已过滤停用词的 query_keywords，避免字符级误删（如"表现"→"现"） query_core = "".join(query_keywords) if query_core: overlap = len(set(query_core) & set(search_text)) score += min(overlap * 0.2, 3.0) # 4. 章节匹配 if section: # 从查询中提取章节关键词（复用 jieba 分词） section_keywords = query_keywords for kw in section_keywords: if kw in section: score += 1.5 # 5. 图片类型加分 if meta.get('chunk_type') == 'chart': score += 2.0 elif meta.get('chunk_type') == 'image': score += 1.0 # 6. 检索相似度（如果有） retrieval_score = meta.get('score', 0) if retrieval_score > 0: score += min(retrieval_score * 2, 2.0) return score # 图片意图关键词：用于判断查询/回答是否需要图片 # 集中管理，便于后续新增文档类型时扩展 # 查询侧关键词（宽松）：用户查询中出现这些词表示想看图片 _FIGURE_QUERY_KEYWORDS = frozenset([ '图', '表', '照片', '图表', '如图', '见表', '示意图', '展示', '示意', '外观', '实物', '结构', '流程图', '架构图', ]) # 回答侧关键词（严格）：LLM 回答中出现这些词表示引用了图片 # 不含单字"图""表"（"力图""企图""表达"等领域词误触发），改用正则图号检测兜底 # 不含"图片"（定义类查询的 LLM 回答也会泛化提到"图片"，导致负面用例误放行） _FIGURE_ANSWER_KEYWORDS = frozenset([ '照片', '图表', '如图', '见图', '见表', '示意图', '流程图', '架构图', '结构图', '实物图', '外观图', ]) def _replace_table_image_placeholders(answer: str, images: List[Dict]) -> str: """ 将回答中 markdown 表格内的 [图片] 占位符替换为实际图片语法。 LLM 看到的上下文里表格单元格是 [图片]（MinerU 解析产物），它也原样输出。 此函数按顺序将每个 [图片] 映射到 images 列表中对应图片的 URL， 生成 ![img](/images/xxx.png) 语法，前端 markdown-it 可直接渲染为 标签。 Args: answer: LLM 生成的回答文本 images: select_images + _filter_images_by_answer 后的最终图片列表 Returns: 替换后的回答文本 """ if not images or '[图片]' not in answer: return answer # 按顺序替换：第 N 个 [图片] → 第 N 张图片 img_iter = iter(images) replaced_count = 0 def _replacer(match): nonlocal replaced_count try: img = next(img_iter) url = img.get('url', '') img_id = img.get('id', 'img') replaced_count += 1 return f'![{img_id}]({url})' except StopIteration: # 图片用完了，保留剩余占位符 return match.group(0) result = re.sub(r'\[图片\]', _replacer, answer) if replaced_count > 0: logger.info(f"[图片占位符替换] 替换了 {replaced_count}/{len(images)} 个 [图片] 为 markdown 图片语法") return result def _filter_images_by_answer(selected_images: List[Dict], answer: str, query: str = "", primary_sections: List[str] = None) -> List[Dict]: """ 后置图片过滤：查询主题一致性校验 + 无图意图早退 + 子章节级过滤。 确保展示的图片与查询/回答主题一致，过滤 section_cluster_boost 过度召回的 不相关章节图片。 过滤规则: 1. 候选 <= 1 张时直接返回（无需过滤） 2. 无图意图早退：回答和查询都不含图片引用词 → 返回空 3. 图号豁免：图片描述包含回答引用的具体图号/表号 → 无条件保留 4. 主题一致性：合并查询+回答关键词，图片描述重叠 >= 阈值才保留 - 子章节惩罚：图片子章节与主要检索章节不一致时，阈值 +2 5. 兜底：过滤后为空且有图片意图 → 保留分数最高的 1 张 Args: selected_images: select_images 返回的候选图片列表 answer: LLM 生成的回答文本 query: 用户查询（拼接 retrieval_query + message，覆盖改写丢失的关键词） primary_sections: 主要检索结果的 section_path 列表（来自 top 文本切片）， 用于子章节级过滤。为空时跳过子章节惩罚，行为与原版一致。 Returns: 过滤后的图片列表 """ if not selected_images or len(selected_images) <= 1: return selected_images import re # 提取回答中引用的图号/表号（用于精确豁免，不再一刀切绕过） _mentioned_refs = set() _mentioned_refs.update(re.findall(r'(?:[见如])?图\s*(\d+[\.\-]?\d*)', answer)) _mentioned_refs.update(re.findall(r'(?:[见如])?表\s*(\d+[\.\-]?\d*)', answer)) # 前置无图意图判断：回答和查询都不需要图片时，直接返回空 # 解决定义/原则类查询（case 18/19）领域关键词导致图片描述高重叠的问题 _has_figure_in_answer = any(kw in answer for kw in _FIGURE_ANSWER_KEYWORDS) if not _has_figure_in_answer: _has_figure_in_answer = bool(re.search(r'[图表]\s*\d+', answer)) # 表格内嵌图片检测：回答中出现 [图片] 占位符或 markdown 图片语法 if not _has_figure_in_answer: _has_figure_in_answer = bool(re.search(r'\[图片\]|!\[', answer)) _has_figure_in_query = any(kw in query for kw in _FIGURE_QUERY_KEYWORDS) if not _has_figure_in_query: _has_figure_in_query = bool(re.search(r'[图表]\s*\d+', query)) if not _has_figure_in_answer and not _has_figure_in_query: logger.info(f"[图片后置过滤] 无图意图前置检查：回答和查询均不含图片意图，返回空 " f"(候选 {len(selected_images)} 张)") return [] # 合并查询+回答关键词：查询提供主题焦点，回答提供细节补充 # 解决纯回答关键词在长回答（500+字）场景下区分度不足的问题 try: import jieba _combined_text = f"{query} {answer}" if query else answer _keywords = set( w for w in jieba.lcut(_combined_text) if len(w) >= 2 and re.search(r'[\u4e00-\u9fff]', w) ) except ImportError: _chars = re.findall(r'[\u4e00-\u9fff]', f"{query} {answer}") _keywords = set(_chars[i] + _chars[i + 1] for i in range(len(_chars) - 1)) if not _keywords: return selected_images # 动态阈值：回答关键词越多，阈值越高（至少匹配 15% 或 2 个关键词，取较小值） threshold = max(2, min(4, round(len(_keywords) * 0.10))) # === 子章节级过滤：从主要检索结果提取主题章节，用于区分同父章节不同子主题的图片 === def _leaf_section_name(section_path: str) -> str: """提取 section_path 的叶子节点名称（去编号前缀）""" if not section_path: return '' parts = [p.strip() for p in section_path.split('>') if p.strip()] leaf = parts[-1] if parts else '' # 去掉编号前缀：(2)、（一）、2.3、第1章、一、 等 return re.sub( r'^(?:\(\d+\)\s*|（[一二三四五六七八九十]+）\s*|\d+[\.\d]*\s*|' r'第\s*\d+\s*章\s*|[一二三四五六七八九十]+、\s*)', '', leaf ).strip() def _section_leaf_match(img_leaf: str, primary_name: str) -> bool: """子章节名称匹配：长名称用子串包含，短名称（<3字）只做精确匹配""" if not img_leaf or not primary_name: return False if img_leaf == primary_name: return True if len(img_leaf) >= 3 and len(primary_name) >= 3: return primary_name in img_leaf or img_leaf in primary_name return False # 构建主要子章节名称集合（从 top 文本切片的 section_path 提取） _primary_leaf_names = set() if primary_sections: for ps in primary_sections: name = _leaf_section_name(ps) if name and len(name) >= 2: _primary_leaf_names.add(name) # 检索发散检测：主要子章节超过 3 个说明 section_cluster_boost 范围过宽 # 子章节惩罚退化为无效（几乎所有图片都能匹配某个 primary section） # 此时对所有图片统一加严阈值，避免不相关图片全部通过 _scattered_bonus = 1 if len(_primary_leaf_names) > 3 else 0 filtered = [] for img in selected_images: desc = img.get('full_description', '') or img.get('description', '') or '' if not desc: # 没有描述的图片，保留（无法判断） filtered.append(img) continue # 图号精确豁免：图片描述包含回答引用的具体图号/表号，无条件保留 # 防御性逻辑：与 P0 答案对齐互补（P0 用 description 100字截断，此处用 full_description） if _mentioned_refs: _ref_matched = False for ref in _mentioned_refs: _ref_norm = ref.replace('-', '.') if (f"图{_ref_norm}" in desc or f"图 {_ref_norm}" in desc or f"表{_ref_norm}" in desc or f"表 {_ref_norm}" in desc): _ref_matched = True break if _ref_matched: filtered.append(img) continue # 主题一致性过滤：图片描述与查询+回答关键词重叠 >= 阈值才保留 overlap = sum(1 for kw in _keywords if kw in desc) # 阈值计算：基础阈值 + 检索发散加严 + 子章节惩罚 effective_threshold = threshold + _scattered_bonus if _primary_leaf_names: img_section_path = img.get('section_path', '') img_leaf = _leaf_section_name(img_section_path) if img_leaf and not any(_section_leaf_match(img_leaf, pn) for pn in _primary_leaf_names): effective_threshold += 2 if overlap >= effective_threshold: filtered.append(img) # 兜底：过滤后为空且有图片意图时，保留分数最高的 1 张 # 注意：无图意图场景已在上方早退（返回 []），此处兜底仅在有图意图时生效 if not filtered and selected_images: filtered = [max(selected_images, key=lambda x: x.get('score', 0))] if len(filtered) < len(selected_images): _sub_info = f", 子章节 {len(_primary_leaf_names)} 个" if _primary_leaf_names else "" _scattered_info = f", 发散加严+{_scattered_bonus}" if _scattered_bonus else "" logger.info(f"[图片后置过滤] {len(selected_images)} → {len(filtered)} 张 " f"(关键词 {len(_keywords)} 个, 阈值 {threshold}{_sub_info}{_scattered_info}, " f"图号豁免 {len(_mentioned_refs)} 个)") return filtered def select_images(contexts: List[Dict], query: str) -> List[Dict]: """ 选择要展示的图片（打分排序 + 预算控制） 根据查询意图动态调整图片数量上限： - 精确查图（指定图号）: 最多 2 张 - 强图片意图（示意图、流程图等）: 最多 3 张 - 列举型查询: 最多 5 张 - 普通查询: 最多 2 张 核心逻辑: 1. 检测查询中的图号引用 2. 从检索文本中提取图表引用 3. 对图片打分并过滤低分图片 4. 通过章节关联排除不相关图片 Args: contexts: 检索上下文列表 query: 用户查询 Returns: 精选图片列表（每项含 score, id, url, type, source 等字段） Example: >>> images = select_images(contexts, "图2.1展示了什么？") >>> print(len(images)) # <= 2 """ import re # 动态预算：数据驱动，不依赖硬编码关键词列表 # 核心策略：宽松预选 + 后置过滤（_filter_images_by_answer）精准裁剪 # 精确查图：用户指定了具体图号（如 "图2.3"）—— 结构化模式匹配，非硬编码 figure_pattern = r'图\s*(\d+\.?\d*)' figure_matches = re.findall(figure_pattern, query) has_figure_query = bool(figure_matches) # 数据驱动的图片意图检测：检查检索结果中是否包含图片/图表类型切片 # 原理：如果向量检索返回了 image/chart 类型 chunk 或含 images_json 的 table chunk， # 说明知识库中存在与查询语义相关的图片内容，应给予展示机会 has_image_data = False _image_chunk_count = 0 _table_image_count = 0 for ctx in contexts: meta = ctx.get('meta', {}) ct = meta.get('chunk_type', '') if ct in ('image', 'chart'): _image_chunk_count += 1 has_image_data = True if ct == 'table' and meta.get('images_json'): _table_image_count += 1 has_image_data = True # 从检索文本中提取图表引用（见表2.2、见图2.5 等） # 重要：只从语义相关的 top 5 文本块提取，避免不相关引用干扰 referenced_figures = {} # {图号: set(文件来源)} referenced_tables = {} # {表号: set(文件来源)} for ctx in contexts[:5]: doc_text = ctx.get('doc', '') source = ctx.get('meta', {}).get('source', '') fig_refs = re.findall(r'(?:[见如])?图\s*(\d+\.?\d*)', doc_text) for fig_num in fig_refs: if fig_num not in referenced_figures: referenced_figures[fig_num] = set() if source: referenced_figures[fig_num].add(source) table_refs = re.findall(r'(?:[见如])?表\s*(\d+\.?\d*)', doc_text) for table_num in table_refs: if table_num not in referenced_tables: referenced_tables[table_num] = set() if source: referenced_tables[table_num].add(source) has_referenced_figures = bool(referenced_figures or referenced_tables) # 获取检索结果中涉及的主要文件来源 primary_sources = set() for ctx in contexts[:5]: source = ctx.get('meta', {}).get('source', '') if source: primary_sources.add(source) # 动态预算：根据检索结果数据驱动设置 # 后置过滤 _filter_images_by_answer 会根据回答内容精准裁剪 if has_figure_query: # 精确查图：用户指定了具体图号 MAX_IMAGES = 2 MIN_SCORE = 5.0 elif has_image_data: # 检索结果中有图片数据 → 宽松预算，给后置过滤留足候选空间 MAX_IMAGES = 5 MIN_SCORE = 2.0 elif has_referenced_figures: # 检索文本中引用了图表编号 MAX_IMAGES = 3 MIN_SCORE = 2.0 else: # 无图片数据 → 保守默认值 MAX_IMAGES = 2 MIN_SCORE = 3.0 # 动态调整：当表格嵌入大量图片时，提升上限以展示完整内容 if has_image_data and _table_image_count > 0: total_table_images = 0 for ctx in contexts: meta = ctx.get('meta', {}) if meta.get('chunk_type') == 'table' and meta.get('images_json'): try: total_table_images += len(json.loads(meta['images_json'])) except (json.JSONDecodeError, TypeError): pass if total_table_images > MAX_IMAGES: MAX_IMAGES = min(total_table_images, 15) # 上限 15，防止图片过多 # 获取检索结果中涉及的主要章节路径（只看前 3 个最相关的文本块） primary_section_paths = set() for ctx in contexts[:3]: section = ctx.get('meta', {}).get('section', '') or ctx.get('meta', {}).get('section_path', '') if section: primary_section_paths.add(section) # ========== P1.5 预计算：表格主题相关性评分 ========== # 从查询中提取关键词片段（使用 jieba 分词 + 2字及以上的词，自动过滤停用词） try: import jieba _query_kw_segments = [w for w in jieba.lcut(query) if len(w) >= 2 and re.search(r'[\u4e00-\u9fff]', w)] except ImportError: # jieba 不可用时回退到 bigram _chars = re.findall(r'[\u4e00-\u9fff]', query) _query_kw_segments = [_chars[i] + _chars[i+1] for i in range(len(_chars) - 1)] # 对所有含 images_json 的表格切片计算主题匹配分 _table_topic_scores = {} for _tc in contexts: _tm = _tc.get('meta', {}) if _tm.get('chunk_type') == 'table' and _tm.get('images_json'): _t_section = (_tm.get('section', '') or _tm.get('section_path', '')) _t_title = _tm.get('title', '') or '' _t_combined = _t_title + _t_section _score = sum(1 for kw in _query_kw_segments if kw in _t_combined) _table_topic_scores[id(_tc)] = _score # 自适应阈值：要求至少匹配 70% 的最佳表格得分（至少 2 分） # 例如查询 "设施设备的参考样式" → 4 个关键词 → 最佳匹配 4 → 阈值 max(2, 2) = 2 # "形象识别标识" 只匹配 "参考"+"样式" = 2 → 但阈值=3(70%of4)时被过滤 _best_table_score = max(_table_topic_scores.values()) if _table_topic_scores else 0 _table_topic_threshold = max(2, round(_best_table_score * 0.7)) if _best_table_score >= 2 else 0 scored_images = [] for ctx in contexts: meta = ctx.get('meta', {}) chunk_type = meta.get('chunk_type', 'text') s = None # P1 评分，用于 P1.5 继承 doc = ctx.get('doc', '') # 默认文档内容 # 处理图片类型和有关联图片的表格类型 if meta.get('image_path') and chunk_type in ('image', 'chart', 'table'): # 优先使用 VLM 详细描述 # 1. lazy_enhance 对 image/chart 更新 ctx['doc'] # 2. lazy_enhance 对 table 更新 ctx['image_description']（而非 doc） doc = ctx.get('image_description', '') or meta.get('vlm_desc', '') or ctx.get('doc', '') # 注入 rerank 分数到 meta，供 score_image_relevance 使用 # （rerank 分数存储在 ctx 顶层而非 meta 中） meta['score'] = ctx.get('score', 0) s = score_image_relevance(query, meta, doc) # ========== VLM 相关性筛选（方案 C）========== # 用 VLM 描述判断图片内容是否与查询相关 # 优先级：meta.vlm_desc（已同步） > .data/cache/vlm/（懒加载缓存） image_path = meta.get('image_path', '') vlm_desc = meta.get('vlm_desc', '') or _get_vlm_cache(image_path) if vlm_desc: vlm_relevance = _check_vlm_relevance(query, vlm_desc) if vlm_relevance < 0.3: # VLM 描述与查询不相关，适度降分 s -= 3.0 logger.debug(f"图片 {image_path} VLM 不相关，降分: {vlm_relevance:.2f}") elif vlm_relevance >= 0.5: # 相关，小幅加分 s += 2.0 # 图片来源 img_source = meta.get('source', '') # 图片章节 img_section = meta.get('section', '') or meta.get('section_path', '') # ========== 章节关联检测：基于层级相似度，无需硬编码格式假设 ========== # 计算图片章节与主要检索结果的最高相似度 section_penalty = 0.0 max_section_sim = 0.0 if primary_section_paths: for ps in primary_section_paths: sim = _section_similarity(img_section, ps) max_section_sim = max(max_section_sim, sim) # 当相似度低于阈值且有足够的章节信息时，判定为不相关 if primary_section_paths and img_section and max_section_sim < 0.3: # 图片章节与主要检索结果不匹配，惩罚 section_penalty = -5.0 # 除非图片被文本切片明确引用 is_referenced = False for fig_num in referenced_figures: if f"图{fig_num}" in doc or f"图 {fig_num}" in doc: is_referenced = True break if not is_referenced: for table_num in referenced_tables: if f"表{table_num}" in doc or f"表 {table_num}" in doc: is_referenced = True break if is_referenced: section_penalty = 0.0 # 被引用则不惩罚 s += section_penalty # 新增：如果图片描述中包含检索文本引用的图号，大幅加分 # 前提：图片章节与主要检索结果的章节相关 if referenced_figures: for fig_num, sources in referenced_figures.items(): # 只检查 doc 字段，不检查 meta（避免 section 中的误匹配） if f"图{fig_num}" in doc or f"图 {fig_num}" in doc: # 使用层级相似度判断章节关联性 section_match = max_section_sim >= 0.3 # 章节匹配时才加分 if section_match: # 图号匹配加分 s += 8.0 # 如果图片来源与引用来源一致，额外加分 if img_source in sources: s += 5.0 # 文件匹配额外加分 break # 新增：如果表格描述中包含检索文本引用的表号，大幅加分 # 同样要求章节相关性 if referenced_tables: for table_num, sources in referenced_tables.items(): # 只检查 doc 字段 if f"表{table_num}" in doc or f"表 {table_num}" in doc: # 使用层级相似度判断章节关联性 section_match = max_section_sim >= 0.3 # 章节匹配时才加分 if section_match: # 表号匹配加分 s += 8.0 # 如果图片来源与引用来源一致，额外加分 if img_source in sources: s += 5.0 # 文件匹配额外加分 break # 新增：如果图片来源在主要检索结果中，加分 if img_source in primary_sources: s += 2.0 if s >= MIN_SCORE: scored_images.append({ 'score': s, 'id': os.path.basename(meta['image_path']), 'chunk_id': meta.get('chunk_id', ''), 'url': f"/images/{os.path.basename(meta['image_path'])}", 'type': meta['chunk_type'], 'source': meta.get('source'), 'page': meta.get('page'), 'section_path': meta.get('section', '') or meta.get('section_path', ''), 'description': doc[:100], # 短描述用于 UI 展示 'full_description': doc # Bug 6b 修复：完整描述用于 LLM 上下文 }) # ========== P1.5：处理表格切片的 images_json（跨页表格多图）========== # 当表格切片有 images_json 字段时，添加所有关联图片 if chunk_type == 'table' and meta.get('images_json'): # 如果 P1 未执行（表格无 image_path），需要独立计算分数 if s is None: doc = ctx.get('image_description', '') or meta.get('vlm_desc', '') or ctx.get('doc', '') meta['score'] = ctx.get('score', 0) s = score_image_relevance(query, meta, doc) # 章节相关性过滤：使用层级相似度，无需硬编码章节格式假设 table_section = meta.get('section', '') or meta.get('section_path', '') section_relevant = True if primary_section_paths and table_section: # 计算表格章节与所有主要章节的最高相似度 max_sim = max( (_section_similarity(table_section, ps) for ps in primary_section_paths), default=0.0 ) if max_sim < 0.3: section_relevant = False elif primary_section_paths and not table_section: # 表格无章节信息时优雅降级：不做章节过滤，仅依赖主题分数 pass if not section_relevant: # 例外：如果表格标题/内容被查询直接提及，仍视为相关 table_title = meta.get('title', '') or '' table_doc = ctx.get('doc', '') or '' if table_title and table_title in query: section_relevant = True elif table_doc and any(kw in table_doc for kw in query.split() if len(kw) >= 2): section_relevant = True if not section_relevant: logger.debug(f"P1.5 跳过无关表格图片: path={table_section}, title={meta.get('title', '')}") continue # 跳过此表格的所有嵌入图片 # 标题/主题相关性过滤：基于预计算的关键词匹配评分 # 如果最佳表格得分 >= 2，则过滤掉得分为 0 的表格 _this_topic_score = _table_topic_scores.get(id(ctx), 0) if _this_topic_score < _table_topic_threshold: logger.debug(f"P1.5 跳过主题不匹配表格: topic_score={_this_topic_score}, threshold={_table_topic_threshold}, title={meta.get('title', '')}") continue try: images_list = json.loads(meta['images_json']) for img_info in images_list: if isinstance(img_info, dict): img_id = img_info.get('id') or img_info.get('path', '') img_page = img_info.get('page', meta.get('page')) else: img_id = str(img_info) img_page = meta.get('page') # 跳过已添加的图片（避免重复） existing_ids = {img['id'] for img in scored_images} if img_id and img_id not in existing_ids: # 为关联图片计算分数（继承主表格分数，略低） assoc_score = s - 1.0 if s >= MIN_SCORE else MIN_SCORE - 1.0 if assoc_score >= MIN_SCORE: scored_images.append({ 'score': assoc_score, 'id': img_id, 'chunk_id': meta.get('chunk_id', ''), 'url': f"/images/{img_id}", 'type': 'table_image', 'source': meta.get('source'), 'page': img_page, 'section_path': meta.get('section', '') or meta.get('section_path', ''), 'description': doc[:100], 'full_description': doc }) except (json.JSONDecodeError, TypeError): pass # ========== P2：通过文本切片的 referenced_images 补充图片 ========== # 检查 top 5 文本切片的 referenced_images，补充未选中的关联图片 existing_image_ids = {img['id'] for img in scored_images} for ctx in contexts[:5]: meta = ctx.get('meta', {}) if meta.get('chunk_type') != 'text': continue referenced = meta.get('referenced_images', []) if not referenced: continue # 查找对应的图片切片 for fig_num in referenced: # 在所有 contexts 中查找匹配的图片 for img_ctx in contexts: img_meta = img_ctx.get('meta', {}) if img_meta.get('chunk_type') not in ('image', 'chart', 'table'): continue img_path = img_meta.get('image_path', '') img_id = os.path.basename(img_path) # 检查是否已存在 if img_id in existing_image_ids: continue # 检查图号/表号是否匹配 img_doc = img_ctx.get('doc', '') if f"图{fig_num}" in img_doc or f"表{fig_num}" in img_doc: # 添加到结果中 scored_images.append({ 'score': 8.0, # 基础分 'id': img_id, 'chunk_id': img_meta.get('chunk_id', ''), 'url': f"/images/{img_id}", 'type': img_meta.get('chunk_type'), 'source': img_meta.get('source'), 'page': img_meta.get('page'), 'section_path': img_meta.get('section', '') or img_meta.get('section_path', ''), 'description': img_doc[:100], # 短描述用于 UI 展示 'full_description': img_doc # Bug 6b 修复：完整描述用于 LLM 上下文 }) existing_image_ids.add(img_id) break # ========== P3: CrossEncoder 语义精排 ========== # 用 reranker 对 (query, image_description) 对做语义评分 # 策略：CE < 0 直接剔除（语义不相关），CE 0~2 保留不加分（弱相关），CE > 2 加分 if scored_images: try: from core.engine import get_engine engine = get_engine() if engine.reranker: # 为每张图片选取最佳描述文本（VLM > full_description > description） # 截断到 512 字符（bge-reranker-base token 上限） pairs = [] for img in scored_images: desc = img.get('full_description', '') or img.get('description', '') or '' if not desc: desc = img.get('id', '') # fallback: 图片文件名 if len(desc) > 512: desc = desc[:512] pairs.append((query, desc)) ce_scores = engine.reranker.predict(pairs) # 分阶段处理：先标记 CE 分数，再过滤 kept_images = [] for img, ce_raw in zip(scored_images, ce_scores): ce_score = float(ce_raw) img['_ce_score'] = round(ce_score, 3) if ce_score < 0: # CE 负分：语义不相关，直接剔除 img['_ce_adj'] = 'removed' logger.debug(f"CE 剔除: {img.get('id','')} (ce={ce_score:.2f})") elif ce_score < 2: # CE 0~2：弱相关，保留但不加分，交给后置过滤判断 img['_ce_adj'] = 0.0 kept_images.append(img) else: # CE > 2：强相关，加分 adjustment = min((ce_score - 2) / 3.0, 1.0) * 5.0 img['score'] = img['score'] + adjustment img['_ce_adj'] = round(adjustment, 2) kept_images.append(img) removed = len(scored_images) - len(kept_images) if removed > 0: logger.debug(f"CrossEncoder 图片精排: 剔除 {removed} 张负分图片") scored_images = kept_images except Exception as e: logger.debug(f"CrossEncoder 图片精排跳过: {e}") scored_images.sort(key=lambda x: x['score'], reverse=True) return scored_images[:MAX_IMAGES] def chat_with_llm(message: str, history: List[Dict] = None, enable_web_search: bool = True) -> Dict[str, Any]: """ 普通聊天 - 使用 LLM 直接回复 当查询不需要知识库检索时，直接调用 LLM 进行回复。 可选启用网络搜索增强。 Args: message: 用户消息 history: 对话历史（由后端传入） enable_web_search: 是否启用网络搜索 Returns: { "answer": str, "sources": list, "web_searched": bool } Example: >>> result = chat_with_llm("你好", enable_web_search=False) >>> print(result["answer"]) """ from config import get_llm_client, LLM_MAX_TOKENS client = get_llm_client() # 构建消息 messages = [] # 添加历史 if history: for h in history[-MAX_HISTORY_ROUNDS:]: messages.append({"role": h["role"], "content": h["content"]}) messages.append({"role": "user", "content": message}) # 调用 LLM answer = call_llm(client, prompt="", model=RAG_CHAT_MODEL, messages=messages, max_tokens=LLM_MAX_TOKENS) return { "answer": answer or "", "sources": [], "web_searched": False } def search_hybrid(query: str, top_k: int = 5, allowed_levels: list = None, allowed_collections: list = None, sub_queries: list = None): """ 混合检索：直接调用生产环境引擎，确保测试效果与生产一致 Args: query: 查询文本 top_k: 返回数量 allowed_levels: 允许的安全级别 allowed_collections: 允许的向量库列表 sub_queries: 意图分析器生成的子查询列表（对比类查询用） Returns: 融合后的检索结果 """ from core.engine import get_engine engine = get_engine() # 直接调用生产环境的检索方法 result = engine.search_knowledge( query=query, top_k=top_k, allowed_levels=allowed_levels, collections=allowed_collections, sub_queries=sub_queries ) # 添加 scores 字段（用于前端显示和上下文过滤） if result and result.get('ids') and result['ids'][0]: distances = result.get('distances', [[]])[0] if result.get('_reranked'): # Rerank 后 distances 中存储的是相关性分数（越高越好），直接使用 scores = [float(d) for d in distances] else: # 未 Rerank：将向量距离转换为相似度分数（距离越小，分数越高） scores = [1.0 - d if d <= 1.0 else 1.0 / (1.0 + d) for d in distances] result['scores'] = [scores] else: result = { 'ids': [[]], 'documents': [[]], 'metadatas': [[]], 'distances': [[]], 'scores': [[]] } return result # ==================== 路由 ==================== @chat_bp.route('/chat', methods=['POST']) @require_gateway_auth def chat(): """ 普通聊天模式 - 直接使用LLM回复 请求体: { "message": "消息内容", "history": [{"role": "user/assistant", "content": "..."}] // 可选 } """ data = request.json or {} message = data.get('message') history = data.get('history', []) if not message: return error_response("MISSING_PARAMS", BAD_REQUEST, "缺少 message", http_status=400) # 输入安全验证 is_valid, reason = validate_query(message) if not is_valid: return error_response("INVALID_QUERY", BAD_REQUEST, reason, http_status=400) # 智能聊天 result = chat_with_llm(message, history) # 过滤敏感信息 answer = filter_response(result["answer"]) return success_response(data={ "answer": answer, "mode": "chat", "sources": result.get("sources", []), "web_searched": result.get("web_searched", False) }) @chat_bp.route('/rag', methods=['POST']) @require_gateway_auth def rag(): """ 知识库问答模式 - SSE 流式返回 请求体: { "message": "消息内容", "history": [{"role": "user/assistant", "content": "..."}], // 可选（开发环境） "chat_history": [{"role": "user/assistant", "content": "..."}], // 可选（生产环境） "collections": ["public_kb"], // 可选，知识库列表 "session_id": "xxx" // 可选，会话ID } SSE 事件序列: 1. start: 开始处理 2. sources: 检索到的来源 3. chunk: 每个 token 4. finish: 完成响应（包含完整 answer 和 sources） 5. error: 错误事件 """ import re from config import ( IS_PROD, IS_DEV, ENABLE_SESSION, RAG_SEARCH_TOP_K, MAX_CONTEXT_CHUNKS, MAX_SOURCES_RETURNED, LLM_TEMPERATURE, LLM_MAX_TOKENS, MAX_HISTORY_ROUNDS, IMAGE_CONTEXT_HISTORY, DIRECT_CONTEXT_MAX_CHARS, RERANK_CONTEXT_MIN_SCORE, CONTEXT_MAX_CHARS, CONTEXT_SOFT_LIMIT, CONFIDENCE_WARN_THRESHOLD, CONFIDENCE_CAUTION_THRESHOLD, SECTION_CLUSTER_RESCUE_ENABLED, CLUSTER_MIN_MEMBERS, CLUSTER_MIN_TYPES, CLUSTER_RESCUE_FLOOR, CLUSTER_MAX_SECTIONS, CLUSTER_MAX_RESCUE_PER_SECTION, CLUSTER_SECTION_PREFIX_LEVELS ) data = request.json or {} message = data.get('message') # 兼容两种参数名：history（旧）和 chat_history（新） # 用 is not None 判断，避免 [] 被 or 吞掉 if 'chat_history' in data: history = data['chat_history'] elif 'history' in data: history = data['history'] else: history = None collections = data.get('collections') session_id = data.get('session_id') if not message: return error_response("MISSING_PARAMS", BAD_REQUEST, "缺少 message", http_status=400) # 生产环境强制校验 chat_history if IS_PROD and history is None: return error_response("MISSING_HISTORY", BAD_REQUEST, "chat_history is required in production", http_status=400) # 输入安全验证 is_valid, reason = validate_query(message) if not is_valid: return error_response("INVALID_QUERY", BAD_REQUEST, reason, http_status=400) # 如果没有指定 collections，使用默认的公开库 if not collections: collections = ['public_kb'] # ==================== 会话历史加载 ==================== # 优先使用传入的 history，否则从 session_repo 加载 user_id = request.current_user.get("user_id") if history is not None: # 使用传入的历史（生产环境必须传入） pass elif session_id and ENABLE_SESSION: # 开发环境：从本地数据库加载 try: session_repo = current_app.session_repo history = session_repo.get_history(session_id) # 限制历史长度 history = history[-MAX_HISTORY_ROUNDS:] if len(history) > MAX_HISTORY_ROUNDS else history except Exception as e: logger.debug(f"解析历史记录失败: {e}") history = [] else: history = [] # 如果没有 session_id，创建新会话（仅开发环境） if not session_id and ENABLE_SESSION: try: session_repo = current_app.session_repo session_id = session_repo.create_session(user_id) except Exception as e: logger.debug(f"创建会话失败: {e}") # ==================== collections 历史推断（开发环境） ==================== # 当 collections 未显式指定（或仅为默认 public_kb）且有会话历史时， # 从历史消息中推断上次使用的 KB，自动恢复以避免用户忘切 KB if (not collections or collections == ['public_kb']) and history: for _msg in reversed(history): if _msg.get("role") == "assistant": _meta = _msg.get("metadata", {}) if isinstance(_meta, dict) and _meta.get("collections"): collections = _meta["collections"] logger.info(f"[KB推断] 从历史推断 collections: {collections}") break # 提前获取 session_repo 引用，避免在生成器内部访问 current_app # （生成器执行时应用上下文可能已结束） session_repo_ref = None if ENABLE_SESSION: try: session_repo_ref = current_app.session_repo except Exception as e: logger.debug(f"获取会话仓库失败: {e}") def generate(): """生成 SSE 流""" import re start_time = _time.time() full_answer = [] try: # 0. 意图分析（改写 + 双层判断） context_images = [] if history: # 从历史中提取图片信息 for msg in reversed(history[-IMAGE_CONTEXT_HISTORY:]): metadata = msg.get("metadata", {}) if isinstance(metadata, dict): images = metadata.get("images", []) if images: context_images.extend(images[:3]) intent = None try: from core.intent_analyzer import analyze_intent intent = analyze_intent(message, history or [], context_images) logger.info(f"[意图分析] use_context={intent.use_context}, need_retrieval={intent.need_retrieval}, intent={intent.intent}, sub_queries={intent.sub_queries}") # 调试事件：意图分析结果 if IS_DEV: yield f"data: {json.dumps({'type': 'intent_result', 'data': {'intent': intent.intent, 'confidence': round(intent.confidence, 2), 'rewritten_query': intent.rewritten_query, 'sub_queries': intent.sub_queries, 'need_retrieval': intent.need_retrieval, 'reason': intent.reason}}, ensure_ascii=False)}\n\n" # 如果不需要检索，直接使用上下文回答 if not intent.need_retrieval and intent.use_context: yield f"data: {json.dumps({'type': 'start', 'message': '正在分析...'}, ensure_ascii=False)}\n\n" # 构建上下文 context_text = "" if history: # 提取最近的助手回答 for msg in reversed(history): if msg.get("role") == "assistant": context_text = msg.get("content", "") break # 构建图片上下文 image_context = "" if context_images: image_context = "\n\n【上下文中的图片】\n" for img in context_images[:5]: if isinstance(img, dict): desc = img.get("description", "") img_type = img.get("type", "图片") image_context += f"- {img_type}: {desc}\n" # 直接调用 LLM from config import get_llm_client, DASHSCOPE_MODEL client = get_llm_client() system_prompt = f"""你是一个专业的知识库问答助手。请根据对话历史和上下文回答用户问题。 如果用户问题是关于图片的，请根据上下文中的图片描述进行分析。 {image_context}""" user_prompt = f"""对话历史： {context_text[:DIRECT_CONTEXT_MAX_CHARS] if context_text else '（无历史上下文）'} 用户问题：{intent.rewritten_query} 请直接回答用户问题。""" # 流式生成回答 for content in call_llm_stream( client, prompt=user_prompt, model=DASHSCOPE_MODEL, messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], temperature=LLM_TEMPERATURE ): full_answer.append(content) yield f"data: {json.dumps({'type': 'chunk', 'content': content}, ensure_ascii=False)}\n\n" # 发送完成事件 yield f"data: {json.dumps({'type': 'finish', 'answer': ''.join(full_answer), 'sources': []}, ensure_ascii=False)}\n\n" return # 直接返回，不执行后续检索 except Exception as e: logger.warning(f"意图分析失败: {e}，继续执行检索流程") # 构建检索查询：使用改写后的完整问题（解决追问偏离问题） retrieval_query = intent.rewritten_query if (intent and intent.rewritten_query) else message # 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'): # 缓存 key 使用 retrieval_query + collections，避免不同上下文的追问命中错误缓存 _cache_collections = ','.join(sorted(collections)) if collections else '' _cache_key_text = f"{retrieval_query}|{_cache_collections}" _semantic_cache_emb = _eng.embedding_model.encode(_cache_key_text) cached = _sc.get(_semantic_cache_emb) # 防御性校验：必须是 RAG 回答缓存（非 intent_analyzer 缓存），且回答非空 if cached is not None and cached.get("cache_type") == "rag_answer" and cached.get("answer"): logger.info(f"[语义缓存] 命中: {message[:50]}...") cached_answer = cached.get("answer", "") # 流式返回缓存的答案 yield f"data: {json.dumps({'type': 'start', 'message': '正在检索知识库...'}, ensure_ascii=False)}\n\n" # 分块发送缓存答案 chunk_size = 20 for i in range(0, len(cached_answer), chunk_size): chunk = cached_answer[i:i+chunk_size] full_answer.append(chunk) yield f"data: {json.dumps({'type': 'chunk', 'content': chunk}, ensure_ascii=False)}\n\n" finish_event = { "type": "finish", "answer": cached_answer, "mode": "rag", "session_id": session_id, "sources": cached.get("sources", []), "citations": cached.get("citations", []), "images": cached.get("images", []), "tables": cached.get("tables", []), "sections": [], "duration_ms": int((_time.time() - start_time) * 1000), "confidence_score": 1.0, "semantic_cache_hit": True } yield f"data: {json.dumps(finish_event, ensure_ascii=False)}\n\n" return 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" # 2. 执行混合检索（扩大召回数量，确保图片切片有机会被召回） # 如果意图分析生成了子查询（对比类），传给搜索引擎并行检索 sub_queries = None if intent and intent.sub_queries and len(intent.sub_queries) > 1: sub_queries = intent.sub_queries search_result = search_hybrid( retrieval_query, top_k=RAG_SEARCH_TOP_K, allowed_collections=collections, sub_queries=sub_queries ) # 调试事件：检索管线详情 if IS_DEV: debug_info = search_result.get('_debug', {}) yield f"data: {json.dumps({'type': 'retrieval_debug', 'data': {'steps': debug_info.get('steps', []), 'collections_searched': collections, 'total_candidates': len(search_result.get('ids', [[]])[0])}}, ensure_ascii=False)}\n\n" # 提取章节聚类提升事件（引擎层）单独发送 for step in debug_info.get('steps', []): if step.get('name') == 'section_cluster_boost': yield f"data: {json.dumps({'type': 'section_cluster_boost', 'data': step}, ensure_ascii=False)}\n\n" # 提取上下文 contexts = [] sources = [] if search_result.get('documents') and search_result['documents'][0]: docs = search_result['documents'][0] metas = search_result.get('metadatas', [[]])[0] scores = search_result.get('scores', [[]])[0] # 图片相关性提升：数据驱动检测（无硬编码关键词） import re figure_pattern = r'图\s*(\d+\.?\d*)' figure_matches = re.findall(figure_pattern, retrieval_query) has_figure_query = bool(figure_matches) # 数据驱动：检查检索结果中是否有图片/图表类型切片 has_image_data = any( m.get('chunk_type') in ('image', 'chart') for m in metas ) has_image_intent = has_figure_query or has_image_data # 给图片/图表切片打 boost 标记，供后续 select_images 使用 if has_image_intent: for i, (doc, meta, score) in enumerate(zip(docs, metas, scores)): if meta.get('chunk_type') in ('image', 'chart'): # 检查 caption 是否与查询相关 caption = meta.get('caption', '') or '' should_boost = False boost_factor = 1.0 # 如果查询包含图片编号，检查 caption 是否匹配 if has_figure_query: for fig_num in figure_matches: if f"图{fig_num}" in caption or f"图 {fig_num}" in caption: should_boost = True boost_factor = 2.0 # 强提升 break # 或者 caption 与查询有足够重叠 if not should_boost: overlap = len(set(message) & set(caption)) if overlap >= 3 or any(word in caption for word in message if len(word) >= 3): should_boost = True boost_factor = 1.5 # 提升 50% if should_boost: meta['_image_boost'] = boost_factor # 打标记，不重排 # 不做 sort！保持检索引擎的原始排序 # 按 source 去重，保留最高分 seen_sources = {} for rank, (doc, meta, score) in enumerate(zip(docs, metas, scores)): meta['_retrieval_rank'] = rank # 确保 _collection 字段存在（单知识库路径下 ChromaDB 原生不返回此字段） if not meta.get('_collection'): # 优先使用入库时写入的 collection 字段 meta['_collection'] = meta.get('collection') or (collections[0] if collections else 'public_kb') source_name = meta.get('source', '未知') if source_name not in seen_sources or score > seen_sources[source_name]['score']: doc_type = meta.get('doc_type', 'other') page = meta.get('page', 0) page_end = meta.get('page_end') # 仅 PDF 的页码是真实可靠的；Word 等的页码是 MinerU 合成的，无意义，不外露 if doc_type == 'pdf' and page: page_range = f"{page}-{page_end}" if (page_end and page_end > page) else str(page) else: page = None page_end = None page_range = '' seen_sources[source_name] = { 'source': source_name, 'page': page, 'page_end': page_end, 'page_range': page_range, 'section': meta.get('section', '') or meta.get('section_path', ''), 'chunk_type': meta.get('chunk_type', 'text'), 'doc_type': doc_type, # 文档类型 'section_chunk_id': meta.get('section_chunk_id'), # 章节内序号 'score': round(score, 3) if isinstance(score, float) else score } # ========== P1：图片使用 full_description ========== # 图片切片使用完整描述（用于 LLM 上下文），而非短摘要 display_doc = doc if meta.get('chunk_type') in ('image', 'chart', 'table'): full_desc = meta.get('full_description', '') if full_desc: display_doc = full_desc # contexts 仍然保留所有结果用于生成答案 contexts.append({'doc': display_doc, 'meta': meta, 'score': score}) sources = list(seen_sources.values()) # 调试事件：召回切片详情 if IS_DEV: chunks_debug = [] for i, ctx in enumerate(contexts[:20]): m = ctx.get('meta', {}) chunks_debug.append({ 'rank': i + 1, 'source': m.get('source', ''), 'page': m.get('page', 0), 'chunk_type': m.get('chunk_type', 'text'), 'section': m.get('section', ''), 'score': round(ctx.get('score', 0), 4) if ctx.get('score') else None, 'content': (ctx.get('doc', '') or '')[:300] }) yield f"data: {json.dumps({'type': 'chunks_retrieved', 'data': {'count': len(contexts), 'chunks': chunks_debug}}, ensure_ascii=False)}\n\n" # 补充检索：从文本切片中提取图号/表号引用，补充检索对应的图片 # 重要：只从最相关的 top 5 文本切片提取引用，避免不相关引用干扰 import re referenced_figures = set() referenced_tables = set() # 只检查 top 5 文本切片（与 select_images 逻辑一致） text_contexts = [ctx for ctx in contexts if ctx.get('meta', {}).get('chunk_type') == 'text'][:5] for ctx in text_contexts: doc_text = ctx.get('doc', '') fig_refs = re.findall(r'(?:[见如及和与])?图\s*(\d+\.?\d*)', doc_text) referenced_figures.update(fig_refs) table_refs = re.findall(r'(?:[见如及和与])?表\s*(\d+\.?\d*)', doc_text) referenced_tables.update(table_refs) # 检查哪些图号/表号对应的图片不在 contexts 中 existing_figure_images = set() existing_table_images = set() for ctx in contexts: doc = ctx.get('doc', '') meta = ctx.get('meta', {}) if meta.get('chunk_type') in ('image', 'chart'): for fig_num in referenced_figures: if f"图{fig_num}" in doc: existing_figure_images.add(fig_num) for table_num in referenced_tables: if f"表{table_num}" in doc: existing_table_images.add(table_num) # 需要补充检索的图号/表号 missing_figures = referenced_figures - existing_figure_images missing_tables = referenced_tables - existing_table_images # 计算主要章节路径（用于补充检索过滤） primary_section_paths_for_supp = set() for ctx in text_contexts[:3]: section = ctx.get('meta', {}).get('section', '') or ctx.get('meta', {}).get('section_path', '') if section: primary_section_paths_for_supp.add(section) if missing_figures or missing_tables: # 补充检索 from knowledge.manager import get_kb_manager kb_manager = get_kb_manager() kb_name = collections[0] if collections else 'public_kb' collection = kb_manager.get_collection(kb_name) if collection: # 构建补充查询 supplement_queries = [] for fig_num in missing_figures: supplement_queries.append(f"图{fig_num}") for table_num in missing_tables: supplement_queries.append(f"表{table_num}") supplement_query = " ".join(supplement_queries) # 使用 embedding 检索 # P4：复用 engine 的 embedding 模型，避免重复加载 try: from core.engine import get_engine engine = get_engine() query_vector = engine.embedding_model.encode(supplement_query).tolist() if isinstance(query_vector[0], list): query_vector = query_vector[0] supplement_result = collection.query( query_embeddings=[query_vector], n_results=10, include=['documents', 'metadatas', 'distances'] ) # 添加匹配的图片切片 for supp_doc, supp_meta, supp_dist in zip( supplement_result['documents'][0], supplement_result['metadatas'][0], supplement_result['distances'][0] ): chunk_type = supp_meta.get('chunk_type', '') if chunk_type in ('image', 'chart'): # 检查是否匹配缺失的图号/表号 is_match = False matched_fig = None for fig_num in missing_figures: if f"图{fig_num}" in supp_doc: is_match = True matched_fig = fig_num break for table_num in missing_tables: if f"表{table_num}" in supp_doc: is_match = True break if is_match: # 额外检查：图片章节是否与主要章节匹配 # 使用层级相似度判断，无需硬编码格式假设 supp_section = supp_meta.get('section', '') or supp_meta.get('section_path', '') if primary_section_paths_for_supp and supp_section: supp_max_sim = max( (_section_similarity(supp_section, ps) for ps in primary_section_paths_for_supp), default=0.0 ) if supp_max_sim < 0.3: continue elif primary_section_paths_for_supp and not supp_section: # 补充检索的图片无章节信息时优雅降级：跳过 continue # Bug 6a 修复：补充检索的图片也要做 full_description 替换 # 与正常检索保持一致 display_doc = supp_doc if supp_meta.get('chunk_type') in ('image', 'chart', 'table'): full_desc = supp_meta.get('full_description', '') if full_desc: display_doc = full_desc contexts.append({ 'doc': display_doc, 'meta': {**supp_meta, '_collection': supp_meta.get('_collection') or (collections[0] if collections else 'public_kb')}, 'score': 1.0 - supp_dist }) logger.info(f"[补充检索] 添加图片: {supp_meta.get('image_path', '')}") except Exception as e: logger.warning(f"补充检索失败: {e}") # 发送来源事件 yield f"data: {json.dumps({'type': 'sources', 'sources': sources[:MAX_SOURCES_RETURNED]}, ensure_ascii=False)}\n\n" # 调试：检查 contexts 中是否有图片切片 image_count = sum(1 for ctx in contexts if ctx.get('meta', {}).get('chunk_type') in ('image', 'chart')) if image_count > 0: logger.info(f"[图片检索] contexts 中包含 {image_count} 个图片/图表切片") for ctx in contexts: meta = ctx.get('meta', {}) if meta.get('chunk_type') in ('image', 'chart'): logging.info(f" - 图片: {meta.get('caption', '')[:50]}, path: {meta.get('image_path', '')}") # 2.5. 懒加载增强（Phase 4）— 异步后台生成 # 当前请求直接使用已有的 VLM 描述（meta/cache），不阻塞响应 # 后台线程异步调用 VLM/LLM 生成缺失描述，写入缓存和 ChromaDB， # 下次查询时缓存命中，响应不受影响 try: import threading from knowledge.lazy_enhance import enhance_retrieved_chunks kb_name = collections[0] if collections else 'public_kb' # 提取后台增强所需字段（替代 deepcopy，避免 100-500ms 递归拷贝） _bg_contexts = [] for _ctx in contexts: _meta = _ctx.get('meta', {}) _bg_contexts.append({ 'meta': { 'chunk_id': _meta.get('chunk_id', ''), 'image_path': _meta.get('image_path', ''), 'chunk_type': _meta.get('chunk_type', 'text'), 'has_vlm_desc': _meta.get('has_vlm_desc', False), 'has_summary': _meta.get('has_summary', False), 'section': _meta.get('section', ''), 'section_path': _meta.get('section_path', ''), 'page': _meta.get('page'), 'caption': _meta.get('caption', ''), 'source': _meta.get('source', ''), }, 'doc': _ctx.get('doc', ''), # 不截断，字符串浅引用无额外开销 'score': _ctx.get('score', 0), 'image_description': _ctx.get('image_description', ''), }) def _background_enhance(): try: import asyncio as _asyncio _asyncio.run(_asyncio.wait_for( enhance_retrieved_chunks(_bg_contexts, retrieval_query, kb_name, defer_chromadb=True), timeout=60.0 )) except Exception as e: logger.debug(f"后台 VLM 增强失败: {e}") _t = threading.Thread(target=_background_enhance, daemon=True) _t.start() except Exception as e: logger.debug(f"懒加载增强启动失败: {e}") # 3. 选择要展示的图片（Phase 5） selected_images = select_images(contexts, retrieval_query) # 调试事件：图片选择详情 if IS_DEV: yield f"data: {json.dumps({'type': 'images_selected', 'data': {'total_scored': len([c for c in contexts if c.get('meta',{}).get('chunk_type') in ('image','chart','table')]), 'selected_count': len(selected_images), 'images': [{'source': img.get('source',''), 'page': img.get('page',0), 'score': round(img.get('_image_boost', 1.0), 2)} for img in selected_images]}}, ensure_ascii=False)}\n\n" # 4. 构建 prompt（Phase 6：LLM 图片感知） # Bug 1 修复：文本切片用于 top 5 名额竞争，图片描述不参与竞争 # BM25 分歧检测救援：BM25 top-3 但 rerank 压制的切片 contexts = _rescue_bm25_divergence(contexts, search_result, RERANK_CONTEXT_MIN_SCORE) # 词法匹配救援：当切片文本精确包含查询关键词但 CrossEncoder 评分低时， # 提升分数使其通过 min_score 过滤（适用于独立切片无法触发聚类救援的场景） contexts = _rescue_lexical_match(contexts, retrieval_query, RERANK_CONTEXT_MIN_SCORE) # 章节聚类救援（路由层安全网）：在 min_score 过滤前， # 检测"全灭 section"并分配保底分数 _rescue_debug = None if SECTION_CLUSTER_RESCUE_ENABLED: _before_scores = {id(ctx): ctx.get('score', 0) for ctx in contexts} contexts = _rescue_section_cluster(contexts, retrieval_query, RERANK_CONTEXT_MIN_SCORE) # 统计救援结果 _rescued = [ctx for ctx in contexts if id(ctx) in _before_scores and ctx.get('score', 0) > _before_scores[id(ctx)]] if _rescued and IS_DEV: _rescue_sections = set() for ctx in _rescued: meta = ctx.get('meta', {}) sp = meta.get('section', '') or meta.get('section_path', '') _rescue_sections.add(_normalize_section_for_rescue(sp, CLUSTER_SECTION_PREFIX_LEVELS)) _rescue_debug = { 'rescued_count': len(_rescued), 'rescued_sections': list(_rescue_sections) } yield f"data: {json.dumps({'type': 'section_cluster_rescue', 'data': _rescue_debug}, ensure_ascii=False)}\n\n" text_contexts = _order_text_contexts_for_prompt(contexts, retrieval_query, MAX_CONTEXT_CHUNKS, min_score=RERANK_CONTEXT_MIN_SCORE) # Phase 2：按字符预算构建上下文 _is_comparison = intent and intent.intent == "comparison" if _is_enum_query(retrieval_query) or _is_comparison: # 列举类 / 对比类查询：保持原始顺序，不做预算截断 # 仍然注入 section_path 标题，帮助 LLM 区分不同章节 enum_parts = [] prev_section = None for ctx in text_contexts: meta = ctx.get('meta', {}) section = meta.get('section', '') or meta.get('section_path', '') if section and section != prev_section: enum_parts.append(f"━ {section} ━") prev_section = section # 精简表格切片：去除冗余的语义增强前缀 doc = _strip_semantic_prefix(ctx.get('doc', ''), meta.get('chunk_type', '')) enum_parts.append(doc) context_text = "\n\n".join(enum_parts) else: context_text = _build_context_with_budget(text_contexts, CONTEXT_MAX_CHARS, CONTEXT_SOFT_LIMIT) # 表格救援：CrossEncoder 对表格评分偏低，导致表格被预算截断 # 当查询涉及表格但上下文中没有表格数据时，从被截断的切片中补回 context_text = _rescue_table_chunks(text_contexts, context_text, retrieval_query) # Phase 4：计算置信度分数（top-3 平均 Rerank 分数） _top_scores = [ctx.get('score', 0) for ctx in text_contexts[:3]] _confidence_score = round(sum(_top_scores) / len(_top_scores), 4) if _top_scores else 0.0 # Bug 6b 优化：直接使用 selected_images 中的 full_description # 这样 LLM 既能看到文本切片，也能知道图片内容 if selected_images: # 区分表格嵌入图片和独立图片 has_table_embedded_images = any( img.get('type') == 'table_image' for img in selected_images ) # 检查上下文中是否有表格含嵌入图片 has_table_with_images = any( ctx.get('meta', {}).get('chunk_type') == 'table' and ctx.get('meta', {}).get('images_json') for ctx in text_contexts ) image_descriptions = [] for i, img in enumerate(selected_images, 1): # 直接使用 select_images 时带上的 full_description 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_descriptions.append(f"【图片{i}】{full_desc}{source_info}") if image_descriptions: context_text += "\n\n【相关图片信息】\n" + "\n\n".join(image_descriptions) # 根据图片类型给出不同的回答指令 if has_table_embedded_images and has_table_with_images: context_text += ( "\n\n【回答要求】参考资料中包含表格及其嵌入图片。" "请以**表格形式**呈现数据（保持原始表格结构），" "并在对应单元格中使用 `![图片](图片URL)` 格式嵌入图片。" "不要将表格内容转为纯文本描述，不要把图片与表格分开展示。" ) else: # 添加指令让 LLM 介绍图片 context_text += "\n\n【回答要求】回答时请简要介绍每张图片的内容和用途。" enhanced_context = context_text # 对比类查询：添加结构化对比指令 if intent and intent.intent == "comparison" and intent.sub_queries: comparison_instruction = ( "\n\n【回答要求】这是一个对比类问题。" "请根据参考资料，从多个角度对比分析，" "使用表格或分点形式清晰呈现差异和共同点。" "如果参考资料中缺少某一方面的信息，请如实说明。" ) enhanced_context = comparison_instruction + "\n\n" + enhanced_context # 推理类查询：添加因果分析指令 elif intent and intent.intent == "reasoning": reasoning_instruction = ( "\n\n【回答要求】这是一个需要分析原因或推理的问题。" "请根据参考资料，先梳理相关事实和数据，再给出逻辑清晰的分析。" "如果涉及因果关系，请明确标注原因和结果；如果资料不足以支撑推理，请如实说明。" ) enhanced_context = reasoning_instruction + "\n\n" + enhanced_context # 操作指导类查询：添加步骤化指令 elif intent and intent.intent == "instruction": instruction_instruction = ( "\n\n【回答要求】这是一个操作指导类问题。" "请根据参考资料，以清晰的步骤或流程形式组织回答。" "如有前置条件或注意事项，请在步骤前说明。" ) enhanced_context = instruction_instruction + "\n\n" + enhanced_context if _is_enum_query(retrieval_query): enum_instruction = ( "\n\n【回答要求】如果参考资料中包含编号列表、禁止情形、要求或条款，" "请按资料中的原始顺序完整列出；不要合并相邻条目，不要跳项，" "资料不足时明确说明缺少哪部分依据。\n\n" ) enhanced_context = enum_instruction + enhanced_context # Phase 4：根据置信度注入谨慎回答指令 if _confidence_score < CONFIDENCE_WARN_THRESHOLD: enhanced_context += ( "\n\n【重要提示】参考资料与问题的相关性较低。" "请仅基于参考资料中明确包含的信息回答，" '如果资料不足以回答问题，请直接说明"知识库中未找到直接相关的信息"。' ) elif _confidence_score < CONFIDENCE_CAUTION_THRESHOLD: enhanced_context += ( "\n\n【提示】参考资料的相关性一般，请优先引用资料中的原文，避免推测。" ) # 调试事件：最终上下文 if IS_DEV: text_used = text_contexts _scores = [round(ctx.get('score', 0), 4) for ctx in text_used] yield f"data: {json.dumps({'type': 'context_built', 'data': {'chunk_count': len(text_used), 'context_length': len(enhanced_context), 'budget_max_chars': CONTEXT_MAX_CHARS, 'min_score_filter': RERANK_CONTEXT_MIN_SCORE, 'confidence_top3': _confidence_score, 'score_stats': {'max': max(_scores) if _scores else 0, 'min': min(_scores) if _scores else 0, 'avg': round(sum(_scores)/len(_scores), 4) if _scores else 0}, 'context_preview': enhanced_context[:500], 'chunks_used': [{'source': ctx.get('meta',{}).get('source',''), 'page': ctx.get('meta',{}).get('page',0), 'score': ctx.get('score',0), 'preview': (ctx.get('doc','') or '')[:100]} for ctx in text_used]}}, ensure_ascii=False)}\n\n" # 5. 流式生成回答 from core.engine import get_engine engine = get_engine() for token in engine.generate_answer_stream(message, enhanced_context, history): full_answer.append(token) yield f"data: {json.dumps({'type': 'chunk', 'content': token}, ensure_ascii=False)}\n\n" # 6. P0：答案对齐过滤器 # 从 LLM 回答中提取图号引用，过滤图片选择结果 full_answer_text = "".join(full_answer) # 提取回答中引用的图号/表号 mentioned = set() # 中文图号：图2.1、图 2-1、见图2.1 等 mentioned.update(re.findall(r'(?:[见如])?图\s*(\d+[\.\-]?\d*)', full_answer_text)) # 中文表号：表2.1、表 2-1、见表2.1 等 mentioned.update(re.findall(r'(?:[见如])?表\s*(\d+[\.\-]?\d*)', full_answer_text)) # 英文图号：Figure 2.1、Fig.2.1 等 mentioned.update(re.findall(r'(?:Fig(?:ure)?\.?\s*)(\d+[\.\-]?\d*)', full_answer_text, re.I)) # 根据回答中的引用过滤图片 if mentioned: aligned_images = [] for img in selected_images: desc = img.get('description', '') # 标准化图号格式（将连字符转为点） for ref in mentioned: ref_normalized = ref.replace('-', '.') if (f"图{ref_normalized}" in desc or f"表{ref_normalized}" in desc or f"图 {ref_normalized}" in desc or f"表 {ref_normalized}" in desc): aligned_images.append(img) break # 如果有匹配的图片，使用对齐后的结果 if aligned_images: selected_images = aligned_images # else: 没有匹配到，保留原选择（不再截断到1张） # else: LLM 没有提图号，保留原选择（不再截断到1张） # 后置图片过滤：用回答内容反向筛选图片，确保图片与回答一致 # query 拼接 retrieval_query + message，防止意图改写丢失图片意图关键词 # 提取主要检索章节路径（top-10 文本切片），用于子章节级图片过滤 _primary_sections = list(set( ctx.get('meta', {}).get('section', '') or ctx.get('meta', {}).get('section_path', '') for ctx in text_contexts[:10] if ctx.get('meta', {}).get('section', '') or ctx.get('meta', {}).get('section_path', '') )) selected_images = _filter_images_by_answer( selected_images, full_answer_text, query=f"{retrieval_query} {message}", primary_sections=_primary_sections ) rich_media = {'images': selected_images, 'tables': [], 'sections': []} # 7. 去掉 LLM 添加的数字引用标记，避免与后端引用重复 clean_answer = re.sub(r'\[\d+\]', '', full_answer_text) # 8. 添加引用标注（自动插入 [ref:chunk_id]） citation_result = _attach_citations(clean_answer, contexts) # 9. 过滤敏感信息（违禁词等） filtered_answer = filter_response(citation_result.get("answer_with_refs", clean_answer)) # 9.5 替换表格内 [图片] 占位符为 markdown 图片语法 if rich_media.get("images"): filtered_answer = _replace_table_image_placeholders(filtered_answer, rich_media["images"]) # 9. 保存消息到会话（仅开发环境） if session_id and session_repo_ref: try: # 保存用户消息 session_repo_ref.add_message(session_id, 'user', message) # 保存 AI 回答（包含完整 metadata：图片、来源、引用等） assistant_metadata = { 'is_rag': True, 'mode': 'rag', } if rich_media.get('images'): assistant_metadata['images'] = rich_media['images'] if sources: assistant_metadata['sources'] = sources if citation_result.get('citations'): assistant_metadata['citations'] = citation_result['citations'] # 记录本次检索使用的向量库（用于后续追问时自动恢复 KB 选择） assistant_metadata['collections'] = collections session_repo_ref.add_message(session_id, 'assistant', filtered_answer, assistant_metadata) # 更新会话最后活跃时间 if hasattr(session_repo_ref, 'update_last_active'): session_repo_ref.update_last_active(session_id) except Exception as e: logger.warning(f"保存会话消息失败: {e}") # 10. 发送完成事件 duration_ms = int((_time.time() - start_time) * 1000) finish_event = { "type": "finish", "answer": filtered_answer, "mode": "rag", "session_id": session_id, "sources": sources, "citations": citation_result.get("citations", []), # 结构化引用列表 "images": rich_media["images"], "tables": rich_media["tables"], "sections": rich_media["sections"], "duration_ms": duration_ms, "confidence_score": _confidence_score # Phase 4：top-3 平均 Rerank 分数 } # 添加分阶段耗时信息（仅开发环境） if IS_DEV and hasattr(search_result, 'get'): debug_info = search_result.get('_debug', {}) timing_info = debug_info.get('timing', {}) # 从 _debug steps 中提取 Rerank 耗时 rerank_time = 0 rerank_cached = False for step in debug_info.get('steps', []): if step.get('name') == 'rerank' and step.get('applied'): rerank_time = step.get('time_ms', 0) rerank_cached = step.get('cached', False) finish_event["timing"] = { "total_search_ms": timing_info.get('total_ms', 0), "rerank_ms": rerank_time, "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'): _cache_collections = ','.join(sorted(collections)) if collections else '' _cache_key_text = f"{retrieval_query}|{_cache_collections}" _semantic_cache_emb = _eng.embedding_model.encode(_cache_key_text) if _semantic_cache_emb is not None: _sc.set(_semantic_cache_emb, { "cache_type": "rag_answer", "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: import logging as _logging _logging.getLogger(__name__).error(f"[SSE] RAG 流异常: {e}", exc_info=True) error_event = { "type": "error", "message": "服务内部错误，请稍后重试" } yield f"data: {json.dumps(error_event, ensure_ascii=False)}\n\n" return Response( generate(), mimetype='text/event-stream', headers={ 'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no' } ) @chat_bp.route('/search', methods=['POST']) @require_gateway_auth def search(): """ 混合检索接口 - 供 Dify 工作流调用 请求体: { "query": "查询文本", "top_k": 5, "collections": ["public_kb"] // 可选 } """ data = request.json or {} query = data.get('query', '') query = sanitize_user_input(query) injection_matches = detect_injection(query) if injection_matches: logger.warning(f"[Chat] 检测到可疑注入: {injection_matches}") top_k = data.get('top_k', 5) collections = data.get('collections') # 后端传入的知识库列表 if not query: return error_response("MISSING_PARAMS", BAD_REQUEST, "query is required", http_status=400) # 输入安全校验（注入检测、违禁词、长度限制） is_valid, reason = validate_query(query) if not is_valid: return error_response("INVALID_QUERY", BAD_REQUEST, reason, http_status=400) # top_k 范围校验 try: top_k = max(1, min(int(top_k), 50)) except (ValueError, TypeError): top_k = 5 # 如果没有指定 collections，使用默认的公开库 if not collections: collections = ['public_kb'] results = search_hybrid(query, top_k=top_k, allowed_collections=collections) return success_response(data={ 'contexts': results['documents'][0], 'metadatas': results['metadatas'][0], 'scores': results['scores'][0], 'ids': results['ids'][0] if 'ids' in results and results['ids'] else [] }) # ==================== 缓存调试接口（临时） ==================== @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})