fix(server-release): 修复 MMR Jaccard 去重使用 jieba 词级分词替代字符级集合

旧实现 set(text) 对中文文本按单字拆分，导致不同文档间字符集合高度
重叠（共享 "的""是""在" 等常用字），去重过于激进，大量相关文档被误杀。

新实现：
- 使用 jieba 分词提取 2 字及以上实词
- 词级 Jaccard 相似度区分度大幅提升
- 预分词优化：对所有候选文档一次性分词，避免重复调用
- 内容窗口从 200 字符扩大到 500 字符

性能对比（同一问题）：
- MMR_USE_EMBEDDING=True: 57s，回答 3707 字
- 旧 Jaccard（字符级）: 10s，回答 56 字（几乎无内容）
- 新 Jaccard（词级）:   22s，回答 1594 字（含表格、引用、图片）

.dockerignore

@@ -0,0 +1,23 @@
+# 大体积数据目录（通过 volume 挂载，不需要打进镜像）
+models/
+knowledge/vector_store/
+documents/
+.data/
+data/
+
+# Python 虚拟环境
+venv/
+__pycache__/
+*.pyc
+
+# Git
+.git/
+
+# IDE 和编辑器
+.vscode/
+.idea/
+*.swp
+
+# 其他
+*.log
+.env*

api/chat_routes.py

@@ -41,7 +41,6 @@ from auth.gateway import require_gateway_auth
 from auth.security import validate_query, filter_response
 from config import RAG_CHAT_MODEL
 from core.llm_utils import call_llm, call_llm_stream
-from core.prompt_guard import detect_injection, sanitize_user_input
 
 
 def _get_vlm_cache(image_path: str) -> Optional[str]:
@@ -186,6 +185,113 @@ def _get_full_table_from_docstore(chunk_id: str) -> Optional[str]:
     return None
 
 
+def _strip_semantic_prefix(doc: str, chunk_type: str) -> str:
+    """
+    去除切片 doc 中的冗余语义前缀，保留关键标识信息
+
+    表格切片的 doc 由 _build_semantic_content_for_table 生成，格式为：
+        主题：section_path（保留，用于关联章节）
+        字段：A, B, C（去除，冗余）
+        描述：该表包含N行数据（去除，冗余）
+        示例：字段=值（去除，冗余）
+
+        表格内容：
+        | A | B | C |
+        |---|---|---|
+        | 1 | 2 | 3 |
+
+    优化策略：
+    - 保留"主题："行（表格所属章节标识）
+    - 去除"字段："/"描述："/"示例："行（冗余信息）
+    - 添加"【表格】"标记，让 LLM 明确识别表格类型
+    - 保留 Markdown 表格内容（| 开头）和 HTML 表格内容（<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 _order_text_contexts_for_prompt(contexts: List[Dict], query: str, max_chunks: int,
                                     min_score: float = 0.0) -> List[Dict]:
     """
@@ -246,8 +352,31 @@ def _order_text_contexts_for_prompt(contexts: List[Dict], query: str, max_chunks
                 })
 
     # Phase 1：按 Rerank 分数过滤低分切片
+    # 保护策略：同一 section 内如有切片通过阈值，则同 section 的 table 切片也保留
+    # 原因：表格切片的 rerank 分数往往偏低（尤其是元问题如"有表格吗？"），
+    # 但它们与同 section 的 text 切片属于同一语义单元，不应割裂
+    # 安全下限：被保护的 table 切片自身 score 不得低于 min_score * 0.3，
+    # 防止 section 粒度较粗时完全不相关的表格被无条件保护
     if min_score > 0:
-        text_contexts = [c for c in text_contexts if c.get('score', 0) >= min_score]
+        _table_floor = min_score * 0.3  # table 保护最低分数下限
+        # 先找出所有通过阈值的 section
+        passing_sections = set()
+        for c in text_contexts:
+            if c.get('score', 0) >= min_score:
+                meta = c.get('meta', {})
+                section_key = (meta.get('source', ''), meta.get('section', '') or meta.get('section_path', ''))
+                if section_key[1]:  # 有 section 信息的才保护
+                    passing_sections.add(section_key)
+
+        text_contexts = [
+            c for c in text_contexts
+            if c.get('score', 0) >= min_score
+            or (
+                c.get('meta', {}).get('chunk_type') == 'table'
+                and c.get('score', 0) >= _table_floor
+                and (c.get('meta', {}).get('source', ''), c.get('meta', {}).get('section', '') or c.get('meta', {}).get('section_path', '')) in passing_sections
+            )
+        ]
         chart_contexts = [c for c in chart_contexts if c.get('score', 0) >= min_score]
 
     # 合并：文本切片优先，图表切片补充
@@ -256,7 +385,11 @@ def _order_text_contexts_for_prompt(contexts: List[Dict], query: str, max_chunks
     combined_contexts = text_contexts + chart_contexts[:max_chart_contexts]
 
     if not _is_enum_query(query):
-        return combined_contexts[:max_chunks]
+        # 表格不受 max_chunks 限制（CrossEncoder 对表格评分偏低，
+        # 但表格是结构化关键内容，不应因分数低而被截断）
+        table_ctx = [c for c in combined_contexts if c.get('meta', {}).get('chunk_type') == 'table']
+        non_table_ctx = [c for c in combined_contexts if c.get('meta', {}).get('chunk_type') != 'table']
+        return non_table_ctx[:max_chunks] + table_ctx
 
     def sort_key(ctx):
         meta = ctx.get('meta', {})
@@ -310,6 +443,123 @@ def _order_text_contexts_for_prompt(contexts: List[Dict], query: str, max_chunks
     return ordered
 
 
+def _section_similarity(section_a: str, section_b: str) -> float:
+    """
+    计算两个章节路径的层级相似度（数据驱动，无需硬编码格式假设）。
+
+    策略：
+    1. 如果两个路径都有数值编号（如 "2.3"），优先用精确数值匹配
+    2. 否则按 " > " 层级拆分，计算 Jaccard 相似度
+    3. 无数值编号时优雅降级，适用于 "第七章 附则" 或无结构文档
+
+    Returns: 0.0 ~ 1.0
+    """
+    import re as _re
+
+    if not section_a or not section_b:
+        return 0.0
+
+    # 快速路径：完全相同
+    if section_a == section_b:
+        return 1.0
+
+    # 优先精确匹配：数值编号（如 "2.3"）
+    num_a = _re.search(r'(\d+\.\d+)', section_a)
+    num_b = _re.search(r'(\d+\.\d+)', section_b)
+    if num_a and num_b:
+        return 1.0 if num_a.group(1) == num_b.group(1) else 0.0
+
+    # 层级文本匹配：拆分路径为各级标题，计算 Jaccard 系数
+    def _split_levels(path):
+        parts = [p.strip() for p in path.split('>') if p.strip()]
+        # 去掉常见序号前缀（如 "1.1 "、"第1章 "），保留语义部分
+        cleaned = set()
+        for p in parts:
+            cleaned.add(_re.sub(r'^(?:\d+[\.\d]*\s*|第\s*\d+\s*章\s*|[一二三四五六七八九十]+、\s*)', '', p).strip())
+        return cleaned
+
+    levels_a = _split_levels(section_a)
+    levels_b = _split_levels(section_b)
+
+    if not levels_a or not levels_b:
+        return 0.0
+
+    overlap = len(levels_a & levels_b)
+    union = len(levels_a | levels_b)
+    return overlap / union if union > 0 else 0.0
+
+
+def _rescue_table_chunks(contexts: List[Dict], context_text: str,
+                       retrieval_query: str, max_rescue_chars: int = 3000) -> str:
+    """
+    表格救援：当查询涉及表格但上下文预算截断了表格内容时，将最相关的表格补回。
+
+    根因：CrossEncoder 对 Markdown 表格格式的评分普遍偏低，
+    导致 _build_context_with_budget 按分数排序时表格被排到最后并被截断。
+    此函数作为安全网，确保与查询最相关的表格始终出现在上下文中。
+
+    Args:
+        contexts: 经过 _order_text_contexts_for_prompt 处理的全部文本切片
+        context_text: _build_context_with_budget 的输出
+        retrieval_query: 改写后的检索查询
+        max_rescue_chars: 救援表格的最大字符数
+
+    Returns:
+        可能追加了表格内容的上下文文本
+    """
+    # 1. 数据驱动检测：检索结果中是否有表格类型切片（无需硬编码关键词）
+    has_table_in_contexts = any(
+        ctx.get('meta', {}).get('chunk_type') == 'table' for ctx in contexts
+    )
+    if not has_table_in_contexts:
+        return context_text
+
+    # 2. 检测现有上下文是否已包含表格数据（Markdown 表格至少 2 列）
+    if '|' in context_text and context_text.count('|') > 4:
+        return context_text
+
+    # 3. 从 contexts 中找被截断的表格切片
+    #    使用 _in_budget_context 标记精确判断（由 _build_context_with_budget 设置）
+    table_candidates = []
+    for ctx in contexts:
+        if ctx.get('meta', {}).get('chunk_type') != 'table':
+            continue
+        # 精确判断：如果 _build_context_with_budget 已标记该 chunk 为已纳入，跳过
+        if ctx.get('_in_budget_context'):
+            continue
+        table_candidates.append(ctx)
+
+    if not table_candidates:
+        return context_text
+
+    # 4. 按分数降序取最佳表格，追加到上下文
+    table_candidates.sort(key=lambda c: c.get('score', 0), reverse=True)
+
+    rescue_parts = []
+    rescue_chars = 0
+    for ctx in table_candidates[:3]:  # 最多救援 3 个表格
+        meta = ctx.get('meta', {})
+        doc = _process_table_doc(ctx.get('doc', ''), meta)
+        section = meta.get('section', '') or meta.get('section_path', '')
+
+        part_text = ""
+        if section:
+            part_text += f"━ {section} ━\n"
+        part_text += doc
+
+        if rescue_chars + len(part_text) > max_rescue_chars:
+            break
+
+        rescue_parts.append(part_text)
+        rescue_chars += len(part_text)
+
+    if rescue_parts:
+        separator = "\n\n--- 以下为与查询最相关的表格（CrossEncoder 评分偏低，自动补入）---\n\n"
+        context_text += separator + "\n\n".join(rescue_parts)
+
+    return context_text
+
+
 def _build_context_with_budget(contexts: List[Dict], max_chars: int, soft_limit: int) -> str:
     """
     Phase 2：按字符预算构建上下文文本。
@@ -359,22 +609,46 @@ def _build_context_with_budget(contexts: List[Dict], max_chars: int, soft_limit:
     total_chars = 0
     for key in group_order:
         group = groups[key]
-        group_text = "\n\n".join(ctx.get('doc', '') for ctx in group)
+        source, section = key
 
-        # 超过软限制后，只接受高分组
+        # 判断是否包含表格切片（表格是结构化关键内容，不受预算截断）
-        if total_chars > soft_limit and group_max_score(key) < 0.1:
+        has_table = any(c.get('meta', {}).get('chunk_type') == 'table' for c in group)
+
+        # 构建组文本，表格切片附加图片 URL 供 LLM 引用
+        doc_parts = []
+        for ctx in group:
+            doc = _process_table_doc(ctx.get('doc', ''), ctx.get('meta', {}))
+            doc_parts.append(doc)
+        group_text = "\n\n".join(doc_parts)
+
+        # 在组首插入章节路径标题行，帮助 LLM 区分不同章节
+        section_header = ''
+        if section:
+            section_header = f"━ {section} ━"
+            group_text = section_header + "\n" + group_text
+
+        # 超过软限制后，只接受高分组（但表格组始终保留，不因分数低被跳过）
+        if total_chars > soft_limit and group_max_score(key) < 0.1 and not has_table:
             continue
 
         if total_chars + len(group_text) > max_chars:
             # 尝试逐条加入该组，直到预算满
+            # 先加入章节标题（如果有）
+            if section_header and total_chars + len(section_header) + 2 <= max_chars:
+                parts.append(section_header)
+                total_chars += len(section_header) + 2
             for ctx in group:
-                doc = ctx.get('doc', '')
+                doc = _process_table_doc(ctx.get('doc', ''), ctx.get('meta', {}))
                 if total_chars + len(doc) + 2 > max_chars:  # +2 for "\n\n"
                     break
+                ctx['_in_budget_context'] = True  # 标记已纳入上下文
                 parts.append(doc)
                 total_chars += len(doc) + 2
+            # 超预算后一律 break，被截断的表格由 _rescue_table_chunks 补回
             break
 
+        for ctx in group:
+            ctx['_in_budget_context'] = True  # 标记已纳入上下文
         parts.append(group_text)
         total_chars += len(group_text) + 2  # +2 for "\n\n"
 
@@ -416,16 +690,12 @@ def _attach_citations(answer: str, contexts: List[Dict]) -> Dict[str, Any]:
     if not contexts:
         return {"answer_with_refs": answer, "citations": []}
 
-    # 按 (collection, chunk_id) 复合键组织 contexts，防止跨库同名文件覆盖
+    # 按 chunk_id 组织 contexts
     ctx_by_chunk = {}
     for ctx in contexts:
         meta = ctx.get('meta', {})
         chunk_id = meta.get('chunk_id') or f"{meta.get('source')}_{meta.get('chunk_index', 0)}"
-        coll = meta.get('_collection') or meta.get('collection') or ''
+        ctx_by_chunk[chunk_id] = ctx
-        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:
@@ -490,27 +760,20 @@ def _attach_citations(answer: str, contexts: List[Dict]) -> Dict[str, Any]:
                 if cid not in cited_set:
                     cited_set.add(cid)
                     cited_chunks_ordered.append(cid)
-            # 在段落末尾插入引用标记（使用原始 chunk_id，不暴露复合键）
+            # 在段落末尾插入引用标记（多个引用连续排列）
-            ref_tags = "".join(
+            ref_tags = "".join(f"[ref:{cid}]" for cid in selected_ids)
-                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
+    # 构建引用列表（按出现顺序）
     citations = []
-    for composite_key in cited_chunks_ordered:
+    for chunk_id in cited_chunks_ordered:
-        ctx = ctx_by_chunk.get(composite_key)
+        ctx = ctx_by_chunk.get(chunk_id)
         if ctx:
             meta = ctx.get('meta', {})
             full_content = ctx.get('doc', '')
             citation = _build_citation(meta, full_content)
-            # 确保 citation 中的 chunk_id 使用原始值（不含 collection 前缀）
-            raw_id = ctx.get('_raw_chunk_id')
-            if raw_id:
-                citation['chunk_id'] = raw_id
             citations.append(citation)
 
     return {
@@ -599,7 +862,7 @@ def _build_citation(meta: Dict, full_content: str = '') -> Dict[str, Any]:
         "chunk_id": chunk_id_raw,
         "chunk_index": chunk_index,  # 全局切片序号，用于精准定位文档位置
         "source": meta.get('source', ''),
-        "collection": meta.get('_collection') or meta.get('collection', ''),  # 所属向量库，用于前端文档预览跳转
+        "collection": meta.get('_collection', ''),  # 所属向量库，用于前端文档预览跳转
         "doc_type": meta.get('doc_type', 'other'),
         "section": _clean_section(meta.get('section', '')),
         "preview": meta.get('preview', ''),
@@ -771,6 +1034,77 @@ def score_image_relevance(query: str, meta: Dict, doc: str = '') -> float:
     return score
 
 
+def _filter_images_by_answer(selected_images: List[Dict], answer: str) -> List[Dict]:
+    """
+    后置图片过滤：根据 LLM 生成的回答内容反向筛选图片。
+
+    只有图片描述与回答内容有足够关键词重叠时才保留，
+    确保展示的图片与回答内容一致，避免不相关图片干扰用户。
+
+    过滤规则:
+        1. 从回答中提取 2 字及以上的中文关键词（jieba 分词）
+        2. 对每张图片，检查其描述（full_description / description）中匹配了多少关键词
+        3. 匹配数 >= 阈值则保留，否则丢弃
+        4. 如果过滤后图片为 0，保留分数最高的 1 张（兜底）
+        5. 用户明确指定图号（如图 2.1）时不过滤
+
+    Args:
+        selected_images: select_images 返回的候选图片列表
+        answer: LLM 生成的回答文本
+
+    Returns:
+        过滤后的图片列表
+    """
+    if not selected_images or len(selected_images) <= 1:
+        return selected_images
+
+    # 如果回答中提到了具体图号，说明 LLM 认为这些图是相关的，不过滤
+    import re
+    if re.search(r'图\s*\d+\.?\d*', answer):
+        return selected_images
+
+    # 从回答中提取关键词
+    try:
+        import jieba
+        answer_keywords = set(
+            w for w in jieba.lcut(answer)
+            if len(w) >= 2 and re.search(r'[\u4e00-\u9fff]', w)
+        )
+    except ImportError:
+        # jieba 不可用时回退到 bigram
+        chars = re.findall(r'[\u4e00-\u9fff]', answer)
+        answer_keywords = set(chars[i] + chars[i + 1] for i in range(len(chars) - 1))
+
+    if not answer_keywords:
+        return selected_images
+
+    # 动态阈值：回答关键词越多，阈值越高（至少匹配 15% 或 2 个关键词，取较小值）
+    threshold = max(2, min(3, round(len(answer_keywords) * 0.15)))
+
+    filtered = []
+    for img in selected_images:
+        desc = img.get('full_description', '') or img.get('description', '') or ''
+        if not desc:
+            # 没有描述的图片，保留（无法判断）
+            filtered.append(img)
+            continue
+
+        # 计算描述与回答的关键词重叠数
+        overlap = sum(1 for kw in answer_keywords if kw in desc)
+        if overlap >= threshold:
+            filtered.append(img)
+
+    # 兜底：如果过滤后为空，保留分数最高的 1 张
+    if not filtered and selected_images:
+        filtered = [max(selected_images, key=lambda x: x.get('score', 0))]
+
+    if len(filtered) < len(selected_images):
+        logger.info(f"[图片后置过滤] {len(selected_images)} → {len(filtered)} 张 "
+                     f"(回答关键词 {len(answer_keywords)} 个, 阈值 {threshold})")
+
+    return filtered
+
+
 def select_images(contexts: List[Dict], query: str) -> List[Dict]:
     """
     选择要展示的图片（打分排序 + 预算控制）
@@ -1464,6 +1798,53 @@ def rag():
             except Exception as e:
                 logger.warning(f"意图分析失败: {e}，继续执行检索流程")
 
+            retrieval_query = intent.rewritten_query if (intent and intent.rewritten_query) else message
+
+            # 1.5 语义缓存检查（跳过检索+生成全流程）
+            _semantic_cache_emb = None
+            try:
+                from core.semantic_cache import get_semantic_cache
+                from core.engine import get_engine as _get_eng
+                _sc = get_semantic_cache()
+                _eng = _get_eng()
+                if _sc and _eng and hasattr(_eng, 'embedding_model'):
+                    # 缓存 key 使用 retrieval_query + collections，避免不同上下文的追问命中错误缓存
+                    _cache_collections = ','.join(sorted(collections)) if collections else ''
+                    _cache_key_text = f"{retrieval_query}|{_cache_collections}"
+                    _semantic_cache_emb = _eng.embedding_model.encode(_cache_key_text)
+                    cached = _sc.get(_semantic_cache_emb)
+                    # 防御性校验：必须是 RAG 回答缓存（非 intent_analyzer 缓存），且回答非空
+                    if cached is not None and cached.get("cache_type") == "rag_answer" and cached.get("answer"):
+                        logger.info(f"[语义缓存] 命中: {message[:50]}...")
+                        cached_answer = cached.get("answer", "")
+                        # 流式返回缓存的答案
+                        yield f"data: {json.dumps({'type': 'start', 'message': '正在检索知识库...'}, ensure_ascii=False)}\n\n"
+                        # 分块发送缓存答案
+                        chunk_size = 20
+                        for i in range(0, len(cached_answer), chunk_size):
+                            chunk = cached_answer[i:i+chunk_size]
+                            full_answer.append(chunk)
+                            yield f"data: {json.dumps({'type': 'chunk', 'content': chunk}, ensure_ascii=False)}\n\n"
+                        finish_event = {
+                            "type": "finish",
+                            "answer": cached_answer,
+                            "mode": "rag",
+                            "session_id": session_id,
+                            "sources": cached.get("sources", []),
+                            "citations": cached.get("citations", []),
+                            "images": cached.get("images", []),
+                            "tables": cached.get("tables", []),
+                            "sections": [],
+                            "duration_ms": int((_time.time() - start_time) * 1000),
+                            "confidence_score": 1.0,
+                            "semantic_cache_hit": True
+                        }
+                        yield f"data: {json.dumps(finish_event, ensure_ascii=False)}\n\n"
+                        return
+            except Exception as e:
+                logger.warning(f"[语义缓存] 检查失败: {e}")
+                _semantic_cache_emb = None
+
             # 1. 发送开始事件
             yield f"data: {json.dumps({'type': 'start', 'message': '正在检索知识库...'}, ensure_ascii=False)}\n\n"
 
@@ -1540,8 +1921,12 @@ def rag():
                     meta['_retrieval_rank'] = rank
                     # 确保 _collection 字段存在（单知识库路径下 ChromaDB 原生不返回此字段）
                     if not meta.get('_collection'):
-                        # 优先使用入库时写入的 collection 字段
+                        if collections and len(collections) == 1:
-                        meta['_collection'] = meta.get('collection') or (collections[0] if collections else 'public_kb')
+                            meta['_collection'] = collections[0]
+                        elif collections:
+                            meta['_collection'] = collections[0]  # 多知识库时回退到第一个
+                        else:
+                            meta['_collection'] = 'public_kb'
                     source_name = meta.get('source', '未知')
                     if source_name not in seen_sources or score > seen_sources[source_name]['score']:
                         doc_type = meta.get('doc_type', 'other')
@@ -1759,11 +2144,27 @@ def rag():
                                                              min_score=RERANK_CONTEXT_MIN_SCORE)
             # Phase 2：按字符预算构建上下文
             _is_comparison = intent and intent.intent == "comparison"
-            if _is_enum_query(message) or _is_comparison:
+            if _is_enum_query(retrieval_query) or _is_comparison:
                 # 列举类 / 对比类查询：保持原始顺序，不做预算截断
-                context_text = "\n\n".join([ctx.get('doc', '') for ctx in text_contexts])
+                # 仍然注入 section_path 标题，帮助 LLM 区分不同章节
+                enum_parts = []
+                prev_section = None
+                for ctx in text_contexts:
+                    meta = ctx.get('meta', {})
+                    section = meta.get('section', '') or meta.get('section_path', '')
+                    if section and section != prev_section:
+                        enum_parts.append(f"━ {section} ━")
+                        prev_section = section
+                    # 精简表格切片：去除冗余的语义增强前缀
+                    doc = _strip_semantic_prefix(ctx.get('doc', ''), meta.get('chunk_type', ''))
+                    enum_parts.append(doc)
+                context_text = "\n\n".join(enum_parts)
             else:
                 context_text = _build_context_with_budget(text_contexts, CONTEXT_MAX_CHARS, CONTEXT_SOFT_LIMIT)
+            # 表格救援：CrossEncoder 对表格评分偏低，导致表格被预算截断
+            # 当查询涉及表格但上下文中没有表格数据时，从被截断的切片中补回
+            context_text = _rescue_table_chunks(text_contexts, context_text, retrieval_query)
+
 
             # Phase 4：计算置信度分数（top-3 平均 Rerank 分数）
             _top_scores = [ctx.get('score', 0) for ctx in text_contexts[:3]]
@@ -1884,6 +2285,9 @@ def rag():
                 # else: 没有匹配到，保留原选择（不再截断到1张）
             # else: LLM 没有提图号，保留原选择（不再截断到1张）
 
+            # 后置图片过滤：用回答内容反向筛选图片，确保图片与回答一致
+            selected_images = _filter_images_by_answer(selected_images, full_answer_text)
+
             rich_media = {'images': selected_images, 'tables': [], 'sections': []}
 
             # 7. 去掉 LLM 添加的数字引用标记，避免与后端引用重复
@@ -1951,14 +2355,39 @@ def rag():
                     "rerank_cached": rerank_cached,
                     "total_ms": duration_ms
                 }
+            # 10.5 写入语义缓存
+            try:
+                from core.semantic_cache import get_semantic_cache
+                _sc = get_semantic_cache()
+                if _sc and filtered_answer:
+                    if _semantic_cache_emb is None:
+                        from core.engine import get_engine as _get_eng
+                        _eng = _get_eng()
+                        if _eng and hasattr(_eng, 'embedding_model'):
+                            _cache_collections = ','.join(sorted(collections)) if collections else ''
+                            _cache_key_text = f"{retrieval_query}|{_cache_collections}"
+                            _semantic_cache_emb = _eng.embedding_model.encode(_cache_key_text)
+                    if _semantic_cache_emb is not None:
+                        _sc.set(_semantic_cache_emb, {
+                            "cache_type": "rag_answer",
+                            "answer": filtered_answer,
+                            "sources": sources,
+                            "citations": citation_result.get("citations", []),
+                            "images": rich_media.get("images", []),
+                            "tables": rich_media.get("tables", []),
+                        })
+                        logger.debug(f"[语义缓存] 写入成功: {message[:50]}...")
+            except Exception as e:
+                logger.info(f"[语义缓存] 写入失败: {e}")
+
             yield f"data: {json.dumps(finish_event, ensure_ascii=False)}\n\n"
 
         except Exception as e:
-            import logging as _logging
+            import traceback
-            _logging.getLogger(__name__).error(f"[SSE] RAG 流异常: {e}", exc_info=True)
             error_event = {
                 "type": "error",
-                "message": "服务内部错误，请稍后重试"
+                "message": str(e),
+                "traceback": traceback.format_exc()
             }
             yield f"data: {json.dumps(error_event, ensure_ascii=False)}\n\n"
 
@@ -1987,27 +2416,12 @@ def search():
     """
     data = request.json or {}
     query = data.get('query', '')
-    query = sanitize_user_input(query)
-    injection_matches = detect_injection(query)
-    if injection_matches:
-        logger.warning(f"[Chat] 检测到可疑注入: {injection_matches}")
     top_k = data.get('top_k', 5)
     collections = data.get('collections')  # 后端传入的知识库列表
 
     if not query:
         return jsonify({'error': 'query is required'}), 400
 
-    # 输入安全校验（注入检测、违禁词、长度限制）
-    is_valid, reason = validate_query(query)
-    if not is_valid:
-        return jsonify({'error': reason}), 400
-
-    # top_k 范围校验
-    try:
-        top_k = max(1, min(int(top_k), 50))
-    except (ValueError, TypeError):
-        top_k = 5
-
     # 如果没有指定 collections，使用默认的公开库
     if not collections:
         collections = ['public_kb']

config.example.py

@@ -109,6 +109,11 @@ USE_RERANK = True
 RERANK_CANDIDATES = 20           # 送入重排序的候选数
 RERANK_TOP_K = 15                # 重排序后保留数
 RERANK_USE_ONNX = os.getenv("RERANK_USE_ONNX", "true").lower() == "true"
+RERANK_BACKEND = os.getenv("RERANK_BACKEND", "local")       # local / cloud / fallback
+RERANK_CLOUD_MODEL = os.getenv("RERANK_CLOUD_MODEL", "qwen3-rerank")
+RERANK_CLOUD_API_KEY = os.getenv("RERANK_CLOUD_API_KEY", "")
+RERANK_CLOUD_BASE_URL = os.getenv("RERANK_CLOUD_BASE_URL", "https://dashscope.aliyuncs.com/compatible-api/v1/reranks")
+RERANK_CLOUD_TIMEOUT = int(os.getenv("RERANK_CLOUD_TIMEOUT", "15"))
 RERANK_CONTEXT_MIN_SCORE = 0.05  # Rerank 分数低于此值的切片不送入 LLM
 
 # ----- RRF 融合 -----

core/engine.py

@@ -29,6 +29,7 @@ RAG 核心引擎
 
 import os
 import gc
+import re
 import time
 import logging
 import threading
@@ -625,7 +626,7 @@ class RAGEngine:
         elif len(conditions) > 1:
             where_filter = {"$and": conditions}
 
-        query_vector = self.embedding_model.encode(query).tolist()
+        query_vector = self._encode_cached(query).tolist()
         recall_k = RERANK_CANDIDATES if (USE_RERANK or USE_HYBRID_SEARCH) else top_k
         recall_k = max(recall_k, top_k * RECALL_MULTIPLIER)
 
@@ -811,6 +812,76 @@ class RAGEngine:
             logger.warning(f"FAQ 集合查询失败: {e}")
             return get_empty_result()
 
+    def _encode_cached(self, text):
+        """
+        带缓存的 embedding 编码
+
+        优先从 Embedding Cache（LRU）读取，未命中再调用模型编码并写入缓存。
+        支持单文本和批量文本输入。
+
+        Args:
+            text: 单个文本字符串 或 文本列表
+
+        Returns:
+            numpy 数组（单文本为一维，批量为二维）
+        """
+        import numpy as _np
+
+        # 检查 embedding 缓存是否启用（缓存配置查询结果，避免每次重复导入）
+        if not hasattr(self, '_emb_cache_enabled'):
+            self._emb_cache_enabled = True  # 默认启用
+            if CACHE_AVAILABLE:
+                try:
+                    from config import EMBEDDING_CACHE_ENABLED
+                    self._emb_cache_enabled = EMBEDDING_CACHE_ENABLED
+                except ImportError:
+                    pass
+
+        if not self._emb_cache_enabled:
+            return self.embedding_model.encode(text)
+
+        try:
+            _cache = get_cache_manager()
+        except Exception:
+            return self.embedding_model.encode(text)
+
+        # 批量输入
+        if isinstance(text, list):
+            try:
+                cached_embs, missed_indices = _cache.get_embeddings_batch(text)
+                if missed_indices:
+                    missed_texts = [text[i] for i in missed_indices]
+                    # encode(list) 始终返回 2D ndarray，直接按行索引即可
+                    new_embs = self.embedding_model.encode(missed_texts)
+                    if len(missed_indices) == 1:
+                        # 单条时 encode 可能返回 1D，需统一处理
+                        if new_embs.ndim == 1:
+                            new_embs = new_embs.reshape(1, -1)
+                    for idx, mi in enumerate(missed_indices):
+                        emb_list = new_embs[idx].tolist()
+                        cached_embs[mi] = emb_list
+                        try:
+                            _cache.set_embedding(text[mi], emb_list)
+                        except Exception:
+                            pass
+                return _np.array(cached_embs)
+            except Exception:
+                # 缓存故障时优雅降级为直接编码
+                return self.embedding_model.encode(text)
+
+        # 单文本输入
+        cached = _cache.get_embedding(text)
+        if cached is not None:
+            return _np.array(cached)
+
+        embedding = self.embedding_model.encode(text)
+        try:
+            emb_list = embedding.tolist() if hasattr(embedding, 'tolist') else list(embedding)
+            _cache.set_embedding(text, emb_list)
+        except Exception:
+            pass
+        return embedding
+
     def _search_image_chunks(self, query_vector: list, top_k: int = 5, where_filter: dict = None) -> dict:
         """
         独立检索图片切片（P0：图片独立召回通道）
@@ -1171,6 +1242,20 @@ class RAGEngine:
             if not neighbors.get('ids') or len(neighbors.get('ids', [])) <= 1:
                 neighbors = _get_neighbors({"$and": [{"source": source}, {"chunk_type": "text"}]})
 
+            # 同时扩展同 section 的 table 邻居（table 切片的 rerank 分数往往偏低，
+            # 但与同 section 的 text 切片属于同一语义单元，不应割裂）
+            table_where = {"$and": [{"source": source}, {"chunk_type": "table"}]}
+            if section:
+                table_where["$and"].append({"section": section})
+            table_neighbors = _get_neighbors(table_where)
+
+            # 当 section 为空时，table 查询只有 source 条件，可能拉入大量无关表格，
+            # 缩小 chunk_index 窗口至 ±1 以降低噪音；有 section 时使用正常窗口
+            if section:
+                _t_before, _t_after = CONTEXT_EXPANSION_BEFORE, CONTEXT_EXPANSION_AFTER
+            else:
+                _t_before, _t_after = 1, 1
+
             neighbor_rows = []
             for n_id, n_doc, n_meta in zip(
                 neighbors.get('ids', []),
@@ -1183,6 +1268,18 @@ class RAGEngine:
                 if seed_index - CONTEXT_EXPANSION_BEFORE <= n_index <= seed_index + CONTEXT_EXPANSION_AFTER:
                     neighbor_rows.append((n_index, n_id, n_doc, n_meta))
 
+            # 同 section 的 table 邻居也加入扩展范围
+            for n_id, n_doc, n_meta in zip(
+                table_neighbors.get('ids', []),
+                table_neighbors.get('documents', []),
+                table_neighbors.get('metadatas', [])
+            ):
+                n_index = self._to_int(n_meta.get('chunk_index'))
+                if n_index is None:
+                    continue
+                if seed_index - _t_before <= n_index <= seed_index + _t_after:
+                    neighbor_rows.append((n_index, n_id, n_doc, n_meta))
+
             seed_neighbors_added = 0
             for n_index, n_id, n_doc, n_meta in sorted(neighbor_rows, key=lambda row: row[0]):
                 if len(items) >= max_chunks:
@@ -1387,7 +1484,7 @@ class RAGEngine:
         if not target_collections:
             return get_empty_result()
 
-        query_vector = self.embedding_model.encode(query).tolist()
+        query_vector = self._encode_cached(query).tolist()
         # 扩大召回数量，以便过滤废止切片后仍有足够结果
         recall_k = RERANK_CANDIDATES if (USE_RERANK or USE_HYBRID_SEARCH) else top_k
         recall_k = max(recall_k, top_k * RECALL_MULTIPLIER)
@@ -1739,12 +1836,12 @@ class RAGEngine:
                 # === 高精度版：基于语义向量 ===
                 from core.mmr import mmr_rerank
 
-                # 获取查询向量
+                # 获取查询向量（使用 embedding 缓存）
-                query_emb = np.array(self.embedding_model.encode(query))
+                query_emb = np.array(self._encode_cached(query))
 
-                # 批量编码所有文档
+                # 批量编码所有文档（使用 embedding 缓存）
                 docs_list = results['documents'][0]
-                all_embeddings = self.embedding_model.encode(docs_list)
+                all_embeddings = self._encode_cached(docs_list)
 
                 # 构建候选列表
                 candidates = []
@@ -2068,21 +2165,33 @@ class RAGEngine:
             "content": (
                 "你是一个严谨的知识库问答助手。"
                 "你必须且只能根据用户提供的【参考资料】回答问题。"
+                "参考资料中每段内容前标有章节路径（━格式），请注意区分不同章节的内容，"
+                "特别当不同章节标题相似或包含相同关键词时，务必根据章节路径准确定位，不要混淆。"
                 "如果参考资料中有答案，必须引用对应内容回答，并在回答末尾标注引用编号（如[1]、[2]）。"
                 "如果参考资料中确实没有相关信息，简短说明即可，不要编造或补充资料外的内容。"
                 "禁止使用参考资料以外的知识进行补充或推测。"
+                "【重要-表格处理规则】当用户询问表格、要求展示表格内容时，你必须将参考资料中的 Markdown 表格原样输出（保留 | 分隔符和表格结构），"
+                "不要仅用文字描述表格存在或仅列出章节名称。如果参考资料中多个章节都有表格，"
+                "优先展示与用户问题最相关的表格完整内容。"
             )
         })
 
         # 添加当前问题（带上下文）- 强化指令
         if context:
+            # 检测用户问题是否涉及表格，加入针对性指令
+            _table_hint = ""
+            # 检测上下文中是否包含 Markdown 表格（数据驱动，无需硬编码关键词）
+            _has_table_in_context = bool(re.search(r'\|.+\|', context)) if context else False
+            if _has_table_in_context:
+                _table_hint = "\n注意：参考资料中包含 Markdown 格式的表格数据，请务必将相关表格以原始 Markdown 表格格式完整展示在回答中，不要仅用文字描述。"
+
             user_message = f"""【参考资料】
 {context}
 
 【用户问题】
 {query}
 
-请仔细阅读以上全部参考资料后回答。如果参考资料中包含相关内容，必须引用回答并标注编号。如果资料中没有相关信息，请明确说明。"""
+请仔细阅读以上全部参考资料后回答。注意参考资料中标有章节路径，请根据章节路径准确定位相关内容。如果参考资料中包含相关内容，必须引用回答并标注编号。如果资料中没有相关信息，请明确说明。{_table_hint}"""
         else:
             user_message = query
 

core/mmr.py

@@ -108,22 +108,44 @@ def mmr_rerank(
     return selected
 
 
+def _tokenize_words(text: str) -> set:
+    """
+    使用 jieba 分词并过滤噪声，返回有意义的词集合。
+
+    过滤规则：
+    - 去除单字符词（如 "的", "了", "在"）—— 这些是停用词，对区分文档无意义
+    - 去除纯数字 / 纯标点
+    - 保留 2 字及以上的实词
+    """
+    import jieba
+    words = set()
+    for w in jieba.cut(text):
+        w = w.strip()
+        if len(w) >= 2 and not w.isdigit():
+            words.add(w)
+    return words
+
+
 def mmr_filter_by_content(
     candidates: List[Dict],
     top_k: int = 30,
-    similarity_threshold: float = 0.9
+    similarity_threshold: float = 0.85
 ) -> List[Dict]:
     """
-    基于内容相似度的去重（简化版，不需要 embedding）
+    基于 jieba 词级 Jaccard 相似度的去重（不需要 embedding）
+
+    与旧版字符级 set(text) 的区别：
+    - 旧版：set("安全生产管理制度") → {'安','全','生','产',...}，中文文档间字符集合高度重叠
+    - 新版：jieba 分词 → {"安全生产", "管理制度", ...}，词级集合区分度高
 
     适用于：
-    - 没有 embedding 的情况
+    - MMR_USE_EMBEDDING=False 时的快速去重
-    - 快速去重场景
+    - 避免 CPU 编码 100+ 文档的 50 秒开销
 
     Args:
         candidates: 候选文档列表
         top_k: 返回数量
-        similarity_threshold: 相似度阈值，超过则视为重复
+        similarity_threshold: 相似度阈值，超过则视为重复（默认 0.85）
 
     Returns:
         去重后的候选文档列表
@@ -134,25 +156,32 @@ def mmr_filter_by_content(
     if len(candidates) <= top_k:
         return candidates
 
-    selected = []
+    # 预分词：对所有候选文档一次性分词，避免重复调用 jieba.cut
-    remaining = candidates.copy()
+    word_sets = []
+    for c in candidates:
+        content = c.get('content', c.get('document', ''))[:500]
+        word_sets.append(_tokenize_words(content))
 
-    while len(selected) < top_k and remaining:
+    selected_indices = []
-        current = remaining.pop(0)
+
+    for i in range(len(candidates)):
+        if len(selected_indices) >= top_k:
+            break
+
+        current_words = word_sets[i]
+        if not current_words:
+            # 空内容直接保留
+            selected_indices.append(i)
+            continue
 
-        # 检查是否与已选内容重复
         is_duplicate = False
-        current_content = current.get('content', current.get('document', ''))[:200]
+        for j in selected_indices:
+            selected_words = word_sets[j]
+            if not selected_words:
+                continue
 
-        for s in selected:
+            intersection = len(current_words & selected_words)
-            s_content = s.get('content', s.get('document', ''))[:200]
+            union = len(current_words | selected_words)
-
-            # 简单的 Jaccard 相似度
-            words1 = set(current_content)
-            words2 = set(s_content)
-            if words1 and words2:
-                intersection = len(words1 & words2)
-                union = len(words1 | words2)
             similarity = intersection / union if union > 0 else 0
 
             if similarity > similarity_threshold:
@@ -160,9 +189,9 @@ def mmr_filter_by_content(
                 break
 
         if not is_duplicate:
-            selected.append(current)
+            selected_indices.append(i)
 
-    return selected
+    return [candidates[i] for i in selected_indices]
 
 
 # ==================== 测试 ====================

knowledge/collection.py

@@ -134,6 +134,9 @@ class CollectionMixin:
         """
         from .base import BM25Index
 
+        # 从磁盘重新加载元数据，确保多 worker 进程间状态一致
+        self._metadata = self._load_metadata()
+
         if not kb_name or not kb_name.replace('_', '').isalnum():
             return False, "向量库名称只能包含字母、数字和下划线"
 
@@ -190,6 +193,9 @@ class CollectionMixin:
         Returns:
             更新成功返回 True，向量库不存在返回 False
         """
+        # 从磁盘重新加载元数据，确保多 worker 进程间状态一致
+        self._metadata = self._load_metadata()
+
         collections = self._metadata.get("collections", {})
         if kb_name not in collections:
             return False
@@ -228,6 +234,9 @@ class CollectionMixin:
         """
         import shutil
 
+        # 从磁盘重新加载元数据，确保多 worker 进程间状态一致
+        self._metadata = self._load_metadata()
+
         if kb_name == PUBLIC_KB_NAME:
             return False, "公开知识库不能删除"
 
@@ -348,41 +357,14 @@ class CollectionMixin:
             - department: 所属部门
             - description: 描述
         """
+        # 从磁盘重新加载元数据，确保多 worker 进程间状态一致
+        self._metadata = self._load_metadata()
         result = []
 
-        # 扫描 base_path 下的所有子目录作为向量库
+        # 以元数据为唯一真相源，不再扫描磁盘自动补充
-        # 每个向量库使用独立目录：base_path/my_ky, base_path/public_kb 等
+        # （扫描磁盘会导致其他 worker 刚创建/删除的集合被误操作）
-        actual_collections = []
-        try:
-            if os.path.exists(self.base_path):
-                for item in os.listdir(self.base_path):
-                    item_path = os.path.join(self.base_path, item)
-                    if os.path.isdir(item_path) and not item.startswith('.'):
-                        # 检查是否包含 chroma.sqlite3（有效的向量库目录）
-                        if os.path.exists(os.path.join(item_path, 'chroma.sqlite3')):
-                            actual_collections.append(item)
-        except Exception as e:
-            logger.warning(f"扫描向量库目录失败: {e}")
-
-        # 如果扫描失败，回退到元数据中的集合列表
-        if not actual_collections:
-            actual_collections = list(self._metadata.get("collections", {}).keys())
-
-        for name in actual_collections:
-            if name not in self._metadata.get("collections", {}):
-                if "collections" not in self._metadata:
-                    self._metadata["collections"] = {}
-                self._metadata["collections"][name] = {
-                    "display_name": name,
-                    "department": "",
-                    "description": "",
-                    "created_at": datetime.now().isoformat()
-                }
-                logger.info(f"自动补充向量库元数据: {name}")
-
-        self._save_metadata()
-
         for name, info in self._metadata.get("collections", {}).items():
+            try:
                 collection = self.get_collection(name)
                 result.append(CollectionInfo(
                     name=name,
@@ -392,6 +374,19 @@ class CollectionMixin:
                     department=info.get("department", ""),
                     description=info.get("description", "")
                 ))
+            except Exception as e:
+                # 只跳过不修改元数据（可能是其他 worker 刚创建的集合）
+                logger.warning(
+                    f"跳过异常向量库 '{name}': {e}"
+                )
+                result.append(CollectionInfo(
+                    name=name,
+                    display_name=info.get("display_name", name),
+                    document_count=0,
+                    created_at=info.get("created_at", ""),
+                    department=info.get("department", ""),
+                    description=info.get("description", "")
+                ))
 
         return result
 
@@ -405,4 +400,6 @@ class CollectionMixin:
         Returns:
             存在返回 True，不存在返回 False
         """
+        # 从磁盘重新加载元数据，确保多 worker 进程间状态一致
+        self._metadata = self._load_metadata()
         return kb_name in self._metadata.get("collections", {})

knowledge/manager.py

@@ -29,6 +29,11 @@
 import os
 import json
 import threading
+try:
+    import fcntl
+    _HAS_FCNTL = True
+except ImportError:
+    _HAS_FCNTL = False
 from typing import List, Dict, Optional, Tuple
 from pathlib import Path
 import logging
@@ -134,22 +139,36 @@ class KnowledgeBaseManager(
         logger.info(f"知识库管理器初始化完成，路径: {self.base_path}，发现 {len(existing_kbs)} 个向量库: {existing_kbs}")
 
     def _load_metadata(self) -> dict:
-        """加载元数据"""
+        """加载元数据（带文件锁）"""
         metadata_path = os.path.join(self.base_path, KB_METADATA_FILE)
         if os.path.exists(metadata_path):
             try:
                 with open(metadata_path, 'r', encoding='utf-8') as f:
+                    if _HAS_FCNTL:
+                        fcntl.flock(f.fileno(), fcntl.LOCK_SH)
+                    try:
                         return json.load(f)
+                    finally:
+                        if _HAS_FCNTL:
+                            fcntl.flock(f.fileno(), fcntl.LOCK_UN)
             except Exception as e:
                 logger.error(f"加载元数据失败: {e}")
         return {"collections": {}}
 
     def _save_metadata(self):
-        """保存元数据"""
+        """保存元数据（带文件锁，防止并发写入覆盖）"""
         metadata_path = os.path.join(self.base_path, KB_METADATA_FILE)
         try:
             with open(metadata_path, 'w', encoding='utf-8') as f:
+                if _HAS_FCNTL:
+                    fcntl.flock(f.fileno(), fcntl.LOCK_EX)
+                try:
                     json.dump(self._metadata, f, ensure_ascii=False, indent=2)
+                    f.flush()
+                    os.fsync(f.fileno())
+                finally:
+                    if _HAS_FCNTL:
+                        fcntl.flock(f.fileno(), fcntl.LOCK_UN)
         except Exception as e:
             logger.error(f"保存元数据失败: {e}")
 
@@ -393,6 +412,11 @@ class KnowledgeBaseManager(
         if len(chunks) < 2:
             return chunks
 
+        # 检测页码是否可靠：若所有 chunk 的 page_start 相同（如 Word 文档 page_idx 全为 0），
+        # 则页码信息不可用，需要启用降级合并规则
+        page_values = set(getattr(c, 'page_start', 0) for c in chunks)
+        pages_unavailable = len(page_values) <= 1
+
         merged_chunks = []
         i = 0
         merge_count = 0
@@ -405,6 +429,7 @@ class KnowledgeBaseManager(
                 # 查找下一个表格（跳过中间的"续表"文本）
                 next_table_idx = None
                 next_chunk = None
+                intermediate_texts = []  # 收集中间文本用于降级判断
 
                 for j in range(i + 1, min(i + 4, len(chunks))):  # 最多向前看3个切片
                     candidate = chunks[j]
@@ -419,7 +444,11 @@ class KnowledgeBaseManager(
                     elif candidate_type == 'text' and ('续表' in candidate_title or '续表' in candidate_content):
                         # 遇到"续表"文本，继续查找下一个表格
                         continue
-                    elif candidate_type not in ('text',):
+                    elif candidate_type == 'text':
+                        # 非"续表"文本，收集后停止查找
+                        intermediate_texts.append(candidate)
+                        break
+                    else:
                         # 遇到非文本类型，停止查找
                         break
 
@@ -439,12 +468,22 @@ class KnowledgeBaseManager(
                     # 获取内容（用于检测"续表"）
                     next_content = getattr(next_chunk, 'content', '')
 
+                    # 通用/无意义标题集合，这些标题不能用于"标题相似"判定
+                    _GENERIC_TITLES = {'表格', 'table', '表格', ''}
+
                     # 判断是否为跨页表格
                     is_cross_page = False
 
                     # 规则1: 页码连续（如果页码有效）
                     page_valid = curr_page_end > 0 and next_page_start > 0
                     if page_valid and curr_page_end + 1 == next_page_start:
+                        # 页码连续时，还需标题匹配或为通用标题才合并
+                        # 避免把不同页面上不相关的表格错误合并
+                        if curr_title == next_title or curr_title in _GENERIC_TITLES and next_title in _GENERIC_TITLES:
+                            is_cross_page = True
+                        elif curr_title and next_title:
+                            clean_next_r1 = next_title.replace('续表', '').strip()
+                            if curr_title in clean_next_r1 or clean_next_r1 in curr_title:
                                 is_cross_page = True
 
                     # 规则2: 第二个表格标题或内容包含"续表"
@@ -452,11 +491,36 @@ class KnowledgeBaseManager(
                         is_cross_page = True
 
                     # 规则3: 标题相似（去掉"续表"后比较）
-                    elif curr_title and next_title:
+                    # 排除通用标题（如"表格"），防止把所有标题为"表格"的相邻表格都误合并
+                    elif (curr_title and next_title
+                          and curr_title not in _GENERIC_TITLES
+                          and next_title not in _GENERIC_TITLES):
                         clean_next = next_title.replace('续表', '').strip()
-                        if curr_title in clean_next or clean_next in curr_title:
+                        if clean_next and (curr_title in clean_next or clean_next in curr_title):
                             is_cross_page = True
 
+                    # 规则4（降级）: 页码不可用（如 Word 文档 page_idx 全为 0）
+                    # 仅当页码信息缺失时才启用此规则，避免 PDF 正常页码时被误合并
+                    if (not is_cross_page
+                            and pages_unavailable
+                            and curr_title in _GENERIC_TITLES
+                            and next_title in _GENERIC_TITLES):
+                        # 检查中间文本是否暗示跨页延续（空、短文本、续表标记等）
+                        has_separating_content = False
+                        for text_chunk in intermediate_texts:
+                            tc = (getattr(text_chunk, 'content', '') or '').strip()
+                            tt = (getattr(text_chunk, 'title', '') or '').strip()
+                            if not tc:
+                                continue  # 空文本不算分隔
+                            if '续表' in tc or '续表' in tt:
+                                continue  # 续表标记，说明是跨页
+                            # 有实质性中间内容（如分类标题"A3类：xxx"），不合并
+                            has_separating_content = True
+                            break
+                        if not has_separating_content:
+                            is_cross_page = True
+                            logger.debug(f"降级合并(页码不可用): '{curr_title}' + '{next_title}'")
+
                     if is_cross_page:
                         # 执行合并
                         merge_count += 1
@@ -469,14 +533,27 @@ class KnowledgeBaseManager(
                             # 合并两个表格的 HTML
                             current.table_html = curr_html + '\n' + next_html
 
-                        # 合并 image_path 到 images
+                        # 合并 image_path 和嵌入图片到 images
                         curr_img = getattr(current, 'image_path', None)
                         next_img = getattr(next_chunk, 'image_path', None)
-                        merged_images = []
+                        curr_images = getattr(current, 'images', None) or []
-                        if curr_img:
+                        next_images = getattr(next_chunk, 'images', None) or []
+
+                        # 合并两个表格的所有图片（image_path + 嵌入图片）
+                        merged_images = list(curr_images)  # 保留当前表格的嵌入图片
+                        # 添加 image_path 图片（如果不在列表中）
+                        existing_ids = {img.get('id', '') for img in merged_images if isinstance(img, dict)}
+                        if curr_img and curr_img not in existing_ids:
                             merged_images.append({'id': curr_img, 'page': curr_page_end})
-                        if next_img:
+                            existing_ids.add(curr_img)
+                        for img in next_images:  # 添加下一个表格的嵌入图片
+                            img_id = img.get('id', '') if isinstance(img, dict) else ''
+                            if img_id and img_id not in existing_ids:
+                                merged_images.append(img)
+                                existing_ids.add(img_id)
+                        if next_img and next_img not in existing_ids:
                             merged_images.append({'id': next_img, 'page': next_page_start})
+
                         if merged_images:
                             current.images = merged_images
                             # 保留第一个图片作为主 image_path
@@ -525,7 +602,7 @@ class KnowledgeBaseManager(
         try:
             from config import get_llm_client, DASHSCOPE_MODEL
             client = get_llm_client()
-            summary = call_llm(client, prompt, DASHSCOPE_MODEL, max_tokens=100)
+            summary = call_llm(client, prompt, DASHSCOPE_MODEL, max_tokens=512)
             return summary.strip() if summary else ""
         except Exception as e:
             logger.warning(f"生成表格摘要失败: {e}")
@@ -584,7 +661,7 @@ class KnowledgeBaseManager(
                         ]
                     }
                 ],
-                max_tokens=200
+                max_tokens=512
             )
 
             description = response.choices[0].message.content

knowledge/router.py

@@ -283,7 +283,7 @@ class KnowledgeBaseRouter:
         content = call_llm(
             self.llm_client, prompt, MODEL,
             temperature=0.1,
-            max_tokens=100
+            max_tokens=512
         )
 
         if content is None:

parsers/heading_rules.py

@@ -0,0 +1,347 @@
+# -*- coding: utf-8 -*-
+"""
+标题识别规则引擎
+
+将 _detect_heading_level 的硬编码正则提取为可配置的规则列表。
+规则按优先级从高到低排序，第一个匹配即返回。
+
+MinerU 解析 DOCX 等 Office 格式时通常不提供 text_level（全部为 0），
+此时需要启发式识别标题层级。本模块提供可配置的规则引擎替代原来的
+硬编码 if-elif 链。
+
+设计要点：
+- HeadingRule 数据类支持正向匹配（pattern）和反向排除（exclude_pattern）
+- 长度约束（min_length / max_length）可精确控制匹配范围
+- 规则可单独禁用（enabled=False），便于调试
+- 全局单例通过 config.py 覆盖默认值
+
+MinerU v2 格式备注：
+    content_list_v2.json 中的 paragraph_content 包含 style=["bold"] 信息，
+    layout.json 中的 spans 也有 style 信息。这些信息比正则匹配 **加粗** 更可靠，
+    但当前代码使用 v1 格式（content_list.json），暂不利用 v2 的 style。
+    HeadingRuleEngine.detect() 签名预留了 style 参数，未来切换到 v2 格式后
+    可直接利用 style 信息辅助判断。
+"""
+
+import re
+import logging
+from dataclasses import dataclass
+from typing import Optional, List, Tuple
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class HeadingRule:
+    """
+    标题识别规则
+
+    每条规则定义一个文本模式到标题级别的映射。
+    规则引擎按列表顺序逐条匹配，第一个命中即返回。
+
+    Attributes:
+        pattern: 编译后的正则（match 语义，从文本开头匹配）
+        level: 匹配时返回的标题级别 (1=h1, 2=h2, 3=h3)
+        name: 规则名称（用于日志和配置覆盖）
+        max_length: 文本最大长度，0=不限
+        min_length: 文本最小长度，0=不限
+        enabled: 是否启用
+        exclude_pattern: 匹配此模式则排除（反向过滤）
+
+    Example:
+        >>> rule = HeadingRule(
+        ...     pattern=re.compile(r'^第[一二三四五六七八九十百千万]+[章节篇部]'),
+        ...     level=1,
+        ...     name="chinese_chapter",
+        ... )
+        >>> rule.match("第一章 总则")
+        1
+        >>> rule.match("这是正文")
+        0
+    """
+
+    pattern: re.Pattern
+    level: int
+    name: str
+    max_length: int = 0
+    min_length: int = 0
+    enabled: bool = True
+    exclude_pattern: Optional[re.Pattern] = None
+
+    def match(self, text: str) -> int:
+        """
+        检查文本是否匹配此规则
+
+        Args:
+            text: 待检测文本（调用前应已 strip）
+
+        Returns:
+            标题级别，0 表示不匹配
+        """
+        if not self.enabled:
+            return 0
+        if self.min_length > 0 and len(text) < self.min_length:
+            return 0
+        if self.max_length > 0 and len(text) > self.max_length:
+            return 0
+        if self.exclude_pattern and self.exclude_pattern.search(text):
+            return 0
+        if self.pattern.match(text):
+            return self.level
+        return 0
+
+
+# 默认规则列表（按优先级从高到低）
+#
+# 注意事项：
+# - 数字三级标题 (1.1.1) 必须在二级 (1.1) 之前，因为 1.1.1 也匹配 ^\d+\.\d+
+# - short_chinese_heading 是最宽泛的规则，放在最后作为兜底
+# - 第 9 条规则相比原版增加了 exclude_pattern，排除以句末标点结尾的短文本
+DEFAULT_HEADING_RULES: List[HeadingRule] = [
+    # 1. 中文章节标题 -> h1
+    #    匹配：第一章、第二章、第十节、第三篇 等
+    HeadingRule(
+        pattern=re.compile(r'^第[一二三四五六七八九十百千万]+[章节篇部]'),
+        level=1,
+        name="chinese_chapter",
+    ),
+    # 2. 中文条款编号 -> h2
+    #    匹配：第一条、第三款 等
+    HeadingRule(
+        pattern=re.compile(r'^第[一二三四五六七八九十百千万]+[条款]'),
+        level=2,
+        name="chinese_article",
+    ),
+    # 3. 数字三级标题 -> h3（必须在二级之前匹配）
+    #    匹配：1.1.1 背景、2.3.4 方案 等
+    HeadingRule(
+        pattern=re.compile(r'^\d+\.\d+\.\d+[\.、\s]'),
+        level=3,
+        name="numeric_level3",
+        max_length=100,
+    ),
+    # 4. 数字二级标题 -> h2（必须在一级之前匹配）
+    #    匹配：1.1 背景、2.3 方案 等
+    HeadingRule(
+        pattern=re.compile(r'^\d+\.\d+[\.、\s]'),
+        level=2,
+        name="numeric_level2",
+        max_length=80,
+    ),
+    # 5. 数字一级标题 -> h1
+    #    匹配：1. 概述、2、背景 等
+    HeadingRule(
+        pattern=re.compile(r'^\d+[\.、\s]'),
+        level=1,
+        name="numeric_level1",
+        max_length=50,
+    ),
+    # 6. 英文章节标题 -> h1
+    #    匹配：Chapter 1、Section 2、Part 3 等
+    HeadingRule(
+        pattern=re.compile(r'^(Chapter|Section|Part|Chapter\s+\d+|Section\s+\d+)', re.IGNORECASE),
+        level=1,
+        name="english_chapter",
+    ),
+    # 7. 分类标题 -> h3（必须在 bold_short_text 之前，否则 **A2类：** 会被加粗规则抢先匹配）
+    #    匹配：A1类：公园、**A2类**：各类卫生医疗机构、**B1类：** 道路 等
+    HeadingRule(
+        pattern=re.compile(r'^\*{0,2}[A-Z]\d+[类類]\*{0,2}[：:]'),
+        level=3,
+        name="category_heading",
+    ),
+    # 8. 加粗短文本 -> h2
+    #    匹配：**重要通知**、**概述** 等（Markdown 加粗标记）
+    #    注意：**A2类：** 已被分类标题规则优先匹配，不会误判为 h2
+    HeadingRule(
+        pattern=re.compile(r'^\*\*.+\*\*$'),
+        level=2,
+        name="bold_short_text",
+        max_length=50,
+    ),
+    # 9. 短中文文本 -> h2（替代原"任何 <20 字符含中文"规则）
+    #    关键改进：排除以句末标点结尾的文本
+    #    原规则将 "这是一段正文。" 也识别为 h2，导致大量误判
+    #    新规则：包含中文 + 长度 2-20 + 不以句末标点结尾 → h2
+    HeadingRule(
+        pattern=re.compile(r'[一-鿿]'),
+        level=2,
+        name="short_chinese_heading",
+        max_length=20,
+        min_length=2,
+        exclude_pattern=re.compile(r'[。！？；…]$'),
+        enabled=True,
+    ),
+]
+
+
+class HeadingRuleEngine:
+    """
+    标题识别规则引擎
+
+    按规则列表顺序逐条匹配，第一个命中即返回标题级别。
+    支持从 config.py 加载自定义规则或覆盖默认规则参数。
+
+    Example:
+        >>> engine = HeadingRuleEngine()
+        >>> engine.detect("第一章 总则")
+        (1, 'chinese_chapter')
+        >>> engine.detect("这是普通正文。")
+        (0, None)
+    """
+
+    def __init__(self, rules: Optional[List[HeadingRule]] = None) -> None:
+        """
+        Args:
+            rules: 规则列表，None 则使用默认规则的深拷贝
+        """
+        if rules is not None:
+            self.rules: List[HeadingRule] = rules
+        else:
+            import copy
+            self.rules = copy.deepcopy(DEFAULT_HEADING_RULES)
+
+    def detect(self, text: str, style: Optional[List[str]] = None) -> Tuple[int, Optional[str]]:
+        """
+        检测文本的标题级别
+
+        Args:
+            text: 待检测文本
+            style: MinerU v2 格式中的 style 信息（如 ["bold"]），
+                   当文本标记为 bold 且较短时，可直接判定为标题，
+                   无需依赖 Markdown **...** 标记。
+
+        Returns:
+            (level, rule_name): 标题级别和匹配的规则名
+            level=0 表示不是标题
+        """
+        text = text.strip()
+        if not text:
+            return 0, None
+
+        # v2 style 信息：如果文本标记为 bold 且较短，优先尝试加粗规则
+        if style and 'bold' in style and 2 <= len(text) <= 50:
+            # 先检查是否匹配更高优先级的分类标题规则
+            for rule in self.rules:
+                if rule.name == 'category_heading' and rule.enabled:
+                    level = rule.match(text)
+                    if level > 0:
+                        logger.debug(f"标题识别(v2 style): '{text[:30]}' -> h{level} (规则: {rule.name})")
+                        return level, rule.name
+
+            # 再检查是否匹配中文章节/条款等高优先级规则
+            for rule in self.rules:
+                if rule.name in ('chinese_chapter', 'chinese_article', 'numeric_level3',
+                                 'numeric_level2', 'numeric_level1', 'english_chapter') and rule.enabled:
+                    level = rule.match(text)
+                    if level > 0:
+                        logger.debug(f"标题识别(v2 style): '{text[:30]}' -> h{level} (规则: {rule.name})")
+                        return level, rule.name
+
+            # 否则作为加粗短文本 → h2（与 bold_short_text 规则对齐，但不依赖 **...** 标记）
+            logger.debug(f"标题识别(v2 style): '{text[:30]}' -> h2 (规则: bold_short_text_via_style)")
+            return 2, 'bold_short_text'
+
+        # 常规规则匹配（v1 格式或无 style 信息时）
+        for rule in self.rules:
+            level = rule.match(text)
+            if level > 0:
+                logger.debug(f"标题识别: '{text[:30]}' -> h{level} (规则: {rule.name})")
+                return level, rule.name
+
+        return 0, None
+
+
+# ==================== 全局单例 ====================
+
+_engine: Optional[HeadingRuleEngine] = None
+
+
+def get_heading_engine() -> HeadingRuleEngine:
+    """获取全局标题识别引擎（延迟初始化，线程安全）"""
+    global _engine
+    if _engine is None:
+        _engine = _create_engine_from_config()
+    return _engine
+
+
+def _create_engine_from_config() -> HeadingRuleEngine:
+    """
+    从 config 创建引擎（支持配置覆盖）
+
+    优先级：
+    1. config.HEADING_RULES_CONFIG 不为 None → 使用自定义规则
+    2. config 细粒度参数覆盖默认规则（如 HEADING_SHORT_TEXT_ENABLED）
+    3. 使用默认规则
+    """
+    # 尝试加载完整自定义规则
+    try:
+        from config import HEADING_RULES_CONFIG
+        if HEADING_RULES_CONFIG is not None:
+            rules = _build_rules_from_config(HEADING_RULES_CONFIG)
+            logger.info(f"使用自定义标题规则: {len(rules)} 条")
+            return HeadingRuleEngine(rules)
+    except ImportError:
+        pass
+
+    # 使用默认规则，应用细粒度配置覆盖
+    rules = list(DEFAULT_HEADING_RULES)
+    try:
+        from config import HEADING_SHORT_TEXT_ENABLED
+        for rule in rules:
+            if rule.name == "short_chinese_heading":
+                rule.enabled = HEADING_SHORT_TEXT_ENABLED
+                logger.debug(f"配置覆盖: short_chinese_heading.enabled={HEADING_SHORT_TEXT_ENABLED}")
+    except ImportError:
+        pass
+
+    try:
+        from config import HEADING_SHORT_TEXT_MAX_LENGTH
+        for rule in rules:
+            if rule.name == "short_chinese_heading":
+                rule.max_length = HEADING_SHORT_TEXT_MAX_LENGTH
+                logger.debug(f"配置覆盖: short_chinese_heading.max_length={HEADING_SHORT_TEXT_MAX_LENGTH}")
+    except ImportError:
+        pass
+
+    return HeadingRuleEngine(rules)
+
+
+def _build_rules_from_config(config: list) -> List[HeadingRule]:
+    """
+    从配置字典列表构建规则列表
+
+    Args:
+        config: 规则配置列表，每项为 dict，包含：
+            - pattern (str): 正则表达式字符串
+            - level (int): 标题级别
+            - name (str): 规则名称
+            - max_length (int, 可选): 文本最大长度
+            - min_length (int, 可选): 文本最小长度
+            - enabled (bool, 可选): 是否启用
+            - exclude_pattern (str, 可选): 排除正则
+
+    Returns:
+        规则列表
+    """
+    rules = []
+    for item in config:
+        exclude = None
+        if 'exclude_pattern' in item:
+            exclude = re.compile(item['exclude_pattern'])
+        rules.append(HeadingRule(
+            pattern=re.compile(item['pattern']),
+            level=item['level'],
+            name=item['name'],
+            max_length=item.get('max_length', 0),
+            min_length=item.get('min_length', 0),
+            enabled=item.get('enabled', True),
+            exclude_pattern=exclude,
+        ))
+    return rules
+
+
+def reset_heading_engine() -> None:
+    """重置引擎（用于测试）"""
+    global _engine
+    _engine = None

parsers/mineru_parser.py

@@ -286,14 +286,138 @@ def parse_with_mineru_online(
         raise RuntimeError(f"MinerU 在线 API 调用失败: {e}")
 
 
+def _parse_v2_content_list(v2_data: list) -> list:
+    """
+    将 MinerU v2 嵌套格式转换为 v1 兼容的扁平 content_list
+
+    v2 格式: [[page0_items], [page1_items], ...]
+    v1 格式: [item, item, ...] 扁平列表
+
+    转换规则:
+    - paragraph → text（拼接 paragraph_content，提取 style 信息）
+    - table → table（保留 table_body/html）
+    - title → text（提取 level 信息）
+    - page_header / page_footer / page_number → 过滤掉
+
+    Args:
+        v2_data: v2 格式的嵌套列表
+
+    Returns:
+        v1 兼容的扁平列表，额外包含 _v2_styles / _v2_table_type 字段
+    """
+    flat_list: List[Dict] = []
+
+    for page_idx, page_items in enumerate(v2_data):
+        if not isinstance(page_items, list):
+            continue
+        for item in page_items:
+            if not isinstance(item, dict):
+                continue
+
+            v2_type: str = item.get('type', '')
+            content = item.get('content', {})
+
+            # 过滤噪音类型（页眉、页脚、页码）
+            if v2_type in ('page_header', 'page_footer', 'page_number'):
+                continue
+
+            if v2_type == 'paragraph':
+                # 拼接段落文本，收集样式信息
+                text_parts: List[str] = []
+                has_bold = False
+                para_content = content.get('paragraph_content', []) if isinstance(content, dict) else []
+                for part in para_content:
+                    if not isinstance(part, dict):
+                        continue
+                    part_text = part.get('content', '')
+                    text_parts.append(part_text)
+                    if 'bold' in part.get('style', []):
+                        has_bold = True
+
+                full_text = ''.join(text_parts).strip()
+                if not full_text:
+                    continue
+
+                flat_item: Dict[str, Any] = {
+                    'type': 'text',
+                    'text': full_text,
+                    'content': full_text,
+                    'page_idx': page_idx,
+                    'bbox': item.get('bbox', []),
+                    'text_level': 0,
+                    '_v2_styles': ['bold'] if has_bold else [],
+                }
+                flat_list.append(flat_item)
+
+            elif v2_type == 'table':
+                flat_item = {
+                    'type': 'table',
+                    'page_idx': page_idx,
+                    'bbox': item.get('bbox', []),
+                    'html': content.get('html', '') if isinstance(content, dict) else '',
+                    'table_body': content.get('html', '') if isinstance(content, dict) else '',
+                    'caption': content.get('caption', '') if isinstance(content, dict) else '',
+                    '_v2_table_type': content.get('table_type', '') if isinstance(content, dict) else '',
+                }
+                flat_list.append(flat_item)
+
+            elif v2_type == 'title':
+                # v2 title 项有 level 信息
+                level = content.get('level', 0) if isinstance(content, dict) else 0
+                text_parts = []
+                para_content = content.get('paragraph_content', []) if isinstance(content, dict) else []
+                for part in para_content:
+                    if isinstance(part, dict):
+                        text_parts.append(part.get('content', ''))
+                full_text = ''.join(text_parts).strip()
+
+                if full_text:
+                    flat_item = {
+                        'type': 'text',
+                        'text': full_text,
+                        'content': full_text,
+                        'page_idx': page_idx,
+                        'bbox': item.get('bbox', []),
+                        'text_level': level,
+                        '_v2_styles': ['bold'],
+                    }
+                    flat_list.append(flat_item)
+
+            elif v2_type in ('image', 'chart'):
+                flat_item = {
+                    'type': v2_type,
+                    'page_idx': page_idx,
+                    'bbox': item.get('bbox', []),
+                    'img_path': content.get('img_path', '') if isinstance(content, dict) else '',
+                    'image_path': content.get('img_path', '') if isinstance(content, dict) else '',
+                    'caption': content.get('caption', '') if isinstance(content, dict) else '',
+                }
+                flat_list.append(flat_item)
+
+            elif v2_type == 'equation':
+                flat_item = {
+                    'type': 'equation',
+                    'page_idx': page_idx,
+                    'bbox': item.get('bbox', []),
+                    'content': content.get('latex', '') if isinstance(content, dict) else '',
+                    'text': content.get('latex', '') if isinstance(content, dict) else '',
+                    'latex': content.get('latex', '') if isinstance(content, dict) else '',
+                    'img_path': content.get('img_path', '') if isinstance(content, dict) else '',
+                }
+                flat_list.append(flat_item)
+
+    logger.info(f"v2 格式转换: {len(v2_data)} 页 → {len(flat_list)} 项（已过滤噪音类型）")
+    return flat_list
+
+
 def _parse_mineru_online_zip(zip_content: bytes, file_path: Path) -> Dict[str, Any]:
     """
     解析 MinerU 在线 API 返回的 zip 包
 
-    zip 包结构：
+    zip 包结构:
     - full.md - Markdown 解析结果
-    - *_content_list.json - 内容列表（扁平格式，优先使用）
+    - *_content_list.json - 内容列表（v1 扁平格式）
-    - *_content_list_v2.json - 内容列表（嵌套格式）
+    - *_content_list_v2.json - 内容列表（v2 嵌套格式，含 style 信息）
     - images/ - 图片目录
 
     Args:
@@ -306,23 +430,44 @@ def _parse_mineru_online_zip(zip_content: bytes, file_path: Path) -> Dict[str, A
     import zipfile
     import io
 
+    # 读取格式偏好配置
+    try:
+        from config import MINERU_PREFER_V2
+    except ImportError:
+        MINERU_PREFER_V2 = True  # 默认优先 v2
+
     with zipfile.ZipFile(io.BytesIO(zip_content), 'r') as zf:
         # 列出所有文件
         file_list = zf.namelist()
         logger.debug(f"zip 包内容: {file_list}")
 
-        # 查找 content_list.json（优先使用扁平格式）
+        # 查找 content_list 文件（v2 含 style 信息，优先使用）
         content_list_path = None
+        is_v2 = False
+
+        if MINERU_PREFER_V2:
+            # 优先使用 v2 格式（含 style 信息）
+            for f in file_list:
+                if f.endswith('_content_list_v2.json'):
+                    content_list_path = f
+                    is_v2 = True
+                    break
+            if not content_list_path:
+                for f in file_list:
+                    if f.endswith('_content_list.json') and not f.endswith('_v2.json'):
+                        content_list_path = f
+                        break
+        else:
+            # 优先使用 v1 格式
             for f in file_list:
-            # 优先使用 content_list.json（扁平格式）
                 if f.endswith('_content_list.json') and not f.endswith('_v2.json'):
                     content_list_path = f
                     break
-        # 如果没有找到，再尝试 v2 格式
             if not content_list_path:
                 for f in file_list:
                     if f.endswith('_content_list_v2.json'):
                         content_list_path = f
+                        is_v2 = True
                         break
 
         # 查找 markdown 文件
@@ -337,7 +482,13 @@ def _parse_mineru_online_zip(zip_content: bytes, file_path: Path) -> Dict[str, A
         if content_list_path:
             with zf.open(content_list_path) as f:
                 content_list = json.load(f)
-            logger.info(f"读取 content_list: {len(content_list)} 项, 来源: {content_list_path}")
+
+            # v2 格式需要转换为 v1 兼容的扁平列表
+            if is_v2 and isinstance(content_list, list) and content_list and isinstance(content_list[0], list):
+                content_list = _parse_v2_content_list(content_list)
+                logger.info(f"v2 格式已转换为扁平列表，共 {len(content_list)} 项")
+
+            logger.info(f"读取 content_list: {len(content_list)} 项, 格式={'v2' if is_v2 else 'v1'}, 来源: {content_list_path}")
 
         # 读取 markdown
         markdown_content = ""
@@ -424,10 +575,14 @@ def _parse_mineru_online_result(result: Dict, file_path: Path) -> Dict[str, Any]
 
         if item_type == "text":
             text = item.get("content", "") or item.get("text", "")
+            v2_styles = item.get("_v2_styles", [])
 
             # 启发式标题识别
             if text_level == 0:
-                text_level = _detect_heading_level(text)
+                from parsers.heading_rules import get_heading_engine
+                engine = get_heading_engine()
+                detected_level, _ = engine.detect(text, style=v2_styles)
+                text_level = detected_level
 
             title = ""
             if text_level > 0:
@@ -471,7 +626,8 @@ def _parse_mineru_online_result(result: Dict, file_path: Path) -> Dict[str, Any]
             else:
                 md_table = ""
 
-            table_images = extract_images_from_markdown(md_table) if md_table else []
+            # 从原始 HTML 提取嵌入图片（md_table 经 get_text 转换后已丢失 <img> 标签）
+            table_images = extract_images_from_markdown(table_body) if table_body else []
 
             chunk = MinerUChunk(
                 content=table_caption or "表格",
@@ -558,8 +714,14 @@ def _parse_mineru_online_result(result: Dict, file_path: Path) -> Dict[str, Any]
         min_merge = 100
         max_size = 1200
 
+    # 表单类型二次校正（在 _post_process_chunks 之前，因为 table 不参与合并）
+    _reclassify_text_chunks(chunks)
+
     chunks = _post_process_chunks(chunks, min_merge_size=min_merge, max_chunk_size=max_size)
 
+    # 验证分类标题是否被规则引擎正确识别（安全网，仅告警不修改）
+    _validate_category_section_paths(chunks)
+
     return {
         'markdown': "\n".join(markdown_parts) if markdown_parts else markdown_content,
         'chunks': chunks,
@@ -697,16 +859,18 @@ def parse_with_mineru(
         logger.error(f"MinerU 解析失败: {e}")
         raise
     finally:
-        # 清理临时目录
+        清理临时目录
         if cleanup_output and os.path.exists(output_dir):
             shutil.rmtree(output_dir, ignore_errors=True)
 
-
 def _detect_heading_level(text: str) -> int:
     """
-    启发式标题识别
+    启发式标题识别（规则引擎版）
 
-    用于 MinerU 解析 DOCX 等 Office 格式时不提供 text_level 的情况。
+    当 MinerU 解析 DOCX 等 Office 格式时不提供 text_level 时使用。
+    规则按优先级从高到低匹配，第一个命中即返回。
+
+    规则定义见 parsers/heading_rules.py，可通过 config.py 覆盖。
 
     Args:
         text: 文本内容
@@ -714,57 +878,91 @@ def _detect_heading_level(text: str) -> int:
     Returns:
         标题级别 (0=正文, 1=h1, 2=h2, 3=h3)
     """
-    import re
+    from parsers.heading_rules import get_heading_engine
+    engine = get_heading_engine()
+    level, _ = engine.detect(text)
+    return level
 
-    text = text.strip()
+def _reclassify_text_chunks(chunks: List[MinerUChunk]) -> None:
+    """
+    二次校正：检测被标记为 text 但实际是表格/表单的 chunk
 
-    # 空文本
+    MinerU 解析 Word 文档时，某些带下划线填空项的表单
+    被标记为 text 类型，需要根据内容特征修正为 table。
+    就地修改 chunks 列表中的 chunk_type 字段。
+
+    检测依据（基于实测数据设计）：
+    - 连续下划线 ___ (3个以上)  —— Word 表单填空项
+    - 冒号后跟下划线 如 "日期：____"  —— 键值对式表单
+    需 >= min_indicators 个指标同时命中才校正，避免误判。
+    """
+    try:
+        from config import FORM_RECLASSIFY_ENABLED
+        if not FORM_RECLASSIFY_ENABLED:
+            return
+    except ImportError:
+        pass
+
+    try:
+        from config import FORM_RECLASSIFY_MIN_INDICATORS
+        min_indicators = FORM_RECLASSIFY_MIN_INDICATORS
+    except ImportError:
+        min_indicators = 2
+
+    # 表单特征指标（编译一次，避免循环内重复编译）
+    # 注意：MinerU 输出的下划线是 Markdown 转义格式 \_\_\_，
+    # 需要同时匹配纯下划线 ___ 和转义下划线 \_\_\_
+    _FORM_INDICATORS = [
+        re.compile(r'(?:\\_|_){3,}'),          # 连续下划线（3个以上，含转义格式）
+        re.compile(r'[：:]\s*(?:\\_|_){2,}'),  # 冒号后跟下划线（含转义格式）
+    ]
+
+    reclassified = 0
+    for chunk in chunks:
+        if chunk.chunk_type != 'text':
+            continue
+        text = (chunk.content or '').strip()
         if not text:
-        return 0
+            continue
 
-    # 中文章节标题模式
+        indicator_count = sum(1 for p in _FORM_INDICATORS if p.search(text))
-    # 第一章、第二章、... -> h1
+        if indicator_count >= min_indicators:
-    if re.match(r'^第[一二三四五六七八九十百千万]+[章节篇部]', text):
+            chunk.chunk_type = 'table'
-        return 1
+            reclassified += 1
+            logger.info(f"表单检测: text -> table, content='{text[:80]}'")
 
-    # 第一条、第二条、... -> h2 (条文编号)
+    if reclassified > 0:
-    if re.match(r'^第[一二三四五六七八九十百千万]+[条款]', text):
+        logger.info(f"表单类型二次校正: 共 {reclassified} 个 text -> table")
-        return 2
 
-    # 数字章节: 1. 2. 3. 或 1、2、3、
-    # 一级标题: 1. 2. 3. (单数字)
-    if re.match(r'^\d+[\.、\s]', text):
-        # 短文本可能是标题
-        if len(text) < 50:
-            return 1
 
-    # 二级标题: 1.1 1.2 2.1 等
+def _validate_category_section_paths(chunks: List[MinerUChunk]) -> None:
-    if re.match(r'^\d+\.\d+[\.、\s]', text):
+    """
-        if len(text) < 80:
+    验证分类标题是否被规则引擎正确识别（安全网）
-            return 2
 
-    # 三级标题: 1.1.1 1.1.2 等
+    当规则引擎正确识别分类标题后，section_stack 自然会更新，
-    if re.match(r'^\d+\.\d+\.\d+[\.、\s]', text):
+    不再需要后处理修改 section_path。此函数仅做验证和告警，
-        if len(text) < 100:
+    便于发现规则引擎的遗漏。
-            return 3
 
-    # 英文章节标题
+    支持的模式：A1类：、B2类：、C1类：等（含可选 ** 粗体标记）。
-    # Chapter 1, Section 2, etc.
+    """
-    if re.match(r'^(Chapter|Section|Part|Chapter\s+\d+|Section\s+\d+)', text, re.IGNORECASE):
+    cat_pattern = re.compile(r'^\*{0,2}[A-Z]\d+[类類]\*{0,2}[：:]')
-        return 1
 
-    # 短文本 + 加粗标记 (**xxx**) 可能是标题
+    missed_count = 0
-    if re.match(r'^\*\*.+\*\*$', text) and len(text) < 50:
+    for chunk in chunks:
-        return 2
+        if chunk.chunk_type == 'text':
+            text = (chunk.content or '').strip()
+            if cat_pattern.match(text) and chunk.text_level == 0:
+                missed_count += 1
+                logger.warning(
+                    f"分类标题未被识别为标题: '{text[:50]}', "
+                    f"section_path='{chunk.section_path}'"
+                )
 
-    # 非常短的文本 (< 20 字符) 可能是标题
+    if missed_count > 0:
-    # 但需要排除常见的非标题短文本
+        logger.warning(
-    if len(text) < 20 and not re.match(r'^[\d\s\.,;:!?，。；：！？、]+$', text):
+            f"发现 {missed_count} 个分类标题未被规则引擎识别，"
-        # 排除纯数字、纯标点
+            f"请检查 heading_rules 配置"
-        if re.search(r'[\u4e00-\u9fff]', text):  # 包含中文
+        )
-            return 2
-
-    return 0
 
 
 def _parse_mineru_output(file_path: Path, output_dir) -> Dict[str, Any]:
@@ -803,16 +1001,45 @@ def _parse_mineru_output(file_path: Path, output_dir) -> Dict[str, Any]:
     else:
         raise RuntimeError(f"MinerU 输出目录不存在: {auto_dir} 或 {office_dir}")
 
-    # 读取 content_list.json
+    # 读取 content_list（v2 含 style 信息，优先使用）
-    content_list_path = output_subdir / f"{doc_name}_content_list.json"
+    try:
-    if not content_list_path.exists():
+        from config import MINERU_PREFER_V2
-        content_list_path = output_subdir / f"{doc_name}_content_list_v2.json"
+    except ImportError:
+        MINERU_PREFER_V2 = True
+
+    v1_path = output_subdir / f"{doc_name}_content_list.json"
+    v2_path = output_subdir / f"{doc_name}_content_list_v2.json"
+
+    content_list_path = None
+    is_v2 = False
+
+    if MINERU_PREFER_V2:
+        # 优先使用 v2 格式
+        if v2_path.exists():
+            content_list_path = v2_path
+            is_v2 = True
+        elif v1_path.exists():
+            content_list_path = v1_path
+    else:
+        # 优先使用 v1 格式
+        if v1_path.exists():
+            content_list_path = v1_path
+        elif v2_path.exists():
+            content_list_path = v2_path
+            is_v2 = True
 
     content_list = []
-    if content_list_path.exists():
+    if content_list_path and content_list_path.exists():
         with open(content_list_path, 'r', encoding='utf-8') as f:
             content_list = json.load(f)
 
+        # v2 格式需要转换为 v1 兼容的扁平列表
+        if is_v2 and isinstance(content_list, list) and content_list and isinstance(content_list[0], list):
+            content_list = _parse_v2_content_list(content_list)
+            logger.info(f"v2 格式已转换为扁平列表，共 {len(content_list)} 项")
+
+        logger.info(f"读取 content_list: {len(content_list)} 项, 格式={'v2' if is_v2 else 'v1'}")
+
     # 读取 Markdown
     md_path = output_subdir / f"{doc_name}.md"
     markdown_content = ""
@@ -857,10 +1084,14 @@ def _parse_mineru_output(file_path: Path, output_dir) -> Dict[str, Any]:
 
         if item_type == "text":
             text = item.get("text", "")
+            v2_styles = item.get("_v2_styles", [])
 
             # 启发式标题识别（当 text_level 为 0 时）
             if text_level == 0:
-                text_level = _detect_heading_level(text)
+                from parsers.heading_rules import get_heading_engine
+                engine = get_heading_engine()
+                detected_level, _ = engine.detect(text, style=v2_styles)
+                text_level = detected_level
 
             # 处理标题
             title = ""
@@ -909,8 +1140,8 @@ def _parse_mineru_output(file_path: Path, output_dir) -> Dict[str, Any]:
             else:
                 md_table = ""
 
-            # 提取表格中的嵌入图片
+            # 从原始 HTML 提取嵌入图片（md_table 经 get_text 转换后已丢失 <img> 标签）
-            table_images = extract_images_from_markdown(md_table) if md_table else []
+            table_images = extract_images_from_markdown(table_body) if table_body else []
 
             chunk = MinerUChunk(
                 content=table_caption or "表格",
@@ -1003,8 +1234,14 @@ def _parse_mineru_output(file_path: Path, output_dir) -> Dict[str, Any]:
         min_merge = 100
         max_size = 1200
 
+    # 表单类型二次校正（在 _post_process_chunks 之前，因为 table 不参与合并）
+    _reclassify_text_chunks(chunks)
+
     chunks = _post_process_chunks(chunks, min_merge_size=min_merge, max_chunk_size=max_size)
 
+    # 验证分类标题是否被规则引擎正确识别（安全网，仅告警不修改）
+    _validate_category_section_paths(chunks)
+
     return {
         'markdown': "\n".join(markdown_parts),
         'chunks': chunks,
@@ -1349,7 +1586,17 @@ def html_table_to_markdown(html_table: str) -> str:
                         del rowspan_tracker[col_idx]
                     col_idx += 1
 
-                # 提取单元格内容
+                # 提取单元格内容（保留图片引用信息）
+                img_tags = cell.find_all('img')
+                if img_tags:
+                    # 单元格包含图片，生成占位标记供 LLM 感知
+                    text_part = cell.get_text(strip=True)
+                    img_count = len(img_tags)
+                    if text_part:
+                        content = f"{text_part} [{'图片' if img_count == 1 else f'{img_count}张图片'}]"
+                    else:
+                        content = f"[{'图片' if img_count == 1 else f'{img_count}张图片'}]"
+                else:
                     content = cell.get_text(strip=True)
 
                 # 处理 rowspan
@@ -1630,6 +1877,21 @@ def parse_with_mineru_persistent(
                         chunk.image_path = new_name
                         break
 
+        # 更新表格嵌入图片的路径映射（images 字段）
+        if hasattr(chunk, 'images') and chunk.images:
+            for img_info in chunk.images:
+                if isinstance(img_info, dict) and 'id' in img_info:
+                    old_id = img_info['id']
+                    if old_id in image_path_map:
+                        img_info['id'] = image_path_map[old_id]
+                    else:
+                        # 兼容：尝试用文件名匹配映射
+                        old_basename = os.path.basename(old_id)
+                        for old_path, new_name in image_path_map.items():
+                            if old_basename == os.path.basename(old_path):
+                                img_info['id'] = new_name
+                                break
+
     # 更新结果中的图片路径列表（供外部使用）
     result['images'] = list(image_path_map.values())