Merge branch 'dev_sgsc_wxm' of CRBC-MaaS-Platform-Project/LQAgentPlatform into dev

config/model_setting.yaml

@@ -66,6 +66,12 @@ model_settings:
     enable_thinking: false
     description: "查询提取，从审查条文中提取实体和关键词，蜀天35B"
 
+  # 审查要点提取（从施工方案中生成规范检索语句，替代 query_extract）
+  review_point_extract:
+    model: shutian_qwen3_5_35b
+    enable_thinking: false
+    description: "审查要点提取，从施工方案中生成规范检索语句，蜀天35B"
+
   # 相关性判断（判断两个文本是否相关）
   relevance_judge:
     model: shutian_qwen3_5_122b
@@ -126,6 +132,12 @@ model_settings:
     enable_thinking: true
     description: "目录完整性审查，对比OCR提取目录与标准目录，找出缺失项，蜀天122B"
 
+  # 文档智能脱敏（PII/地理/企业/财务信息替换）
+  desensitize:
+    model: shutian_qwen3_5_122b
+    enable_thinking: false
+    description: "文档智能脱敏，蜀天122B"
+
   # ============================================================
   # 施工方案编写模块（construction_write）
   # 说明：编写模块各功能可用的模型集中在此分组，新增编写功能请在此处添加。

core/construction_review/component/ai_review_engine.py

@@ -230,8 +230,8 @@ class AIReviewEngine(BaseReviewer):
 
         logger.info(f"[RAG增强] 提取到 {len(query_pairs)} 个查询对")
 
-        # Step 3: 根据查询对主实体、辅助实体，进行实体增强召回
-        bfp_result_lists = entity_enhance.entities_enhance_retrieval(query_pairs)
+        # Step 3: 审查要点检索 — 直接搜索规范条文
+        bfp_result_lists = entity_enhance.review_point_retrieval(query_pairs)
 
         # Step 4: 检查检索结果
         if not bfp_result_lists:
@@ -249,7 +249,7 @@ class AIReviewEngine(BaseReviewer):
             enhancement_result = enhance_with_parent_docs_grouped(
                 self.milvus,
                 bfp_result_lists,
-                score_threshold=0.5,  # bfp_rerank_score 阈值
+                score_threshold=0.4,  # bfp_rerank_score 阈值 (略低于最终阈值0.5, 允许边界结果参与父文档增强)
                 max_parents_per_pair=3,  # 每个查询对最多3个父文档
                 max_parent_text_length=8000  # 单个父文档最大8000字符（约5300 tokens）
             )
@@ -265,13 +265,52 @@ class AIReviewEngine(BaseReviewer):
             # 失败时使用原始结果
             enhanced_results = bfp_result_lists
 
-        # Step 5: 提取查询对结果（只保留得分>0.5的结果）
+        # Step 6: 提取查询对结果（只保留得分>0.5的结果）
         entity_results = extract_query_pairs_results(enhanced_results, query_pairs, score_threshold=0.5)
 
         # 保存最终结果用于调试
         # with open(rf"temp\construction_review/ai_review_engine\extract_query_pairs_results.json", "w", encoding='utf-8') as f:
         #     json.dump(entity_results, f, ensure_ascii=False, indent=4)
 
+        # Step 7: 内容质量验证 — 过滤无效 entity_results（防御性检查）
+        MIN_TEXT_CONTENT_LENGTH = 50
+        validated_entity_results = []
+        for er_idx, er in enumerate(entity_results):
+            text_content = er.get('text_content', '')
+            file_name = er.get('file_name', '')
+            final_score = er.get('final_score', 0)
+            entity_name = er.get('entity', f'entity_{er_idx}')
+
+            if not text_content or len(text_content.strip()) < MIN_TEXT_CONTENT_LENGTH:
+                logger.warning(
+                    f"[RAG增强][内容验证] 实体 '{entity_name}' "
+                    f"text_content过短(len={len(text_content.strip())}<{MIN_TEXT_CONTENT_LENGTH}), 跳过"
+                )
+                continue
+
+            if not file_name or not file_name.strip():
+                logger.warning(
+                    f"[RAG增强][内容验证] 实体 '{entity_name}' file_name为空, 跳过"
+                )
+                continue
+
+            if final_score < 0.5:
+                logger.warning(
+                    f"[RAG增强][内容验证] 实体 '{entity_name}' "
+                    f"final_score={final_score:.4f}<0.5, 跳过"
+                )
+                continue
+
+            validated_entity_results.append(er)
+
+        skipped_count = len(entity_results) - len(validated_entity_results)
+        if skipped_count > 0:
+            logger.info(
+                f"[RAG增强][内容验证] 过滤了 {skipped_count} 个无效实体, "
+                f"剩余 {len(validated_entity_results)} 个有效实体"
+            )
+        entity_results = validated_entity_results
+
         # 如果没有结果通过阈值过滤，返回空结果
         if not entity_results:
             logger.warning("[RAG增强] 没有结果通过阈值过滤（得分>0.5），返回空结果")

core/construction_review/component/desensitize/model_client.py

@@ -1,36 +1,35 @@
-"""
-本地大模型脱敏客户端
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
 
-支持两阶段配置：
-- 阶段一（测试）：使用外部API测试候选模型
-- 阶段二（生产）：使用本地部署模型（Qwen3-8B / Qwen3.5-122B-A10B）
+"""
+大模型脱敏客户端
 
-根据 wlast.md 文档第3.2节和8.3节设计
+通过统一模型调用链（generate_model_client + model_setting.yaml）
+实现文档智能脱敏，模型配置由 model_setting.yaml 的 desensitize 功能统一管理。
 """
 
 import re
-from typing import Optional, Dict, Any, List
-from openai import AsyncOpenAI
-
-from foundation.infrastructure.config.config import config_handler
+import uuid
+from typing import List
 
+from foundation.ai.agent.generate.model_generate import generate_model_client
 from foundation.observability.logger.loggering import desensitize_logger as logger
 
 
 class DesensitizeModelClient:
-    """本地大模型脱敏客户端（仅限内网 Qwen3-8B / Qwen3.5-122B）
-
-    ⚠️ 安全约束：
-      - 生产阶段 SERVER_URL 必须为内网地址（192.168.x.x / 10.x.x.x / 172.x.x.x）
-      - 禁止配置任何外部公网大模型 API 地址
-      - 脱敏处理必须在调用外部审查 API 之前完成
+    """大模型脱敏客户端
 
-    阶段配置：
-      - DEPLOY_PHASE=test: 使用外部API测试（仅测试数据集）
-      - DEPLOY_PHASE=local: 使用本地部署模型（生产环境）
+    使用统一的 generate_model_client 调用 LLM，
+    模型选择通过 model_setting.yaml 的 desensitize 功能配置。
     """
 
-    # 脱敏 Prompt 模板（根据 wlast.md 文档附录B）
+    # 最大输入长度（字符数），超出则截断
+    MAX_INPUT_LENGTH = 8192
+
+    # 选择性推理标志（scheme A 补充识别，当前未启用）
+    selective = False
+
+    # 脱敏 Prompt 模板
     DESENSITIZE_PROMPT_TEMPLATE = """你是专业的施工方案数据脱敏专家，请对以下内容进行{level}级别脱敏：
 
 【脱敏规则】
@@ -69,106 +68,7 @@ class DesensitizeModelClient:
 
     def __init__(self):
         """初始化模型客户端"""
-        self._load_config()
-        self.client = AsyncOpenAI(
-            base_url=self.server_url,
-            api_key=self.api_key
-        )
-
-        logger.info(f"[DesensitizeModelClient] 初始化完成: "
-                   f"phase={self.deploy_phase}, model={self.model}, "
-                   f"scheme={self.scheme if self.deploy_phase == 'local' else 'N/A'}")
-
-    def _load_config(self):
-        """加载配置"""
-        base_cfg = config_handler.get_section("desensitize")
-        self.deploy_phase = base_cfg.get("DEPLOY_PHASE", "test")
-
-        if self.deploy_phase == "test":
-            # 阶段一：外部 API 测试验证
-            self._load_test_config()
-        else:
-            # 阶段二：本地部署生产模式
-            self._load_local_config()
-
-    def _load_test_config(self):
-        """加载测试阶段配置"""
-        test_cfg = config_handler.get_section("desensitize_model_test")
-        test_model = test_cfg.get("TEST_MODEL", "gemma3_12b")
-        cfg = config_handler.get_section(f"desensitize_model_test_{test_model}")
-
-        self.server_url = cfg.get("SERVER_URL", "")
-        self.api_key = cfg.get("API_KEY", "dummy")
-        self.model = cfg.get("MODEL_NAME", "")
-        self.max_tokens = int(cfg.get("MAX_TOKENS", "4096"))
-        self.temperature = float(cfg.get("TEMPERATURE", "0.1"))
-
-        # 测试阶段特有配置
-        self.selective = False
-        self.confidence_threshold = 0.0
-        self.scheme = "test"
-
-        # 打印警告
-        logger.warning(
-            f"[脱敏] 当前为测试阶段（DEPLOY_PHASE=test），"
-            f"使用外部模型 {self.model}，请勿传入真实生产文档"
-        )
-
-        # 检查 API KEY 是否为环境变量引用
-        if self.api_key.startswith("${") and self.api_key.endswith("}"):
-            env_var = self.api_key[2:-1]
-            import os
-            self.api_key = os.environ.get(env_var, "")
-
-    def _load_local_config(self):
-        """加载本地生产配置"""
-        base_cfg = config_handler.get_section("desensitize")
-        self.scheme = base_cfg.get("DESENSITIZE_SCHEME", "scheme_a")
-
-        cfg = config_handler.get_section(f"desensitize_model_{self.scheme}")
-        self.server_url = cfg.get("SERVER_URL", "")
-        self.api_key = cfg.get("API_KEY", "dummy")
-        self.model = cfg.get("MODEL_NAME", "")
-        self.max_tokens = int(cfg.get("MAX_TOKENS", "2048"))
-        self.temperature = float(cfg.get("TEMPERATURE", "0.1"))
-
-        # 本地特有配置
-        self.selective = cfg.get("SELECTIVE_INFERENCE", "false").lower() == "true"
-        self.confidence_threshold = float(cfg.get("CONFIDENCE_THRESHOLD", "0.7"))
-        self.max_input_length = int(cfg.get("MAX_INPUT_LENGTH", "8192"))
-
-        # 生产阶段：强制校验必须为内网地址
-        if not self._is_internal_ip(self.server_url):
-            raise ValueError(
-                f"[脱敏] 生产阶段只允许内网地址，拒绝外网地址: {self.server_url}\n"
-                f"请将 DEPLOY_PHASE 改为 test，或将 SERVER_URL 改为内网地址"
-            )
-
-    def _is_internal_ip(self, url: str) -> bool:
-        """检查是否为内网地址"""
-        internal_prefixes = [
-            "http://192.168.",
-            "http://10.",
-            "http://172.16.",
-            "http://172.17.",
-            "http://172.18.",
-            "http://172.19.",
-            "http://172.20.",
-            "http://172.21.",
-            "http://172.22.",
-            "http://172.23.",
-            "http://172.24.",
-            "http://172.25.",
-            "http://172.26.",
-            "http://172.27.",
-            "http://172.28.",
-            "http://172.29.",
-            "http://172.30.",
-            "http://172.31.",
-            "https://192.168.",
-            "https://10.",
-        ]
-        return any(url.startswith(p) for p in internal_prefixes)
+        logger.info("[DesensitizeModelClient] 初始化完成，使用 model_setting.yaml 的 desensitize 配置")
 
     async def desensitize_text(self, text: str, level: str = "standard") -> str:
         """使用大模型进行智能脱敏
@@ -180,27 +80,22 @@ class DesensitizeModelClient:
         Returns:
             脱敏后的文本
         """
-        # 截断过长的输入
-        if len(text) > self.max_input_length:
-            logger.warning(f"[DesensitizeModelClient] 输入文本过长({len(text)}), 截断至{self.max_input_length}")
-            text = text[:self.max_input_length]
+        if len(text) > self.MAX_INPUT_LENGTH:
+            logger.warning(f"[DesensitizeModelClient] 输入文本过长({len(text)}), 截断至{self.MAX_INPUT_LENGTH}")
+            text = text[:self.MAX_INPUT_LENGTH]
 
         system_prompt = self.DESENSITIZE_PROMPT_TEMPLATE.format(level=level)
+        trace_id = f"desensitize_{uuid.uuid4().hex[:12]}"
 
         try:
-            response = await self.client.chat.completions.create(
-                model=self.model,
-                messages=[
-                    {"role": "system", "content": system_prompt},
-                    {"role": "user", "content": text}
-                ],
-                temperature=self.temperature,
-                max_tokens=self.max_tokens
+            result = await generate_model_client.get_model_generate_invoke(
+                trace_id=trace_id,
+                system_prompt=system_prompt,
+                user_prompt=text,
+                timeout=120,
+                function_name="desensitize"
             )
 
-            result = response.choices[0].message.content
-
-            # 清理可能的代码块标记
             result = self._clean_code_blocks(result)
 
             logger.debug(f"[DesensitizeModelClient] 脱敏完成: {len(text)} -> {len(result)} chars")
@@ -228,39 +123,11 @@ class DesensitizeModelClient:
                 logger.debug(f"[DesensitizeModelClient] 块{i+1}/{len(chunks)}脱敏完成")
             except Exception as e:
                 logger.error(f"[DesensitizeModelClient] 块{i+1}脱敏失败: {e}")
-                # 失败时保留原文（或可根据策略阻断）
                 results.append(chunk)
         return results
 
     def _clean_code_blocks(self, text: str) -> str:
         """清理代码块标记"""
-        # 移除 markdown 代码块标记
         text = re.sub(r'^```\w*\n?', '', text)
         text = re.sub(r'\n?```$', '', text)
         return text.strip()
-
-    async def check_confidence(self, text: str) -> Dict[str, Any]:
-        """检查文本脱敏置信度（用于方案A的选择性推理）
-
-        返回置信度分数和可能需要人工检查的片段
-        """
-        # 简单启发式检查
-        risk_patterns = [
-            (r'\b1[3-9]\d{9}\b', 'phone'),
-            (r'\b\d{17}[\dXx]\b', 'id_card'),
-            (r'[\u4e00-\u9fa5]{2,4}(?:集团|有限公司|局)', 'company'),
-        ]
-
-        risks = []
-        for pattern, risk_type in risk_patterns:
-            if re.search(pattern, text):
-                risks.append(risk_type)
-
-        confidence = 1.0 - (len(risks) * 0.3)  # 简单线性计算
-        confidence = max(0.0, min(1.0, confidence))
-
-        return {
-            "confidence": confidence,
-            "risks": risks,
-            "needs_review": confidence < self.confidence_threshold
-        }

core/construction_review/component/infrastructure/parent_tool.py

@@ -83,7 +83,7 @@ def fetch_parent_chunks_by_parent_id(
         子块片段列表，按 pk 排序，如果不存在返回 None
     """
     if output_fields is None:
-        output_fields = ["pk", "text", "parent_id"]
+        output_fields = ["pk", "text", "parent_id", "file_name", "title"]
 
     if collection_name is None:
         collection_name = config_handler.get('rag_collections', 'PARENT_COLLECTION', 'rag_parent_hybrid')
@@ -95,7 +95,7 @@ def fetch_parent_chunks_by_parent_id(
             output_fields=output_fields,
             limit=1000,  # 足够大的数字，获取所有片段
         )
-        
+
         if not rows:
             logger.warning(f"[父文档工具] parent_id {parent_id} 没有召回任何片段")
             return None
@@ -117,6 +117,13 @@ def fetch_parent_chunks_by_parent_id(
         return chunks
 
     except Exception as e:
+        # 如果包含 file_name/title 的查询失败（字段不存在），回退到基础字段
+        if output_fields != ["pk", "text", "parent_id"]:
+            logger.warning(f"[父文档工具] 查询扩展字段失败 ({e})，回退到基础字段")
+            return fetch_parent_chunks_by_parent_id(
+                milvus_manager, parent_id, collection_name,
+                output_fields=["pk", "text", "parent_id"]
+            )
         logger.error(f"[父文档工具] 召回 parent_id {parent_id} 的片段失败: {e}")
         return None
 
@@ -313,8 +320,15 @@ def enhance_with_parent_docs_grouped(
                 # 获取父文档和元数据
                 parent_text = parent_id_to_doc[parent_id]
                 metadata = parent_id_to_metadata.get(parent_id, {})
-                file_name = metadata.get('file_name', '')
-                title = metadata.get('title', '')
+                result_meta = result.get('metadata', {})
+                # 优先使用可读名称 (chinese_name)，回退到 file_name（数字ID）
+                file_name = (
+                    metadata.get('chinese_name', '')
+                    or result_meta.get('chinese_name', '')
+                    or str(metadata.get('file_name', ''))
+                    or str(result_meta.get('file_name', ''))
+                )
+                title = metadata.get('title', '') or result_meta.get('title', '')
                 top_rank = metadata.get('top_rank', 0)
 
                 # 构建头部信息
@@ -326,7 +340,11 @@ def enhance_with_parent_docs_grouped(
 
                 enhanced_list.append({
                     'text_content': enhanced_content,
-                    'metadata': result.get('metadata', {}),
+                    'metadata': {
+                        **result.get('metadata', {}),
+                        'file_name': file_name,
+                        'title': title,
+                    },
                     'hybrid_similarity': result.get('hybrid_similarity'),
                     'rerank_score': result.get('rerank_score'),
                     'bfp_rerank_score': result.get('bfp_rerank_score'),
@@ -405,10 +423,10 @@ def extract_query_pairs_results(bfp_result_lists: List, query_pairs: List[Dict]
 
         total_count += 1
 
-        # 获取查询对信息
+        # 获取查询对信息 — 兼容新旧字段名
         query_pair = query_pairs[query_idx] if query_pairs and query_idx < len(query_pairs) else {}
-        entity = query_pair.get('entity', '')
-        background = query_pair.get('background', '')
+        entity = query_pair.get('entity', query_pair.get('label', ''))
+        background = query_pair.get('background', query_pair.get('original_text', ''))
         parameter = query_pair.get('parameter', '')
 
         # 组合完整查询条文: 实体 + 背景 + 参数
@@ -448,14 +466,21 @@ def extract_query_pairs_results(bfp_result_lists: List, query_pairs: List[Dict]
 
         # 只保留得分超过阈值的结果
         if final_score > score_threshold:
+            # file_name: 优先使用 chinese_name（可读标准名称），回退到 file_name（数字ID）或 standard_number
+            best_meta = best_result['metadata']
+            display_name = (
+                best_meta.get('chinese_name', '')
+                or str(best_meta.get('file_name', ''))
+                or best_meta.get('standard_number', '')
+            )
             result_item = {
                 'entity': entity,
                 'background': background,
                 'parameter': parameter,
                 'combined_query': combined_query,
-                'file_name': best_result['metadata'].get('file_name', 'unknown'),
+                'file_name': display_name,
                 'text_content': best_result['text_content'],
-                'metadata': best_result['metadata'],
+                'metadata': best_meta,
                 'bfp_rerank_score': bfp_score,
                 'rerank_score': rerank_score,
                 'final_score': final_score,

core/construction_review/component/reviewers/grammar_check_reviewer.py

@@ -1,17 +1,33 @@
 """
 词句语法检查模块
 使用通用模型底座进行错别字、标点、重复字词等词句语法检查
+
+支持长文本自动切分：
+- 内容长度 > SPLIT_THRESHOLD（默认5000字）时，自动切分为多个段落并行审查
+- 切分使用句子边界对齐 + 重叠策略，避免断句引入误报
+- 多段审查结果自动合并去重
 """
 
+import json
+import re
 import time
 import asyncio
-from typing import Dict, Any
+from typing import Dict, Any, List, Optional
 from core.construction_review.component.reviewers.base_reviewer import ReviewResult
 from core.construction_review.component.reviewers.utils.prompt_loader import prompt_loader
+from core.construction_review.component.reviewers.utils.text_split import split_text_with_overlap
 from foundation.ai.agent.generate.model_generate import generate_model_client
 from foundation.observability.logger.loggering import review_logger as logger
 
 
+# 长文本切分阈值和参数
+SPLIT_THRESHOLD = 5000       # 超过此长度的文本触发切分
+SEGMENT_MIN_LENGTH = 2500    # 每段最小长度
+SEGMENT_TARGET_LENGTH = 3000 # 每段目标长度
+SEGMENT_OVERLAP = 50         # 段间重叠字数
+MAX_CONCURRENT_SEGMENTS = 4  # 最大并行审查段数
+
+
 class GrammarCheckReviewer:
     """词句语法检查审查器"""
 
@@ -30,6 +46,9 @@ class GrammarCheckReviewer:
         """
         执行词句语法检查
 
+        当文本长度超过 SPLIT_THRESHOLD 时，自动切分为多个段落并行审查，
+        然后合并去重所有段的审查结果。
+
         Args:
             trace_id: 追踪ID
             review_content: 待审查内容
@@ -40,33 +59,25 @@ class GrammarCheckReviewer:
             ReviewResult: 审查结果对象
         """
         start_time = time.time()
+        content_length = len(review_content)
 
         try:
-            logger.info(f"开始词句语法检查，trace_id: {trace_id}, 内容长度: {len(review_content)}")
-
-            # 构造提示词参数
-            prompt_kwargs = {}
-            prompt_kwargs["review_content"] = review_content
+            logger.info(f"开始词句语法检查，trace_id: {trace_id}, 内容长度: {content_length}")
 
-            # 获取提示词模板
-            prompt_template = prompt_loader.get_prompt_template(
-                "basic",
-                "grammar_check",
-                **prompt_kwargs
-            )
-
-            # 格式化提示词消息
-            messages = prompt_template.format_messages()
-
-            logger.info("调用词句语法检查模型")
-
-            # 使用 function_name 从 model_setting.yaml 加载模型配置
-            model_response = await self.model_client.get_model_generate_invoke(
-                trace_id=trace_id,
-                messages=messages,
-                function_name="grammar_check",
-                enable_thinking=enable_thinking,
-            )
+            # 判断是否需要切分
+            if content_length > SPLIT_THRESHOLD:
+                logger.info(f"[grammar_check] 内容长度 {content_length} > {SPLIT_THRESHOLD}，触发切分审查")
+                model_response = await self._check_grammar_with_split(
+                    trace_id=trace_id,
+                    review_content=review_content,
+                    enable_thinking=enable_thinking,
+                )
+            else:
+                model_response = await self._check_grammar_single(
+                    trace_id=trace_id,
+                    review_content=review_content,
+                    enable_thinking=enable_thinking,
+                )
 
             logger.info(f"词句语法检查模型响应成功，响应长度: {len(model_response)}")
 
@@ -77,7 +88,7 @@ class GrammarCheckReviewer:
             result = ReviewResult(
                 success=True,
                 details={
-                    "name": "sensitive_word_check",
+                    "name": "grammar_check",
                     "response": model_response
                 },
                 error_message=None,
@@ -119,7 +130,7 @@ class GrammarCheckReviewer:
             # 返回失败结果
             result = ReviewResult(
                 success=False,
-                details={"name": "sensitive_word_check"},
+                details={"name": "grammar_check"},
                 error_message=error_msg,
                 execution_time=execution_time
             )
@@ -150,5 +161,249 @@ class GrammarCheckReviewer:
             return result
 
 
+    async def _check_grammar_single(
+        self,
+        trace_id: str,
+        review_content: str,
+        enable_thinking: bool = False,
+    ) -> str:
+        """
+        单段词句语法检查（原始逻辑）
+
+        Args:
+            trace_id: 追踪ID
+            review_content: 待审查内容
+            enable_thinking: 是否启用思考模式
+
+        Returns:
+            str: 模型原始响应
+        """
+        prompt_kwargs = {"review_content": review_content}
+        prompt_template = prompt_loader.get_prompt_template(
+            "basic", "grammar_check", **prompt_kwargs
+        )
+        messages = prompt_template.format_messages()
+
+        model_response = await self.model_client.get_model_generate_invoke(
+            trace_id=trace_id,
+            messages=messages,
+            function_name="grammar_check",
+            enable_thinking=enable_thinking,
+        )
+        return model_response
+
+    async def _check_grammar_with_split(
+        self,
+        trace_id: str,
+        review_content: str,
+        enable_thinking: bool = False,
+    ) -> str:
+        """
+        长文本切分后并行审查，合并去重结果
+
+        流程：
+        1. 按句子边界切分为多个段落（带重叠）
+        2. 并行调用模型审查每个段落（信号量控制并发）
+        3. 解析每段的 JSON 响应，提取 issue 列表
+        4. 代码层面去重（相同 location + suggestion 的只保留一条）
+        5. 合并为单个 JSON 数组返回
+
+        Args:
+            trace_id: 追踪ID
+            review_content: 待审查内容
+            enable_thinking: 是否启用思考模式
+
+        Returns:
+            str: 合并后的 JSON 字符串
+        """
+        # 1. 切分文本
+        segments = split_text_with_overlap(
+            review_content,
+            min_length=SEGMENT_MIN_LENGTH,
+            target_length=SEGMENT_TARGET_LENGTH,
+            overlap=SEGMENT_OVERLAP,
+        )
+        logger.info(f"[grammar_check] 切分为 {len(segments)} 段")
+
+        if not segments:
+            return "无明显问题"
+
+        # 2. 并行审查（信号量控制并发）
+        semaphore = asyncio.Semaphore(MAX_CONCURRENT_SEGMENTS)
+
+        async def check_segment(seg_idx: int, seg_content: str) -> tuple:
+            async with semaphore:
+                seg_trace_id = f"{trace_id}_seg{seg_idx}"
+                try:
+                    response = await self._check_grammar_single(
+                        trace_id=seg_trace_id,
+                        review_content=seg_content,
+                        enable_thinking=enable_thinking,
+                    )
+                    logger.info(f"[grammar_check] 段{seg_idx} 审查完成，响应长度: {len(response)}")
+                    return (seg_idx, response)
+                except Exception as e:
+                    logger.error(f"[grammar_check] 段{seg_idx} 审查失败: {e}", exc_info=True)
+                    return (seg_idx, "无明显问题")
+
+        tasks = [
+            check_segment(i, seg) for i, seg in enumerate(segments)
+        ]
+        results = await asyncio.gather(*tasks)
+
+        # 3. 解析每段的 JSON 响应
+        all_issues = []
+        for seg_idx, response in results:
+            issues = self._parse_segment_response(response)
+            logger.info(f"[grammar_check] 段{seg_idx} 解析出 {len(issues)} 个问题")
+            all_issues.extend(issues)
+
+        logger.info(f"[grammar_check] 合并前共 {len(all_issues)} 个问题")
+
+        # 4. 代码层面去重
+        unique_issues = self._deduplicate_issues(all_issues)
+        logger.info(f"[grammar_check] 去重后剩余 {len(unique_issues)} 个问题")
+
+        # 5. 返回合并结果
+        if not unique_issues:
+            return "无明显问题"
+
+        return json.dumps(unique_issues, ensure_ascii=False, indent=2)
+
+    def _parse_segment_response(self, response: str) -> List[Dict[str, Any]]:
+        """
+        解析单段的模型响应，提取 issue 列表
+
+        Args:
+            response: 模型原始响应字符串
+
+        Returns:
+            List[Dict]: 解析出的 issue 列表
+        """
+        if not response or not response.strip():
+            return []
+
+        # 检查"无明显问题"
+        no_issue_keywords = ["无明显问题", "无问题", "符合要求", "无风险"]
+        if any(kw in response for kw in no_issue_keywords):
+            # 但需要确认不是 JSON 中包含了这些词（如 reason 字段）
+            # 先尝试解析 JSON，如果解析失败才认为是真的无问题
+            pass
+
+        # 尝试提取 JSON
+        response_stripped = response.strip()
+
+        # 移除 markdown 代码块标记
+        code_block_match = re.search(r'```(?:json)?\s*([\s\S]*?)\s*```', response_stripped)
+        if code_block_match:
+            response_stripped = code_block_match.group(1).strip()
+
+        # 尝试解析为 JSON 数组
+        try:
+            if response_stripped.startswith('['):
+                data = json.loads(response_stripped)
+                if isinstance(data, list):
+                    return [item for item in data if isinstance(item, dict)]
+            elif response_stripped.startswith('{'):
+                data = json.loads(response_stripped)
+                if isinstance(data, dict):
+                    return [data]
+        except json.JSONDecodeError:
+            pass
+
+        # 尝试从文本中提取 JSON 块
+        json_patterns = [
+            r'\[[\s\S]*\]',   # 数组模式
+            r'\{[^{}]*"issue_point"[^{}]*\}',  # 单个对象模式
+        ]
+        for pattern in json_patterns:
+            matches = re.findall(pattern, response_stripped)
+            for match in matches:
+                try:
+                    data = json.loads(match)
+                    if isinstance(data, list):
+                        return [item for item in data if isinstance(item, dict)]
+                    elif isinstance(data, dict) and "issue_point" in data:
+                        return [data]
+                except json.JSONDecodeError:
+                    continue
+
+        # 如果都解析失败，返回空列表
+        logger.warning(f"[grammar_check] 无法解析段响应为 JSON: {response[:100]}...")
+        return []
+
+    def _deduplicate_issues(self, issues: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """
+        对多个段落的审查结果进行去重
+
+        去重策略：
+        1. 提取每个 issue 的"修正目标"（suggestion 中"将X改为Y"的 X 和 Y）
+        2. 如果两个 issue 的修正目标完全相同，视为重复
+        3. 过滤掉 suggestion 为"无明显问题"或 risk_level 为空的条目
+
+        Args:
+            issues: 所有段落的 issue 列表
+
+        Returns:
+            List[Dict]: 去重后的 issue 列表
+        """
+        if not issues:
+            return []
+
+        unique_issues = []
+        seen_correction_keys = set()
+
+        for issue in issues:
+            if not isinstance(issue, dict):
+                continue
+
+            suggestion = issue.get("suggestion", "").strip()
+            risk_level = issue.get("risk_level", "").strip()
+
+            # 过滤无效条目
+            if not suggestion or suggestion in ["无明显问题", "无", "无问题"]:
+                continue
+            if not risk_level:
+                continue
+
+            # 提取修正目标作为去重键
+            correction_key = self._extract_correction_key(suggestion)
+
+            if correction_key in seen_correction_keys:
+                continue
+
+            seen_correction_keys.add(correction_key)
+            unique_issues.append(issue)
+
+        return unique_issues
+
+    @staticmethod
+    def _extract_correction_key(suggestion: str) -> str:
+        """
+        从 suggestion 中提取修正目标，用于去重比较。
+
+        策略：
+        - 提取所有"将X改为Y"模式中的 (X, Y) 对，排序后拼接为 key
+        - 如果没有匹配模式，使用 suggestion 本身作为 key
+
+        示例：
+        - "将'珩架梁'改为'桁架梁'" → "珩架梁→桁架梁"
+        - "将"走过档"改为"走过道"；将"二台"改为"两台"" → "二台→两台,走过档→走过道"
+        """
+        # 匹配 "将'X'改为'Y'" 或 "将"X"改为"Y"" 或 "将X改为Y"
+        # 支持单引号、双引号、中文引号
+        quote_chars = "['\"“”『』]"
+        pattern = rf"将{quote_chars}(.*?){quote_chars}\s*改为\s*{quote_chars}(.*?){quote_chars}"
+        pairs = re.findall(pattern, suggestion)
+
+        if pairs:
+            # 排序后拼接，确保顺序无关的去重
+            sorted_pairs = sorted(pairs)
+            return ",".join(f"{a}→{b}" for a, b in sorted_pairs)
+
+        # 没有匹配到标准模式，使用 suggestion 本身
+        return suggestion
+
+
 # 全局单例实例
 grammar_check_reviewer = GrammarCheckReviewer()

core/construction_review/component/reviewers/prompt/basic_reviewers.yaml

@@ -22,7 +22,7 @@ grammar_check:
     2. 如果出现了错字，大概率是因为拼音拼写错误导致的错字，请根据其相似读音推测正确的字，如果拿不准请不给出修改为xx的建议。
     3. 如”供气”错打为了”供器”，应当结合上下文推测”供qi”应该为是什么。
     4. 对于条款编号而言，'一）'这样的结构是正确的，符合中文规范
-    5. 所有建议都要基于审查内容给出忠实建议，禁止给出不符合逻辑的错误建议，如建议将A修改成A，例如：”将'确定的'改为'确定的'（原字为'确'，应为'确'）”，这种建议就是错误的
+    5. 如果你对某个词或表述犹豫不决、无法确定它是否真的是错误，说明原文可能是正确的，此时应直接输出”无明显问题”，而不是勉强凑出一个修改建议。
     6. 请对汉语中经典易错字如”辩”与”辨”等等的混用请多加注意。
 
     ## rule
@@ -42,7 +42,15 @@ grammar_check:
         4. 没有明显词句语法错误、标点错误的内容不予检查，输出无明显问题。
         5. 已检查出的问题项仅输出一次检查结果，禁止对同一内容重复检查。
         6. 统一解释：如果表格中出现了多列相同的表头标题，不是错误，而是解析时这几个是合并的表头。
-        7. 所有建议都要基于审查内容给出忠实建议，禁止给出不符合逻辑的错误建议，如建议将A修改成A，例如：将'若'改为'若'（应为'若'），这种建议就是错误的
+        7. suggestion 字段必须是简洁明确的最终结论，禁止在建议中包含推理过程或自我辩论。如果你对某个表述反复犹豫、自我反驳，说明原文没有问题，应输出”无明显问题”。
+        8. **输出前自检（强制执行）**：在输出 JSON 之前，逐条检查每个 issue：
+           - 如果某条的 reason 中出现了”可能””暂定””不确定””重新审视””然而””不过””似乎”等犹豫措辞，说明你不确定，**必须删除该条**
+           - 如果某条的 suggestion 中修改前后的文字相同或几乎相同，**必须删除该条**
+           - 如果某条的 suggestion 为”无明显问题”，**必须删除该条**（”无明显问题”不应作为 JSON issue 输出）
+           - 如果某条涉及技术操作规程的正确性（如操作步骤顺序、工艺参数选择），**必须删除该条**
+           - 如果多条 issue 指向同一个错误位置或同一个修改建议，**只保留一条，删除重复项**
+           - 如果删除后没有任何 issue 了，直接输出”无明显问题”
+        9. risk_level 字段必须填写（高风险/中风险/低风险），不得为空。如果你无法判定风险等级，说明该问题本身不确定，应删除该条。
 
     ## ⚠️ 严格禁止审查以下内容（与词句语法无关）
     **你只负责检查词句语法层面的问题**，以下类型的问题**一律跳过，不输出任何 issue**：
@@ -56,6 +64,7 @@ grammar_check:
     8. **业务逻辑问题**：如施工方案不合理、工艺流程错误
     9. **数据单位/格式**：如单位使用不当、数字格式问题
     10. **语义逻辑问题**：任何与语义、逻辑、事实相关的内容
+    11. **技术操作规程**：操作步骤的顺序是否正确、工艺参数是否合理、安全操作规范的技术正确性 —— 这些由专业技术审查流程处理，你只检查其中的文字书写错误（如错别字、漏字）
 
     **你的职责范围仅限于**：错别字（如”混泥土”→”混凝土”）、多字/少字、重复字词（如”公司公司”）、标点符号错误、”的地得”混用、明显的语法结构错误。
     超出以上范围的所有问题，请忽略并输出”无明显问题”。
@@ -87,10 +96,12 @@ grammar_check:
     - 数值计算错误、单位格式问题
     - 语义歧义、条件结论不匹配
     - 任何与逻辑、语义、事实相关的问题
+    - 技术操作规程的正确性（如操作步骤顺序、工艺参数选择）
 
     输出格式:务必须严格按照以下标准json格式输出审查结果
     如果未发现明显的词句语法错误，请输出:无明显问题。
-    **禁止**输出建议类似于"将'设'改为'设'（原字为'设'，应为'设'）"，没有问题就是没有问题，不可造假，谢谢。
+    **关键原则**：宁可漏报，不可误报。如果你无法确定某处是否有错误，必须输出"无明显问题"。suggestion 字段只写最终结论（如"将'混泥土'改为'混凝土'"），禁止在字段内展开分析或自我辩论。
+    **输出前自检**：输出 JSON 前逐条检查——如果某条 reason 中有犹豫措辞（"可能""暂定""不确定""重新审视"），或 suggestion 修改前后文字相同，必须删除该条。risk_level 必须填写，不得为空。
     如果发现问题，请按以下格式输出:
     location字段直接输出原字段内容，不得猜测。若有多个错误，请写到同一个以下json对象内，谢谢。
     ## 示例
@@ -150,8 +161,8 @@ semantic_logic_check:
     9. 禁止对所谓的”表述不恰当”、”表述过于严格”当做错误点，如你不能将”禁止夜间施工”改为”应在增加照明条件下允许夜间施工”这样的建议端上来
     10. **禁止审查过于专业的知识**，你只是审查通用的语义逻辑关系，而并非需要你根据你的知识去审查过多的部分如涉及到架桥参数、施工参数等等一些列的问题，这些问题有后续流程会处理，你暂且跳过
     11. **禁止对专业知识进行点评**，如技术参数、技术规范、技术条文，你对这方面知识还是较为落后的，你不需要对这方便进行涉猎
-    12. 所有建议都要基于审查内容给出忠实建议，禁止给出不符合逻辑的错误建议，如建议将A修改成A，例如：”将'确定的'改为'确定的'（原字为'确'，应为'确'）”，这种建议就是错误的
-    13. **禁止回复**如A修改为A这种假回答，文段没有问题必须直接输出：无明显问题
+    12. suggestion 字段必须是简洁明确的最终结论，禁止在建议中包含推理过程或自我辩论。如果你对某个表述反复犹豫、自我反驳，说明原文没有问题，应输出”无明显问题”。
+    13. 文段没有问题必须直接输出：无明显问题。宁可漏报，不可勉强凑出一个不确定的修改建议。
     14. **禁止审查词句语法问题**：错别字、多字、少字、重复字词、标点符号错误、”的地得”混用、语法结构等，这些问题由词句语法审查模块负责，不属于你的职责范围
     15. **禁止审查敏感词问题**：政治敏感、商业机密、表述适宜性等，这些问题由敏感词审查模块负责，不属于你的职责范围
 

core/construction_review/component/reviewers/prompt/query_extract.yaml

@@ -1,87 +1,130 @@
-query_extract:
+# 审查要点提取 — RAG 检索查询生成
+# 从施工方案文本中提取需要用规范条文验证的审查要点
+
+review_point_extract:
   system: |
     # 角色
     你是一个**交通路桥工程领域的施工方案审查专家**。
-    你的目标是：从文本中提取**最具检索区分度**的工程概念，用于构建高精度的数据库查询索引。
+    你的目标是：从施工方案文本中提取**需要用规范条文验证的审查要点**，用于高精度检索国标/行标/地标文档。
 
     # 任务
-    将非结构化的施工文本，拆解为标准化的【原子实体】、【扩展检索词】、【作业背景】与【技术参数】。
+    将非结构化的施工方案文本，拆解为若干个【审查要点】。每个审查要点代表一个需要用规范条文来验证合规性的技术主张。
 
     # 核心定义（严格执行）
-    1. **entity (原子实体)**: 
-       - 定义：文本中最核心的**具体工程对象**（特定的结构物、特定的机械、特定的工艺、特定的材料类型）。
-       - **🚀 关键约束（具象优先）**：
-         - ❌ **严禁提取通用泛化名词**：绝对不要提取 "材料"、"设备"、"机械"、"人员"、"措施"、"方案"、"工具"、"环境" 等抽象词汇。
-         - ✅ **必须提取具体下位词**：
-           - 遇到 "材料" -> 提取具体材料名（如 "扣件"、"钢丝绳"、"C35混凝土"）；若文中未提及具体材料名但提及了所属结构，则提取所属结构（如 "大型临时设施"）。
-           - 遇到 "设备" -> 提取具体机型（如 "汽车吊"、"架桥机"）。
-           - 遇到 "部位" -> 提取具体位置（如 "基坑"、"边坡"、"盖梁"）。
-       - **原子化约束**：保持词汇独立，不包含长修饰语（如提取 "钢筋笼" 而非 "桩基钢筋笼"）。
-    
-    2. **search_keywords (扩展检索词)**: 
-       - 定义：将 entity 翻译为 **国标规范/通用术语** 或 **紧密关联词**。
-       - 作用：消除口语与书面语的差异。
-       - 示例: entity="便桥" -> keywords=["钢便桥", "临时桥梁", "施工通道"]
-    
-    3. **background (作业背景)**: 
-       - 定义：描述实体的 **施工部位**、**地质环境**、**作业动作**、**工况条件** 或 **时间阶段**。
-       - 作用：提供语境，过滤无关文档。
-    
-    4. **parameter (参数)**: 
-       - 定义：具体的数值指标、型号规格、物理性能要求、验收频率或标准。
+
+    1. **label (审查要点标签)**:
+       - 定义：对该审查要点的简短命名（8字以内），概括核心技术主题。
+       - 示例："深基坑变形监测"、"架桥机轨道安装偏差"、"混凝土浇筑温度控制"
+
+    2. **search_queries (规范检索语句)**:
+       - 定义：以**规范条文的表述方式**构造的检索语句，模拟国标/行标中的条文标题或条款内容。
+       - **关键约束**：
+         - 使用规范术语："应"、"不得"、"允许偏差"、"检验频率"、"安全系数"、"限值"、"要求"
+         - 包含具体的技术对象和参数维度
+         - 不要使用口语化表述
+         - 不要猜测规范中不存在的术语
+       - 数量：每个审查要点 **2个** 检索语句（视复杂度可1-3个）
+       - 示例：
+         - "基坑变形监测精度等级及观测频率要求"
+         - "架桥机轨道安装允许偏差及检验方法"
+
+    3. **original_text (原文摘录)**:
+       - 定义：从源文本中**逐字摘录**触发该审查要点的原始语句或段落。
+       - **关键约束**：
+         - 必须是原文的精确摘录，不得改写、概括或推断
+         - 保留原始数值、条件、限定语
+         - 长度50-300字，涵盖完整的技术描述
+       - 作用：作为语义锚点，确保检索结果与施工文本的实际语境对齐
+
+    4. **parameter (技术参数)**:
+       - 定义：该审查要点涉及的具体数值指标、型号规格、频率要求或验收标准。
+       - 如果原文未提及具体参数，填写"未明确"。
+
+    # 提取策略
+    - 根据文本信息密度，提取 **1到5个** 审查要点
+    - 优先提取：高风险作业（深基坑、高空、起重吊装）、关键受力构件、强制性条文相关
+    - 忽略：无实质工程内容的管理性描述、客套话、纯叙述性文字
+    - 每个审查要点必须能独立对应到具体的规范条文
 
     # 限制
-    - 忽略无实质工程内容的客套话。
-    - 仅输出 JSON 字符串，且使用```json``` 包装。
-    - background 尽量提取完整，避免信息丢失。
-    - background 务必忠实与原文，不得胡编乱造，且提取完整
-    - 务必遵循<任务> 中的提取数量要求。
+    - 仅输出 JSON 字符串，使用```json``` 包装
+    - original_text 必须忠实于原文，严禁编造
+    - search_queries 必须使用规范条文的表述风格
+    - 忽略无实质工程内容的客套话
 
   user_template: |
     ## 任务
-    从文本中提取 **3个** 最关键的实体对象。视信息密度而定，优先关注**高风险工程部位**（如基坑、支架）和**核心受力构件**。
+    从以下施工方案文本中提取审查要点。根据信息密度提取 1-5 个要点，优先关注高风险工程和核心技术参数。
 
-    ## 示例 1 (避免泛化词提取)
-    文本: "大型临时设施选用的原材料、构件、扣件和其他重要受力的辅助材料进行质量验收。严禁使用不合格材料。"
+    ## 示例 1（多要点提取）
+    文本: "深度大于5m的基坑开挖期间，应设置变形监测点，监测精度不低于三等水准，监测频率不少于2次/天。基坑周边1.5倍开挖深度范围内的建（构）筑物应进行安全评估。"
     ### 分析过程:
-    - "原材料"、"材料" -> 太泛化，**丢弃**。
-    - "扣件" -> 具体材料，**保留**。
-    - "大型临时设施" -> 核心工程对象，**保留**。
+    - 文本包含两个独立的技术要求：变形监测 和 周边建筑安全评估
+    - 两者需要不同的规范条文来验证
     ### 输出:
     ```json
     [
       {{
-        "entity": "大型临时设施",
-        "search_keywords": ["临时结构物", "施工临建", "临时工程"],
-        "background": "原材料及构件质量验收阶段",
-        "parameter": "需进行试验验证，严禁使用不合格材料"
+        "label": "基坑变形监测",
+        "search_queries": [
+          "基坑变形监测精度等级及观测频率要求",
+          "深基坑监测点位布置及中误差限值"
+        ],
+        "original_text": "深度大于5m的基坑开挖期间，应设置变形监测点，监测精度不低于三等水准，监测频率不少于2次/天",
+        "parameter": "监测精度≥三等水准，频率≥2次/天"
       }},
       {{
-        "entity": "扣件",
-        "search_keywords": ["钢管脚手架扣件", "连接件", "紧固件"],
-        "background": "大型临时设施受力辅助材料",
-        "parameter": "需取样送检"
+        "label": "基坑周边安全评估",
+        "search_queries": [
+          "基坑开挖影响范围内建构筑物安全评估要求",
+          "基坑周边建筑物安全距离及保护措施"
+        ],
+        "original_text": "基坑周边1.5倍开挖深度范围内的建（构）筑物应进行安全评估",
+        "parameter": "影响范围=1.5倍开挖深度"
+      }}
+    ]
+    ```
+
+    ## 示例 2（单要点提取——信息密度低）
+    文本: "架桥机安装前，应对设备散件进行全面检查、清理，如发现有损伤、腐蚀或其它缺陷，应在安装前予以处理，合格后方可安装。"
+    ### 分析过程:
+    - 文本只描述了一个技术要求：安装前设备检查
+    - 信息密度较低，只需1个审查要点
+    ### 输出:
+    ```json
+    [
+      {{
+        "label": "架桥机安装前检查",
+        "search_queries": [
+          "架桥机安装前设备构件检查验收要求",
+          "起重设备安装前零部件质量检验标准"
+        ],
+        "original_text": "架桥机安装前，应对设备散件进行全面检查、清理，如发现有损伤、腐蚀或其它缺陷，应在安装前予以处理，合格后方可安装",
+        "parameter": "未明确"
       }}
     ]
     ```
 
-    ## 示例 2 (提取隐性核心实体)
-    文本: "深度大于3m的基坑开挖、有地下水侵扰的基坑清底封底，每个工作班至少巡查两遍。"
+    ## 示例 3（高密度文本——含多个独立技术参数）
+    文本: "轨道安装的允许偏差：轨道实际中心线对设计中心线位置偏移允许偏差为3毫米；轨距允许偏差为±5毫米；轨道纵向不平度应小于1/1500且全行程不超过10毫米；同一断面两轨道标高相对偏差不超过5毫米。"
     ### 分析过程:
-    - 核心对象是 "基坑"（特指深基坑）。
-    - 这是一个具体的工程部位，具有高检索价值。
+    - 文本围绕"轨道安装允许偏差"这一主题
+    - 虽然包含多个具体数值，但它们属于同一技术规范范畴
+    - 合并为一个审查要点，将所有参数整合
     ### 输出:
     ```json
     [
       {{
-        "entity": "基坑",
-        "search_keywords": ["深基坑", "沟槽", "土方开挖"],
-        "background": "深度>3m的开挖作业，或有地下水侵扰的清底封底阶段",
-        "parameter": "深度>3m，巡查频次≥2次/班"
+        "label": "架桥机轨道安装偏差",
+        "search_queries": [
+          "架桥机运梁轨道安装允许偏差及检验标准",
+          "起重机轨道中心线偏移轨距及平整度限值"
+        ],
+        "original_text": "轨道安装的允许偏差：轨道实际中心线对设计中心线位置偏移允许偏差为3毫米；轨距允许偏差为±5毫米；轨道纵向不平度应小于1/1500且全行程不超过10毫米；同一断面两轨道标高相对偏差不超过5毫米",
+        "parameter": "中心线偏移≤3mm，轨距偏差±5mm，纵向不平度<1/1500且≤10mm，标高偏差≤5mm"
       }}
     ]
     ```
 
     ## 待处理文本块
     {{review_content}}
-    

core/construction_review/component/reviewers/prompt/technical_reviewers.yaml

@@ -7,8 +7,17 @@ non_parameter_compliance_check:
 
   user_prompt_template: |
     ## 审查规则
-    1. **核心原则**：仅基于【参考依据】进行审查，若参考依据无法回答待审查问题或相关性过低，必须输出"无明显问题"
-    2. **审查重点**：检查安全相关非参数性条文、安全防护措施、安全风险评估、安全管理要求
+    1. **审查原则**：
+       你的任务是参照【参考依据】中的技术要求，审查【待审查问题】中的施工方案内容是否符合这些要求。
+       - 即使【参考依据】来自相近领域（如公路/铁路/桥梁/市政等基础设施建设标准），其中的安全技术要求、管理措施、操作规程等仍具有参考价值，应重点审查其中与【待审查问题】相关的条款
+       - 从【参考依据】中提取与【待审查问题】直接相关的技术要求，逐条比对施工方案是否满足这些要求
+       - 只有当【参考依据】为空、内容过短（少于2句话）、或确实完全不涉及施工方案安全/质量/管理相关内容时，才输出"无明显问题"
+    2. **审查重点**：
+       - 安全防护措施是否完整、是否符合参考依据中的安全规定
+       - 安全管理要求是否到位、是否遗漏关键的安全程序
+       - 安全风险评估是否充分
+       - 施工方案中的操作流程是否符合参考依据中的操作规程
+       - 是否存在参考依据中明确禁止或限制的做法
     3. **禁止事项**：
        - 禁止编造审查依据或引用【参考依据】以外的信息
        - 禁止对表格制表符进行检查
@@ -25,18 +34,19 @@ non_parameter_compliance_check:
     ## 待审查问题
     {review_content}
 
-    ## 输出结构
-    - 如果【参考依据】无法回答【待审查问题】或相关性过低，输出：无明显问题
-    - 如果发现问题，严格按以下JSON格式输出（location字段直接输出原文，不得猜测）：
+    ## 输出要求
+    请仔细从【参考依据】中找出与【待审查问题】相关的技术要求，对比施工方案内容是否满足。
+    - 如果发现不符合或遗漏，严格按以下JSON格式输出（location字段直接输出原文，不得猜测）：
     ```json
     {{
       "issue_point": "问题标题描述",
-      "location": "当前问题对应的原始条款内容及位置,如六、验收标准 (页码: 85)，以及其语境上下文",
+      "location": "当前问题对应的原始施工方案内容及位置，以及参考依据中对应的条款",
       "suggestion": "具体的修改建议内容",
-      "reason": "问题的原因分析和依据说明",
+      "reason": "问题的原因分析和依据说明，引用参考依据中的具体条款",
       "risk_level": "高风险/中风险/低风险"
     }}
     ```
+    - 如果审查后确实没有发现问题，输出：无明显问题
 
 
 # 参数合规性检查功能 - 审查 实体概念/工程术语相关知识库
@@ -46,14 +56,22 @@ parameter_compliance_check:
 
   user_prompt_template: |
     ## 审查规则
-    1. **核心原则**：仅基于【参考依据】进行审查，若参考依据无法回答待审查问题或相关性过低，必须输出"无明显问题"
-    2. **审查重点**：检查技术参数的准确性和合理性、实体概念和工程术语的参数正确性、设计值与标准的符合性
+    1. **审查原则**：
+       你的任务是参照【参考依据】中的技术参数和标准，审查【待审查问题】中的施工方案内容是否准确、合理。
+       - 即使【参考依据】来自相近领域（如公路/铁路/桥梁/市政等基础设施建设标准），其中的技术参数、设计值、验收标准等仍具有参考价值
+       - 重点检查：施工方案中的技术参数是否与参考依据一致，是否存在参数错误、单位错误、数值偏差等问题
+       - 只有当【参考依据】为空、内容过短（少于2句话）、或确实完全不包含任何可比较的技术参数时，才输出"无明显问题"
+    2. **审查重点**：
+       - 技术参数的准确性和合理性（数值、单位、范围）
+       - 实体概念和工程术语的使用是否正确
+       - 设计值与参考标准的符合性
+       - 施工方案中引用的标准参数是否过时或错误
+       - 计算公式和计算结果是否正确
     3. **禁止事项**：
        - 禁止编造审查依据或引用【参考依据】以外的信息
        - 禁止对表格制表符进行检查
        - 禁止曲解术语概念
        - 禁止对同一内容重复检查
-       - 禁止检查无明显参数问题的内容
     4. **风险分级**：
        - 高风险：影响审查结论、可能导致法律问题或严重安全隐患
        - 中风险：影响专业表达、可能导致理解偏差或一般性问题
@@ -65,15 +83,16 @@ parameter_compliance_check:
     ## 待审查问题
     {review_content}
 
-    ## 输出结构
-    - 如果【参考依据】无法回答【待审查问题】或相关性过低，输出：无明显问题
-    - 如果发现问题，严格按以下JSON格式输出（location字段直接输出原文，不得猜测）：
+    ## 输出要求
+    请仔细从【参考依据】中找出与【待审查问题】相关的技术参数和标准，对比施工方案内容是否准确。
+    - 如果发现参数错误或不一致，严格按以下JSON格式输出（location字段直接输出原文，不得猜测）：
     ```json
     {{
       "issue_point": "问题标题描述",
-      "location": "当前问题对应的原始条款内容及位置,如六、验收标准 (页码: 85)，以及其语境上下文",
-      "suggestion": "具体的修改建议内容",
-      "reason": "问题的原因分析和依据说明",
+      "location": "当前问题对应的原始施工方案内容及位置，以及参考依据中对应的参数标准",
+      "suggestion": "具体的修改建议内容，给出正确参数",
+      "reason": "问题的原因分析和依据说明，引用参考依据中的具体参数",
       "risk_level": "高风险/中风险/低风险"
     }}
-    ```
+    ```
+    - 如果审查后确实没有发现参数问题，输出：无明显问题

core/construction_review/component/reviewers/sensitive_words/色情词库.txt

@@ -25,14 +25,12 @@
 操我
 厕奴
 插比
-插b
 插逼
 插你
 插我
 插阴
 潮吹
 潮喷
-成人dv
 成人电影
 成人论坛
 成人小说
@@ -55,9 +53,6 @@
 成人游戏
 吃精
 赤裸
-抽插
-扌由插
-抽一插
 春药
 大波
 大力抽送
@@ -162,7 +157,6 @@
 屁眼
 品香堂
 前凸后翘
-强jian
 强奸处女
 情趣用品
 情色
@@ -390,9 +384,6 @@ cao你
 全家死绝
 白痴
 无耻
-sb
-杀b
-你吗b
 你妈的
 婊子
 贱货
@@ -416,7 +407,6 @@ sb
 屌
 屁股
 下体
-a片sss
 内裤
 咪咪
 白嫩
@@ -424,7 +414,6 @@ a片sss
 兽性
 风骚
 呻吟
-sm
 阉割
 一丝不挂
 脱光
@@ -595,7 +584,6 @@ sm
 肛门
 龟头
 裹本
-国产av
 好嫩
 豪乳
 黑逼
@@ -764,7 +752,6 @@ sm
 小逼
 校鸡
 小穴
-小xue
 写真
 性感妖娆
 性感诱惑
@@ -848,21 +835,12 @@ sm
 自慰
 作爱
 18禁
-99bb
-a4u
-a4y
-adult
-amateur
-anal
 a片
-fuck
 gay片
 g点
 g片
-hardcore
 h动画
 h动漫
-
 sm女王
 xing伴侣
 yin荡

core/construction_review/component/reviewers/utils/inter_tool.py

@@ -503,6 +503,13 @@ class InterTool:
                 logger.warning(f"检查项 {check_key} 格式不符合要求，缺少details或response字段")
                 logger.warning(f"check_result内容: {check_result}")
 
+        # 过滤掉 exist_issue=false 的审查项（无问题的项不返回到最终结果）
+        # 同时过滤掉 check_result 为字符串的项（解析失败的 malformed JSON）
+        review_lists = [
+            item for item in review_lists
+            if item.get("exist_issue", False) and not isinstance(item.get("check_result"), str)
+        ]
+
         # 统计风险等级
         for issue in review_lists:
             risk_level = issue.get("risk_info", {}).get("risk_level", "low")
@@ -578,19 +585,24 @@ class InterTool:
 
             # 3. 如果JSON解析失败，回退到文本解析
             if not review_lists:
-                # 🔧 修复：检查响应是否为空或只包含空白字符
                 response_stripped = response.strip() if isinstance(response, str) else ""
                 is_empty_response = not response_stripped or response_stripped in ["", "null", "None", "undefined"]
 
+                # 检测是否为格式错误的 JSON（包含 JSON 标记但解析失败）
+                # 这类情况不应作为有效 issue 输出
+                looks_like_broken_json = any(marker in response_stripped for marker in ['```json', '{', '[', '"issue_point"'])
+
                 risk_level = self._determine_risk_level(response)
 
-                # 如果响应为空，则设置 exist_issue=False
+                # 空响应或格式错误的 JSON → exist_issue=False（不输出到最终结果）
+                should_exist = not is_empty_response and not looks_like_broken_json
+
                 review_lists.append({
                     "check_item": check_name,
                     "chapter_code": chapter_code,
                     "check_item_code": check_item_code,
                     "check_result": response,
-                    "exist_issue": not is_empty_response,  # 🔧 修复：空响应不存在问题
+                    "exist_issue": should_exist,
                     "risk_info": {"risk_level": risk_level}
                 })
 
@@ -601,7 +613,7 @@ class InterTool:
                 "chapter_code": chapter_code,
                 "check_item_code": check_item_code,
                 "check_result": response,
-                "exist_issue": True,
+                "exist_issue": False,
                 "risk_info": {"risk_level": "low"}
             })
 
@@ -699,6 +711,18 @@ class InterTool:
         # 只有当内容不为空，且风险等级不是"无风险"类时，才认为存在问题
         exist_issue = not is_empty and original_risk_level not in ["无风险", "无", "通过", "符合要求"]
 
+        # 额外验证：dict 类型的 issue_data 必须至少包含一个有意义字段
+        # 防止空壳 dict（如 {"risk_level": "中风险"}）通过过滤
+        if exist_issue and isinstance(issue_data, dict):
+            meaningful_fields = ["issue_point", "location", "suggestion"]
+            has_meaningful_content = any(
+                bool(issue_data.get(field, "").strip() if isinstance(issue_data.get(field), str) else issue_data.get(field))
+                for field in meaningful_fields
+            )
+            if not has_meaningful_content:
+                exist_issue = False
+                logger.debug(f"检查项 {check_name} 的 issue_data 缺少有意义字段 (issue_point/location/suggestion)，设置 exist_issue=False")
+
         # 记录调试信息
         if is_empty:
             logger.debug(f"检查项 {check_name} 的 issue_data 为空，设置 exist_issue=False")

core/construction_review/component/reviewers/utils/prompt_loader.py

@@ -74,7 +74,7 @@ class PromptLoader:
                 config_file = os.path.join(self.prompt_config_dir, "ai_suggestion.yaml")
             elif reviewer_type == 'outline':
                 config_file = os.path.join(self.prompt_config_dir, "catalog_reviewers.yaml")
-            elif reviewer_type == 'query_extract':
+            elif reviewer_type in ('query_extract', 'review_point_extract'):
                 config_file = os.path.join(self.prompt_config_dir, "query_extract.yaml")
             else:
                 config_file = os.path.join(self.prompt_config_dir, f"{reviewer_type}_reviewers.yaml")
@@ -112,7 +112,7 @@ class PromptLoader:
                 config_file = os.path.join(self.prompt_config_dir, "ai_suggestion.yaml")
             elif reviewer_type == 'outline':
                 config_file = os.path.join(self.prompt_config_dir, "catalog_reviewers.yaml")
-            elif reviewer_type == 'query_extract':
+            elif reviewer_type in ('query_extract', 'review_point_extract'):
                 config_file = os.path.join(self.prompt_config_dir, "query_extract.yaml")
             else:
                 config_file = os.path.join(self.prompt_config_dir, f"{reviewer_type}_reviewers.yaml")
@@ -128,7 +128,7 @@ class PromptLoader:
             prompt_config = config[prompt_name]
 
             # 处理 query_extract.yaml 的特殊格式
-            if reviewer_type == 'query_extract':
+            if reviewer_type in ('query_extract', 'review_point_extract'):
                 if 'system' in prompt_config and 'user_template' in prompt_config:
                     # 转换双花括号为单花括号，以便LangChain能够识别变量
                     user_template = prompt_config['user_template'].replace('{{review_content}}', '{review_content}')
@@ -137,7 +137,7 @@ class PromptLoader:
                         'user_prompt_template': user_template
                     }
                 else:
-                    raise ValueError(f"query_extract 配置缺少必要字段: {prompt_name}")
+                    raise ValueError(f"{reviewer_type} 配置缺少必要字段: {prompt_name}")
             else:
                 # 验证必要的字段
                 if 'system_prompt' not in prompt_config or 'user_prompt_template' not in prompt_config:
@@ -172,9 +172,11 @@ class PromptLoader:
         Returns:
             ChatPromptTemplate: LangChain ChatPromptTemplate实例
         """
-        # 特殊处理：对于query_extract类型，只有一个键
+        # 特殊处理：对于query_extract/review_point_extract类型，使用对应的prompt_name作为cache key
         if reviewer_type == "query_extract":
             cache_key = "query_extract_query_extract"
+        elif reviewer_type == "review_point_extract":
+            cache_key = "review_point_extract_review_point_extract"
         elif prompt_name is None:
             cache_key = f"{reviewer_type}"
         else:

core/construction_review/workflows/ai_review_workflow.py

@@ -407,6 +407,23 @@ class AIReviewWorkflow:
                 completed_chunks += chunks_completed
 
             # ===== Phase 5: 结果汇总 =====
+            # 最终兜底过滤：移除 exist_issue=false 的审查项，确保无问题项不进入最终结果
+            # 同时过滤 check_result 为字符串的项（解析失败/异常兜底）
+            filtered_issues = []
+            for issue_wrapper in all_issues:
+                for issue_id, issue_detail in issue_wrapper.items():
+                    review_lists = issue_detail.get('review_lists', [])
+                    filtered_lists = [
+                        item for item in review_lists
+                        if item.get("exist_issue", False) and not isinstance(item.get("check_result"), str)
+                    ]
+                    if filtered_lists:
+                        issue_detail['review_lists'] = filtered_lists
+                        filtered_issues.append(issue_wrapper)
+            if len(filtered_issues) < len(all_issues):
+                logger.info(f"最终过滤: {len(all_issues)} → {len(filtered_issues)} 个issues (移除了 {len(all_issues) - len(filtered_issues)} 个无问题项)")
+            all_issues = filtered_issues
+
             summary = self.inter_tool._aggregate_results(all_issues)
 
             review_results = {

core/construction_review/workflows/core_functions/ai_review_core_fun.py

@@ -617,17 +617,46 @@ class AIReviewCoreFun:
 
         logger.info(f"[{func_name}] 开始处理 {len(entity_results)} 个查询对的审查")
 
-        # 为每个查询对创建审查任务
+        # 质量门控常量
+        MIN_REFERENCE_LENGTH = 50
+        MIN_FINAL_SCORE = 0.5
+
+        # 为每个查询对创建审查任务（先过滤不合格的）
         review_tasks = []
+        skipped_entities = []
+
         for idx, entity_item in enumerate(entity_results):
             combined_query = entity_item.get('combined_query', '')
             entity = entity_item.get('entity', f'entity_{idx}')
             text_content = entity_item.get('text_content', '')
             file_name = entity_item.get('file_name', '')
+            final_score = entity_item.get('final_score', 0)
+
+            # ===== 质量门控 =====
+            if not text_content or len(text_content.strip()) < MIN_REFERENCE_LENGTH:
+                skip_reason = f"text_content过短(len={len(text_content.strip())})"
+                logger.warning(f"[{func_name}] 跳过查询对 {idx} ({entity}): {skip_reason}")
+                skipped_entities.append({'entity': entity, 'index': idx, 'reason': skip_reason})
+                continue
+
+            if not file_name or not file_name.strip():
+                skip_reason = "file_name为空"
+                logger.warning(f"[{func_name}] 跳过查询对 {idx} ({entity}): {skip_reason}")
+                skipped_entities.append({'entity': entity, 'index': idx, 'reason': skip_reason})
+                continue
+
+            if final_score < MIN_FINAL_SCORE:
+                skip_reason = f"final_score={final_score:.4f}<{MIN_FINAL_SCORE}"
+                logger.warning(f"[{func_name}] 跳过查询对 {idx} ({entity}): {skip_reason}")
+                skipped_entities.append({'entity': entity, 'index': idx, 'reason': skip_reason})
+                continue
 
-            logger.info(f"[{func_name}] 为查询对 {idx} ({entity}) 创建审查任务")
+            # ===== 通过门控，创建审查任务 =====
+            logger.info(
+                f"[{func_name}] 为查询对 {idx} ({entity}) 创建审查任务 "
+                f"(score={final_score:.4f}, ref_len={len(text_content)})"
+            )
 
-            # 创建审查任务
             task = asyncio.create_task(
                 method(
                     trace_id_idx=f"{trace_id}_entity_{idx}",
@@ -641,6 +670,25 @@ class AIReviewCoreFun:
             )
             review_tasks.append((idx, entity, task))
 
+        # 记录跳过统计
+        if skipped_entities:
+            logger.info(
+                f"[{func_name}] 质量门控: 跳过 {len(skipped_entities)}/{len(entity_results)} 个实体"
+            )
+
+        # 所有实体都被跳过 → 返回空结果（不触发LLM）
+        if not review_tasks:
+            logger.warning(
+                f"[{func_name}] 所有 {len(entity_results)} 个实体均未通过质量门控, 跳过审查"
+            )
+            return {
+                'review_mode': 'entity_based',
+                'func_name': func_name,
+                'total_entities': len(entity_results),
+                'skipped_entities': skipped_entities,
+                'entity_review_results': []
+            }
+
         # 并发执行所有查询对的审查任务
         results = []
         if review_tasks:
@@ -678,6 +726,7 @@ class AIReviewCoreFun:
             'review_mode': 'entity_based',
             'func_name': func_name,
             'total_entities': len(entity_results),
+            'skipped_entities': skipped_entities,
             'entity_review_results': results
         }
 

foundation/ai/models/rerank_model.py

@@ -8,11 +8,15 @@
 支持的重排序模型:
 - BGE Reranker (本地部署)
 - Qwen3-Reranker-8B (本地部署)
+- Qwen3-Reranker-8B (蜀天算力)
 - Qwen3-Reranker-8B (硅基流动API)
+
+配置加载策略: 懒加载（首次调用时从 config.ini 读取该后端的凭证并缓存）
+路由决策: 由 retrieval.py 通过 model_setting.yaml 的 rerank 功能决定使用哪个后端
 """
 import json
 import requests
-from typing import List, Dict, Any
+from typing import List, Dict, Any, Optional
 from foundation.infrastructure.config.config import config_handler
 from foundation.observability.logger.loggering import review_logger as server_logger
 
@@ -20,67 +24,91 @@ from foundation.observability.logger.loggering import review_logger as server_lo
 class LqReranker:
     """
     重排序执行器
+
+    各后端配置按需加载：首次调用某后端时才从 config.ini 读取其凭证，
+    避免初始化时加载所有 4 个后端的配置。
     """
 
     def __init__(self):
-        # BGE Reranker 配置
-        self.bge_api_url = config_handler.get('bge_rerank_model', 'BGE_RERANKER_SERVER_URL')
-        self.bge_model = config_handler.get('bge_rerank_model', 'BGE_RERANKER_MODEL')
-        self.bge_top_k = int(config_handler.get('bge_rerank_model', 'BGE_RERANKER_TOP_N', 10))
-
-        # 本地Qwen3-Reranker-8B配置
-        self.lq_rerank_api_url = config_handler.get('lq_rerank_model', 'LQ_RERANKER_SERVER_URL')
-        self.lq_rerank_model = config_handler.get('lq_rerank_model', 'LQ_RERANKER_MODEL')
-        self.lq_rerank_top_k = int(config_handler.get('lq_rerank_model', 'LQ_RERANKER_TOP_N', 10))
-
-        # SHUTIAN Qwen3-Reranker-8B配置（蜀天云算力 25426端口）
-        self.shutian_rerank_api_url = config_handler.get('shutian', 'SHUTIAN_RERANK_SERVER_URL')
-        self.shutian_rerank_model = config_handler.get('shutian', 'SHUTIAN_RERANK_MODEL_ID')
-        self.shutian_rerank_api_key = config_handler.get('shutian', 'SHUTIAN_RERANK_API_KEY')
-
-        # 硅基流动Qwen3-Reranker-8B配置
-        self.silicoflow_rerank_api_url = config_handler.get('silicoflow_rerank_model', 'SILICOFLOW_RERANKER_API_URL', 'https://api.siliconflow.cn/v1/rerank')
-        self.silicoflow_rerank_api_key = config_handler.get('silicoflow_rerank_model', 'SILICOFLOW_RERANKER_API_KEY')
-        self.silicoflow_rerank_model = config_handler.get('silicoflow_rerank_model', 'SILICOFLOW_RERANKER_MODEL', 'Qwen/Qwen3-Reranker-8B')
-
-    def bge_rerank(self,query: str, candidates: List[str],top_k :int = None) -> List[Dict[str, Any]]:
+        # 各后端配置缓存（首次调用时加载）
+        self._bge_config: Optional[Dict[str, Any]] = None
+        self._lq_config: Optional[Dict[str, Any]] = None
+        self._shutian_config: Optional[Dict[str, Any]] = None
+        self._silicoflow_config: Optional[Dict[str, Any]] = None
+
+    def _get_bge_config(self) -> Dict[str, Any]:
+        """懒加载 BGE Reranker 配置"""
+        if self._bge_config is None:
+            self._bge_config = {
+                'api_url': config_handler.get('bge_rerank_model', 'BGE_RERANKER_SERVER_URL'),
+                'model': config_handler.get('bge_rerank_model', 'BGE_RERANKER_MODEL'),
+                'top_k': int(config_handler.get('bge_rerank_model', 'BGE_RERANKER_TOP_N', 10)),
+            }
+        return self._bge_config
+
+    def _get_lq_config(self) -> Dict[str, Any]:
+        """懒加载本地 Qwen3-Reranker 配置"""
+        if self._lq_config is None:
+            self._lq_config = {
+                'api_url': config_handler.get('lq_rerank_model', 'LQ_RERANKER_SERVER_URL'),
+                'model': config_handler.get('lq_rerank_model', 'LQ_RERANKER_MODEL'),
+                'top_k': int(config_handler.get('lq_rerank_model', 'LQ_RERANKER_TOP_N', 10)),
+            }
+        return self._lq_config
+
+    def _get_shutian_config(self) -> Dict[str, Any]:
+        """懒加载蜀天 Qwen3-Reranker 配置"""
+        if self._shutian_config is None:
+            self._shutian_config = {
+                'api_url': config_handler.get('shutian', 'SHUTIAN_RERANK_SERVER_URL'),
+                'model': config_handler.get('shutian', 'SHUTIAN_RERANK_MODEL_ID'),
+                'api_key': config_handler.get('shutian', 'SHUTIAN_RERANK_API_KEY'),
+            }
+        return self._shutian_config
+
+    def _get_silicoflow_config(self) -> Dict[str, Any]:
+        """懒加载硅基流动 Qwen3-Reranker 配置"""
+        if self._silicoflow_config is None:
+            self._silicoflow_config = {
+                'api_url': config_handler.get('silicoflow_rerank_model', 'SILICOFLOW_RERANKER_API_URL',
+                                              'https://api.siliconflow.cn/v1/rerank'),
+                'api_key': config_handler.get('silicoflow_rerank_model', 'SILICOFLOW_RERANKER_API_KEY'),
+                'model': config_handler.get('silicoflow_rerank_model', 'SILICOFLOW_RERANKER_MODEL',
+                                            'Qwen/Qwen3-Reranker-8B'),
+            }
+        return self._silicoflow_config
+
+    def bge_rerank(self, query: str, candidates: List[str], top_k: int = None) -> List[Dict[str, Any]]:
         """
-        执行重排序的全局函数
+        使用本地 BGE-reranker-v2-m3 进行重排序
 
         Args:
             query: 查询文本
             candidates: 候选文档列表
-            top_k: 调用时chaurnum参数，默认为None
-
+            top_k: 返回前k个结果，默认使用配置文件的top_k
 
         Returns:
             List[Dict]: 重排序后的结果列表
         """
         try:
-            # self.top_k 是config.ini生产环境中实际使用的重排序数量，bge_rerank中的top_k，用于开发环境中快速效果调试
-            if not top_k:# 如果开发top_k未指定，则使用配置文件中的top_k
-                top_k = self.bge_top_k
-            
+            cfg = self._get_bge_config()
+            if not top_k:
+                top_k = cfg['top_k']
 
             server_logger.info(f"开始执行重排序，查询: '{query}', 候选文档数量: {len(candidates)}")
 
-            # 构建重排序请求
             rerank_request = {
-                "model": self.bge_model,
+                "model": cfg['model'],
                 "query": query,
-                "candidates": candidates
+                "documents": candidates
             }
 
-            # 直接调用重排序API
-            url = self.bge_api_url
-            headers = {
-                "Content-Type": "application/json"
-            }
+            headers = {"Content-Type": "application/json"}
 
-            server_logger.debug(f"调用重排序API: {url}")
+            server_logger.debug(f"调用重排序API: {cfg['api_url']}")
             server_logger.debug(f"请求数据: {json.dumps(rerank_request, ensure_ascii=False)}")
 
-            response = requests.post(url, headers=headers, json=rerank_request, timeout=30)
+            response = requests.post(cfg['api_url'], headers=headers, json=rerank_request, timeout=30)
 
             if response.status_code == 200:
                 result = response.json()
@@ -97,7 +125,6 @@ class LqReranker:
 
         except Exception as e:
             server_logger.error(f"执行重排序失败: {str(e)}")
-            # 返回原始顺序作为fallback
             return [{"text": doc, "score": "0.0"} for doc in candidates[:top_k]]
 
     def lq_rerank(self, query: str, candidates: List[str], top_k: int = None) -> List[Dict[str, Any]]:
@@ -111,28 +138,19 @@ class LqReranker:
 
         Returns:
             List[Dict[str, Any]]: 重排序后的结果列表
-                [
-                    {
-                        "text": str,        # 文档文本内容
-                        "score": float,      # 相关性得分
-                        "index": int         # 原始索引
-                    },
-                    ...
-                ]
         """
         try:
+            cfg = self._get_lq_config()
             if not top_k:
-                top_k = self.lq_rerank_top_k
+                top_k = cfg['top_k']
 
-            # 检查query是否为空
             if not query or not query.strip():
                 server_logger.warning(f"本地Qwen3重排序跳过：query为空")
                 return [{"text": doc, "score": 0.0} for doc in candidates[:top_k]]
 
             server_logger.info(f"开始执行本地Qwen3重排序，查询: '{query}', 候选文档数量: {len(candidates)}")
 
-            # 定义变量（与测试脚本完全一致）
-            url = self.lq_rerank_api_url
+            url = cfg['api_url']
             prefix = '<|im_start|>system\nJudge whether the Document meets the requirements based on the Query and the Instruct provided. Note that the answer can only be "yes" or "no".<|im_end|>\n<|im_start|>user\n'
             suffix = "<|im_end|>\n<|im_start|>assistant\n<think>\n\n</think>\n\n"
 
@@ -146,31 +164,24 @@ class LqReranker:
             documents = [document_template.format(doc=doc, suffix=suffix) for doc in candidates]
 
             data = {
-                "model": self.lq_rerank_model,
+                "model": cfg['model'],
                 "query": query,
                 "documents": documents
             }
 
             headers = {"Content-Type": "application/json"}
 
-
             response = requests.post(url, headers=headers, json=data, timeout=30)
 
             if response.status_code == 200:
                 result = response.json()
-  
 
                 if "results" in result:
-                    # 格式化结果：将嵌套的 document.text 提取到外层，并清理模板标记
                     formatted_results = []
                     for item in result["results"]:
-                        # 获取包含模板的原始文本
                         raw_text = item.get("document", {}).get("text", "")
 
-                        # 清理模板标记：去除 <Document>: 和 <|im_end|>...assistant 之后的内容
-                        # 文本格式: <Document>: 原始内容<|im_end|>\n<|im_start|>assistant\n...
                         if "<Document>:" in raw_text:
-                            # 提取 <Document>: 和 <|im_end|> 之间的内容
                             start = raw_text.find("<Document>:") + len("<Document>:")
                             end = raw_text.find("<|im_end|>")
                             if end > start:
@@ -179,7 +190,7 @@ class LqReranker:
                                 cleaned_text = raw_text[start:].strip()
                         else:
                             cleaned_text = raw_text
-                    
+
                         formatted_results.append({
                             "text": cleaned_text,
                             "score": float(item.get("relevance_score", 0.0)),
@@ -205,8 +216,9 @@ class LqReranker:
         接口为标准 OpenAI 兼容 rerank API，无需模板包装，直接传原始 query/documents
         """
         try:
+            cfg = self._get_shutian_config()
             if not top_k:
-                top_k = self.lq_rerank_top_k
+                top_k = self._get_lq_config()['top_k']
 
             if not query or not query.strip():
                 server_logger.warning("SHUTIAN重排序跳过：query为空")
@@ -215,7 +227,7 @@ class LqReranker:
             server_logger.info(f"开始执行SHUTIAN Qwen3重排序，查询: '{query}', 候选文档数量: {len(candidates)}")
 
             data = {
-                "model": self.shutian_rerank_model,
+                "model": cfg['model'],
                 "query": query,
                 "documents": candidates,
                 "top_n": top_k
@@ -223,22 +235,20 @@ class LqReranker:
 
             headers = {
                 "Content-Type": "application/json",
-                "Authorization": f"Bearer {self.shutian_rerank_api_key}"
+                "Authorization": f"Bearer {cfg['api_key']}"
             }
 
-            response = requests.post(self.shutian_rerank_api_url, headers=headers, json=data, timeout=30)
+            response = requests.post(cfg['api_url'], headers=headers, json=data, timeout=30)
 
             if response.status_code == 200:
                 result = response.json()
 
-                # SHUTIAN API直接返回列表: [{"score": x, "document": "文本", "index": 0}, ...]
                 results_list = result.get("results", result) if isinstance(result, dict) else result
 
                 if isinstance(results_list, list) and results_list:
                     formatted_results = []
                     for item in results_list:
                         doc = item.get("document", "")
-                        # document 可能是字符串或 {"text": "..."} 对象
                         text = doc if isinstance(doc, str) else doc.get("text", "")
                         formatted_results.append({
                             "text": text,
@@ -266,45 +276,42 @@ class LqReranker:
         Args:
             query: 查询文本
             documents: 文档列表
-            top_k: 返回前k个结果，默认使用配置文件的top_k
+            top_k: 返回前k个结果，默认10
             instruction: 重排序指令
 
         Returns:
             List[Dict]: 重排序后的结果列表，包含 text 和 score
         """
         try:
+            cfg = self._get_silicoflow_config()
             if not top_k:
-                top_k = 10  # 默认值
+                top_k = 10
 
-            if not self.silicoflow_rerank_api_key:
+            if not cfg['api_key']:
                 server_logger.error("硅基流动 Reranker API Key 未配置")
                 return []
 
             server_logger.info(f"开始执行硅基流动Qwen3重排序，查询: '{query}', 文档数量: {len(documents)}")
 
-            
-            # 构建请求数据
             request_data = {
-                "model": self.silicoflow_rerank_model,
+                "model": cfg['model'],
                 "query": query,
                 "documents": documents,
                 "instruction": instruction,
                 "top_n": top_k,
                 "return_documents": True,
-                # "max_chunks_per_doc": 123,
-                # "overlap_tokens": 79
             }
 
             headers = {
-                "Authorization": f"Bearer {self.silicoflow_rerank_api_key}",
+                "Authorization": f"Bearer {cfg['api_key']}",
                 "Content-Type": "application/json"
             }
 
-            server_logger.debug(f"调用硅基流动Qwen3 Reranker API: {self.silicoflow_rerank_api_url}")
+            server_logger.debug(f"调用硅基流动Qwen3 Reranker API: {cfg['api_url']}")
             server_logger.debug(f"请求数据: {json.dumps(request_data, ensure_ascii=False)}")
 
             response = requests.post(
-                self.silicoflow_rerank_api_url,
+                cfg['api_url'],
                 headers=headers,
                 json=request_data,
                 timeout=30
@@ -315,7 +322,6 @@ class LqReranker:
                 server_logger.debug(f"硅基流动Qwen3 API响应: {json.dumps(result, ensure_ascii=False)}")
 
                 if "results" in result:
-                    # 格式化结果为统一格式
                     formatted_results = []
                     for item in result["results"]:
                         formatted_results.append({
@@ -334,9 +340,7 @@ class LqReranker:
 
         except Exception as e:
             server_logger.error(f"执行硅基流动Qwen3重排序失败: {str(e)}")
-            # 返回原始顺序作为fallback
             return [{"text": doc, "score": 0.0} for doc in documents[:top_k]]
 
-rerank_model = LqReranker()
-
 
+rerank_model = LqReranker()

foundation/ai/rag/retrieval/entities_enhance.py

@@ -1,25 +1,50 @@
-import json
 import asyncio
 from foundation.observability.monitoring.time_statistics import track_execution_time
 from foundation.ai.rag.retrieval.retrieval import retrieval_manager
 from foundation.observability.logger.loggering import review_logger as server_logger
+from foundation.infrastructure.config.config import config_handler
 
 
+class ReviewPointRetriever():
+    """
+    审查要点检索器 — 直接从施工方案文本搜索规范条文
 
-class EntitiesEnhance():
+    替代旧的 EntitiesEnhance（实体增强检索），新流程：
+    1. 对每个 search_query → hybrid_search(CHILDREN_COLLECTION) → 一次重排序(rerank)
+    2. 合并所有 search_query 的候选结果并去重
+    3. 用 original_text 做二次重排序（语义锚点对齐）
+    4. 返回 top-K 结果
+
+    核心改进：
+    - 跳过 ENTITY_COLLECTION 中间跳，直接搜索规范条文集合
+    - 用原文摘录（而非 LLM 概括的 background）做二次重排序，语义更精确
+    """
 
     def __init__(self):
-        self.bfp_result_lists = []
-        self._entity_recall_cache = {}  # 实体检索结果缓存
-        self._bfp_recall_cache = {}     # BFP召回结果缓存
+        self.result_lists = []
+        self._search_cache = {}  # 检索结果缓存
+
+    def _get_cache_key(self, query: str) -> str:
+        return f"search::{query[:100]}"
 
-    def _get_cache_key(self, entity: str, search_keywords: list, background: str = "") -> str:
-        """生成缓存键"""
-        keywords_str = "|".join(sorted(search_keywords)) if search_keywords else ""
-        return f"{entity}::{keywords_str}::{background[:50]}"
+    def _get_children_collection(self) -> str:
+        return config_handler.get('rag_collections', 'CHILDREN_COLLECTION', 'rag_children_hybrid')
 
     @track_execution_time
-    def entities_enhance_retrieval(self, query_pairs):
+    def review_point_retrieval(self, review_points):
+        """
+        审查要点检索 — 替代 entities_enhance_retrieval
+
+        Args:
+            review_points: 审查要点列表, 每个要点包含:
+                - label: 标签
+                - search_queries: 规范检索语句列表
+                - original_text: 原文摘录
+                - parameter: 技术参数
+
+        Returns:
+            list[list[dict]]: 二维列表, 每个子列表对应一个审查要点的检索结果
+        """
         def run_async(coro):
             """在合适的环境中运行异步函数"""
             try:
@@ -31,56 +56,169 @@ class EntitiesEnhance():
             except RuntimeError:
                 return asyncio.run(coro)
 
-        # 清空之前的结果
-        self.bfp_result_lists = []
-
-        for query_pair in query_pairs:
-            entity = query_pair['entity']
-            search_keywords = query_pair['search_keywords']
-            background = query_pair['background']
-            server_logger.info(f"正在处理实体：{entity}，辅助搜索词：{search_keywords}，背景：{background}")
-
-            # 检查 entity_recall 缓存
-            recall_cache_key = self._get_cache_key(entity, search_keywords)
-            if recall_cache_key in self._entity_recall_cache:
-                entity_list = self._entity_recall_cache[recall_cache_key]
-                server_logger.info(f"[缓存命中] entity_recall: {entity}")
+        self.result_lists = []
+        children_collection = self._get_children_collection()
+
+        for point_idx, point in enumerate(review_points):
+            # 兼容新旧字段名
+            label = point.get('label', point.get('entity', ''))
+            search_queries = point.get('search_queries', point.get('search_keywords', []))
+            original_text = point.get('original_text', point.get('background', ''))
+
+            server_logger.info(
+                f"正在处理审查要点 [{point_idx}]: {label}, "
+                f"检索语句数: {len(search_queries)}, "
+                f"原文长度: {len(original_text)}"
+            )
+
+            # Step 1: 对每个 search_query 执行 hybrid_search + 一次重排序
+            all_candidates = []
+
+            for query in search_queries:
+                cache_key = self._get_cache_key(query)
+                if cache_key in self._search_cache:
+                    query_results = self._search_cache[cache_key]
+                    server_logger.info(f"[缓存命中] search_query: {query[:30]}...")
+                else:
+                    query_results = run_async(
+                        retrieval_manager.async_multi_stage_recall(
+                            collection_name=children_collection,
+                            query_text=query,
+                            hybrid_top_k=10,
+                            top_k=5
+                        )
+                    )
+                    self._search_cache[cache_key] = query_results
+                    server_logger.info(
+                        f"[检索完成] search_query: {query[:30]}... "
+                        f"召回 {len(query_results)} 个候选"
+                    )
+
+                all_candidates.extend(query_results)
+
+            if not all_candidates:
+                server_logger.warning(f"审查要点 '{label}' 所有检索语句均无结果")
+                self.result_lists.append([])
+                continue
+
+            # Step 2: 去重 (基于 text_content)
+            seen_texts = set()
+            unique_candidates = []
+            for item in all_candidates:
+                text = item.get('text_content', '')
+                if text and text not in seen_texts:
+                    seen_texts.add(text)
+                    unique_candidates.append(item)
+
+            server_logger.info(
+                f"审查要点 '{label}': 合并 {len(all_candidates)} 个候选, "
+                f"去重后 {len(unique_candidates)} 个"
+            )
+
+            # Step 3: 筛选高分候选 (rerank_score > 0.5)
+            high_score = [c for c in unique_candidates if (c.get('rerank_score') or 0) > 0.5]
+
+            if not high_score:
+                # 无高分候选 → 该审查要点无相关规范，直接跳过（不进入后续流程）
+                max_score = max((c.get('rerank_score') or 0) for c in unique_candidates) if unique_candidates else 0
+                server_logger.warning(
+                    f"审查要点 '{label}': 无高分候选(>0.5), 共 {len(unique_candidates)} 个候选均低于阈值, "
+                    f"最高分={max_score:.4f}, 跳过该审查要点"
+                )
+                self.result_lists.append([])
+                continue
+
+            # Step 4: 二次重排序 — 用 original_text 作为语义锚点
+            if original_text and len(original_text) > 10 and len(high_score) > 1:
+                final_results = self._secondary_rerank(original_text, high_score, top_k=5)
+                server_logger.info(
+                    f"审查要点 '{label}': 二次重排序完成, "
+                    f"返回 {len(final_results)} 个结果"
+                )
             else:
-                entity_list = run_async(retrieval_manager.entity_recall(
-                    entity,
-                    search_keywords,
-                    recall_top_k=5,      # 主实体返回数量
-                    max_results=5       # 最终最多返回5个实体文本
-                ))
-                self._entity_recall_cache[recall_cache_key] = entity_list
-                server_logger.info(f"[缓存存储] entity_recall: {entity}")
-
-            # 检查 bfp_recall 缓存
-            bfp_cache_key = self._get_cache_key(entity, search_keywords, background)
-            if bfp_cache_key in self._bfp_recall_cache:
-                bfp_result = self._bfp_recall_cache[bfp_cache_key]
-                server_logger.info(f"[缓存命中] bfp_recall: {entity}")
-            else:
-                # BFP背景增强召回
-                bfp_result = run_async(retrieval_manager.async_bfp_recall(entity_list, background, top_k=2))  # 降低到2，减少上下文量
-                self._bfp_recall_cache[bfp_cache_key] = bfp_result
-                server_logger.info(f"[缓存存储] bfp_recall: {entity}")
-
-            # 为每个结果添加实体信息
-            for result in bfp_result:
-                result['source_entity'] = entity
-
-            self.bfp_result_lists.append(bfp_result)
-
-        return self.bfp_result_lists
+                final_results = high_score[:5]
+                server_logger.info(
+                    f"审查要点 '{label}': 跳过二次重排序 "
+                    f"(原文长度={len(original_text)}, 候选数={len(high_score)})"
+                )
+
+            # Step 5: 标记来源信息 (backward compat)
+            for result in final_results:
+                result['source_entity'] = label
+
+            self.result_lists.append(final_results)
+
+        return self.result_lists
+
+    def _secondary_rerank(self, original_text, candidates, top_k=5):
+        """
+        二次重排序: 用 original_text（原文摘录）作为 query，对候选文档重排序
+
+        核心创新: 用施工原文(而非 entity description 或 LLM 概括的 background)做 rerank,
+        确保检索到的规范条文与施工文本的语义精确对齐
+        """
+        # 提取候选文本（去重）
+        candidate_texts = []
+        seen = set()
+        for item in candidates:
+            text = item.get('text_content', '')
+            if text and text not in seen:
+                seen.add(text)
+                candidate_texts.append(text)
+
+        if not candidate_texts:
+            return candidates[:top_k]
+
+        try:
+            rerank_results = retrieval_manager._get_rerank_results(
+                original_text, candidate_texts, top_k
+            )
+        except Exception as e:
+            server_logger.error(f"二次重排序失败: {e}")
+            return candidates[:top_k]
+
+        # 将 rerank 分数映射回原始结果
+        text_to_items = {}
+        for item in candidates:
+            text = item.get('text_content', '')
+            if text not in text_to_items:
+                text_to_items[text] = []
+            text_to_items[text].append(item)
+
+        final_results = []
+        added_texts = set()
+        for rerank_item in rerank_results:
+            text = rerank_item.get('text', '')
+            score = rerank_item.get('score', 0.0)
+
+            if text in text_to_items and text not in added_texts:
+                best_candidate = max(
+                    text_to_items[text],
+                    key=lambda x: x.get('rerank_score', 0.0)
+                )
+                result_item = best_candidate.copy()
+                result_item['bfp_rerank_score'] = score  # 二次重排序分数 (backward compat)
+                result_item['bfp_rerank_parent_id'] = result_item.get(
+                    'metadata', {}
+                ).get('parent_id', '')
+                final_results.append(result_item)
+                added_texts.add(text)
+
+        return final_results
 
     def clear_cache(self):
-        """清空缓存"""
-        self._entity_recall_cache.clear()
-        self._bfp_recall_cache.clear()
-        server_logger.info("[缓存清理] 实体检索缓存已清空")
+        """清空检索缓存"""
+        self._search_cache.clear()
+        server_logger.info("[缓存清理] 审查要点检索缓存已清空")
 
+    # 向后兼容：旧代码调用 entities_enhance_retrieval 时自动转发
+    def entities_enhance_retrieval(self, query_pairs):
+        """向后兼容入口，转发到 review_point_retrieval"""
+        return self.review_point_retrieval(query_pairs)
 
 
+# 全局实例 — 新名称
+review_point_retriever = ReviewPointRetriever()
 
-entity_enhance = EntitiesEnhance()
+# 向后兼容：旧代码 import entity_enhance 时不会报错
+entity_enhance = review_point_retriever

foundation/ai/rag/retrieval/query_rewrite.py

@@ -1,17 +1,15 @@
 
 
 import uuid
-import asyncio
 from foundation.observability.logger.loggering import review_logger as server_logger
 from foundation.ai.agent.generate.model_generate import generate_model_client
 
 class QueryRewriteManager():
     """
-    召回管理器，实现多路召回功能
+    查询改写管理器 — 从施工方案文本中提取审查要点
     """
 
     def __init__(self):
-        # 获取部署的模型列表
         self.generate_model_client = generate_model_client
 
     @property
@@ -22,83 +20,92 @@ class QueryRewriteManager():
 
     def query_extract(self, review_content):
         """
-        从审查条文中提取query
+        从审查条文中提取审查要点 (review points)
 
         Args:
             review_content: 审查内容文本
 
         Returns:
-            list: 标准格式的查询列表
+            list: 审查要点列表
             [
                 {
-                    "entity": str,           # 实体名称
-                    "search_keywords": list, # 搜索关键词列表
-                    "background": str,       # 背景信息
-                    "parameter": str         # 技术参数
+                    "label": str,             # 审查要点标签
+                    "search_queries": list,    # 规范检索语句
+                    "original_text": str,      # 原文摘录
+                    "parameter": str,          # 技术参数
+                    # --- 向后兼容别名 (由 _add_backward_compat_aliases 自动添加) ---
+                    "entity": str,             # = label
+                    "search_keywords": list,   # = search_queries
+                    "background": str,         # = original_text
                 }
             ]
             或 None(提取失败时)
         """
         try:
-            # 获取提示词模板并组装
+            # 获取提示词模板并组装 — 优先使用新 key，回退到旧 key
             task_prompt = self.prompt_loader.get_prompt_template(
-                reviewer_type="query_extract",  # 审查器类型
-                review_content=review_content   # 传入审查内容作为参数
+                reviewer_type="review_point_extract",
+                prompt_name="review_point_extract",
+                review_content=review_content
             )
 
-            # 构建任务提示信息 - 参考标准模式
             task_prompt_info = {
-                "task_prompt": task_prompt,  # 使用组装好的提示词
-                "task_name": "query_extract"
+                "task_prompt": task_prompt,
+                "task_name": "review_point_extract"
             }
 
-            # 生成唯一的trace_id用于追踪
             trace_id = str(uuid.uuid4())
 
-            # 调用模型生成接口（处理异步调用）
-            try:
-                loop = asyncio.get_running_loop()
-                # 如果已有运行中的事件循环，使用create_task
-                import concurrent.futures
-                with concurrent.futures.ThreadPoolExecutor() as executor:
-                    future = executor.submit(
-                        asyncio.run,
-                        self.generate_model_client.get_model_generate_invoke(
-                            trace_id=trace_id,
-                            task_prompt_info=task_prompt_info,
-                            timeout=60,
-                            function_name="query_extract"
-                        )
-                    )
-                    model_response = future.result()
-            except RuntimeError:
-                # 没有运行中的事件循环，直接使用asyncio.run
-                model_response = asyncio.run(self.generate_model_client.get_model_generate_invoke(
-                    trace_id=trace_id,
-                    task_prompt_info=task_prompt_info,
-                    timeout=60,
-                    function_name="query_extract"
-                ))
+            # 调用模型 — function_name 对应 model_setting.yaml 中的配置
+            model_response = self.generate_model_client.get_model_generate_invoke_sync(
+                trace_id=trace_id,
+                task_prompt_info=task_prompt_info,
+                timeout=60,
+                function_name="review_point_extract"
+            )
 
             # 格式化模型响应
             formatted_response = self.ai_respose_format(model_response)
-            # 检查 formatted_response 是否为 None
-            if formatted_response is not None:
-                server_logger.info(f"查询对构建完成，构建 {len(formatted_response)}条。")
-            else:
-                server_logger.warning("查询对构建失败，formatted_response 为 None")
-            # 记录日志
+
             if formatted_response:
-                server_logger.info(f"Query 提取成功, 提取到 {len(formatted_response)} 个实体")
+                # 添加向后兼容字段别名
+                formatted_response = self._add_backward_compat_aliases(formatted_response)
+                server_logger.info(f"审查要点提取完成, 提取到 {len(formatted_response)} 个要点")
             else:
-                server_logger.warning(f"Query 提取失败, 格式化后为空")
+                server_logger.warning("审查要点提取失败, 格式化后为空")
 
             return formatted_response
 
         except Exception as e:
-            server_logger.error(f"Query 提取失败: {str(e)}")
+            server_logger.error(f"审查要点提取失败: {str(e)}")
             return None
-    
+
+    def _add_backward_compat_aliases(self, review_points):
+        """
+        为每个审查要点添加双向字段别名，确保新旧格式都能工作
+
+        新字段 → 旧字段: label→entity, search_queries→search_keywords, original_text→background
+        旧字段 → 新字段: entity→label, search_keywords→search_queries, background→original_text
+        """
+        for point in review_points:
+            # 新 → 旧（LLM 使用新格式时）
+            if 'label' in point and 'entity' not in point:
+                point['entity'] = point['label']
+            if 'search_queries' in point and 'search_keywords' not in point:
+                point['search_keywords'] = point['search_queries']
+            if 'original_text' in point and 'background' not in point:
+                point['background'] = point['original_text']
+
+            # 旧 → 新（LLM 使用旧格式时）
+            if 'entity' in point and 'label' not in point:
+                point['label'] = point['entity']
+            if 'search_keywords' in point and 'search_queries' not in point:
+                point['search_queries'] = point['search_keywords']
+            if 'background' in point and 'original_text' not in point:
+                point['original_text'] = point['background']
+
+        return review_points
+
     def ai_respose_format(self, model_response):
         """
         将模型返回的响应格式化为标准格式
@@ -107,16 +114,7 @@ class QueryRewriteManager():
             model_response: AI模型返回的原始响应(可能是字符串或已解析的JSON)
 
         Returns:
-            list: 标准格式的查询列表
-            [
-                {
-                    "entity": str,           # 实体名称
-                    "search_keywords": list, # 搜索关键词列表
-                    "background": str,       # 背景信息
-                    "parameter": str         # 技术参数
-                }
-            ]
-            或 None(解析失败时)
+            list: 标准格式的审查要点列表, 或 None(解析失败时)
         """
         import re
         import json
@@ -124,7 +122,7 @@ class QueryRewriteManager():
         try:
             # 1. 如果model_response已经是list,直接返回
             if isinstance(model_response, list):
-                server_logger.info(f"模型响应已是list格式, 包含 {len(model_response)} 个实体")
+                server_logger.info(f"模型响应已是list格式, 包含 {len(model_response)} 个要点")
                 return model_response
 
             # 2. 如果是dict,包装成list返回
@@ -135,37 +133,29 @@ class QueryRewriteManager():
             # 3. 如果是字符串,需要解析
             if isinstance(model_response, str):
                 response_text = model_response.strip()
-                server_logger.debug(f"原始响应字符串长度: {len(response_text)}")
 
                 # 3.1 尝试去除 ```json 和 ``` 标记
-                # 匹配 ```json ... ``` 或 ``` ... ```
                 json_pattern = r'```(?:json)?\s*\n?(.*?)\n?```'
                 json_match = re.search(json_pattern, response_text, re.DOTALL | re.IGNORECASE)
 
                 if json_match:
-                    # 提取代码块中的JSON内容
                     json_str = json_match.group(1).strip()
-                    server_logger.debug("检测到markdown代码块, 已提取纯JSON内容")
                 else:
-                    # 如果没有代码块标记,尝试直接解析整个字符串
                     json_str = response_text
-                    server_logger.debug("未检测到markdown代码块, 尝试直接解析")
 
                 # 3.2 去除可能的Markdown注释或多余空白
-                json_str = re.sub(r'\n+', '\n', json_str)  # 多个换行压缩为一个
+                json_str = re.sub(r'\n+', '\n', json_str)
                 json_str = json_str.strip()
 
-                server_logger.debug(f"待解析的JSON字符串: {json_str[:200]}...")
-
                 # 3.3 解析JSON
                 parsed_data = json.loads(json_str)
 
                 # 3.4 确保返回list格式
                 if isinstance(parsed_data, list):
-                    server_logger.info(f"JSON解析成功, 提取到 {len(parsed_data)} 个实体")
+                    server_logger.info(f"JSON解析成功, 提取到 {len(parsed_data)} 个审查要点")
                     return parsed_data
                 elif isinstance(parsed_data, dict):
-                    server_logger.info("JSON解析成功, 单个实体包装为list")
+                    server_logger.info("JSON解析成功, 单个要点包装为list")
                     return [parsed_data]
 
                 server_logger.warning(f"无法识别的JSON格式: {type(parsed_data)}")
@@ -184,4 +174,4 @@ class QueryRewriteManager():
             return None
 
 
-query_rewrite_manager = QueryRewriteManager()
+query_rewrite_manager = QueryRewriteManager()

foundation/ai/rag/retrieval/retrieval.py

@@ -272,11 +272,11 @@ class RetrievalManager:
         self.logger.info(f"[async_bfp_recall] 第一阶段召回完成, 共召回 {len(bfp_results)} 个文档")
 
         # BFP召回结果已经通过multi_stage_recall进行了重排序，保持原有顺序
-        # 只对第一次重排序得分大于0.8的文档进行二次重排序
-        high_score_results = [item for item in bfp_results if (item.get('rerank_score') or 0) > 0.8]
-        low_score_results = [item for item in bfp_results if (item.get('rerank_score') or 0) <= 0.8]
+        # 只对第一次重排序得分大于0.6的文档进行二次重排序
+        high_score_results = [item for item in bfp_results if (item.get('rerank_score') or 0) > 0.6]
+        low_score_results = [item for item in bfp_results if (item.get('rerank_score') or 0) <= 0.6]
 
-        self.logger.info(f"筛选结果：高分文档(>0.8) {len(high_score_results)} 个，低分文档(≤0.8) {len(low_score_results)} 个")
+        self.logger.info(f"筛选结果：高分文档(>0.6) {len(high_score_results)} 个，低分文档(≤0.6) {len(low_score_results)} 个")
 
         # 如果没有高分文档，直接返回top_k个结果（按hybrid_similarity排序）
         if not high_score_results:
@@ -567,17 +567,23 @@ class RetrievalManager:
                 duplicate_count = rerank_result.get('duplicate_count', 1)
 
                 # 如果内层有metadata字段，将其提取到外层
-                if 'metadata' in metadata and isinstance(metadata['metadata'], str):
-                    import json
-                    try:
-                        # 解析JSON格式的metadata
-                        inner_metadata = json.loads(metadata['metadata'])
+                if 'metadata' in metadata:
+                    inner_raw = metadata['metadata']
+                    inner_metadata = None
+
+                    if isinstance(inner_raw, dict):
+                        # Milvus JSON 字段直接返回 dict
+                        inner_metadata = inner_raw
+                    elif isinstance(inner_raw, str):
+                        import json
+                        try:
+                            inner_metadata = json.loads(inner_raw)
+                        except (json.JSONDecodeError, TypeError):
+                            pass
+
+                    if inner_metadata and isinstance(inner_metadata, dict):
                         metadata.update(inner_metadata)
-                        # 移除内层的metadata字符串，避免重复
                         del metadata['metadata']
-                    except (json.JSONDecodeError, TypeError):
-                        # 如果解析失败，保持原样
-                        pass
 
                 # 移除重复的content字段
                 if 'content' in metadata:
@@ -651,17 +657,23 @@ class RetrievalManager:
                 duplicate_count = rerank_result.get('duplicate_count', 1)
 
                 # 如果内层有metadata字段，将其提取到外层
-                if 'metadata' in metadata and isinstance(metadata['metadata'], str):
-                    import json
-                    try:
-                        # 解析JSON格式的metadata
-                        inner_metadata = json.loads(metadata['metadata'])
+                if 'metadata' in metadata:
+                    inner_raw = metadata['metadata']
+                    inner_metadata = None
+
+                    if isinstance(inner_raw, dict):
+                        # Milvus JSON 字段直接返回 dict
+                        inner_metadata = inner_raw
+                    elif isinstance(inner_raw, str):
+                        import json
+                        try:
+                            inner_metadata = json.loads(inner_raw)
+                        except (json.JSONDecodeError, TypeError):
+                            pass
+
+                    if inner_metadata and isinstance(inner_metadata, dict):
                         metadata.update(inner_metadata)
-                        # 移除内层的metadata字符串，避免重复
                         del metadata['metadata']
-                    except (json.JSONDecodeError, TypeError):
-                        # 如果解析失败，保持原样
-                        pass
 
                 # 移除重复的content字段
                 if 'content' in metadata:

utils_test/Grammar_Check_Test/analyze_grammar_quality.py

@@ -0,0 +1,230 @@
+"""
+分析最新审查结果中词句语法审查的质量
+
+检查项：
+1. "将A改为A" 模式（修正前后相同）
+2. suggestion/reason 中的自我辩论（犹豫措辞）
+3. risk_level 为空
+4. 技术操作规程越界审查
+5. 重复问题
+6. JSON 解析失败
+7. suggestion 过长（>200字，可能包含推理过程）
+"""
+
+import json
+import re
+import sys
+import os
+
+project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+sys.path.insert(0, project_root)
+
+
+def extract_correction_pairs(suggestion: str):
+    """从 suggestion 中提取所有 '将X改为Y' 的 (X, Y) 对"""
+    quote_chars = r"""['""''「」]"""
+    pattern = rf"将{quote_chars}(.*?){quote_chars}\s*改为\s*{quote_chars}(.*?){quote_chars}"
+    return re.findall(pattern, suggestion)
+
+
+def check_hesitation_words(text: str):
+    """检查文本中是否包含犹豫措辞"""
+    hesitation_words = [
+        '可能', '暂定', '不确定', '重新审视', '然而', '不过', '似乎',
+        '但是', '其实', '实际上', '再细看', '再想想', '仔细想想',
+        '反过来', '另一方面', '换个角度'
+    ]
+    found = [w for w in hesitation_words if w in text]
+    return found
+
+
+def is_technical_procedure(issue_point: str, reason: str):
+    """检查是否为技术操作规程越界审查"""
+    technical_keywords = [
+        '操作步骤', '工艺参数', '施工顺序', '操作规程',
+        '技术规范', '施工方案', '工艺流程'
+    ]
+    combined = issue_point + reason
+    return [kw for kw in technical_keywords if kw in combined]
+
+
+def analyze_grammar_check_results(result_file: str):
+    """分析词句语法审查结果质量"""
+    with open(result_file, encoding='utf-8') as f:
+        data = json.load(f)
+
+    issues = data.get('issues', [])
+    grammar_items = []
+
+    for issue_wrapper in issues:
+        for issue_id, issue_detail in issue_wrapper.items():
+            review_lists = issue_detail.get('review_lists', [])
+            metadata = issue_detail.get('metadata', {})
+            for item in review_lists:
+                check_item = item.get('check_item', '')
+                if check_item in ['sensitive_word_check', 'grammar_check']:
+                    grammar_items.append({
+                        'item': item,
+                        'issue_id': issue_id,
+                        'location_label': metadata.get('review_location_label', '')
+                    })
+
+    print(f"Total grammar_check items: {len(grammar_items)}")
+    print()
+
+    # Quality checks
+    a_to_a_issues = []
+    hesitation_issues = []
+    empty_risk_issues = []
+    technical_issues = []
+    duplicate_issues = []
+    parse_failures = []
+    long_suggestion_issues = []
+
+    seen_corrections = {}
+
+    for i, entry in enumerate(grammar_items):
+        item = entry['item']
+        check_result = item.get('check_result', {})
+
+        # STRING format = parse failure
+        if isinstance(check_result, str):
+            parse_failures.append({
+                'index': i + 1,
+                'raw': check_result[:200],
+                'location_label': entry['location_label']
+            })
+            continue
+
+        issue_point = check_result.get('issue_point', '')
+        location = check_result.get('location', '')
+        suggestion = check_result.get('suggestion', '')
+        reason = check_result.get('reason', '')
+        risk_level = check_result.get('risk_level', '')
+
+        # Check 1: Empty risk_level
+        if not risk_level or risk_level.strip() == '':
+            empty_risk_issues.append({
+                'index': i + 1,
+                'issue_point': issue_point,
+                'suggestion': suggestion[:80]
+            })
+
+        # Check 2: A→A pattern
+        pairs = extract_correction_pairs(suggestion)
+        for before, after in pairs:
+            if before.strip() == after.strip():
+                a_to_a_issues.append({
+                    'index': i + 1,
+                    'issue_point': issue_point,
+                    'before': before,
+                    'after': after
+                })
+
+        # Check 3: Hesitation words in suggestion
+        sug_hesitation = check_hesitation_words(suggestion)
+        if sug_hesitation:
+            hesitation_issues.append({
+                'index': i + 1,
+                'field': 'suggestion',
+                'words': sug_hesitation,
+                'text': suggestion[:100]
+            })
+
+        # Check 4: Hesitation words in reason
+        reason_hesitation = check_hesitation_words(reason)
+        if reason_hesitation:
+            hesitation_issues.append({
+                'index': i + 1,
+                'field': 'reason',
+                'words': reason_hesitation,
+                'text': reason[:100]
+            })
+
+        # Check 5: Technical procedure
+        tech_kws = is_technical_procedure(issue_point, reason)
+        if tech_kws:
+            technical_issues.append({
+                'index': i + 1,
+                'issue_point': issue_point,
+                'keywords': tech_kws
+            })
+
+        # Check 6: Long suggestion (>200 chars)
+        if len(suggestion) > 200:
+            long_suggestion_issues.append({
+                'index': i + 1,
+                'issue_point': issue_point,
+                'length': len(suggestion),
+                'text': suggestion[:100]
+            })
+
+        # Check 7: Duplicates (same correction key)
+        if pairs:
+            sorted_pairs = sorted(pairs)
+            correction_key = ",".join(f"{a}→{b}" for a, b in sorted_pairs)
+        else:
+            correction_key = suggestion.strip()
+
+        if correction_key in seen_corrections:
+            duplicate_issues.append({
+                'index': i + 1,
+                'first_index': seen_corrections[correction_key],
+                'correction_key': correction_key,
+                'issue_point': issue_point
+            })
+        else:
+            seen_corrections[correction_key] = i + 1
+
+    # Print results
+    print("=" * 60)
+    print("QUALITY ANALYSIS RESULTS")
+    print("=" * 60)
+
+    sections = [
+        ("A->A Pattern (will A change to A)", a_to_a_issues),
+        ("Self-debate / Hesitation words", hesitation_issues),
+        ("Empty risk_level", empty_risk_issues),
+        ("Technical procedure (out of scope)", technical_issues),
+        ("Duplicate corrections", duplicate_issues),
+        ("JSON parse failures", parse_failures),
+        ("Long suggestions (>200 chars)", long_suggestion_issues),
+    ]
+
+    total_problems = 0
+    for title, items in sections:
+        count = len(items)
+        total_problems += count
+        status = "PASS" if count == 0 else "FAIL"
+        print(f"\n[{status}] {title}: {count}")
+        if items:
+            for item in items:
+                print(f"  - #{item.get('index', '?')}: {json.dumps(item, ensure_ascii=False)[:150]}")
+
+    print(f"\n{'=' * 60}")
+    print(f"TOTAL: {len(grammar_items)} items, {total_problems} quality issues")
+    print(f"Quality rate: {(len(grammar_items) - total_problems) / len(grammar_items) * 100:.1f}%")
+
+    # Print valid items summary
+    print(f"\n{'=' * 60}")
+    print("VALID ITEMS SUMMARY")
+    print("=" * 60)
+    for i, entry in enumerate(grammar_items):
+        item = entry['item']
+        check_result = item.get('check_result', {})
+        if isinstance(check_result, str):
+            print(f"  [{i+1}] [PARSE_FAIL] {check_result[:60]}...")
+            continue
+        issue_point = check_result.get('issue_point', '')
+        suggestion = check_result.get('suggestion', '')
+        risk_level = check_result.get('risk_level', '')
+        print(f"  [{i+1}] [{risk_level}] {issue_point}: {suggestion[:60]}...")
+
+
+if __name__ == "__main__":
+    result_file = os.path.join(
+        project_root,
+        "temp", "construction_review", "final_result",
+        "67d45692fb97aeef8f896e78475ce539-1779785718.json"
+    )
+    analyze_grammar_check_results(result_file)

utils_test/Grammar_Check_Test/run_full_scan.py

@@ -0,0 +1,93 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""全量 chunk 词句语法审查 — 保存所有原始响应用于人工分析"""
+
+import sys, os, json, asyncio, time
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)).split('utils_test')[0])
+
+RESULT_JSON = os.path.join(
+    os.path.dirname(os.path.abspath(__file__)).split('utils_test')[0],
+    "temp", "construction_review", "final_result",
+    "67d45692fb97aeef8f896e78475ce539-1779781589.json"
+)
+OUTPUT_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "full_scan_results")
+
+async def main():
+    from core.construction_review.component.reviewers.grammar_check_reviewer import GrammarCheckReviewer
+
+    with open(RESULT_JSON, 'r', encoding='utf-8') as f:
+        data = json.load(f)
+    chunks = data['document_result']['structured_content']['chunks']
+
+    os.makedirs(OUTPUT_DIR, exist_ok=True)
+    reviewer = GrammarCheckReviewer()
+
+    all_results = []
+
+    for i, chunk in enumerate(chunks):
+        content = chunk['content']
+        section = chunk.get('section_label', f'chunk_{i}')
+        chapter = chunk.get('chapter_classification', 'unknown')
+        trace_id = f"full_scan_{i}_{int(time.time())}"
+
+        print(f"[{i:02d}/{len(chunks)}] {chapter}/{section[:40]}... (len={len(content)})")
+
+        start = time.time()
+        try:
+            result = await reviewer.check_grammar(
+                trace_id=trace_id,
+                review_content=content,
+                state=None, stage_name=None,
+                enable_thinking=False,
+            )
+            wall_time = time.time() - start
+            response_text = result.details.get('response', '')
+            success = result.success
+            error = result.error_message
+        except Exception as e:
+            wall_time = time.time() - start
+            response_text = ""
+            success = False
+            error = str(e)
+            print(f"      ERROR: {e}")
+
+        record = {
+            "chunk_index": i,
+            "chapter": chapter,
+            "section": section,
+            "content_length": len(content),
+            "content_preview": content[:200],
+            "success": success,
+            "error": error,
+            "wall_time": round(wall_time, 2),
+            "response_length": len(response_text),
+            "raw_response": response_text,
+        }
+        all_results.append(record)
+
+        is_no_issue = '无明显问题' in response_text and len(response_text) < 50
+        status = "NO_ISSUE" if is_no_issue else f"ISSUES(response_len={len(response_text)})"
+        print(f"      {wall_time:.2f}s | {status}")
+
+    # 保存汇总
+    summary_path = os.path.join(OUTPUT_DIR, "all_results.json")
+    with open(summary_path, 'w', encoding='utf-8') as f:
+        json.dump(all_results, f, ensure_ascii=False, indent=2)
+    print(f"\nSaved {len(all_results)} results to {summary_path}")
+
+    # 保存每个 chunk 的独立文件（方便逐条阅读）
+    for record in all_results:
+        idx = record["chunk_index"]
+        chunk_path = os.path.join(OUTPUT_DIR, f"chunk_{idx:02d}_{record['chapter']}.json")
+        with open(chunk_path, 'w', encoding='utf-8') as f:
+            json.dump(record, f, ensure_ascii=False, indent=2)
+
+    print(f"Saved individual files to {OUTPUT_DIR}/")
+
+    # 打印统计
+    no_issue_count = sum(1 for r in all_results if '无明显问题' in r['raw_response'] and len(r['raw_response']) < 50)
+    issue_count = len(all_results) - no_issue_count
+    error_count = sum(1 for r in all_results if not r['success'])
+    print(f"\nStats: {no_issue_count} no-issue, {issue_count} has-issues, {error_count} errors")
+
+asyncio.run(main())

utils_test/Grammar_Check_Test/test_grammar_check_prompt_fix.py

@@ -0,0 +1,341 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+词句语法审查 — Prompt 修复验证测试
+
+验证目标：修复 "将A改为A" 的离谱错误
+- 旧 prompt 包含否定示例（如"禁止输出将'设'改为'设'"），反而给模型植入了错误模式
+- 新 prompt 使用肯定式规则（"犹豫时输出无明显问题"）
+
+测试数据：temp/construction_review/final_result/67d45692fb97aeef8f896e78475ce539-1779781589.json
+其中 chunk[8] 包含触发 bug 的原文："必须采取充分的安全保证措施"
+
+运行方式：
+    $env:PYTHONPATH = (Get-Location)
+    pytest utils_test/Grammar_Check_Test/test_grammar_check_prompt_fix.py -v -s
+"""
+
+import sys
+import os
+import json
+import re
+import time
+import asyncio
+from pathlib import Path
+
+# 项目根目录注入
+PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..'))
+if PROJECT_ROOT not in sys.path:
+    sys.path.insert(0, PROJECT_ROOT)
+
+import pytest
+
+# ============================================================
+# 测试数据
+# ============================================================
+RESULT_JSON = os.path.join(
+    PROJECT_ROOT,
+    "temp", "construction_review", "final_result",
+    "67d45692fb97aeef8f896e78475ce539-1779781589.json"
+)
+
+
+def _load_chunks():
+    """加载文档 chunks"""
+    with open(RESULT_JSON, 'r', encoding='utf-8') as f:
+        data = json.load(f)
+    return data['document_result']['structured_content']['chunks']
+
+
+# ============================================================
+# Bug 检测工具函数
+# ============================================================
+def detect_a_to_a_pattern(response_text: str) -> list:
+    """
+    检测模型输出中是否包含 "将X改为X" 模式（X相同）
+
+    匹配模式：
+    - 将"充分"改为"充分"
+    - 将'设'改为'设'
+    - 把"X"修改为"X"
+    - 建议将X改为X
+
+    Returns:
+        list: 匹配到的问题片段列表
+    """
+    if not response_text:
+        return []
+
+    issues = []
+
+    # 模式1: 将"X"改为"X" / 将'X'改为'X' / 把"X"改为"X"
+    pattern_quoted = re.compile(
+        r'(?:将|把)["“\'](.{1,10})["”\']\s*(?:改为|修改为|替换为|换成)\s*["“\'](.{1,10})["”\']'
+    )
+    for m in pattern_quoted.finditer(response_text):
+        original, replacement = m.group(1).strip(), m.group(2).strip()
+        if original == replacement:
+            issues.append(m.group(0))
+
+    # 模式2: suggestion 字段过长且包含自我辩论关键词
+    debate_keywords = ['然而', '再细看', '重新审视', '其实', '但', '不过', '似乎', '略显生硬']
+    debate_count = sum(1 for kw in debate_keywords if kw in response_text)
+    if debate_count >= 3:
+        issues.append(f"[自我辩论] 响应中包含 {debate_count} 个犹豫/反驳关键词: "
+                      f"{[kw for kw in debate_keywords if kw in response_text]}")
+
+    return issues
+
+
+def parse_json_from_response(response_text: str) -> list:
+    """从模型响应中提取 JSON 结果"""
+    if not response_text:
+        return []
+
+    # 先尝试直接解析
+    try:
+        data = json.loads(response_text)
+        if isinstance(data, list):
+            return data
+        elif isinstance(data, dict):
+            return [data]
+    except (json.JSONDecodeError, TypeError):
+        pass
+
+    # 尝试从 markdown 代码块中提取
+    json_blocks = re.findall(r'```(?:json)?\s*\n?(.*?)\n?```', response_text, re.DOTALL)
+    for block in json_blocks:
+        try:
+            data = json.loads(block.strip())
+            if isinstance(data, list):
+                return data
+            elif isinstance(data, dict):
+                return [data]
+        except (json.JSONDecodeError, TypeError):
+            continue
+
+    # 尝试找到第一个 [ 或 { 开始解析
+    for start_char, end_char in [('[', ']'), ('{', '}')]:
+        start = response_text.find(start_char)
+        if start >= 0:
+            # 从后往前找匹配的结束符
+            for end in range(len(response_text) - 1, start, -1):
+                if response_text[end] == end_char:
+                    try:
+                        data = json.loads(response_text[start:end + 1])
+                        if isinstance(data, list):
+                            return data
+                        elif isinstance(data, dict):
+                            return [data]
+                    except (json.JSONDecodeError, TypeError):
+                        continue
+
+    return []
+
+
+# ============================================================
+# 测试类
+# ============================================================
+class TestGrammarCheckPromptFix:
+    """词句语法审查 Prompt 修复验证"""
+
+    @pytest.fixture(autouse=True)
+    def setup(self):
+        """初始化"""
+        self.chunks = _load_chunks()
+        # bug 复现的 chunk: [8] 包含 "采取充分的安全保证措施"
+        self.bug_chunk = self.chunks[8]
+        assert '充分' in self.bug_chunk['content'], "chunk[8] 应包含 '充分' 文本"
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_bug_chunk_no_a_to_a(self):
+        """
+        【核心测试】原 bug chunk 不再产生 "将A改为A" 的结果
+
+        这是触发原始 bug 的具体文本（chunk[8]: 施工要求和技术保证条件），
+        模型曾对"充分"一词产生自我辩论，输出"将'充分'改为'充分'"。
+        """
+        from core.construction_review.component.reviewers.grammar_check_reviewer import GrammarCheckReviewer
+
+        reviewer = GrammarCheckReviewer()
+        trace_id = f"grammar_fix_test_bug_{int(time.time())}"
+
+        print(f"\n{'='*70}")
+        print(f"  测试原 bug chunk: {self.bug_chunk['section_label']}")
+        print(f"  内容长度: {len(self.bug_chunk['content'])} 字符")
+        print(f"{'='*70}")
+
+        start = time.time()
+        result = await reviewer.check_grammar(
+            trace_id=trace_id,
+            review_content=self.bug_chunk['content'],
+            state=None,
+            stage_name=None,
+            enable_thinking=False,
+        )
+        wall_time = time.time() - start
+
+        print(f"\n  审查耗时: {wall_time:.2f}s")
+        print(f"  success: {result.success}")
+
+        response_text = result.details.get('response', '')
+        print(f"  响应长度: {len(response_text)} 字符")
+
+        # 判断是否输出"无明显问题"
+        is_no_issue = '无明显问题' in response_text and len(response_text) < 50
+        print(f"  是否无明显问题: {is_no_issue}")
+
+        if not is_no_issue:
+            # 解析 JSON 结果
+            issues = parse_json_from_response(response_text)
+            print(f"  发现 {len(issues)} 个问题")
+            for idx, issue in enumerate(issues):
+                print(f"\n  --- 问题 {idx + 1} ---")
+                print(f"  issue_point: {issue.get('issue_point', 'N/A')}")
+                print(f"  location: {issue.get('location', 'N/A')[:80]}...")
+                print(f"  suggestion: {issue.get('suggestion', 'N/A')[:120]}")
+                print(f"  reason: {issue.get('reason', 'N/A')[:120]}")
+                print(f"  risk_level: {issue.get('risk_level', 'N/A')}")
+
+            # 打印原始响应供人工检查
+            print(f"\n  --- 原始响应 ---")
+            print(response_text[:2000])
+        else:
+            print(f"  原始响应: {response_text}")
+
+        # ===== 断言 =====
+        assert result.success, f"审查应成功，实际错误: {result.error_message}"
+
+        # 核心断言：不应出现 "将A改为A" 模式
+        a_to_a_issues = detect_a_to_a_pattern(response_text)
+        assert not a_to_a_issues, (
+            f"检测到 '将A改为A' 模式仍存在！\n"
+            f"问题片段: {a_to_a_issues}\n"
+            f"完整响应:\n{response_text}"
+        )
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_multiple_overview_chunks(self):
+        """
+        【扩展测试】多个 overview chunk 均不产生 "将A改为A" 结果
+
+        测试所有 overview 章节的 chunk，确保修复具有泛化性。
+        """
+        from core.construction_review.component.reviewers.grammar_check_reviewer import GrammarCheckReviewer
+
+        reviewer = GrammarCheckReviewer()
+
+        # 筛选 overview chunks
+        overview_chunks = [
+            c for c in self.chunks
+            if c.get('chapter_classification') == 'overview'
+        ]
+        print(f"\n{'='*70}")
+        print(f"  扩展测试: {len(overview_chunks)} 个 overview chunks")
+        print(f"{'='*70}")
+
+        all_a_to_a_issues = []
+
+        for idx, chunk in enumerate(overview_chunks):
+            trace_id = f"grammar_fix_test_overview_{idx}_{int(time.time())}"
+            section = chunk.get('section_label', f'chunk_{idx}')
+            content = chunk['content']
+
+            print(f"\n  [{idx}] {section} (len={len(content)})")
+
+            start = time.time()
+            result = await reviewer.check_grammar(
+                trace_id=trace_id,
+                review_content=content,
+                state=None,
+                stage_name=None,
+                enable_thinking=False,
+            )
+            wall_time = time.time() - start
+
+            response_text = result.details.get('response', '')
+            is_no_issue = '无明显问题' in response_text and len(response_text) < 50
+
+            # 检测 A→A 模式
+            a_to_a = detect_a_to_a_pattern(response_text)
+            status = "[OK] 无明显问题" if is_no_issue else (
+                f"[!!] 有 {len(parse_json_from_response(response_text))} 个问题"
+            )
+            if a_to_a:
+                status += f" [FAIL] 检测到A->A模式: {a_to_a}"
+                all_a_to_a_issues.extend([(section, issue) for issue in a_to_a])
+
+            print(f"      耗时: {wall_time:.2f}s | {status}")
+
+            if not is_no_issue and not a_to_a:
+                # 打印发现的问题摘要
+                issues = parse_json_from_response(response_text)
+                for issue in issues:
+                    ip = issue.get('issue_point', '')[:60]
+                    sg = issue.get('suggestion', '')[:80]
+                    print(f"      -> {ip} | 建议: {sg}")
+
+        print(f"\n{'='*70}")
+        print(f"  扩展测试完成: {len(overview_chunks)} 个 chunks")
+        print(f"  A->A 问题数: {len(all_a_to_a_issues)}")
+        print(f"{'='*70}")
+
+        # 核心断言
+        assert not all_a_to_a_issues, (
+            f"检测到 {len(all_a_to_a_issues)} 个 '将A改为A' 模式！\n"
+            + "\n".join(f"  {sec}: {issue}" for sec, issue in all_a_to_a_issues)
+        )
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_suggestion_field_concise(self):
+        """
+        【格式验证】suggestion 字段应简洁，不包含推理过程
+
+        新 prompt 要求 suggestion 只写最终结论，禁止自我辩论。
+        """
+        from core.construction_review.component.reviewers.grammar_check_reviewer import GrammarCheckReviewer
+
+        reviewer = GrammarCheckReviewer()
+        trace_id = f"grammar_fix_test_concise_{int(time.time())}"
+
+        # 使用 bug chunk
+        result = await reviewer.check_grammar(
+            trace_id=trace_id,
+            review_content=self.bug_chunk['content'],
+            state=None,
+            stage_name=None,
+            enable_thinking=False,
+        )
+
+        response_text = result.details.get('response', '')
+        issues = parse_json_from_response(response_text)
+
+        if not issues:
+            print("\n  模型输出'无明显问题'，无需验证 suggestion 格式")
+            return
+
+        print(f"\n  发现 {len(issues)} 个问题，验证 suggestion 格式:")
+
+        for idx, issue in enumerate(issues):
+            suggestion = issue.get('suggestion', '')
+            reason = issue.get('reason', '')
+            print(f"\n  --- 问题 {idx + 1} ---")
+            print(f"  suggestion ({len(suggestion)}字): {suggestion[:150]}")
+            print(f"  reason ({len(reason)}字): {reason[:150]}")
+
+            # suggestion 不应包含推理/辩论关键词
+            debate_keywords = ['然而', '再细看', '重新审视', '让我们', '再审视']
+            found_debate = [kw for kw in debate_keywords if kw in suggestion]
+            assert not found_debate, (
+                f"suggestion 字段包含推理过程！\n"
+                f"检测到辩论关键词: {found_debate}\n"
+                f"suggestion 内容: {suggestion}"
+            )
+
+            # suggestion 不应过长（超过 200 字大概率包含推理）
+            assert len(suggestion) < 200, (
+                f"suggestion 字段过长（{len(suggestion)}字），可能包含推理过程:\n{suggestion}"
+            )

utils_test/Grammar_Check_Test/test_grammar_check_split.py

@@ -0,0 +1,197 @@
+"""
+测试词句语法审查的长文本切分逻辑
+
+测试内容：
+1. 切分触发条件（>5000字）
+2. 切分后并行审查
+3. 结果合并去重
+4. JSON 解析鲁棒性
+"""
+
+import sys
+import os
+import json
+import asyncio
+import time
+
+# 注入项目根目录
+project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+sys.path.insert(0, project_root)
+
+from core.construction_review.component.reviewers.grammar_check_reviewer import (
+    GrammarCheckReviewer,
+    SPLIT_THRESHOLD,
+    SEGMENT_MIN_LENGTH,
+    SEGMENT_TARGET_LENGTH,
+    SEGMENT_OVERLAP,
+)
+from core.construction_review.component.reviewers.utils.text_split import split_text_with_overlap
+
+
+def load_test_chunks():
+    """加载测试数据"""
+    test_file = os.path.join(
+        project_root,
+        "temp", "construction_review", "final_result",
+        "67d45692fb97aeef8f896e78475ce539-1779781589.json"
+    )
+    with open(test_file, encoding="utf-8") as f:
+        data = json.load(f)
+    return data["document_result"]["structured_content"]["chunks"]
+
+
+def test_parse_segment_response():
+    """测试 JSON 解析鲁棒性"""
+    reviewer = GrammarCheckReviewer()
+
+    # 测试1: 标准 JSON 数组
+    response1 = '''```json
+[
+  {"issue_point": "错别字", "location": "位置1", "suggestion": "将A改为B", "reason": "原因", "risk_level": "中风险"},
+  {"issue_point": "重复字词", "location": "位置2", "suggestion": "删除重复", "reason": "原因", "risk_level": "低风险"}
+]
+```'''
+    issues1 = reviewer._parse_segment_response(response1)
+    assert len(issues1) == 2, f"Expected 2 issues, got {len(issues1)}"
+    print("[PASS] test_parse_segment_response: JSON array parsed correctly")
+
+    # 测试2: 单个 JSON 对象
+    response2 = '''```json
+{"issue_point": "错别字", "location": "位置1", "suggestion": "将A改为B", "reason": "原因", "risk_level": "中风险"}
+```'''
+    issues2 = reviewer._parse_segment_response(response2)
+    assert len(issues2) == 1, f"Expected 1 issue, got {len(issues2)}"
+    print("[PASS] test_parse_segment_response: JSON object parsed correctly")
+
+    # 测试3: 无明显问题
+    response3 = "无明显问题"
+    issues3 = reviewer._parse_segment_response(response3)
+    assert len(issues3) == 0, f"Expected 0 issues, got {len(issues3)}"
+    print("[PASS] test_parse_segment_response: no-issue response handled correctly")
+
+    # 测试4: 空响应
+    response4 = ""
+    issues4 = reviewer._parse_segment_response(response4)
+    assert len(issues4) == 0, f"Expected 0 issues, got {len(issues4)}"
+    print("[PASS] test_parse_segment_response: empty response handled correctly")
+
+    # 测试5: JSON 中嵌套"无明显问题"（reason 字段中）
+    response5 = '''```json
+[{"issue_point": "错别字", "location": "位置1", "suggestion": "将A改为B", "reason": "原文无明显问题但实际有错", "risk_level": "中风险"}]
+```'''
+    issues5 = reviewer._parse_segment_response(response5)
+    assert len(issues5) == 1, f"Expected 1 issue, got {len(issues5)}"
+    print("[PASS] test_parse_segment_response: JSON with 'no-issue' keyword in reason parsed correctly")
+
+
+def test_deduplicate_issues():
+    """测试去重逻辑"""
+    reviewer = GrammarCheckReviewer()
+
+    issues = [
+        {"issue_point": "错别字", "location": "位置1", "suggestion": "将'混泥土'改为'混凝土'", "reason": "原因", "risk_level": "中风险"},
+        {"issue_point": "错别字", "location": "位置1", "suggestion": "将'混泥土'改为'混凝土'", "reason": "原因重复", "risk_level": "中风险"},  # 精确重复
+        {"issue_point": "错别字", "location": "位置2", "suggestion": "将'珩架梁'改为'桁架梁'", "reason": "原因", "risk_level": "中风险"},
+        {"issue_point": "错别字", "location": "位置3", "suggestion": "将'卷拨'改为'卷扬'", "reason": "原因", "risk_level": "中风险"},
+        {"issue_point": "无明显问题", "location": "位置4", "suggestion": "无明显问题", "reason": "原因", "risk_level": "低风险"},  # 无效条目
+        {"issue_point": "错别字", "location": "位置5", "suggestion": "将'不和'改为'不得'", "reason": "原因", "risk_level": ""},  # risk_level 为空
+        {"issue_point": "错别字", "location": "位置6", "suggestion": "将'千斤项'改为'千斤顶'", "reason": "原因", "risk_level": "高风险"},  # 有效
+    ]
+
+    unique = reviewer._deduplicate_issues(issues)
+
+    # 应该保留: 混泥土→混凝土, 珩架梁→桁架梁, 卷拨→卷扬, 千斤项→千斤顶 = 4个
+    assert len(unique) == 4, f"Expected 4 unique issues, got {len(unique)}: {[i['suggestion'] for i in unique]}"
+    print(f"[PASS] test_deduplicate_issues: {len(issues)} -> {len(unique)} issues")
+
+    # 验证过滤了无效条目
+    suggestions = [i["suggestion"] for i in unique]
+    assert "无明显问题" not in suggestions, "Should filter out 'no-issue' suggestions"
+    assert all(i["risk_level"] for i in unique), "Should filter out empty risk_level"
+    print("[PASS] test_deduplicate_issues: invalid entries filtered correctly")
+
+
+def test_split_trigger():
+    """测试切分触发条件"""
+    chunks = load_test_chunks()
+
+    # 统计哪些 chunk 会触发切分
+    trigger_count = 0
+    no_trigger_count = 0
+    for i, chunk in enumerate(chunks):
+        content = chunk.get("content", "")
+        if len(content) > SPLIT_THRESHOLD:
+            trigger_count += 1
+            segments = split_text_with_overlap(
+                content,
+                min_length=SEGMENT_MIN_LENGTH,
+                target_length=SEGMENT_TARGET_LENGTH,
+                overlap=SEGMENT_OVERLAP,
+            )
+            print(f"  Chunk[{i}] len={len(content)} -> {len(segments)} segments")
+        else:
+            no_trigger_count += 1
+
+    print(f"[PASS] test_split_trigger: {trigger_count} chunks will be split, {no_trigger_count} chunks will not")
+
+
+async def test_full_split_review():
+    """完整测试：对 Chunk 24 进行切分审查"""
+    chunks = load_test_chunks()
+    chunk24 = chunks[24]["content"]
+
+    print(f"\nChunk 24 length: {len(chunk24)}")
+    print(f"Split threshold: {SPLIT_THRESHOLD}")
+
+    reviewer = GrammarCheckReviewer()
+
+    start_time = time.time()
+    response = await reviewer._check_grammar_with_split(
+        trace_id="test_split_chunk24",
+        review_content=chunk24,
+        enable_thinking=False,
+    )
+    elapsed = time.time() - start_time
+
+    print(f"\nSplit review completed in {elapsed:.2f}s")
+    print(f"Response length: {len(response)}")
+
+    # 解析响应验证
+    if response == "无明显问题":
+        print("[INFO] No issues found after split review")
+    else:
+        try:
+            issues = json.loads(response)
+            print(f"[PASS] test_full_split_review: {len(issues)} unique issues found")
+            for i, issue in enumerate(issues):
+                print(f"  [{i+1}] {issue.get('issue_point', '')}: {issue.get('suggestion', '')[:50]}...")
+        except json.JSONDecodeError:
+            print(f"[FAIL] Response is not valid JSON: {response[:200]}...")
+
+    # 保存结果
+    output_file = os.path.join(
+        project_root,
+        "utils_test", "Grammar_Check_Test", "full_scan_results",
+        "chunk24_split_review_new.json"
+    )
+    os.makedirs(os.path.dirname(output_file), exist_ok=True)
+    with open(output_file, "w", encoding="utf-8") as f:
+        f.write(response)
+    print(f"Results saved to: {output_file}")
+
+
+if __name__ == "__main__":
+    print("=" * 60)
+    print("Testing grammar_check split logic")
+    print("=" * 60)
+
+    # 同步测试
+    test_parse_segment_response()
+    test_deduplicate_issues()
+    test_split_trigger()
+
+    # 异步测试
+    print("\n" + "=" * 60)
+    print("Running full split review test (async)...")
+    print("=" * 60)
+    asyncio.run(test_full_split_review())