From 99f5cf519e297d616fc1df45a4b7c9eee2554840 Mon Sep 17 00:00:00 2001 From: lacerate551 <128470311+lacerate551@users.noreply.github.com> Date: Wed, 10 Jun 2026 13:54:38 +0800 Subject: [PATCH] =?UTF-8?q?feat(server-release):=20=E4=BB=8E=20main=20?= =?UTF-8?q?=E8=BF=81=E7=A7=BB=20RAG=20=E6=A3=80=E7=B4=A2=E8=B4=A8=E9=87=8F?= =?UTF-8?q?=E5=A2=9E=E5=BC=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 5 个辅助函数：_strip_semantic_prefix（语义前缀清理）、 _process_table_doc（表格文档处理）、_section_similarity（章节相似度）、 _rescue_table_chunks（表格救援）、_filter_images_by_answer（图片后置过滤） - 替换 _order_text_contexts_for_prompt（添加表格保护逻辑：passing_sections + table_floor） - 替换 _build_context_with_budget（集成 _process_table_doc + section headers） - rag() 函数添加 5 个调用点：语义缓存检查/写入、strip_semantic_prefix、 rescue_table_chunks、filter_images_by_answer - 新增 retrieval_query 变量（使用改写后的查询替代原始 message） - 不涉及任何 API 端点变更 --- api/chat_routes.py | 510 ++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 462 insertions(+), 48 deletions(-) diff --git a/api/chat_routes.py b/api/chat_routes.py index 7a389bd..ac59de0 100644 --- a/api/chat_routes.py +++ b/api/chat_routes.py @@ -41,7 +41,6 @@ from auth.gateway import require_gateway_auth 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]: @@ -186,6 +185,113 @@ def _get_full_table_from_docstore(chunk_id: str) -> Optional[str]: 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 表格内容（ 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 _order_text_contexts_for_prompt(contexts: List[Dict], query: str, max_chunks: int, min_score: float = 0.0) -> List[Dict]: """ @@ -246,8 +352,31 @@ def _order_text_contexts_for_prompt(contexts: List[Dict], query: str, max_chunks }) # Phase 1：按 Rerank 分数过滤低分切片 + # 保护策略：同一 section 内如有切片通过阈值，则同 section 的 table 切片也保留 + # 原因：表格切片的 rerank 分数往往偏低（尤其是元问题如"有表格吗？"）， + # 但它们与同 section 的 text 切片属于同一语义单元，不应割裂 + # 安全下限：被保护的 table 切片自身 score 不得低于 min_score * 0.3， + # 防止 section 粒度较粗时完全不相关的表格被无条件保护 if min_score > 0: - text_contexts = [c for c in text_contexts if c.get('score', 0) >= min_score] + _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] # 合并：文本切片优先，图表切片补充 @@ -256,7 +385,11 @@ def _order_text_contexts_for_prompt(contexts: List[Dict], query: str, max_chunks combined_contexts = text_contexts + chart_contexts[:max_chart_contexts] if not _is_enum_query(query): - return combined_contexts[:max_chunks] + # 表格不受 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', {}) @@ -310,6 +443,123 @@ def _order_text_contexts_for_prompt(contexts: List[Dict], query: str, max_chunks return ordered +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_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：按字符预算构建上下文文本。 @@ -359,22 +609,46 @@ def _build_context_with_budget(contexts: List[Dict], max_chars: int, soft_limit: total_chars = 0 for key in group_order: group = groups[key] - group_text = "\n\n".join(ctx.get('doc', '') for ctx in group) + source, section = key - # 超过软限制后，只接受高分组 - if total_chars > soft_limit and group_max_score(key) < 0.1: + # 判断是否包含表格切片（表格是结构化关键内容，不受预算截断） + 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 = ctx.get('doc', '') + 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" @@ -416,16 +690,12 @@ def _attach_citations(answer: str, contexts: List[Dict]) -> Dict[str, Any]: if not contexts: return {"answer_with_refs": answer, "citations": []} - # 按 (collection, chunk_id) 复合键组织 contexts，防止跨库同名文件覆盖 + # 按 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 + ctx_by_chunk[chunk_id] = ctx # jieba 分词函数（fallback 到字符级） try: @@ -490,27 +760,20 @@ def _attach_citations(answer: str, contexts: List[Dict]) -> Dict[str, Any]: 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 - ) + # 在段落末尾插入引用标记（多个引用连续排列） + ref_tags = "".join(f"[ref:{cid}]" for cid in selected_ids) result_parts.append(f"{para}{ref_tags}{sep}") else: result_parts.append(para + sep) - # 构建引用列表（按出现顺序），使用原始 chunk_id 构建 citation + # 构建引用列表（按出现顺序） citations = [] - for composite_key in cited_chunks_ordered: - ctx = ctx_by_chunk.get(composite_key) + for chunk_id in cited_chunks_ordered: + ctx = ctx_by_chunk.get(chunk_id) if ctx: meta = ctx.get('meta', {}) full_content = ctx.get('doc', '') citation = _build_citation(meta, full_content) - # 确保 citation 中的 chunk_id 使用原始值（不含 collection 前缀） - raw_id = ctx.get('_raw_chunk_id') - if raw_id: - citation['chunk_id'] = raw_id citations.append(citation) return { @@ -599,7 +862,7 @@ def _build_citation(meta: Dict, full_content: str = '') -> Dict[str, Any]: "chunk_id": chunk_id_raw, "chunk_index": chunk_index, # 全局切片序号，用于精准定位文档位置 "source": meta.get('source', ''), - "collection": meta.get('_collection') or meta.get('collection', ''), # 所属向量库，用于前端文档预览跳转 + "collection": meta.get('_collection', ''), # 所属向量库，用于前端文档预览跳转 "doc_type": meta.get('doc_type', 'other'), "section": _clean_section(meta.get('section', '')), "preview": meta.get('preview', ''), @@ -771,6 +1034,77 @@ def score_image_relevance(query: str, meta: Dict, doc: str = '') -> float: return score +def _filter_images_by_answer(selected_images: List[Dict], answer: str) -> List[Dict]: + """ + 后置图片过滤：根据 LLM 生成的回答内容反向筛选图片。 + + 只有图片描述与回答内容有足够关键词重叠时才保留， + 确保展示的图片与回答内容一致，避免不相关图片干扰用户。 + + 过滤规则: + 1. 从回答中提取 2 字及以上的中文关键词（jieba 分词） + 2. 对每张图片，检查其描述（full_description / description）中匹配了多少关键词 + 3. 匹配数 >= 阈值则保留，否则丢弃 + 4. 如果过滤后图片为 0，保留分数最高的 1 张（兜底） + 5. 用户明确指定图号（如图 2.1）时不过滤 + + Args: + selected_images: select_images 返回的候选图片列表 + answer: LLM 生成的回答文本 + + Returns: + 过滤后的图片列表 + """ + if not selected_images or len(selected_images) <= 1: + return selected_images + + # 如果回答中提到了具体图号，说明 LLM 认为这些图是相关的，不过滤 + import re + if re.search(r'图\s*\d+\.?\d*', answer): + return selected_images + + # 从回答中提取关键词 + try: + import jieba + answer_keywords = set( + w for w in jieba.lcut(answer) + if len(w) >= 2 and re.search(r'[\u4e00-\u9fff]', w) + ) + except ImportError: + # jieba 不可用时回退到 bigram + chars = re.findall(r'[\u4e00-\u9fff]', answer) + answer_keywords = set(chars[i] + chars[i + 1] for i in range(len(chars) - 1)) + + if not answer_keywords: + return selected_images + + # 动态阈值：回答关键词越多，阈值越高（至少匹配 15% 或 2 个关键词，取较小值） + threshold = max(2, min(3, round(len(answer_keywords) * 0.15))) + + filtered = [] + for img in selected_images: + desc = img.get('full_description', '') or img.get('description', '') or '' + if not desc: + # 没有描述的图片，保留（无法判断） + filtered.append(img) + continue + + # 计算描述与回答的关键词重叠数 + overlap = sum(1 for kw in answer_keywords if kw in desc) + if overlap >= 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): + logger.info(f"[图片后置过滤] {len(selected_images)} → {len(filtered)} 张 " + f"(回答关键词 {len(answer_keywords)} 个, 阈值 {threshold})") + + return filtered + + def select_images(contexts: List[Dict], query: str) -> List[Dict]: """ 选择要展示的图片（打分排序 + 预算控制） @@ -1464,6 +1798,53 @@ def rag(): 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" @@ -1540,8 +1921,12 @@ def rag(): 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') + if collections and len(collections) == 1: + meta['_collection'] = collections[0] + elif collections: + meta['_collection'] = collections[0] # 多知识库时回退到第一个 + else: + meta['_collection'] = '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') @@ -1759,11 +2144,27 @@ def rag(): min_score=RERANK_CONTEXT_MIN_SCORE) # Phase 2：按字符预算构建上下文 _is_comparison = intent and intent.intent == "comparison" - if _is_enum_query(message) or _is_comparison: + if _is_enum_query(retrieval_query) or _is_comparison: # 列举类 / 对比类查询：保持原始顺序，不做预算截断 - context_text = "\n\n".join([ctx.get('doc', '') for ctx in text_contexts]) + # 仍然注入 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]] @@ -1884,6 +2285,9 @@ def rag(): # else: 没有匹配到，保留原选择（不再截断到1张） # else: LLM 没有提图号，保留原选择（不再截断到1张） + # 后置图片过滤：用回答内容反向筛选图片，确保图片与回答一致 + selected_images = _filter_images_by_answer(selected_images, full_answer_text) + rich_media = {'images': selected_images, 'tables': [], 'sections': []} # 7. 去掉 LLM 添加的数字引用标记，避免与后端引用重复 @@ -1951,14 +2355,39 @@ def rag(): "rerank_cached": rerank_cached, "total_ms": duration_ms } + # 10.5 写入语义缓存 + try: + from core.semantic_cache import get_semantic_cache + _sc = get_semantic_cache() + if _sc and filtered_answer: + if _semantic_cache_emb is None: + from core.engine import get_engine as _get_eng + _eng = _get_eng() + if _eng and hasattr(_eng, 'embedding_model'): + _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) + import traceback error_event = { "type": "error", - "message": "服务内部错误，请稍后重试" + "message": str(e), + "traceback": traceback.format_exc() } yield f"data: {json.dumps(error_event, ensure_ascii=False)}\n\n" @@ -1987,27 +2416,12 @@ def search(): """ 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 jsonify({'error': 'query is required'}), 400 - # 输入安全校验（注入检测、违禁词、长度限制） - is_valid, reason = validate_query(query) - if not is_valid: - return jsonify({'error': reason}), 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']