d6d27b70f3
- 修复 P1.5 使用原始检索分数(0-1)对比 MIN_SCORE(2-5)导致表格图片永远无法返回的 bug - 改为使用 score_image_relevance() 计算表格图片相关性分数 - 添加 s=None 初始化，使 P1.5 可独立计算分数 - 添加 MAX_IMAGES 动态调整，支持表格多图场景（上限15）
2838 lines
120 KiB
Python
2838 lines
120 KiB
Python
"""
|
||
核心聊天与检索 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 auth.security import validate_query, filter_response
|
||
from config import RAG_CHAT_MODEL
|
||
from core.llm_utils import call_llm, call_llm_stream
|
||
|
||
|
||
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 表格内容（<table/<tr 等）
|
||
|
||
Args:
|
||
doc: 原始 doc 内容
|
||
chunk_type: 切片类型
|
||
|
||
Returns:
|
||
精简后的 doc 内容
|
||
"""
|
||
if chunk_type != 'table' or not doc:
|
||
return doc
|
||
|
||
import re
|
||
_html_table_re = re.compile(r'^\s*<\s*(table|tr|td|th|tbody|thead|caption)', re.IGNORECASE)
|
||
|
||
lines = doc.split('\n')
|
||
result_lines = []
|
||
in_table_section = False
|
||
theme_line = None
|
||
|
||
for line in lines:
|
||
# 保留"主题："行（表格所属章节标识）
|
||
if line.startswith('主题：'):
|
||
theme_line = line
|
||
continue
|
||
|
||
# 跳过冗余的语义增强行
|
||
if line.startswith('字段：') or line.startswith('描述：') or line.startswith('示例：'):
|
||
continue
|
||
|
||
# 遇到"表格内容："标记，进入表格区域
|
||
if line.strip() == '表格内容：':
|
||
in_table_section = True
|
||
continue
|
||
|
||
# 表格区域的内容直接保留
|
||
if in_table_section:
|
||
result_lines.append(line)
|
||
elif line.startswith('|'):
|
||
# 没有"表格内容："标记时，直接遇到 Markdown 表格行
|
||
in_table_section = True
|
||
result_lines.append(line)
|
||
elif _html_table_re.match(line):
|
||
# 没有"表格内容："标记时，直接遇到 HTML 表格标签
|
||
in_table_section = True
|
||
result_lines.append(line)
|
||
|
||
# 构建结果：主题行 + 【表格】标记 + 表格内容
|
||
if result_lines:
|
||
output_parts = []
|
||
if theme_line:
|
||
output_parts.append(theme_line)
|
||
output_parts.append('【表格】')
|
||
output_parts.extend(result_lines)
|
||
return '\n'.join(output_parts)
|
||
|
||
return doc
|
||
|
||
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 _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 _order_text_contexts_for_prompt(contexts: List[Dict], query: str, max_chunks: int,
|
||
min_score: float = 0.0) -> 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 _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：按字符预算构建上下文文本。
|
||
|
||
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 _get_agentic_rag() -> 'AgenticRAG':
|
||
"""
|
||
获取 AgenticRAG 实例
|
||
|
||
Returns:
|
||
AgenticRAG: 当前应用中的 AgenticRAG 实例
|
||
"""
|
||
return current_app.config['AGENTIC_RAG']
|
||
|
||
|
||
def _attach_citations(answer: str, contexts: List[Dict]) -> Dict[str, Any]:
|
||
"""
|
||
自动为回答添加引用标记（按段落级别匹配，jieba 分词精准匹配）
|
||
|
||
流程:
|
||
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": []}
|
||
|
||
# 按 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)}"
|
||
ctx_by_chunk[chunk_id] = 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])
|
||
|
||
if selected_ids:
|
||
for cid in selected_ids:
|
||
if cid not in cited_set:
|
||
cited_set.add(cid)
|
||
cited_chunks_ordered.append(cid)
|
||
# 在段落末尾插入引用标记（多个引用连续排列）
|
||
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)
|
||
|
||
# 构建引用列表（按出现顺序）
|
||
citations = []
|
||
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)
|
||
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', ''), # 所属向量库，用于前端文档预览跳转
|
||
"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_core = re.sub(r'[图表图片如图所示]', '', query) # 去掉泛词
|
||
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 看到的上下文里表格单元格图片以 [图片] 或 [N张图片]（MinerU 解析产物）表示，
|
||
LLM 通常会原样输出，有时还会加上  格式。
|
||
|
||
此函数按顺序将占位符映射到 images 列表中的 URL，
|
||
生成  语法，前端 markdown-it 可直接渲染为 <img> 标签。
|
||
|
||
处理的占位符格式：
|
||
- [图片] →  （单张）
|
||
-  →  （LLM 加了 ! 和假 URL）
|
||
- [3张图片] →   
|
||
-  → 同上 （LLM 加了前缀和假 URL）
|
||
|
||
Args:
|
||
answer: LLM 生成的回答文本
|
||
images: select_images + _filter_images_by_answer 后的最终图片列表
|
||
|
||
Returns:
|
||
替换后的回答文本
|
||
"""
|
||
if not images or '图片' not in answer:
|
||
return answer
|
||
|
||
img_iter = iter(images)
|
||
replaced_count = 0
|
||
|
||
def _replacer(match):
|
||
nonlocal replaced_count
|
||
n_str = match.group(1) # 数字部分，None 表示单张
|
||
n = int(n_str) if n_str else 1
|
||
|
||
parts = []
|
||
for _ in range(n):
|
||
try:
|
||
img = next(img_iter)
|
||
url = img.get('url', '')
|
||
img_id = img.get('id', 'img')
|
||
parts.append(f'')
|
||
replaced_count += 1
|
||
except StopIteration:
|
||
parts.append('[图片]') # 图片用完了
|
||
return ' '.join(parts)
|
||
|
||
# !? — 可选前导 !（LLM 有时加）
|
||
# \[(\d+)?张?图片\] — [图片] 或 [3张图片]（张 可选）
|
||
# (?:\((?:图片URL|https?://[^)]+)\))? — 可选的 LLM 从上下文复制的假 URL
|
||
# 匹配 (图片URL) 或 (https://via.placeholder.com/150) 等
|
||
result = re.sub(r'!?\[(\d+)?张?图片\](?:\((?:图片URL|https?://[^)]+)\))?', _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}" if query else answer)
|
||
_keywords = set(chars[i] + chars[i + 1] for i in range(len(chars) - 1))
|
||
|
||
if not _keywords:
|
||
return selected_images
|
||
|
||
# 动态阈值：关键词越多阈值越高，上限 4（比原版略宽松，因子章节惩罚承担更多过滤）
|
||
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
|
||
|
||
# 动态预算：列举型查询允许更多图片
|
||
list_keywords = ["哪些", "有什么", "包含", "列出", "所有", "全部"]
|
||
is_list_query = any(kw in query for kw in list_keywords)
|
||
|
||
# 检测查询中的图片编号（如 "图2.1"）
|
||
figure_pattern = r'图\s*(\d+\.?\d*)'
|
||
figure_matches = re.findall(figure_pattern, query)
|
||
has_figure_query = bool(figure_matches)
|
||
|
||
# 新增：从检索文本中提取图表引用（见表2.2、见图2.5 等）
|
||
# 同时记录引用所在的文件来源
|
||
# 重要：只从语义相关的 top 5 文本块提取，避免不相关引用干扰
|
||
referenced_figures = {} # {图号: set(文件来源)}
|
||
referenced_tables = {} # {表号: set(文件来源)}
|
||
|
||
for ctx in contexts[:5]: # 只从前5个最相关的文本块提取引用
|
||
doc_text = ctx.get('doc', '')
|
||
source = ctx.get('meta', {}).get('source', '')
|
||
|
||
# 提取 "见图X.X"、"如图X.X" 或单独的 "图X.X"
|
||
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)
|
||
|
||
# 提取 "见表X.X"、"如表X.X" 或单独的 "表X.X"
|
||
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]: # 只看前5个最相关的
|
||
source = ctx.get('meta', {}).get('source', '')
|
||
if source:
|
||
primary_sources.add(source)
|
||
|
||
# 图片意图检测：区分精确查图和泛指
|
||
strong_image_keywords = ["示意图", "流程图", "结构图", "过程线", "曲线图", "分布图", "图示", "看图", "显示图"]
|
||
weak_image_keywords = ["图片", "图表", "如图", "图", "统计"]
|
||
|
||
has_strong_image_intent = any(kw in query for kw in strong_image_keywords)
|
||
has_weak_image_intent = any(kw in query for kw in weak_image_keywords)
|
||
|
||
# 精确查图：用户指定了具体图号（如 "图2.3"），只返回最匹配的 1-2 张
|
||
if has_figure_query:
|
||
MAX_IMAGES = 2
|
||
MIN_SCORE = 5.0 # 精确匹配应该高分
|
||
# 强图片意图：明确要看某种图
|
||
elif has_strong_image_intent:
|
||
MAX_IMAGES = 3
|
||
MIN_SCORE = 5.0 # 提高阈值，避免不相关图片通过
|
||
# 列举型查询
|
||
elif is_list_query:
|
||
MAX_IMAGES = 5
|
||
MIN_SCORE = 3.0
|
||
# 有图表引用：检索文本中提到了图表
|
||
elif has_referenced_figures:
|
||
MAX_IMAGES = 3
|
||
MIN_SCORE = 2.0 # 降低阈值，让引用的图表能通过
|
||
# 弱图片意图：只是提到"图"字，可能是泛指（如 "发电量图"）
|
||
elif has_weak_image_intent:
|
||
MAX_IMAGES = 1 # 只返回最相关的一张
|
||
MIN_SCORE = 2.0 # 降低阈值，让语义相关的图片能通过
|
||
# 普通查询
|
||
else:
|
||
MAX_IMAGES = 2
|
||
MIN_SCORE = 3.0
|
||
|
||
# 动态调整：当表格嵌入大量图片时，提升上限以展示完整内容
|
||
_table_image_count = 0
|
||
for _ctx in contexts:
|
||
_meta = _ctx.get('meta', {})
|
||
if _meta.get('chunk_type') == 'table' and _meta.get('images_json'):
|
||
try:
|
||
_table_image_count += len(json.loads(_meta['images_json']))
|
||
except (json.JSONDecodeError, TypeError):
|
||
pass
|
||
if _table_image_count > MAX_IMAGES:
|
||
MAX_IMAGES = min(_table_image_count, 15) # 上限 15
|
||
|
||
# 获取检索结果中涉及的主要章节（只看前 3 个最相关的文本块）
|
||
primary_sections = set()
|
||
for ctx in contexts[:3]:
|
||
section = ctx.get('meta', {}).get('section', '') or ctx.get('meta', {}).get('section_path', '')
|
||
if section:
|
||
# 提取章节编号
|
||
# 优先匹配 X.X 格式（如 "2.3发电"），再匹配 第X章 格式
|
||
section_num = re.search(r'(\d+\.\d+)', section)
|
||
if not section_num:
|
||
# 尝试匹配 "第X章" 格式
|
||
chapter_match = re.search(r'第\s*(\d+)\s*章', section)
|
||
if chapter_match:
|
||
section_num = chapter_match
|
||
if section_num:
|
||
primary_sections.add(section_num.group(1))
|
||
|
||
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', '')
|
||
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', '')
|
||
# 只匹配 X.X 格式的章节号，避免匹配年份
|
||
img_section_num = re.search(r'(\d+\.\d+)', img_section)
|
||
img_section_id = img_section_num.group(1) if img_section_num else None
|
||
|
||
# ========== 核心修复：图片必须与主要文本切片章节关联 ==========
|
||
# 如果有主要章节，且图片章节不在其中，大幅降低分数
|
||
section_penalty = 0.0
|
||
if primary_sections and img_section_id and img_section_id not in primary_sections:
|
||
# 图片章节与主要检索结果不匹配，惩罚
|
||
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 = img_section_id and img_section_id in primary_sections
|
||
|
||
# P2：只有章节匹配才加分，移除"s >= 5.0"漏洞
|
||
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 = img_section_id and img_section_id in primary_sections
|
||
|
||
# P2：只有章节匹配才加分，移除"s >= 5.0"漏洞
|
||
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)
|
||
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,
|
||
'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,
|
||
'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
|
||
|
||
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 reciprocal_rank_fusion(results_list, weights=None, k=60):
|
||
"""
|
||
倒数排名融合算法
|
||
|
||
Args:
|
||
results_list: 多个检索结果列表
|
||
weights: 各结果权重
|
||
k: RRF 参数
|
||
|
||
Returns:
|
||
融合后的排序结果
|
||
"""
|
||
if weights is None:
|
||
weights = [1.0] * len(results_list)
|
||
|
||
fused_scores = {}
|
||
doc_data = {}
|
||
|
||
for results, weight in zip(results_list, weights):
|
||
if not results or not results.get('ids'):
|
||
continue
|
||
|
||
ids = results['ids'][0]
|
||
docs = results['documents'][0] if results.get('documents') else [''] * len(ids)
|
||
metas = results['metadatas'][0] if results.get('metadatas') else [{}] * len(ids)
|
||
distances = results['distances'][0] if results.get('distances') else [0] * len(ids)
|
||
|
||
for rank, (doc_id, doc, meta, dist) in enumerate(zip(ids, docs, metas, distances)):
|
||
if doc_id not in fused_scores:
|
||
fused_scores[doc_id] = 0
|
||
doc_data[doc_id] = {'doc': doc, 'meta': meta, 'dist': dist}
|
||
|
||
# RRF 分数
|
||
fused_scores[doc_id] += weight / (rank + k)
|
||
|
||
# 按分数排序
|
||
sorted_ids = sorted(fused_scores.keys(), key=lambda x: fused_scores[x], reverse=True)
|
||
|
||
return {
|
||
'ids': sorted_ids,
|
||
'documents': [doc_data[i]['doc'] for i in sorted_ids],
|
||
'metadatas': [doc_data[i]['meta'] for i in sorted_ids],
|
||
'scores': [fused_scores[i] for i in sorted_ids],
|
||
'distances': [doc_data[i]['dist'] for i in sorted_ids]
|
||
}
|
||
|
||
|
||
def search_hybrid(query: str, top_k: int = 5, candidates: int = 15,
|
||
allowed_levels: list = None, allowed_collections: list = None,
|
||
sub_queries: list = None):
|
||
"""
|
||
混合检索：直接调用生产环境引擎，确保测试效果与生产一致
|
||
|
||
Args:
|
||
query: 查询文本
|
||
top_k: 返回数量
|
||
candidates: 候选数量（用于 RERANK_CANDIDATES，由 config 控制）
|
||
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 jsonify({"error": "缺少 message"}), 400
|
||
|
||
# 输入安全验证
|
||
is_valid, reason = validate_query(message)
|
||
if not is_valid:
|
||
return jsonify({"error": reason}), 400
|
||
|
||
# 智能聊天
|
||
result = chat_with_llm(message, history)
|
||
|
||
# 过滤敏感信息
|
||
answer = filter_response(result["answer"])
|
||
|
||
return jsonify({
|
||
"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, RAG_SEARCH_CANDIDATES,
|
||
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
|
||
)
|
||
|
||
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 jsonify({"error": "缺少 message"}), 400
|
||
|
||
# 生产环境强制校验 chat_history
|
||
if IS_PROD and history is None:
|
||
return jsonify({
|
||
"error": "chat_history is required in production",
|
||
"code": "MISSING_HISTORY"
|
||
}), 400
|
||
|
||
# 输入安全验证
|
||
is_valid, reason = validate_query(message)
|
||
if not is_valid:
|
||
return jsonify({"error": reason}), 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}")
|
||
|
||
# 提前获取 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(
|
||
message,
|
||
top_k=RAG_SEARCH_TOP_K,
|
||
candidates=RAG_SEARCH_CANDIDATES,
|
||
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"
|
||
|
||
# 提取上下文
|
||
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, message)
|
||
has_figure_query = bool(figure_matches)
|
||
|
||
# Bug 4 修复：缩小 strong_image_keywords，移除歧义词
|
||
# "过程线" 是水文术语不是图片意图，"图片"/"图表"/"如图" 太泛
|
||
strong_image_keywords = ["示意图", "流程图", "结构图", "曲线图", "分布图", "图示", "看图", "显示图", "给我看"]
|
||
has_image_intent = has_figure_query or any(kw in message for kw in strong_image_keywords)
|
||
|
||
# Bug 2 修复：不重排 contexts，只在 meta 里打标记给后续的 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'):
|
||
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')
|
||
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_sections_for_supplement = set()
|
||
for ctx in text_contexts[:3]:
|
||
section = ctx.get('meta', {}).get('section', '') or ctx.get('meta', {}).get('section_path', '')
|
||
if section:
|
||
section_num = re.search(r'(\d+\.\d+)', section)
|
||
if not section_num:
|
||
chapter_match = re.search(r'第\s*(\d+)\s*章', section)
|
||
if chapter_match:
|
||
section_num = chapter_match
|
||
if section_num:
|
||
primary_sections_for_supplement.add(section_num.group(1))
|
||
|
||
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', '')
|
||
supp_section_num = re.search(r'(\d+\.\d+)', supp_section)
|
||
supp_section_id = supp_section_num.group(1) if supp_section_num else None
|
||
|
||
# 如果图片章节不在主要章节中，跳过
|
||
if primary_sections_for_supplement and supp_section_id and supp_section_id not in primary_sections_for_supplement:
|
||
# 不是主要章节的图片，跳过
|
||
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 调用耗时过长，可能导致请求超时
|
||
# TODO: 后续可改为异步后台任务
|
||
try:
|
||
import asyncio
|
||
from knowledge.lazy_enhance import enhance_retrieved_chunks
|
||
kb_name = collections[0] if collections else 'public_kb'
|
||
asyncio.run(enhance_retrieved_chunks(contexts, message, kb_name))
|
||
except Exception as e:
|
||
logger.warning(f"懒加载增强失败: {e}")
|
||
|
||
# 3. 选择要展示的图片（Phase 5）
|
||
selected_images = select_images(contexts, message)
|
||
|
||
# 调试事件：图片选择详情
|
||
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 名额竞争，图片描述不参与竞争
|
||
|
||
# 词法匹配救援：当切片文本精确包含查询关键词但 CrossEncoder 评分低时，
|
||
# 提升分数使其通过 min_score 过滤（适用于独立切片无法触发聚类救援的场景）
|
||
contexts = _rescue_lexical_match(contexts, retrieval_query, RERANK_CONTEXT_MIN_SCORE)
|
||
|
||
# 章节聚类救援（路由层安全网）：在 min_score 过滤前，
|
||
# 检测"全灭 section"并分配保底分数
|
||
if SECTION_CLUSTER_RESCUE_ENABLED:
|
||
contexts = _rescue_section_cluster(contexts, retrieval_query, RERANK_CONTEXT_MIN_SCORE)
|
||
|
||
text_contexts = _order_text_contexts_for_prompt(contexts, message, 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:
|
||
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)
|
||
# 添加指令让 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(message):
|
||
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']
|
||
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 traceback
|
||
error_event = {
|
||
"type": "error",
|
||
"message": str(e),
|
||
"traceback": traceback.format_exc()
|
||
}
|
||
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', '')
|
||
top_k = data.get('top_k', 5)
|
||
collections = data.get('collections') # 后端传入的知识库列表
|
||
|
||
if not query:
|
||
return jsonify({'error': 'query is required'}), 400
|
||
|
||
# 如果没有指定 collections，使用默认的公开库
|
||
if not collections:
|
||
collections = ['public_kb']
|
||
|
||
results = search_hybrid(query, top_k=top_k, allowed_collections=collections)
|
||
|
||
return jsonify({
|
||
'contexts': results['documents'][0],
|
||
'metadatas': results['metadatas'][0],
|
||
'scores': results['scores'][0]
|
||
})
|