Files
rag/api/chat_routes.py
lacerate551 8adb550931 feat(rag): 数据驱动的检索优化与表格/图片处理增强
- 移除硬编码关键词列表,改为数据驱动的图片意图检测
- 新增 _section_similarity() 层级章节相似度函数,替代正则匹配
- 新增 _rescue_table_chunks() 表格救援机制,基于 _in_budget_context 标记
- 修复 _build_context_with_budget 表格组超预算不 break 的问题
- 新增 _filter_images_by_answer() 后置图片过滤
- 表格切片 rerank 分数保护策略(同 section 表格不被误删)
- 跨页表格合并逻辑改进(通用标题检测 + 页码不可用降级策略)
- 意图分析增强追问补全 + 缓存类型隔离
- 语义缓存 key 加入 collections 防止跨上下文误命中
- 使用 retrieval_query 替代原始 message 进行检索
- engine.py: CloudReranker 模型更新 + 移除 top_n + table 邻居扩展
- LLM prompt 增加表格处理规则和章节路径提示

🤖 Generated with [Qoder][https://qoder.com]
2026-06-08 15:44:22 +08:00

2640 lines
111 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
核心聊天与检索 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 threading
import time as _time
from typing import List, Dict, Any, Optional, Tuple
from pathlib import Path
import numpy as np
from flask import Blueprint, request, jsonify, Response, current_app
import logging
logger = logging.getLogger(__name__)
from auth.gateway import require_gateway_auth
from core.status_codes import SUCCESS, BAD_REQUEST, NOT_FOUND, INTERNAL_ERROR
from api.response_utils import success_response, error_response
from auth.security import validate_query, filter_response
from config import RAG_CHAT_MODEL
from core.llm_utils import call_llm, call_llm_stream
from core.prompt_guard import detect_injection, sanitize_user_input
def _get_vlm_cache(image_path: str) -> Optional[str]:
"""
从缓存获取图片的 VLM 描述
Args:
image_path: 图片文件名(不含路径),如 '0cd5e156ff0a.png'
Returns:
VLM 描述文本,无缓存返回 None
"""
import hashlib
try:
images_dir = Path(".data/images")
full_path = images_dir / image_path
if not full_path.exists():
return None
img_hash = hashlib.md5(full_path.read_bytes()).hexdigest()
cache_file = Path(f".data/cache/vlm/{img_hash}.txt")
if cache_file.exists():
return cache_file.read_text(encoding='utf-8')
except Exception as e:
logger.debug(f"读取 VLM 缓存失败: {e}")
return None
def _check_vlm_relevance(query: str, vlm_desc: str) -> float:
"""
使用 LLM 评估图片描述与查询的相关性
Args:
query: 用户查询
vlm_desc: 图片的 VLM 描述
Returns:
相关性分数 (0-1)0=不相关1=高度相关
"""
if not vlm_desc or len(vlm_desc) < 20:
return 0.5 # 无有效描述,中性评分
# 使用 jieba 分词提取关键词
import jieba
stop_words = {'', '', '图片', '图表', '如图', '所示', '', '', '', '', '', '', '', '', '', '', '', ''}
query_keywords = [w for w in jieba.lcut(query) if len(w) >= 2 and w not in stop_words]
if not query_keywords:
return 0.5
# 检查关键词是否在 VLM 描述中出现
matches = sum(1 for kw in query_keywords if kw in vlm_desc)
if matches == 0:
return 0.0 # 完全不相关
# 按匹配比例评分
ratio = matches / len(query_keywords)
return min(ratio, 1.0)
chat_bp = Blueprint('chat', __name__)
def _safe_int(value: Any, default: int = 10**9) -> int:
"""
安全的整数转换
Args:
value: 待转换的值
default: 转换失败时的默认值
Returns:
转换后的整数值,失败返回默认值
"""
try:
if value is None or value == "":
return default
return int(value)
except (TypeError, ValueError):
return default
def _is_enum_query(query: str) -> bool:
"""
判断是否为枚举型查询
枚举型查询通常包含"哪些""列出"等关键词,
需要特殊的上下文排序策略。
Args:
query: 用户查询文本
Returns:
是枚举型查询返回 True否则返回 False
"""
try:
from core.query_classifier import is_enumeration_query
return is_enumeration_query(query)
except Exception as e:
logger.debug(f"枚举查询检测失败: {e}")
markers = ("哪些", "有哪些", "列出", "严禁", "禁止", "不得", "包括", "要求", "情形", "场景")
return bool(query) and any(marker in query for marker in markers)
def _get_full_table_from_docstore(chunk_id: str) -> Optional[str]:
"""
从 DocStore 获取表格的完整 Markdown 内容
Args:
chunk_id: 切片 ID'组织架构.xlsx_0''test_report.pdf_3'
Returns:
完整表格 Markdown未找到返回 None
"""
docstore_dir = Path(".data/docstore")
if not docstore_dir.exists():
return None
# chunk_id 格式: {filename}_{index}
# DocStore 格式: {filename}_table_{index}.json 或 {filename}_{index}.json
possible_paths = [
docstore_dir / f"{chunk_id}.json", # 直接匹配
docstore_dir / f"{chunk_id.replace('_', '_table_', 1)}.json", # xxx_0 -> xxx_table_0
]
# 如果 chunk_id 是 {filename}_{num} 格式,尝试 {filename}_table_{num}
parts = chunk_id.rsplit('_', 1)
if len(parts) == 2:
filename, idx = parts
possible_paths.append(docstore_dir / f"{filename}_table_{idx}.json")
for doc_path in possible_paths:
if doc_path.exists():
try:
with open(doc_path, 'r', encoding='utf-8') as f:
record = json.load(f)
return record.get('markdown', '')
except Exception as e:
logger.debug(f"读取 DocStore 失败: {doc_path}, {e}")
return None
def _strip_semantic_prefix(doc: str, chunk_type: str) -> str:
"""
去除切片 doc 中的冗余语义前缀,保留关键标识信息
表格切片的 doc 由 _build_semantic_content_for_table 生成,格式为:
主题section_path保留用于关联章节
字段A, B, C去除冗余
描述该表包含N行数据去除冗余
示例:字段=值(去除,冗余)
表格内容:
| A | B | C |
|---|---|---|
| 1 | 2 | 3 |
优化策略:
- 保留"主题:"行(表格所属章节标识)
- 去除"字段:"/"描述:"/"示例:"行(冗余信息)
- 添加"【表格】"标记,让 LLM 明确识别表格类型
- 保留 Markdown 表格内容(| 开头)和 HTML 表格内容(<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 _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 _process_table_doc(doc: str, meta: Dict) -> str:
"""
处理单个切片的 doc精简表格语义前缀 + 注入嵌入图片 URL
集中处理两处共用逻辑(正常路径和预算截断路径),避免重复代码。
Args:
doc: 切片原始 doc
meta: 切片 metadata
Returns:
处理后的 doc
"""
doc = _strip_semantic_prefix(doc, meta.get('chunk_type', ''))
if meta.get('chunk_type') == 'table' and meta.get('images_json'):
try:
img_list = json.loads(meta['images_json'])
if img_list:
img_urls = [
f"/images/{img.get('id', '')}"
for img in img_list
if isinstance(img, dict) and img.get('id')
]
if img_urls:
doc += "\n\n[该表格包含以下图片,可在回答中引用]: " + ", ".join(img_urls)
except (json.JSONDecodeError, TypeError):
pass
return doc
def _section_similarity(section_a: str, section_b: str) -> float:
"""
计算两个章节路径的层级相似度(数据驱动,无需硬编码格式假设)。
策略:
1. 如果两个路径都有数值编号(如 "2.3"),优先用精确数值匹配
2. 否则按 " > " 层级拆分,计算 Jaccard 相似度
3. 无数值编号时优雅降级,适用于 "第七章 附则" 或无结构文档
Returns: 0.0 ~ 1.0
"""
import re as _re
if not section_a or not section_b:
return 0.0
# 快速路径:完全相同
if section_a == section_b:
return 1.0
# 优先精确匹配:数值编号(如 "2.3"
num_a = _re.search(r'(\d+\.\d+)', section_a)
num_b = _re.search(r'(\d+\.\d+)', section_b)
if num_a and num_b:
return 1.0 if num_a.group(1) == num_b.group(1) else 0.0
# 层级文本匹配:拆分路径为各级标题,计算 Jaccard 系数
def _split_levels(path):
parts = [p.strip() for p in path.split('>') if p.strip()]
# 去掉常见序号前缀(如 "1.1 "、"第1章 "),保留语义部分
cleaned = set()
for p in parts:
cleaned.add(_re.sub(r'^(?:\d+[\.\d]*\s*|第\s*\d+\s*章\s*|[一二三四五六七八九十]+、\s*)', '', p).strip())
return cleaned
levels_a = _split_levels(section_a)
levels_b = _split_levels(section_b)
if not levels_a or not levels_b:
return 0.0
overlap = len(levels_a & levels_b)
union = len(levels_a | levels_b)
return overlap / union if union > 0 else 0.0
def _rescue_table_chunks(contexts: List[Dict], context_text: str,
retrieval_query: str, max_rescue_chars: int = 3000) -> str:
"""
表格救援:当查询涉及表格但上下文预算截断了表格内容时,将最相关的表格补回。
根因CrossEncoder 对 Markdown 表格格式的评分普遍偏低,
导致 _build_context_with_budget 按分数排序时表格被排到最后并被截断。
此函数作为安全网,确保与查询最相关的表格始终出现在上下文中。
Args:
contexts: 经过 _order_text_contexts_for_prompt 处理的全部文本切片
context_text: _build_context_with_budget 的输出
retrieval_query: 改写后的检索查询
max_rescue_chars: 救援表格的最大字符数
Returns:
可能追加了表格内容的上下文文本
"""
# 1. 数据驱动检测:检索结果中是否有表格类型切片(无需硬编码关键词)
has_table_in_contexts = any(
ctx.get('meta', {}).get('chunk_type') == 'table' for ctx in contexts
)
if not has_table_in_contexts:
return context_text
# 2. 检测现有上下文是否已包含表格数据Markdown 表格至少 2 列)
if '|' in context_text and context_text.count('|') > 4:
return context_text
# 3. 从 contexts 中找被截断的表格切片
# 使用 _in_budget_context 标记精确判断(由 _build_context_with_budget 设置)
table_candidates = []
for ctx in contexts:
if ctx.get('meta', {}).get('chunk_type') != 'table':
continue
# 精确判断:如果 _build_context_with_budget 已标记该 chunk 为已纳入,跳过
if ctx.get('_in_budget_context'):
continue
table_candidates.append(ctx)
if not table_candidates:
return context_text
# 4. 按分数降序取最佳表格,追加到上下文
table_candidates.sort(key=lambda c: c.get('score', 0), reverse=True)
rescue_parts = []
rescue_chars = 0
for ctx in table_candidates[:3]: # 最多救援 3 个表格
meta = ctx.get('meta', {})
doc = _process_table_doc(ctx.get('doc', ''), meta)
section = meta.get('section', '') or meta.get('section_path', '')
part_text = ""
if section:
part_text += f"{section}\n"
part_text += doc
if rescue_chars + len(part_text) > max_rescue_chars:
break
rescue_parts.append(part_text)
rescue_chars += len(part_text)
if rescue_parts:
separator = "\n\n--- 以下为与查询最相关的表格CrossEncoder 评分偏低,自动补入)---\n\n"
context_text += separator + "\n\n".join(rescue_parts)
return context_text
def _build_context_with_budget(contexts: List[Dict], max_chars: int, soft_limit: int) -> str:
"""
Phase 2按字符预算构建上下文文本。
1. 按 (source, section) 分组,组内按 chunk_index 保持连续
2. 组间按组内最高 Rerank 分数降序排列
3. 逐组加入直到达到 max_chars 预算
4. 超过 soft_limit 后仅接受组内最高分 >= soft_min_score 的组
对于列举类查询_is_enum_query保持原始顺序直接拼接
因为 _order_text_contexts_for_prompt 已经优化了排序。
Args:
contexts: 已排序的上下文列表
max_chars: 硬性字符上限
soft_limit: 软限制,超过后收紧准入
Returns:
拼接好的上下文文本
"""
if not contexts:
return ""
# 按 (source, section) 分组
groups = {}
group_order = []
for ctx in contexts:
meta = ctx.get('meta', {})
key = (meta.get('source', ''), meta.get('section', '') or meta.get('section_path', ''))
if key not in groups:
groups[key] = []
group_order.append(key)
groups[key].append(ctx)
# 组内按 chunk_index 排序
for key in groups:
groups[key].sort(key=lambda c: _safe_int(c.get('meta', {}).get('chunk_index')))
# 组间按组内最高 score 降序
def group_max_score(key):
return max((c.get('score', 0) for c in groups[key]), default=0)
group_order.sort(key=group_max_score, reverse=True)
# 贪心加入直到预算满
parts = []
total_chars = 0
for key in group_order:
group = groups[key]
source, section = key
# 判断是否包含表格切片(表格是结构化关键内容,不受预算截断)
has_table = any(c.get('meta', {}).get('chunk_type') == 'table' for c in group)
# 构建组文本,表格切片附加图片 URL 供 LLM 引用
doc_parts = []
for ctx in group:
doc = _process_table_doc(ctx.get('doc', ''), ctx.get('meta', {}))
doc_parts.append(doc)
group_text = "\n\n".join(doc_parts)
# 在组首插入章节路径标题行,帮助 LLM 区分不同章节
section_header = ''
if section:
section_header = f"{section}"
group_text = section_header + "\n" + group_text
# 超过软限制后,只接受高分组(但表格组始终保留,不因分数低被跳过)
if total_chars > soft_limit and group_max_score(key) < 0.1 and not has_table:
continue
if total_chars + len(group_text) > max_chars:
# 尝试逐条加入该组,直到预算满
# 先加入章节标题(如果有)
if section_header and total_chars + len(section_header) + 2 <= max_chars:
parts.append(section_header)
total_chars += len(section_header) + 2
for ctx in group:
doc = _process_table_doc(ctx.get('doc', ''), ctx.get('meta', {}))
if total_chars + len(doc) + 2 > max_chars: # +2 for "\n\n"
break
ctx['_in_budget_context'] = True # 标记已纳入上下文
parts.append(doc)
total_chars += len(doc) + 2
# 超预算后一律 break被截断的表格由 _rescue_table_chunks 补回
break
for ctx in group:
ctx['_in_budget_context'] = True # 标记已纳入上下文
parts.append(group_text)
total_chars += len(group_text) + 2 # +2 for "\n\n"
return "\n\n".join(parts)
def _attach_citations(answer: str, contexts: List[Dict]) -> Dict[str, Any]:
"""
自动为回答添加引用标记按段落级别匹配jieba 分词精准匹配)
流程:
1. 将 answer 按自然段落分割(\n\n
2. 对每个段落用 jieba 分词后计算词级重叠度
3. Phase 6动态阈值短段落 0.55 / 长段落 0.45),每段最多 2 个引用
4. 前端负责对引用进行重新编号
Args:
answer: LLM 生成的回答
contexts: 检索到的上下文列表
Returns:
{
"answer_with_refs": "回答文本(含 [ref:chunk_id] 标记)",
"citations": [引用列表]
}
"""
import re
if not contexts:
return {"answer_with_refs": answer, "citations": []}
# 按 (collection, chunk_id) 复合键组织 contexts防止跨库同名文件覆盖
ctx_by_chunk = {}
for ctx in contexts:
meta = ctx.get('meta', {})
chunk_id = meta.get('chunk_id') or f"{meta.get('source')}_{meta.get('chunk_index', 0)}"
coll = meta.get('_collection') or meta.get('collection') or ''
composite_key = f"{coll}/{chunk_id}" if coll else chunk_id
# 保存原始 chunk_id用于对外输出ref tag / citation
ctx['_raw_chunk_id'] = chunk_id
ctx_by_chunk[composite_key] = ctx
# jieba 分词函数fallback 到字符级)
try:
import jieba
def _tokenize(text: str) -> set:
return set(w for w in jieba.lcut(text) if len(w) >= 2)
except ImportError:
def _tokenize(text: str) -> set:
return set(text)
# 按自然段落分割(\n\n 为分隔符,保留分隔符用于重组)
parts = re.split(r'(\n\n+)', answer)
cited_chunks_ordered: List[str] = [] # 保序去重的 chunk_id 列表
cited_set: set = set()
result_parts = []
for i in range(0, len(parts), 2):
para = parts[i]
sep = parts[i + 1] if i + 1 < len(parts) else ''
stripped = para.strip()
# 跳过过短段落或 markdown 表格/标题行
is_table = stripped.startswith('|') or '|---' in stripped
is_short = len(stripped) < 15
if is_table or is_short:
result_parts.append(para + sep)
continue
# 词级重叠匹配:找最相关的 chunk
para_words = _tokenize(stripped[:300])
# Phase 6动态阈值 — 短段落用更高阈值避免误匹配
overlap_threshold = 0.55 if len(stripped) < 50 else 0.45
# 收集所有超过阈值的候选 chunk按分数降序
candidates = []
for chunk_id, ctx in ctx_by_chunk.items():
ctx_doc = ctx.get('doc', '')
if not ctx_doc:
continue
ctx_words = _tokenize(ctx_doc[:400])
if not para_words:
continue
overlap = len(para_words & ctx_words)
score = overlap / len(para_words)
if score >= overlap_threshold:
candidates.append((chunk_id, score))
candidates.sort(key=lambda x: x[1], reverse=True)
# Phase 6允许最多 2 个引用(分数差距 < 0.1 时附加第二引用)
selected_ids = []
if candidates:
selected_ids.append(candidates[0][0])
if len(candidates) > 1 and (candidates[0][1] - candidates[1][1]) < 0.1:
selected_ids.append(candidates[1][0])
# 按 _raw_chunk_id 去重:不同 composite_key 可能指向同一个底层 chunk
# 避免同一 chunk 产生重复引用标记(如 [3][3]
seen_raw_ids = set()
deduped_ids = []
for cid in selected_ids:
raw_id = ctx_by_chunk[cid].get('_raw_chunk_id', cid)
if raw_id not in seen_raw_ids:
seen_raw_ids.add(raw_id)
deduped_ids.append(cid)
selected_ids = deduped_ids
if selected_ids:
for cid in selected_ids:
if cid not in cited_set:
cited_set.add(cid)
cited_chunks_ordered.append(cid)
# 在段落末尾插入引用标记(使用原始 chunk_id不暴露复合键
ref_tags = "".join(
f"[ref:{ctx_by_chunk[cid].get('_raw_chunk_id', cid)}]"
for cid in selected_ids
)
result_parts.append(f"{para}{ref_tags}{sep}")
else:
result_parts.append(para + sep)
# 构建引用列表(按出现顺序),使用原始 chunk_id 构建 citation
# 按 _raw_chunk_id 去重,避免同一 chunk 产生重复引用条目
citations = []
seen_citation_raw_ids = set()
for composite_key in cited_chunks_ordered:
ctx = ctx_by_chunk.get(composite_key)
if ctx:
raw_id = ctx.get('_raw_chunk_id') or composite_key
if raw_id in seen_citation_raw_ids:
continue # 同一 chunk 已在引用列表中,跳过
seen_citation_raw_ids.add(raw_id)
meta = ctx.get('meta', {})
full_content = ctx.get('doc', '')
citation = _build_citation(meta, full_content)
citation['chunk_id'] = raw_id
citations.append(citation)
return {
"answer_with_refs": "".join(result_parts),
"citations": citations
}
def _clean_section(raw: str) -> str:
"""清洗 section 字段:过滤掉像正文内容而非章节路径的值"""
if not raw:
return ''
import re as _re
def _is_heading(part: str) -> bool:
"""判断一个字符串是否像章节标题(而非正文内容)"""
p = part.strip()
if not p:
return False
# 以句号/问号/感叹号结尾 → 正文句子
if p.endswith(('', '', '', '.', '!', '?')):
return False
# 含冒号且偏长 → 更像是内容摘要而非标题
if '' in p and len(p) > 25:
return False
# 长度超过 30 字 → 不像是标题
if len(p) > 30:
return False
return True
# 去掉 Markdown 加粗标记
cleaned = raw.replace('**', '').strip()
# 如果含 ' > ' 分隔符,验证每一级都是合法标题
if ' > ' in cleaned:
parts = [p.strip() for p in cleaned.split('>') if p.strip()]
valid = [p for p in parts if _is_heading(p)]
if valid:
return ' > '.join(valid[:3])
# 所有部分都不像标题 → 清空
return ''
# 匹配 【xxx】 或 第X篇/章/节 格式
if _re.match(r'^(【[^】]+】|第[一二三四五六七八九十\d]+[篇章节部])', cleaned):
return cleaned[:40]
# 短字符串(< 30字且不像句子无句号/逗号),保留
if len(cleaned) < 30 and not any(c in cleaned for c in '。,;:'):
return cleaned
# 其余视为正文内容,清空
return ''
def _build_citation(meta: Dict, full_content: str = '') -> Dict[str, Any]:
"""
根据文档类型构建定位信息
不同文档类型使用不同的定位策略:
- PDF: 坐标定位page + bbox
- Word: 语义定位section + section_chunk_id + preview
- Excel: 表格定位sheet + preview
Args:
meta: 切片元数据
full_content: 完整切片内容(可选)
Returns:
引用信息字典,包含定位信息
Example:
>>> citation = _build_citation(meta, "完整内容...")
>>> print(citation["page"]) # PDF: 页码
>>> print(citation["section"]) # Word: 章节
"""
# 从 chunk_id 中提取全局切片序号(格式: "filename_N"
chunk_id_raw = meta.get('chunk_id', '')
chunk_index = None
if chunk_id_raw and '_' in str(chunk_id_raw):
try:
chunk_index = int(str(chunk_id_raw).rsplit('_', 1)[-1])
except (ValueError, IndexError):
chunk_index = meta.get('chunk_index')
else:
chunk_index = meta.get('chunk_index')
citation = {
"chunk_id": chunk_id_raw,
"chunk_index": chunk_index, # 全局切片序号,用于精准定位文档位置
"source": meta.get('source', ''),
"collection": meta.get('_collection') or meta.get('collection', ''), # 所属向量库,用于前端文档预览跳转
"doc_type": meta.get('doc_type', 'other'),
"section": _clean_section(meta.get('section', '')),
"preview": meta.get('preview', ''),
"content": (full_content or meta.get('preview', ''))[:300], # 截断至 300 字避免冒返大量数据
"chunk_type": meta.get('chunk_type', 'text'),
}
doc_type = meta.get('doc_type', 'other')
if doc_type == 'pdf':
# PDF: 坐标定位
bbox_raw = meta.get('bbox')
bbox = None
if bbox_raw:
try:
bbox = json.loads(bbox_raw) if isinstance(bbox_raw, str) else bbox_raw
except (json.JSONDecodeError, TypeError):
bbox = bbox_raw
citation.update({
"page": meta.get('page'),
"page_end": meta.get('page_end'),
"bbox": bbox,
"bbox_mode": meta.get('bbox_mode'),
})
elif doc_type == 'word':
# Word: 语义定位
citation.update({
"section_chunk_id": meta.get('section_chunk_id'), # 章节内段落序号
})
elif doc_type == 'excel':
# Excel: 表格定位
citation.update({
"page": meta.get('page'), # 工作表序号
})
else:
# 其他类型:返回所有可用信息
bbox_raw = meta.get('bbox')
bbox = None
if bbox_raw:
try:
bbox = json.loads(bbox_raw) if isinstance(bbox_raw, str) else bbox_raw
except (json.JSONDecodeError, TypeError):
bbox = bbox_raw
citation.update({
"page": meta.get('page'),
"page_end": meta.get('page_end'),
"bbox": bbox,
"bbox_mode": meta.get('bbox_mode'),
})
return citation
def score_image_relevance(query: str, meta: Dict, doc: str = '') -> float:
"""
图片相关性打分(语义增强版)
通过多维度特征评估图片与查询的相关性:
1. 图片编号精确匹配(如 "图2.1"
2. 关键词匹配(年份、数值+单位、中文词组)
3. 整体文本相似度
4. 章节匹配
5. 图片类型加分
Args:
query: 用户查询
meta: 切片元数据
doc: document 字段(包含图片描述和上下文)
Returns:
相关性分数(>= 3.0 推荐展示)
Example:
>>> score = score_image_relevance("图2.1是什么?", meta, doc)
>>> if score >= 3.0:
... # 推荐展示该图片
"""
import re
score = 0.0
# 优先使用 doc 字段(包含完整描述和上下文)
search_text = doc or meta.get('caption', '')
section = meta.get('section', '') or meta.get('section_path', '')
source = meta.get('source', '')
# 1. 图片编号精确匹配(最高优先级)
figure_pattern = r'\s*(\d+\.?\d*)'
figure_matches = re.findall(figure_pattern, query)
if figure_matches:
for fig_num in figure_matches:
# 在所有文本中查找图号
all_text = f"{search_text} {section} {source}"
if f"{fig_num}" in all_text or f"{fig_num}" in all_text or f"见图{fig_num}" in all_text:
score += 10.0 # 精确匹配,直接返回
return score
# 1.5. 表格编号精确匹配(新增:支持表格图片)
table_pattern = r'\s*(\d+\.?\d*)'
table_matches = re.findall(table_pattern, query)
if table_matches:
for table_num in table_matches:
# 在所有文本中查找表号
all_text = f"{search_text} {section} {source}"
if f"{table_num}" in all_text or f"{table_num}" in all_text or f"见表{table_num}" in all_text:
score += 10.0 # 精确匹配,直接返回
return score
# 2. 查询词匹配(通用方式,不硬编码关键词)
# 从查询中提取有意义的词:中文词组、数字+单位、年份等
# 使用 jieba 分词(如果可用)或简单的正则提取
query_keywords = []
# 提取年份(如 "2003年"
year_matches = re.findall(r'(\d{4})\s*年', query)
query_keywords.extend(year_matches)
# 提取数值+单位(如 "100亿"、"50万千瓦时"
num_unit_matches = re.findall(r'(\d+\.?\d*\s*[亿万万千百吨米秒])', query)
query_keywords.extend(num_unit_matches)
# 使用 jieba 分词提取中文关键词
import jieba
jieba_words = [w for w in jieba.lcut(query) if len(w) >= 2]
query_keywords.extend(jieba_words)
# 过滤掉泛词(图、表、图片等)
stop_words = {'', '', '图片', '图表', '如图', '所示', '如下', '如下表', '如下图', '', '', '', '', '', '', '', '', '', '', '', ''}
query_keywords = [kw for kw in query_keywords if kw not in stop_words]
# 在图片描述中匹配关键词
keyword_match_score = 0.0
for kw in query_keywords:
if kw in search_text or kw in section:
keyword_match_score += 2.0
score += min(keyword_match_score, 8.0) # 最多加 8 分
# 3. 整体文本相似度(字符级别)
if search_text:
# 复用已过滤停用词的 query_keywords避免字符级误删如"表现"→"现"
query_core = "".join(query_keywords)
if query_core:
overlap = len(set(query_core) & set(search_text))
score += min(overlap * 0.2, 3.0)
# 4. 章节匹配
if section:
# 从查询中提取章节关键词(复用 jieba 分词)
section_keywords = query_keywords
for kw in section_keywords:
if kw in section:
score += 1.5
# 5. 图片类型加分
if meta.get('chunk_type') == 'chart':
score += 2.0
elif meta.get('chunk_type') == 'image':
score += 1.0
# 6. 检索相似度(如果有)
retrieval_score = meta.get('score', 0)
if retrieval_score > 0:
score += min(retrieval_score * 2, 2.0)
return score
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]:
"""
选择要展示的图片(打分排序 + 预算控制)
根据查询意图动态调整图片数量上限:
- 精确查图(指定图号): 最多 2 张
- 强图片意图(示意图、流程图等): 最多 3 张
- 列举型查询: 最多 5 张
- 普通查询: 最多 2 张
核心逻辑:
1. 检测查询中的图号引用
2. 从检索文本中提取图表引用
3. 对图片打分并过滤低分图片
4. 通过章节关联排除不相关图片
Args:
contexts: 检索上下文列表
query: 用户查询
Returns:
精选图片列表(每项含 score, id, url, type, source 等字段)
Example:
>>> images = select_images(contexts, "图2.1展示了什么?")
>>> print(len(images)) # <= 2
"""
import re
# 动态预算:数据驱动,不依赖硬编码关键词列表
# 核心策略:宽松预选 + 后置过滤_filter_images_by_answer精准裁剪
# 精确查图:用户指定了具体图号(如 "图2.3")—— 结构化模式匹配,非硬编码
figure_pattern = r'\s*(\d+\.?\d*)'
figure_matches = re.findall(figure_pattern, query)
has_figure_query = bool(figure_matches)
# 数据驱动的图片意图检测:检查检索结果中是否包含图片/图表类型切片
# 原理:如果向量检索返回了 image/chart 类型 chunk 或含 images_json 的 table chunk
# 说明知识库中存在与查询语义相关的图片内容,应给予展示机会
has_image_data = False
_image_chunk_count = 0
_table_image_count = 0
for ctx in contexts:
meta = ctx.get('meta', {})
ct = meta.get('chunk_type', '')
if ct in ('image', 'chart'):
_image_chunk_count += 1
has_image_data = True
if ct == 'table' and meta.get('images_json'):
_table_image_count += 1
has_image_data = True
# 从检索文本中提取图表引用见表2.2、见图2.5 等)
# 重要:只从语义相关的 top 5 文本块提取,避免不相关引用干扰
referenced_figures = {} # {图号: set(文件来源)}
referenced_tables = {} # {表号: set(文件来源)}
for ctx in contexts[:5]:
doc_text = ctx.get('doc', '')
source = ctx.get('meta', {}).get('source', '')
fig_refs = re.findall(r'(?:[见如])?图\s*(\d+\.?\d*)', doc_text)
for fig_num in fig_refs:
if fig_num not in referenced_figures:
referenced_figures[fig_num] = set()
if source:
referenced_figures[fig_num].add(source)
table_refs = re.findall(r'(?:[见如])?表\s*(\d+\.?\d*)', doc_text)
for table_num in table_refs:
if table_num not in referenced_tables:
referenced_tables[table_num] = set()
if source:
referenced_tables[table_num].add(source)
has_referenced_figures = bool(referenced_figures or referenced_tables)
# 获取检索结果中涉及的主要文件来源
primary_sources = set()
for ctx in contexts[:5]:
source = ctx.get('meta', {}).get('source', '')
if source:
primary_sources.add(source)
# 动态预算:根据检索结果数据驱动设置
# 后置过滤 _filter_images_by_answer 会根据回答内容精准裁剪
if has_figure_query:
# 精确查图:用户指定了具体图号
MAX_IMAGES = 2
MIN_SCORE = 5.0
elif has_image_data:
# 检索结果中有图片数据 → 宽松预算,给后置过滤留足候选空间
MAX_IMAGES = 5
MIN_SCORE = 2.0
elif has_referenced_figures:
# 检索文本中引用了图表编号
MAX_IMAGES = 3
MIN_SCORE = 2.0
else:
# 无图片数据 → 保守默认值
MAX_IMAGES = 2
MIN_SCORE = 3.0
# 动态调整:当表格嵌入大量图片时,提升上限以展示完整内容
if has_image_data and _table_image_count > 0:
total_table_images = 0
for ctx in contexts:
meta = ctx.get('meta', {})
if meta.get('chunk_type') == 'table' and meta.get('images_json'):
try:
total_table_images += len(json.loads(meta['images_json']))
except (json.JSONDecodeError, TypeError):
pass
if total_table_images > MAX_IMAGES:
MAX_IMAGES = min(total_table_images, 15) # 上限 15防止图片过多
# 获取检索结果中涉及的主要章节路径(只看前 3 个最相关的文本块)
primary_section_paths = set()
for ctx in contexts[:3]:
section = ctx.get('meta', {}).get('section', '') or ctx.get('meta', {}).get('section_path', '')
if section:
primary_section_paths.add(section)
# ========== P1.5 预计算:表格主题相关性评分 ==========
# 从查询中提取关键词片段(使用 jieba 分词 + 2字及以上的词自动过滤停用词
try:
import jieba
_query_kw_segments = [w for w in jieba.lcut(query)
if len(w) >= 2 and re.search(r'[\u4e00-\u9fff]', w)]
except ImportError:
# jieba 不可用时回退到 bigram
_chars = re.findall(r'[\u4e00-\u9fff]', query)
_query_kw_segments = [_chars[i] + _chars[i+1] for i in range(len(_chars) - 1)]
# 对所有含 images_json 的表格切片计算主题匹配分
_table_topic_scores = {}
for _tc in contexts:
_tm = _tc.get('meta', {})
if _tm.get('chunk_type') == 'table' and _tm.get('images_json'):
_t_section = (_tm.get('section', '') or _tm.get('section_path', ''))
_t_title = _tm.get('title', '') or ''
_t_combined = _t_title + _t_section
_score = sum(1 for kw in _query_kw_segments if kw in _t_combined)
_table_topic_scores[id(_tc)] = _score
# 自适应阈值:要求至少匹配 70% 的最佳表格得分(至少 2 分)
# 例如查询 "设施设备的参考样式" → 4 个关键词 → 最佳匹配 4 → 阈值 max(2, 2) = 2
# "形象识别标识" 只匹配 "参考"+"样式" = 2 → 但阈值=3(70%of4)时被过滤
_best_table_score = max(_table_topic_scores.values()) if _table_topic_scores else 0
_table_topic_threshold = max(2, round(_best_table_score * 0.7)) if _best_table_score >= 2 else 0
scored_images = []
for ctx in contexts:
meta = ctx.get('meta', {})
chunk_type = meta.get('chunk_type', 'text')
s = None # P1 评分,用于 P1.5 继承
doc = ctx.get('doc', '') # 默认文档内容
# 处理图片类型和有关联图片的表格类型
if meta.get('image_path') and chunk_type in ('image', 'chart', 'table'):
# 优先使用 VLM 详细描述
# 1. lazy_enhance 对 image/chart 更新 ctx['doc']
# 2. lazy_enhance 对 table 更新 ctx['image_description'](而非 doc
doc = ctx.get('image_description', '') or meta.get('vlm_desc', '') or ctx.get('doc', '')
s = score_image_relevance(query, meta, doc)
# ========== VLM 相关性筛选(方案 C==========
# 用 VLM 描述判断图片内容是否与查询相关
# 优先级meta.vlm_desc已同步 > .data/cache/vlm/(懒加载缓存)
image_path = meta.get('image_path', '')
vlm_desc = meta.get('vlm_desc', '') or _get_vlm_cache(image_path)
if vlm_desc:
vlm_relevance = _check_vlm_relevance(query, vlm_desc)
if vlm_relevance < 0.3:
# VLM 描述与查询不相关,适度降分
s -= 3.0
logger.debug(f"图片 {image_path} VLM 不相关,降分: {vlm_relevance:.2f}")
elif vlm_relevance >= 0.5:
# 相关,小幅加分
s += 2.0
# 图片来源
img_source = meta.get('source', '')
# 图片章节
img_section = meta.get('section', '') or meta.get('section_path', '')
# ========== 章节关联检测:基于层级相似度,无需硬编码格式假设 ==========
# 计算图片章节与主要检索结果的最高相似度
section_penalty = 0.0
max_section_sim = 0.0
if primary_section_paths:
for ps in primary_section_paths:
sim = _section_similarity(img_section, ps)
max_section_sim = max(max_section_sim, sim)
# 当相似度低于阈值且有足够的章节信息时,判定为不相关
if primary_section_paths and img_section and max_section_sim < 0.3:
# 图片章节与主要检索结果不匹配,惩罚
section_penalty = -5.0
# 除非图片被文本切片明确引用
is_referenced = False
for fig_num in referenced_figures:
if f"{fig_num}" in doc or f"{fig_num}" in doc:
is_referenced = True
break
if not is_referenced:
for table_num in referenced_tables:
if f"{table_num}" in doc or f"{table_num}" in doc:
is_referenced = True
break
if is_referenced:
section_penalty = 0.0 # 被引用则不惩罚
s += section_penalty
# 新增:如果图片描述中包含检索文本引用的图号,大幅加分
# 前提:图片章节与主要检索结果的章节相关
if referenced_figures:
for fig_num, sources in referenced_figures.items():
# 只检查 doc 字段,不检查 meta避免 section 中的误匹配)
if f"{fig_num}" in doc or f"{fig_num}" in doc:
# 使用层级相似度判断章节关联性
section_match = max_section_sim >= 0.3
# 章节匹配时才加分
if section_match:
# 图号匹配加分
s += 8.0
# 如果图片来源与引用来源一致,额外加分
if img_source in sources:
s += 5.0 # 文件匹配额外加分
break
# 新增:如果表格描述中包含检索文本引用的表号,大幅加分
# 同样要求章节相关性
if referenced_tables:
for table_num, sources in referenced_tables.items():
# 只检查 doc 字段
if f"{table_num}" in doc or f"{table_num}" in doc:
# 使用层级相似度判断章节关联性
section_match = max_section_sim >= 0.3
# 章节匹配时才加分
if section_match:
# 表号匹配加分
s += 8.0
# 如果图片来源与引用来源一致,额外加分
if img_source in sources:
s += 5.0 # 文件匹配额外加分
break
# 新增:如果图片来源在主要检索结果中,加分
if img_source in primary_sources:
s += 2.0
if s >= MIN_SCORE:
scored_images.append({
'score': s,
'id': os.path.basename(meta['image_path']),
'url': f"/images/{os.path.basename(meta['image_path'])}",
'type': meta['chunk_type'],
'source': meta.get('source'),
'page': meta.get('page'),
'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', '')
s = score_image_relevance(query, meta, doc)
# 章节相关性过滤:使用层级相似度,无需硬编码章节格式假设
table_section = meta.get('section', '') or meta.get('section_path', '')
section_relevant = True
if primary_section_paths and table_section:
# 计算表格章节与所有主要章节的最高相似度
max_sim = max(
(_section_similarity(table_section, ps) for ps in primary_section_paths),
default=0.0
)
if max_sim < 0.3:
section_relevant = False
elif primary_section_paths and not table_section:
# 表格无章节信息时优雅降级:不做章节过滤,仅依赖主题分数
pass
if not section_relevant:
# 例外:如果表格标题/内容被查询直接提及,仍视为相关
table_title = meta.get('title', '') or ''
table_doc = ctx.get('doc', '') or ''
if table_title and table_title in query:
section_relevant = True
elif table_doc and any(kw in table_doc for kw in query.split() if len(kw) >= 2):
section_relevant = True
if not section_relevant:
logger.debug(f"P1.5 跳过无关表格图片: path={table_section}, title={meta.get('title', '')}")
continue # 跳过此表格的所有嵌入图片
# 标题/主题相关性过滤:基于预计算的关键词匹配评分
# 如果最佳表格得分 >= 2则过滤掉得分为 0 的表格
_this_topic_score = _table_topic_scores.get(id(ctx), 0)
if _this_topic_score < _table_topic_threshold:
logger.debug(f"P1.5 跳过主题不匹配表格: topic_score={_this_topic_score}, threshold={_table_topic_threshold}, title={meta.get('title', '')}")
continue
try:
images_list = json.loads(meta['images_json'])
for img_info in images_list:
if isinstance(img_info, dict):
img_id = img_info.get('id') or img_info.get('path', '')
img_page = img_info.get('page', meta.get('page'))
else:
img_id = str(img_info)
img_page = meta.get('page')
# 跳过已添加的图片(避免重复)
existing_ids = {img['id'] for img in scored_images}
if img_id and img_id not in existing_ids:
# 为关联图片计算分数(继承主表格分数,略低)
assoc_score = s - 1.0 if s >= MIN_SCORE else MIN_SCORE - 1.0
if assoc_score >= MIN_SCORE:
scored_images.append({
'score': assoc_score,
'id': img_id,
'url': f"/images/{img_id}",
'type': 'table_image',
'source': meta.get('source'),
'page': img_page,
'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'),
'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 error_response("MISSING_PARAMS", BAD_REQUEST, "缺少 message", http_status=400)
# 输入安全验证
is_valid, reason = validate_query(message)
if not is_valid:
return error_response("INVALID_QUERY", BAD_REQUEST, reason, http_status=400)
# 智能聊天
result = chat_with_llm(message, history)
# 过滤敏感信息
answer = filter_response(result["answer"])
return success_response(data={
"answer": answer,
"mode": "chat",
"sources": result.get("sources", []),
"web_searched": result.get("web_searched", False)
})
@chat_bp.route('/rag', methods=['POST'])
@require_gateway_auth
def rag():
"""
知识库问答模式 - SSE 流式返回
请求体:
{
"message": "消息内容",
"history": [{"role": "user/assistant", "content": "..."}], // 可选(开发环境)
"chat_history": [{"role": "user/assistant", "content": "..."}], // 可选(生产环境)
"collections": ["public_kb"], // 可选,知识库列表
"session_id": "xxx" // 可选会话ID
}
SSE 事件序列:
1. start: 开始处理
2. sources: 检索到的来源
3. chunk: 每个 token
4. finish: 完成响应(包含完整 answer 和 sources
5. error: 错误事件
"""
import re
from config import (
IS_PROD, IS_DEV, ENABLE_SESSION,
RAG_SEARCH_TOP_K, 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
)
data = request.json or {}
message = data.get('message')
# 兼容两种参数名history和 chat_history
# 用 is not None 判断,避免 [] 被 or 吞掉
if 'chat_history' in data:
history = data['chat_history']
elif 'history' in data:
history = data['history']
else:
history = None
collections = data.get('collections')
session_id = data.get('session_id')
if not message:
return error_response("MISSING_PARAMS", BAD_REQUEST, "缺少 message", http_status=400)
# 生产环境强制校验 chat_history
if IS_PROD and history is None:
return error_response("MISSING_HISTORY", BAD_REQUEST, "chat_history is required in production", http_status=400)
# 输入安全验证
is_valid, reason = validate_query(message)
if not is_valid:
return error_response("INVALID_QUERY", BAD_REQUEST, reason, http_status=400)
# 如果没有指定 collections使用默认的公开库
if not collections:
collections = ['public_kb']
# ==================== 会话历史加载 ====================
# 优先使用传入的 history否则从 session_repo 加载
user_id = request.current_user.get("user_id")
if history is not None:
# 使用传入的历史(生产环境必须传入)
pass
elif session_id and ENABLE_SESSION:
# 开发环境:从本地数据库加载
try:
session_repo = current_app.session_repo
history = session_repo.get_history(session_id)
# 限制历史长度
history = history[-MAX_HISTORY_ROUNDS:] if len(history) > MAX_HISTORY_ROUNDS else history
except Exception as e:
logger.debug(f"解析历史记录失败: {e}")
history = []
else:
history = []
# 如果没有 session_id创建新会话仅开发环境
if not session_id and ENABLE_SESSION:
try:
session_repo = current_app.session_repo
session_id = session_repo.create_session(user_id)
except Exception as e:
logger.debug(f"创建会话失败: {e}")
# ==================== collections 历史推断(开发环境) ====================
# 当 collections 未显式指定(或仅为默认 public_kb且有会话历史时
# 从历史消息中推断上次使用的 KB自动恢复以避免用户忘切 KB
if (not collections or collections == ['public_kb']) and history:
for _msg in reversed(history):
if _msg.get("role") == "assistant":
_meta = _msg.get("metadata", {})
if isinstance(_meta, dict) and _meta.get("collections"):
collections = _meta["collections"]
logger.info(f"[KB推断] 从历史推断 collections: {collections}")
break
# 提前获取 session_repo 引用,避免在生成器内部访问 current_app
# (生成器执行时应用上下文可能已结束)
session_repo_ref = None
if ENABLE_SESSION:
try:
session_repo_ref = current_app.session_repo
except Exception as e:
logger.debug(f"获取会话仓库失败: {e}")
def generate():
"""生成 SSE 流"""
import re
start_time = _time.time()
full_answer = []
try:
# 0. 意图分析(改写 + 双层判断)
context_images = []
if history:
# 从历史中提取图片信息
for msg in reversed(history[-IMAGE_CONTEXT_HISTORY:]):
metadata = msg.get("metadata", {})
if isinstance(metadata, dict):
images = metadata.get("images", [])
if images:
context_images.extend(images[:3])
intent = None
try:
from core.intent_analyzer import analyze_intent
intent = analyze_intent(message, history or [], context_images)
logger.info(f"[意图分析] use_context={intent.use_context}, need_retrieval={intent.need_retrieval}, intent={intent.intent}, sub_queries={intent.sub_queries}")
# 调试事件:意图分析结果
if IS_DEV:
yield f"data: {json.dumps({'type': 'intent_result', 'data': {'intent': intent.intent, 'confidence': round(intent.confidence, 2), 'rewritten_query': intent.rewritten_query, 'sub_queries': intent.sub_queries, 'need_retrieval': intent.need_retrieval, 'reason': intent.reason}}, ensure_ascii=False)}\n\n"
# 如果不需要检索,直接使用上下文回答
if not intent.need_retrieval and intent.use_context:
yield f"data: {json.dumps({'type': 'start', 'message': '正在分析...'}, ensure_ascii=False)}\n\n"
# 构建上下文
context_text = ""
if history:
# 提取最近的助手回答
for msg in reversed(history):
if msg.get("role") == "assistant":
context_text = msg.get("content", "")
break
# 构建图片上下文
image_context = ""
if context_images:
image_context = "\n\n【上下文中的图片】\n"
for img in context_images[:5]:
if isinstance(img, dict):
desc = img.get("description", "")
img_type = img.get("type", "图片")
image_context += f"- {img_type}: {desc}\n"
# 直接调用 LLM
from config import get_llm_client, DASHSCOPE_MODEL
client = get_llm_client()
system_prompt = f"""你是一个专业的知识库问答助手。请根据对话历史和上下文回答用户问题。
如果用户问题是关于图片的,请根据上下文中的图片描述进行分析。
{image_context}"""
user_prompt = f"""对话历史:
{context_text[:DIRECT_CONTEXT_MAX_CHARS] if context_text else '(无历史上下文)'}
用户问题:{intent.rewritten_query}
请直接回答用户问题。"""
# 流式生成回答
for content in call_llm_stream(
client,
prompt=user_prompt,
model=DASHSCOPE_MODEL,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=LLM_TEMPERATURE
):
full_answer.append(content)
yield f"data: {json.dumps({'type': 'chunk', 'content': content}, ensure_ascii=False)}\n\n"
# 发送完成事件
yield f"data: {json.dumps({'type': 'finish', 'answer': ''.join(full_answer), 'sources': []}, ensure_ascii=False)}\n\n"
return # 直接返回,不执行后续检索
except Exception as e:
logger.warning(f"意图分析失败: {e},继续执行检索流程")
# 构建检索查询:使用改写后的完整问题(解决追问偏离问题)
retrieval_query = intent.rewritten_query if (intent and intent.rewritten_query) else message
# 1.5 语义缓存检查(跳过检索+生成全流程)
_semantic_cache_emb = None
try:
from core.semantic_cache import get_semantic_cache
from core.engine import get_engine as _get_eng
_sc = get_semantic_cache()
_eng = _get_eng()
if _sc and _eng and hasattr(_eng, 'embedding_model'):
# 缓存 key 使用 retrieval_query + collections避免不同上下文的追问命中错误缓存
_cache_collections = ','.join(sorted(collections)) if collections else ''
_cache_key_text = f"{retrieval_query}|{_cache_collections}"
_semantic_cache_emb = _eng.embedding_model.encode(_cache_key_text)
cached = _sc.get(_semantic_cache_emb)
# 防御性校验:必须是 RAG 回答缓存(非 intent_analyzer 缓存),且回答非空
if cached is not None and cached.get("cache_type") == "rag_answer" and cached.get("answer"):
logger.info(f"[语义缓存] 命中: {message[:50]}...")
cached_answer = cached.get("answer", "")
# 流式返回缓存的答案
yield f"data: {json.dumps({'type': 'start', 'message': '正在检索知识库...'}, ensure_ascii=False)}\n\n"
# 分块发送缓存答案
chunk_size = 20
for i in range(0, len(cached_answer), chunk_size):
chunk = cached_answer[i:i+chunk_size]
full_answer.append(chunk)
yield f"data: {json.dumps({'type': 'chunk', 'content': chunk}, ensure_ascii=False)}\n\n"
finish_event = {
"type": "finish",
"answer": cached_answer,
"mode": "rag",
"session_id": session_id,
"sources": cached.get("sources", []),
"citations": cached.get("citations", []),
"images": cached.get("images", []),
"tables": cached.get("tables", []),
"sections": [],
"duration_ms": int((_time.time() - start_time) * 1000),
"confidence_score": 1.0,
"semantic_cache_hit": True
}
yield f"data: {json.dumps(finish_event, ensure_ascii=False)}\n\n"
return
except Exception as e:
logger.warning(f"[语义缓存] 检查失败: {e}")
_semantic_cache_emb = None
# 1. 发送开始事件
yield f"data: {json.dumps({'type': 'start', 'message': '正在检索知识库...'}, ensure_ascii=False)}\n\n"
# 2. 执行混合检索(扩大召回数量,确保图片切片有机会被召回)
# 如果意图分析生成了子查询(对比类),传给搜索引擎并行检索
sub_queries = None
if intent and intent.sub_queries and len(intent.sub_queries) > 1:
sub_queries = intent.sub_queries
search_result = search_hybrid(
retrieval_query,
top_k=RAG_SEARCH_TOP_K,
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, retrieval_query)
has_figure_query = bool(figure_matches)
# 数据驱动:检查检索结果中是否有图片/图表类型切片
has_image_data = any(
m.get('chunk_type') in ('image', 'chart') for m in metas
)
has_image_intent = has_figure_query or has_image_data
# 给图片/图表切片打 boost 标记,供后续 select_images 使用
if has_image_intent:
for i, (doc, meta, score) in enumerate(zip(docs, metas, scores)):
if meta.get('chunk_type') in ('image', 'chart'):
# 检查 caption 是否与查询相关
caption = meta.get('caption', '') or ''
should_boost = False
boost_factor = 1.0
# 如果查询包含图片编号,检查 caption 是否匹配
if has_figure_query:
for fig_num in figure_matches:
if f"{fig_num}" in caption or f"{fig_num}" in caption:
should_boost = True
boost_factor = 2.0 # 强提升
break
# 或者 caption 与查询有足够重叠
if not should_boost:
overlap = len(set(message) & set(caption))
if overlap >= 3 or any(word in caption for word in message if len(word) >= 3):
should_boost = True
boost_factor = 1.5 # 提升 50%
if should_boost:
meta['_image_boost'] = boost_factor # 打标记,不重排
# 不做 sort保持检索引擎的原始排序
# 按 source 去重,保留最高分
seen_sources = {}
for rank, (doc, meta, score) in enumerate(zip(docs, metas, scores)):
meta['_retrieval_rank'] = rank
# 确保 _collection 字段存在(单知识库路径下 ChromaDB 原生不返回此字段)
if not meta.get('_collection'):
# 优先使用入库时写入的 collection 字段
meta['_collection'] = meta.get('collection') or (collections[0] if collections else 'public_kb')
source_name = meta.get('source', '未知')
if source_name not in seen_sources or score > seen_sources[source_name]['score']:
doc_type = meta.get('doc_type', 'other')
page = meta.get('page', 0)
page_end = meta.get('page_end')
# 仅 PDF 的页码是真实可靠的Word 等的页码是 MinerU 合成的,无意义,不外露
if doc_type == 'pdf' and page:
page_range = f"{page}-{page_end}" if (page_end and page_end > page) else str(page)
else:
page = None
page_end = None
page_range = ''
seen_sources[source_name] = {
'source': source_name,
'page': page,
'page_end': page_end,
'page_range': page_range,
'section': meta.get('section', '') or meta.get('section_path', ''),
'chunk_type': meta.get('chunk_type', 'text'),
'doc_type': doc_type, # 文档类型
'section_chunk_id': meta.get('section_chunk_id'), # 章节内序号
'score': round(score, 3) if isinstance(score, float) else score
}
# ========== P1图片使用 full_description ==========
# 图片切片使用完整描述(用于 LLM 上下文),而非短摘要
display_doc = doc
if meta.get('chunk_type') in ('image', 'chart', 'table'):
full_desc = meta.get('full_description', '')
if full_desc:
display_doc = full_desc
# contexts 仍然保留所有结果用于生成答案
contexts.append({'doc': display_doc, 'meta': meta, 'score': score})
sources = list(seen_sources.values())
# 调试事件:召回切片详情
if IS_DEV:
chunks_debug = []
for i, ctx in enumerate(contexts[:20]):
m = ctx.get('meta', {})
chunks_debug.append({
'rank': i + 1,
'source': m.get('source', ''),
'page': m.get('page', 0),
'chunk_type': m.get('chunk_type', 'text'),
'section': m.get('section', ''),
'score': round(ctx.get('score', 0), 4) if ctx.get('score') else None,
'content': (ctx.get('doc', '') or '')[:300]
})
yield f"data: {json.dumps({'type': 'chunks_retrieved', 'data': {'count': len(contexts), 'chunks': chunks_debug}}, ensure_ascii=False)}\n\n"
# 补充检索:从文本切片中提取图号/表号引用,补充检索对应的图片
# 重要:只从最相关的 top 5 文本切片提取引用,避免不相关引用干扰
import re
referenced_figures = set()
referenced_tables = set()
# 只检查 top 5 文本切片(与 select_images 逻辑一致)
text_contexts = [ctx for ctx in contexts if ctx.get('meta', {}).get('chunk_type') == 'text'][:5]
for ctx in text_contexts:
doc_text = ctx.get('doc', '')
fig_refs = re.findall(r'(?:[见如及和与])?图\s*(\d+\.?\d*)', doc_text)
referenced_figures.update(fig_refs)
table_refs = re.findall(r'(?:[见如及和与])?表\s*(\d+\.?\d*)', doc_text)
referenced_tables.update(table_refs)
# 检查哪些图号/表号对应的图片不在 contexts 中
existing_figure_images = set()
existing_table_images = set()
for ctx in contexts:
doc = ctx.get('doc', '')
meta = ctx.get('meta', {})
if meta.get('chunk_type') in ('image', 'chart'):
for fig_num in referenced_figures:
if f"{fig_num}" in doc:
existing_figure_images.add(fig_num)
for table_num in referenced_tables:
if f"{table_num}" in doc:
existing_table_images.add(table_num)
# 需要补充检索的图号/表号
missing_figures = referenced_figures - existing_figure_images
missing_tables = referenced_tables - existing_table_images
# 计算主要章节路径(用于补充检索过滤)
primary_section_paths_for_supp = set()
for ctx in text_contexts[:3]:
section = ctx.get('meta', {}).get('section', '') or ctx.get('meta', {}).get('section_path', '')
if section:
primary_section_paths_for_supp.add(section)
if missing_figures or missing_tables:
# 补充检索
from knowledge.manager import get_kb_manager
kb_manager = get_kb_manager()
kb_name = collections[0] if collections else 'public_kb'
collection = kb_manager.get_collection(kb_name)
if collection:
# 构建补充查询
supplement_queries = []
for fig_num in missing_figures:
supplement_queries.append(f"{fig_num}")
for table_num in missing_tables:
supplement_queries.append(f"{table_num}")
supplement_query = " ".join(supplement_queries)
# 使用 embedding 检索
# P4复用 engine 的 embedding 模型,避免重复加载
try:
from core.engine import get_engine
engine = get_engine()
query_vector = engine.embedding_model.encode(supplement_query).tolist()
if isinstance(query_vector[0], list):
query_vector = query_vector[0]
supplement_result = collection.query(
query_embeddings=[query_vector],
n_results=10,
include=['documents', 'metadatas', 'distances']
)
# 添加匹配的图片切片
for supp_doc, supp_meta, supp_dist in zip(
supplement_result['documents'][0],
supplement_result['metadatas'][0],
supplement_result['distances'][0]
):
chunk_type = supp_meta.get('chunk_type', '')
if chunk_type in ('image', 'chart'):
# 检查是否匹配缺失的图号/表号
is_match = False
matched_fig = None
for fig_num in missing_figures:
if f"{fig_num}" in supp_doc:
is_match = True
matched_fig = fig_num
break
for table_num in missing_tables:
if f"{table_num}" in supp_doc:
is_match = True
break
if is_match:
# 额外检查:图片章节是否与主要章节匹配
# 使用层级相似度判断,无需硬编码格式假设
supp_section = supp_meta.get('section', '') or supp_meta.get('section_path', '')
if primary_section_paths_for_supp and supp_section:
supp_max_sim = max(
(_section_similarity(supp_section, ps) for ps in primary_section_paths_for_supp),
default=0.0
)
if supp_max_sim < 0.3:
continue
elif primary_section_paths_for_supp and not supp_section:
# 补充检索的图片无章节信息时优雅降级:跳过
continue
# Bug 6a 修复:补充检索的图片也要做 full_description 替换
# 与正常检索保持一致
display_doc = supp_doc
if supp_meta.get('chunk_type') in ('image', 'chart', 'table'):
full_desc = supp_meta.get('full_description', '')
if full_desc:
display_doc = full_desc
contexts.append({
'doc': display_doc,
'meta': {**supp_meta, '_collection': supp_meta.get('_collection') or (collections[0] if collections else 'public_kb')},
'score': 1.0 - supp_dist
})
logger.info(f"[补充检索] 添加图片: {supp_meta.get('image_path', '')}")
except Exception as e:
logger.warning(f"补充检索失败: {e}")
# 发送来源事件
yield f"data: {json.dumps({'type': 'sources', 'sources': sources[:MAX_SOURCES_RETURNED]}, ensure_ascii=False)}\n\n"
# 调试:检查 contexts 中是否有图片切片
image_count = sum(1 for ctx in contexts if ctx.get('meta', {}).get('chunk_type') in ('image', 'chart'))
if image_count > 0:
logger.info(f"[图片检索] contexts 中包含 {image_count} 个图片/图表切片")
for ctx in contexts:
meta = ctx.get('meta', {})
if meta.get('chunk_type') in ('image', 'chart'):
logging.info(f" - 图片: {meta.get('caption', '')[:50]}, path: {meta.get('image_path', '')}")
# 2.5. 懒加载增强Phase 4
# 暂时禁用VLM 调用耗时过长,可能导致请求超时
# 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, retrieval_query, kb_name))
except Exception as e:
logger.warning(f"懒加载增强失败: {e}")
# 3. 选择要展示的图片Phase 5
selected_images = select_images(contexts, retrieval_query)
# 调试事件:图片选择详情
if IS_DEV:
yield f"data: {json.dumps({'type': 'images_selected', 'data': {'total_scored': len([c for c in contexts if c.get('meta',{}).get('chunk_type') in ('image','chart','table')]), 'selected_count': len(selected_images), 'images': [{'source': img.get('source',''), 'page': img.get('page',0), 'score': round(img.get('_image_boost', 1.0), 2)} for img in selected_images]}}, ensure_ascii=False)}\n\n"
# 4. 构建 promptPhase 6LLM 图片感知)
# Bug 1 修复:文本切片用于 top 5 名额竞争,图片描述不参与竞争
text_contexts = _order_text_contexts_for_prompt(contexts, retrieval_query, MAX_CONTEXT_CHUNKS,
min_score=RERANK_CONTEXT_MIN_SCORE)
# Phase 2按字符预算构建上下文
_is_comparison = intent and intent.intent == "comparison"
if _is_enum_query(retrieval_query) or _is_comparison:
# 列举类 / 对比类查询:保持原始顺序,不做预算截断
# 仍然注入 section_path 标题,帮助 LLM 区分不同章节
enum_parts = []
prev_section = None
for ctx in text_contexts:
meta = ctx.get('meta', {})
section = meta.get('section', '') or meta.get('section_path', '')
if section and section != prev_section:
enum_parts.append(f"{section}")
prev_section = section
# 精简表格切片:去除冗余的语义增强前缀
doc = _strip_semantic_prefix(ctx.get('doc', ''), meta.get('chunk_type', ''))
enum_parts.append(doc)
context_text = "\n\n".join(enum_parts)
else:
context_text = _build_context_with_budget(text_contexts, CONTEXT_MAX_CHARS, CONTEXT_SOFT_LIMIT)
# 表格救援CrossEncoder 对表格评分偏低,导致表格被预算截断
# 当查询涉及表格但上下文中没有表格数据时,从被截断的切片中补回
context_text = _rescue_table_chunks(text_contexts, context_text, retrieval_query)
# Phase 4计算置信度分数top-3 平均 Rerank 分数)
_top_scores = [ctx.get('score', 0) for ctx in text_contexts[:3]]
_confidence_score = round(sum(_top_scores) / len(_top_scores), 4) if _top_scores else 0.0
# Bug 6b 优化:直接使用 selected_images 中的 full_description
# 这样 LLM 既能看到文本切片,也能知道图片内容
if selected_images:
# 区分表格嵌入图片和独立图片
has_table_embedded_images = any(
img.get('type') == 'table_image' for img in selected_images
)
# 检查上下文中是否有表格含嵌入图片
has_table_with_images = any(
ctx.get('meta', {}).get('chunk_type') == 'table'
and ctx.get('meta', {}).get('images_json')
for ctx in text_contexts
)
image_descriptions = []
for i, img in enumerate(selected_images, 1):
# 直接使用 select_images 时带上的 full_description
full_desc = img.get('full_description', '') or img.get('description', '')
if full_desc:
# 添加图片来源信息
img_source = img.get('source', '')
img_page = img.get('page', '')
source_info = f"(来源:{img_source}{img_page}页)" if img_source and img_page else ""
image_descriptions.append(f"【图片{i}{full_desc}{source_info}")
if image_descriptions:
context_text += "\n\n【相关图片信息】\n" + "\n\n".join(image_descriptions)
# 根据图片类型给出不同的回答指令
if has_table_embedded_images and has_table_with_images:
context_text += (
"\n\n【回答要求】参考资料中包含表格及其嵌入图片。"
"请以**表格形式**呈现数据(保持原始表格结构),"
"并在对应单元格中使用 `![图片](图片URL)` 格式嵌入图片。"
"不要将表格内容转为纯文本描述,不要把图片与表格分开展示。"
)
else:
# 添加指令让 LLM 介绍图片
context_text += "\n\n【回答要求】回答时请简要介绍每张图片的内容和用途。"
enhanced_context = context_text
# 对比类查询:添加结构化对比指令
if intent and intent.intent == "comparison" and intent.sub_queries:
comparison_instruction = (
"\n\n【回答要求】这是一个对比类问题。"
"请根据参考资料,从多个角度对比分析,"
"使用表格或分点形式清晰呈现差异和共同点。"
"如果参考资料中缺少某一方面的信息,请如实说明。"
)
enhanced_context = comparison_instruction + "\n\n" + enhanced_context
# 推理类查询:添加因果分析指令
elif intent and intent.intent == "reasoning":
reasoning_instruction = (
"\n\n【回答要求】这是一个需要分析原因或推理的问题。"
"请根据参考资料,先梳理相关事实和数据,再给出逻辑清晰的分析。"
"如果涉及因果关系,请明确标注原因和结果;如果资料不足以支撑推理,请如实说明。"
)
enhanced_context = reasoning_instruction + "\n\n" + enhanced_context
# 操作指导类查询:添加步骤化指令
elif intent and intent.intent == "instruction":
instruction_instruction = (
"\n\n【回答要求】这是一个操作指导类问题。"
"请根据参考资料,以清晰的步骤或流程形式组织回答。"
"如有前置条件或注意事项,请在步骤前说明。"
)
enhanced_context = instruction_instruction + "\n\n" + enhanced_context
if _is_enum_query(retrieval_query):
enum_instruction = (
"\n\n【回答要求】如果参考资料中包含编号列表、禁止情形、要求或条款,"
"请按资料中的原始顺序完整列出;不要合并相邻条目,不要跳项,"
"资料不足时明确说明缺少哪部分依据。\n\n"
)
enhanced_context = enum_instruction + enhanced_context
# Phase 4根据置信度注入谨慎回答指令
if _confidence_score < CONFIDENCE_WARN_THRESHOLD:
enhanced_context += (
"\n\n【重要提示】参考资料与问题的相关性较低。"
"请仅基于参考资料中明确包含的信息回答,"
'如果资料不足以回答问题,请直接说明"知识库中未找到直接相关的信息"'
)
elif _confidence_score < CONFIDENCE_CAUTION_THRESHOLD:
enhanced_context += (
"\n\n【提示】参考资料的相关性一般,请优先引用资料中的原文,避免推测。"
)
# 调试事件:最终上下文
if IS_DEV:
text_used = text_contexts
_scores = [round(ctx.get('score', 0), 4) for ctx in text_used]
yield f"data: {json.dumps({'type': 'context_built', 'data': {'chunk_count': len(text_used), 'context_length': len(enhanced_context), 'budget_max_chars': CONTEXT_MAX_CHARS, 'min_score_filter': RERANK_CONTEXT_MIN_SCORE, 'confidence_top3': _confidence_score, 'score_stats': {'max': max(_scores) if _scores else 0, 'min': min(_scores) if _scores else 0, 'avg': round(sum(_scores)/len(_scores), 4) if _scores else 0}, 'context_preview': enhanced_context[:500], 'chunks_used': [{'source': ctx.get('meta',{}).get('source',''), 'page': ctx.get('meta',{}).get('page',0), 'score': ctx.get('score',0), 'preview': (ctx.get('doc','') or '')[:100]} for ctx in text_used]}}, ensure_ascii=False)}\n\n"
# 5. 流式生成回答
from core.engine import get_engine
engine = get_engine()
for token in engine.generate_answer_stream(message, enhanced_context, history):
full_answer.append(token)
yield f"data: {json.dumps({'type': 'chunk', 'content': token}, ensure_ascii=False)}\n\n"
# 6. P0答案对齐过滤器
# 从 LLM 回答中提取图号引用,过滤图片选择结果
full_answer_text = "".join(full_answer)
# 提取回答中引用的图号/表号
mentioned = set()
# 中文图号图2.1、图 2-1、见图2.1 等
mentioned.update(re.findall(r'(?:[见如])?图\s*(\d+[\.\-]?\d*)', full_answer_text))
# 中文表号表2.1、表 2-1、见表2.1 等
mentioned.update(re.findall(r'(?:[见如])?表\s*(\d+[\.\-]?\d*)', full_answer_text))
# 英文图号Figure 2.1、Fig.2.1 等
mentioned.update(re.findall(r'(?:Fig(?:ure)?\.?\s*)(\d+[\.\-]?\d*)', full_answer_text, re.I))
# 根据回答中的引用过滤图片
if mentioned:
aligned_images = []
for img in selected_images:
desc = img.get('description', '')
# 标准化图号格式(将连字符转为点)
for ref in mentioned:
ref_normalized = ref.replace('-', '.')
if (f"{ref_normalized}" in desc or
f"{ref_normalized}" in desc or
f"{ref_normalized}" in desc or
f"{ref_normalized}" in desc):
aligned_images.append(img)
break
# 如果有匹配的图片,使用对齐后的结果
if aligned_images:
selected_images = aligned_images
# else: 没有匹配到保留原选择不再截断到1张
# else: LLM 没有提图号保留原选择不再截断到1张
# 后置图片过滤:用回答内容反向筛选图片,确保图片与回答一致
selected_images = _filter_images_by_answer(selected_images, full_answer_text)
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. 保存消息到会话(仅开发环境)
if session_id and session_repo_ref:
try:
# 保存用户消息
session_repo_ref.add_message(session_id, 'user', message)
# 保存 AI 回答(包含完整 metadata图片、来源、引用等
assistant_metadata = {
'is_rag': True,
'mode': 'rag',
}
if rich_media.get('images'):
assistant_metadata['images'] = rich_media['images']
if sources:
assistant_metadata['sources'] = sources
if citation_result.get('citations'):
assistant_metadata['citations'] = citation_result['citations']
# 记录本次检索使用的向量库(用于后续追问时自动恢复 KB 选择)
assistant_metadata['collections'] = collections
session_repo_ref.add_message(session_id, 'assistant', filtered_answer, assistant_metadata)
# 更新会话最后活跃时间
if hasattr(session_repo_ref, 'update_last_active'):
session_repo_ref.update_last_active(session_id)
except Exception as e:
logger.warning(f"保存会话消息失败: {e}")
# 10. 发送完成事件
duration_ms = int((_time.time() - start_time) * 1000)
finish_event = {
"type": "finish",
"answer": filtered_answer,
"mode": "rag",
"session_id": session_id,
"sources": sources,
"citations": citation_result.get("citations", []), # 结构化引用列表
"images": rich_media["images"],
"tables": rich_media["tables"],
"sections": rich_media["sections"],
"duration_ms": duration_ms,
"confidence_score": _confidence_score # Phase 4top-3 平均 Rerank 分数
}
# 添加分阶段耗时信息(仅开发环境)
if IS_DEV and hasattr(search_result, 'get'):
debug_info = search_result.get('_debug', {})
timing_info = debug_info.get('timing', {})
# 从 _debug steps 中提取 Rerank 耗时
rerank_time = 0
rerank_cached = False
for step in debug_info.get('steps', []):
if step.get('name') == 'rerank' and step.get('applied'):
rerank_time = step.get('time_ms', 0)
rerank_cached = step.get('cached', False)
finish_event["timing"] = {
"total_search_ms": timing_info.get('total_ms', 0),
"rerank_ms": rerank_time,
"rerank_cached": rerank_cached,
"total_ms": duration_ms
}
# 10.5 写入语义缓存
try:
from core.semantic_cache import get_semantic_cache
_sc = get_semantic_cache()
if _sc and filtered_answer:
if _semantic_cache_emb is None:
from core.engine import get_engine as _get_eng
_eng = _get_eng()
if _eng and hasattr(_eng, 'embedding_model'):
_cache_collections = ','.join(sorted(collections)) if collections else ''
_cache_key_text = f"{retrieval_query}|{_cache_collections}"
_semantic_cache_emb = _eng.embedding_model.encode(_cache_key_text)
if _semantic_cache_emb is not None:
_sc.set(_semantic_cache_emb, {
"cache_type": "rag_answer",
"answer": filtered_answer,
"sources": sources,
"citations": citation_result.get("citations", []),
"images": rich_media.get("images", []),
"tables": rich_media.get("tables", []),
})
logger.debug(f"[语义缓存] 写入成功: {message[:50]}...")
except Exception as e:
logger.info(f"[语义缓存] 写入失败: {e}")
yield f"data: {json.dumps(finish_event, ensure_ascii=False)}\n\n"
except Exception as e:
import logging as _logging
_logging.getLogger(__name__).error(f"[SSE] RAG 流异常: {e}", exc_info=True)
error_event = {
"type": "error",
"message": "服务内部错误,请稍后重试"
}
yield f"data: {json.dumps(error_event, ensure_ascii=False)}\n\n"
return Response(
generate(),
mimetype='text/event-stream',
headers={
'Cache-Control': 'no-cache',
'X-Accel-Buffering': 'no'
}
)
@chat_bp.route('/search', methods=['POST'])
@require_gateway_auth
def search():
"""
混合检索接口 - 供 Dify 工作流调用
请求体:
{
"query": "查询文本",
"top_k": 5,
"collections": ["public_kb"] // 可选
}
"""
data = request.json or {}
query = data.get('query', '')
query = sanitize_user_input(query)
injection_matches = detect_injection(query)
if injection_matches:
logger.warning(f"[Chat] 检测到可疑注入: {injection_matches}")
top_k = data.get('top_k', 5)
collections = data.get('collections') # 后端传入的知识库列表
if not query:
return error_response("MISSING_PARAMS", BAD_REQUEST, "query is required", http_status=400)
# 输入安全校验(注入检测、违禁词、长度限制)
is_valid, reason = validate_query(query)
if not is_valid:
return error_response("INVALID_QUERY", BAD_REQUEST, reason, http_status=400)
# top_k 范围校验
try:
top_k = max(1, min(int(top_k), 50))
except (ValueError, TypeError):
top_k = 5
# 如果没有指定 collections使用默认的公开库
if not collections:
collections = ['public_kb']
results = search_hybrid(query, top_k=top_k, allowed_collections=collections)
return success_response(data={
'contexts': results['documents'][0],
'metadatas': results['metadatas'][0],
'scores': results['scores'][0]
})
# ==================== 缓存调试接口(临时) ====================
@chat_bp.route('/cache/stats', methods=['GET'])
def cache_stats():
"""缓存统计接口(调试用)"""
stats = {}
try:
from core.cache import get_cache_manager
cache = get_cache_manager()
all_stats = cache.get_all_stats()
for name, s in all_stats.items():
stats[name] = {
'total_entries': s.total_entries,
'hits': s.hits,
'misses': s.misses,
'hit_rate': f"{s.hit_rate:.2%}",
'evictions': s.evictions,
}
except Exception as e:
stats['_error'] = str(e)
# 语义缓存统计(全局单例 + 意图分析器实例)
try:
from core.semantic_cache import get_semantic_cache
sc = get_semantic_cache()
if sc:
stats['semantic_cache'] = sc.get_stats()
except Exception:
pass
try:
from core.intent_analyzer import IntentAnalyzer
if hasattr(IntentAnalyzer, '_instance'):
ia = IntentAnalyzer._instance
if hasattr(ia, 'semantic_cache') and ia.semantic_cache:
stats['semantic_cache_intent'] = ia.semantic_cache.get_stats()
except Exception:
pass
return jsonify(stats)
@chat_bp.route('/cache/clear', methods=['POST'])
def cache_clear():
"""缓存清空接口(调试用)"""
result = {}
try:
from core.cache import get_cache_manager
cache = get_cache_manager()
cache.clear_all()
result['lru_cache'] = 'cleared'
except Exception as e:
result['lru_cache'] = f'error: {e}'
try:
from core.semantic_cache import get_semantic_cache
sc = get_semantic_cache()
sc.clear()
result['semantic_cache'] = 'cleared'
except Exception:
result['semantic_cache'] = 'not_available'
return jsonify({'status': 'ok', 'cleared': result})