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

config/model_setting.yaml

@@ -118,6 +118,12 @@ model_settings:
     enable_thinking: false
     description: "时效规范审查抽取（目录提取），蜀天35B"
 
+  # 目录完整性审查（对比实际目录与标准目录）
+  catalog_integrity_review:
+    model: shutian_qwen3_5_122b
+    enable_thinking: true
+    description: "目录完整性审查，对比OCR提取目录与标准目录，找出缺失项，蜀天35B"
+
   # Embedding 模型（用于相似度计算）
   embedding:
     model: shutian_qwen3_embed # 或 lq_qwen3_8b_emd 

core/construction_review/component/minimal_pipeline/catalog_reviewer.py

@@ -0,0 +1,420 @@
+"""
+目录完整性审查模块
+
+使用LLM对比实际目录（OCR提取）和标准目录，找出缺失项。
+"""
+
+import json
+import re
+from pathlib import Path
+from typing import Dict, Any, List, Optional
+
+import yaml
+
+from foundation.observability.logger.loggering import review_logger as logger
+
+
+class CatalogReviewer:
+    """目录审查器"""
+
+    # 默认标准目录模板路径
+    DEFAULT_TEMPLATE_PATH = Path(__file__).parent.parent / 'doc_worker' / 'config' / 'StandardCatalogTemplate.yaml'
+
+    # JSON 格式示例（避免 f-string 嵌套问题，单独定义）
+    _JSON_EXAMPLE_TEMPLATE = '''{
+  "details": {
+    "name": "catalog_check",
+    "response": [
+      {
+        "check_item": "check_completeness",
+        "chapter_code": "catalog",
+        "check_item_code": "catalog_check_completeness",
+        "check_result": {
+          "issue_point": "【一级缺失】第四章 施工工艺技术",
+          "location": "目录页",
+          "suggestion": "建议补充'第四章 施工工艺技术'章节",
+          "reason": "目录页缺少该章节",
+          "risk_level": "高风险"
+        },
+        "exist_issue": true,
+        "risk_info": {"risk_level": "high"}
+      },
+      {
+        "check_item": "check_completeness",
+        "chapter_code": "catalog",
+        "check_item_code": "catalog_check_completeness",
+        "check_result": {
+          "issue_point": "【一级缺失】第十章 其他资料",
+          "location": "目录页",
+          "suggestion": "建议补充'第十章 其他资料'章节",
+          "reason": "目录页缺少该章节",
+          "risk_level": "高风险"
+        },
+        "exist_issue": true,
+        "risk_info": {"risk_level": "high"}
+      },
+      {
+        "check_item": "check_completeness",
+        "chapter_code": "catalog",
+        "check_item_code": "catalog_check_completeness",
+        "check_result": {
+          "issue_point": "【二级缺失】第一章 编制依据 - 四、编制原则",
+          "location": "第一章",
+          "suggestion": "建议补充'四、编制原则'",
+          "reason": "第一章缺少该二级目录",
+          "risk_level": "中风险"
+        },
+        "exist_issue": true,
+        "risk_info": {"risk_level": "medium"}
+      }
+    ],
+    "review_location_label": "目录完整性审查",
+    "chapter_code": "catalog"
+  },
+  "success": true
+}'''
+
+    def __init__(self, template_path: Optional[Path] = None):
+        self.template_path = template_path or self.DEFAULT_TEMPLATE_PATH
+        self.standard_text = self._load_standard_template()
+
+    def _load_standard_template(self) -> str:
+        """加载标准目录模板"""
+        try:
+            with open(self.template_path, 'r', encoding='utf-8') as f:
+                template = yaml.safe_load(f)
+            return template.get('text_template', '')
+        except Exception as e:
+            logger.warning(f"[CatalogReviewer] 加载标准模板失败: {e}")
+            return self._default_template()
+
+    def _default_template(self) -> str:
+        """默认标准目录模板"""
+        return """第一章 编制依据
+一、法律法规
+二、标准规范
+三、文件制度
+四、编制原则
+五、编制范围
+
+第二章 工程概况
+一、设计概况
+二、工程地质与水文气象
+三、周边环境
+四、施工平面及立面布置
+五、施工要求和技术保证条件
+六、风险辨识与分级
+七、参建各方责任主体单位
+
+第三章 施工计划
+一、施工进度计划
+二、施工材料计划
+三、施工设备计划
+四、劳动力计划
+五、安全生产费用使用计划
+
+第四章 施工工艺技术
+一、主要施工方法概述
+二、技术参数
+三、工艺流程
+四、施工准备
+五、施工方法及操作要求
+六、检查要求
+
+第五章 安全保证措施
+一、安全保证体系
+二、组织保证措施
+三、技术保证措施
+四、监测监控措施
+五、应急处置措施
+
+第六章 质量保证措施
+一、质量保证体系
+二、质量目标
+三、工程创优规划
+四、质量控制程序与具体措施
+
+第七章 环境保证措施
+一、环境保证体系
+二、环境保护组织机构
+三、环境保护及文明施工措施
+
+第八章 施工管理及作业人员配备与分工
+一、施工管理人员
+二、专职安全生产管理人员
+三、其他作业人员
+
+第九章 验收要求
+一、验收标准
+二、验收程序
+三、验收内容
+四、验收时间
+五、验收人员
+
+第十章 其他资料
+一、计算书
+二、相关施工图纸
+三、附图附表
+四、编制及审核人员情况"""
+
+    async def review(self, actual_catalog_text: str, trace_id_idx: str = "") -> Dict[str, Any]:
+        """
+        审查目录完整性
+
+        Args:
+            actual_catalog_text: 实际目录文本（标准格式）
+            trace_id_idx: 追踪ID索引
+
+        Returns:
+            对齐 check_completeness 格式的结果字典
+        """
+        import time
+        start_time = time.time()
+
+        try:
+            from foundation.ai.agent.generate.model_generate import generate_model_client
+
+            prompt = self._build_prompt(actual_catalog_text)
+
+            # 重试机制：最多3次
+            max_retries = 3
+            last_error = None
+
+            for attempt in range(max_retries):
+                try:
+                    logger.info(f"[DEBUG][CatalogReviewer] 调用模型 catalog_integrity_review，第 {attempt + 1} 次尝试")
+
+                    # 使用 generate_model_client 调用模型
+                    content = await generate_model_client.get_model_generate_invoke(
+                        trace_id=f"{trace_id_idx or 'catalog_review'}_attempt{attempt}",
+                        system_prompt="你是一位施工方案文档审查专家，负责对比实际目录和标准目录，找出缺失项。请按JSON格式输出最终结果。",
+                        user_prompt=prompt,
+                        function_name="catalog_integrity_review",
+                        timeout=120
+                    )
+
+                    logger.info(f"[DEBUG][CatalogReviewer] 模型返回，开始解析")
+                    logger.info(f"[DEBUG][CatalogReviewer] content length: {len(content)}")
+
+                    # 直接解析 LLM 返回的 check_completeness 格式
+                    result = self._extract_json(content)
+                    if result and "details" in result:
+                        logger.info(f"[DEBUG][CatalogReviewer] 成功解析 LLM 返回的格式")
+                        execution_time = time.time() - start_time
+                        return {
+                            "details": result["details"],
+                            "success": result.get("success", True),
+                            "execution_time": execution_time
+                        }
+                    else:
+                        logger.warning(f"[DEBUG][CatalogReviewer] 第 {attempt + 1} 次：LLM 返回格式不正确")
+                        last_error = "LLM 返回格式不正确"
+                        if attempt < max_retries - 1:
+                            import asyncio
+                            await asyncio.sleep(1)  # 短暂等待后重试
+
+                except Exception as e:
+                    logger.warning(f"[DEBUG][CatalogReviewer] 第 {attempt + 1} 次调用失败: {e}")
+                    last_error = str(e)
+                    if attempt < max_retries - 1:
+                        import asyncio
+                        await asyncio.sleep(1)
+
+            # 所有重试都失败
+            raise ValueError(f"重试 {max_retries} 次后仍失败: {last_error}")
+
+        except Exception as e:
+            logger.error(f"[CatalogReviewer] LLM审查失败（已重试3次）: {e}")
+            execution_time = time.time() - start_time
+            return {
+                "details": {
+                    "name": "catalog_check",
+                    "response": [{
+                        "check_item": "check_completeness",
+                        "chapter_code": "catalog",
+                        "check_item_code": "catalog_check_completeness",
+                        "check_result": {
+                            "issue_point": "目录审查失败",
+                            "location": "目录页",
+                            "suggestion": "请检查OCR识别结果或手动确认目录",
+                            "reason": str(e),
+                            "risk_level": "中风险"
+                        },
+                        "exist_issue": True,
+                        "risk_info": {"risk_level": "medium"}
+                    }],
+                    "review_location_label": "目录完整性审查",
+                    "chapter_code": "catalog",
+                    "original_content": f"审查失败: {str(e)}"
+                },
+                "success": False,
+                "execution_time": execution_time
+            }
+
+    def _build_prompt(self, actual_catalog_text: str) -> str:
+        """构建审查Prompt"""
+        json_example = self._JSON_EXAMPLE_TEMPLATE
+
+        # 基础 JSON 模板（使用单引号字符串避免 f-string 转义问题）
+        base_template = '''{
+  "details": {
+    "name": "catalog_check",
+    "response": [
+      {
+        "check_item": "check_completeness",
+        "chapter_code": "catalog",
+        "check_item_code": "catalog_check_completeness",
+        "check_result": {
+          "issue_point": "【一级缺失】xxx",
+          "location": "目录页",
+          "suggestion": "建议补充'xxx'章节",
+          "reason": "简要说明",
+          "risk_level": "高风险"
+        },
+        "exist_issue": true,
+        "risk_info": {"risk_level": "high"}
+      }
+    ],
+    "review_location_label": "目录完整性审查",
+    "chapter_code": "catalog"
+  },
+  "success": true
+}'''
+
+        return f"""你是一位施工方案文档审查专家。请对比【实际目录】和【标准目录】，找出缺失项。
+
+## 审查原则
+1. **语义匹配**：实际目录与标准目录含义相同即算匹配，不要求文字完全一致
+2. **常见同义表述**（示例）：
+   - "编制依据" ≈ "方案编制依据" ≈ "编制原则及依据"
+   - "工程概况" ≈ "工程基本情况" ≈ "项目概况"
+   - "施工计划" ≈ "施工进度计划" ≈ "施工安排"
+   - "法律法规" ≈ "相关法律" ≈ "法规依据"
+3. **容错范围**：
+   - 一级标题必须严格对应（如"编制依据"不能变成"引用标准"）
+   - 二级标题允许一定变通，但核心含义必须一致
+
+## 实际目录（来自OCR识别）
+```
+{actual_catalog_text}
+```
+
+## 标准目录（必须包含的完整结构）
+```
+{self.standard_text}
+```
+
+## 输出规则
+1. **一级缺失判定**：实际目录中完全没有对应的章，或章节标题完全不匹配
+2. **二级缺失判定**：只有当父级一级目录**存在**时，才检查其下的二级目录是否缺失
+3. **重要**：如果某个一级目录缺失，**不要报告**该章节下的二级目录缺失（避免重复提醒）
+
+## 输出要求
+**重要：最终答案只输出 JSON，不要添加 markdown 代码块标记（```json）。**
+
+请直接输出 check_completeness 格式的 JSON 结果：
+{base_template}
+
+**重要输出规则**：
+1. **每个缺失项必须单独输出**：一级缺失和二级缺失要分开，不同的缺失项也要分开
+2. **禁止合并**：不要将多个缺失项写在一个 `issue_point` 里
+3. **列表格式**：`response` 必须是一个列表，每个缺失项是列表中的一个独立对象
+
+**正确示例**（多个缺失项分开）：
+```json
+{json_example}
+```
+
+**风险等级规则**：
+- 一级缺失：risk_level 为 "高风险", risk_info.risk_level 为 "high"
+- 二级缺失：risk_level 为 "中风险", risk_info.risk_level 为 "medium"
+- 如无缺失，response 中放一条 "issue_point": "【目录完整】一二级目录结构完整", "exist_issue": false
+"""
+
+    def _extract_json(self, content: str) -> Optional[Dict[str, Any]]:
+        """从LLM响应中提取JSON，增强健壮性（支持思考模式输出）"""
+        try:
+            # 清理内容：移除 markdown 代码块标记
+            content = content.strip()
+            original_preview = content[:500]
+
+            # 移除 markdown 代码块
+            content = re.sub(r'^```json\s*', '', content, flags=re.IGNORECASE | re.MULTILINE)
+            content = re.sub(r'\s*```\s*$', '', content, flags=re.MULTILINE)
+            content = re.sub(r'^```\s*', '', content, flags=re.MULTILINE)
+
+            # 处理思考模式输出：跳过思考部分，提取最终答案
+            # 检查 <think>...</think> 标签 (Qwen3.5 思考模式标准格式)
+            think_end = content.find("</think>")
+            if think_end != -1:
+                # 提取 </think> 之后的内容
+                content = content[think_end + len("</think>"):].strip()
+                logger.debug(f"[CatalogReviewer] 检测到 <think> 标签，从 </think> 后提取内容，长度: {len(content)}")
+
+            # 找到第一个 { 开始的位置
+            json_start = content.find('{')
+            if json_start == -1:
+                logger.warning(f"[CatalogReviewer] 未找到 JSON 开始标记 '{{'")
+                return None
+            content = content[json_start:]
+
+            # 找到最后一个 } 结束的位置
+            json_end = content.rfind('}')
+            if json_end == -1:
+                logger.warning(f"[CatalogReviewer] 未找到 JSON 结束标记 '}}'")
+                return None
+            content = content[:json_end + 1]
+
+            # 尝试直接解析
+            try:
+                return json.loads(content)
+            except json.JSONDecodeError as e:
+                logger.debug(f"[CatalogReviewer] 直接解析失败: {e}")
+
+            # 尝试修复常见问题后重新解析
+            fixed_content = self._fix_json_content(content)
+            try:
+                return json.loads(fixed_content)
+            except json.JSONDecodeError as e:
+                logger.debug(f"[CatalogReviewer] 修复后解析失败: {e}")
+
+            logger.error(f"[CatalogReviewer] JSON解析失败")
+            logger.error(f"[CatalogReviewer] 原始内容: {original_preview}")
+            return None
+
+        except Exception as e:
+            logger.error(f"[CatalogReviewer] JSON解析异常: {e}")
+            logger.error(f"[CatalogReviewer] 内容前500字: {content[:500]}")
+            return None
+
+    def _fix_json_content(self, content: str) -> str:
+        """尝试修复常见的 JSON 格式问题"""
+        # 1. 移除多余的空白和换行
+        content = content.strip()
+
+        # 2. 修复属性名未加引号的问题（简单情况）
+        # 将 { key: value } 转换为 { "key": value }
+        content = re.sub(r'(\{|,)\s*([a-zA-Z_][a-zA-Z0-9_]*)\s*:', r'\1"\2":', content)
+
+        # 3. 修复尾随逗号
+        content = re.sub(r',\s*([}\]])', r'\1', content)
+
+        # 4. 修复单引号为双引号
+        content = content.replace("'", '"')
+
+        return content
+
+
+async def review_catalog_integrity(actual_catalog_text: str, template_path: Optional[Path] = None) -> Dict[str, Any]:
+    """
+    便捷函数：审查目录完整性
+
+    Args:
+        actual_catalog_text: 实际目录文本（标准格式）
+        template_path: 可选的自定义模板路径
+
+    Returns:
+        审查结果字典
+    """
+    reviewer = CatalogReviewer(template_path)
+    return await reviewer.review(actual_catalog_text)

utils_test/Model_Test/test_thinking_vs_nonthinking.py

@@ -0,0 +1,554 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+"""
+测试 qwen3.5 模型在思考模式与非思考模式下的输出区别
+
+测试场景：目录完整性审查（catalog_integrity_review）
+模型：shutian_qwen3_5_122b
+
+运行方式:
+    cd D:/wx_work/sichuan_luqiao/LQAgentPlatform
+    python utils_test/Model_Test/test_thinking_vs_nonthinking.py
+"""
+
+import asyncio
+import time
+import sys
+import json
+from pathlib import Path
+from datetime import datetime
+
+# 添加项目根目录到 Python 路径
+project_root = Path(__file__).parent.parent.parent
+sys.path.insert(0, str(project_root))
+
+from foundation.ai.agent.generate.model_generate import generate_model_client
+from foundation.observability.logger.loggering import review_logger as logger
+
+
+# 测试用的实际目录文本（模拟OCR识别结果）
+TEST_ACTUAL_CATALOG = """第一章 编制依据
+一、法律法规
+二、标准规范
+三、文件制度
+
+第二章 工程概况
+一、设计概况
+二、工程地质与水文气象
+三、周边环境
+
+第三章 施工计划
+一、施工进度计划
+二、施工材料计划
+
+第四章 施工工艺技术
+一、主要施工方法概述
+二、技术参数
+三、工艺流程
+
+第五章 安全保证措施
+一、安全保证体系
+二、组织保证措施
+
+第六章 质量保证措施
+一、质量保证体系
+
+第七章 施工管理及作业人员配备与分工
+一、施工管理人员
+
+第八章 验收要求
+一、验收标准"""
+
+# 标准目录模板
+STANDARD_CATALOG = """第一章 编制依据
+一、法律法规
+二、标准规范
+三、文件制度
+四、编制原则
+五、编制范围
+
+第二章 工程概况
+一、设计概况
+二、工程地质与水文气象
+三、周边环境
+四、施工平面及立面布置
+五、施工要求和技术保证条件
+六、风险辨识与分级
+七、参建各方责任主体单位
+
+第三章 施工计划
+一、施工进度计划
+二、施工材料计划
+三、施工设备计划
+四、劳动力计划
+五、安全生产费用使用计划
+
+第四章 施工工艺技术
+一、主要施工方法概述
+二、技术参数
+三、工艺流程
+四、施工准备
+五、施工方法及操作要求
+六、检查要求
+
+第五章 安全保证措施
+一、安全保证体系
+二、组织保证措施
+三、技术保证措施
+四、监测监控措施
+五、应急处置措施
+
+第六章 质量保证措施
+一、质量保证体系
+二、质量目标
+三、工程创优规划
+四、质量控制程序与具体措施
+
+第七章 环境保证措施
+一、环境保证体系
+二、环境保护组织机构
+三、环境保护及文明施工措施
+
+第八章 施工管理及作业人员配备与分工
+一、施工管理人员
+二、专职安全生产管理人员
+三、其他作业人员
+
+第九章 验收要求
+一、验收标准
+二、验收程序
+三、验收内容
+四、验收时间
+五、验收人员
+
+第十章 其他资料
+一、计算书
+二、相关施工图纸
+三、附图附表
+四、编制及审核人员情况"""
+
+# System Prompt（更新后，允许思考）
+SYSTEM_PROMPT = "你是一位施工方案文档审查专家，负责对比实际目录和标准目录，找出缺失项。请按JSON格式输出最终结果。"
+
+# User Prompt（更新后，允许思考）
+USER_PROMPT_TEMPLATE = """你是一位施工方案文档审查专家。请对比【实际目录】和【标准目录】，找出缺失项。
+
+## 审查原则
+1. **语义匹配**：实际目录与标准目录含义相同即算匹配，不要求文字完全一致
+2. **常见同义表述**（示例）：
+   - "编制依据" ≈ "方案编制依据" ≈ "编制原则及依据"
+   - "工程概况" ≈ "工程基本情况" ≈ "项目概况"
+   - "施工计划" ≈ "施工进度计划" ≈ "施工安排"
+   - "法律法规" ≈ "相关法律" ≈ "法规依据"
+3. **容错范围**：
+   - 一级标题必须严格对应（如"编制依据"不能变成"引用标准"）
+   - 二级标题允许一定变通，但核心含义必须一致
+
+## 实际目录（来自OCR识别）
+```
+{actual_catalog}
+```
+
+## 标准目录（必须包含的完整结构）
+```
+{standard_catalog}
+```
+
+## 输出规则
+1. **一级缺失判定**：实际目录中完全没有对应的章，或章节标题完全不匹配
+2. **二级缺失判定**：只有当父级一级目录**存在**时，才检查其下的二级目录是否缺失
+3. **重要**：如果某个一级目录缺失，**不要报告**该章节下的二级目录缺失（避免重复提醒）
+
+## 输出要求
+**重要：最终答案只输出 JSON，不要添加 markdown 代码块标记（```json）。**
+
+请直接输出 check_completeness 格式的 JSON 结果：
+{{
+  "details": {{
+    "name": "catalog_check",
+    "response": [
+      {{
+        "check_item": "check_completeness",
+        "chapter_code": "catalog",
+        "check_item_code": "catalog_check_completeness",
+        "check_result": {{
+          "issue_point": "【一级缺失】xxx",
+          "location": "目录页",
+          "suggestion": "建议补充'xxx'章节",
+          "reason": "简要说明",
+          "risk_level": "高风险"
+        }},
+        "exist_issue": true,
+        "risk_info": {{"risk_level": "high"}}
+      }}
+    ],
+    "review_location_label": "目录完整性审查",
+    "chapter_code": "catalog"
+  }},
+  "success": true
+}}
+
+**注意**：
+- 一级缺失：risk_level 为 "高风险", risk_info.risk_level 为 "high"
+- 二级缺失：risk_level 为 "中风险", risk_info.risk_level 为 "medium"
+- 如无缺失，response 中放一条 "issue_point": "【目录完整】一二级目录结构完整", "exist_issue": false
+"""
+
+
+def build_prompt(actual_catalog: str, standard_catalog: str) -> str:
+    """构建测试用的 prompt"""
+    return USER_PROMPT_TEMPLATE.format(
+        actual_catalog=actual_catalog,
+        standard_catalog=standard_catalog
+    )
+
+
+async def test_with_mode(mode_name: str, enable_thinking: bool, output_dir: Path) -> dict:
+    """
+    测试指定模式
+
+    Args:
+        mode_name: 模式名称（用于输出）
+        enable_thinking: 是否启用思考模式
+        output_dir: 输出目录
+
+    Returns:
+        测试结果字典
+    """
+    print(f"\n{'='*70}")
+    print(f" 测试模式: {mode_name} (enable_thinking={enable_thinking})")
+    print(f"{'='*70}")
+
+    model_name = "shutian_qwen3_5_122b"
+    trace_id = f"test_{mode_name}_{int(time.time())}"
+
+    # 构建 prompt
+    user_prompt = build_prompt(TEST_ACTUAL_CATALOG, STANDARD_CATALOG)
+
+    # 记录开始时间
+    start_time = time.time()
+
+    try:
+        # 调用模型
+        print(f"⏳ 正在调用模型 {model_name}...")
+        print(f"   enable_thinking={enable_thinking}")
+        print(f"   trace_id={trace_id}")
+
+        response = await generate_model_client.get_model_generate_invoke(
+            trace_id=trace_id,
+            system_prompt=SYSTEM_PROMPT,
+            user_prompt=user_prompt,
+            model_name=model_name,
+            enable_thinking=enable_thinking,
+            timeout=180
+        )
+
+        elapsed_time = time.time() - start_time
+
+        print(f"✅ 调用成功")
+        print(f"   响应时间: {elapsed_time:.2f}s")
+        print(f"   响应长度: {len(response)} 字符")
+
+        # 保存完整响应到文件
+        output_file = output_dir / f"{mode_name}_response.txt"
+        with open(output_file, 'w', encoding='utf-8') as f:
+            f.write(f"模式: {mode_name}\n")
+            f.write(f"enable_thinking: {enable_thinking}\n")
+            f.write(f"模型: {model_name}\n")
+            f.write(f"响应时间: {elapsed_time:.2f}s\n")
+            f.write(f"响应长度: {len(response)} 字符\n")
+            f.write(f"trace_id: {trace_id}\n")
+            f.write("="*70 + "\n")
+            f.write("原始响应内容:\n")
+            f.write(response)
+
+        print(f"   完整响应已保存: {output_file}")
+
+        # 分析响应特征
+        analysis = analyze_response(response)
+
+        # 保存分析结果
+        analysis_file = output_dir / f"{mode_name}_analysis.json"
+        with open(analysis_file, 'w', encoding='utf-8') as f:
+            json.dump({
+                "mode": mode_name,
+                "enable_thinking": enable_thinking,
+                "model": model_name,
+                "elapsed_time": elapsed_time,
+                "response_length": len(response),
+                "trace_id": trace_id,
+                "analysis": analysis
+            }, f, ensure_ascii=False, indent=2)
+
+        print(f"   分析结果已保存: {analysis_file}")
+
+        return {
+            "success": True,
+            "mode": mode_name,
+            "enable_thinking": enable_thinking,
+            "elapsed_time": elapsed_time,
+            "response_length": len(response),
+            "analysis": analysis,
+            "response_preview": response[:500] + "..." if len(response) > 500 else response
+        }
+
+    except Exception as e:
+        elapsed_time = time.time() - start_time
+        print(f"❌ 调用失败: {e}")
+
+        # 保存错误信息
+        error_file = output_dir / f"{mode_name}_error.txt"
+        with open(error_file, 'w', encoding='utf-8') as f:
+            f.write(f"模式: {mode_name}\n")
+            f.write(f"enable_thinking: {enable_thinking}\n")
+            f.write(f"错误信息: {str(e)}\n")
+            f.write(f"响应时间: {elapsed_time:.2f}s\n")
+
+        return {
+            "success": False,
+            "mode": mode_name,
+            "enable_thinking": enable_thinking,
+            "elapsed_time": elapsed_time,
+            "error": str(e)
+        }
+
+
+def analyze_response(response: str) -> dict:
+    """分析响应内容的特征"""
+    analysis = {
+        "has_thinking_process": False,
+        "has_answer_marker": False,
+        "has_json_structure": False,
+        "thinking_length": 0,
+        "json_start_index": -1,
+        "detected_patterns": []
+    }
+
+    # 检测思考过程标记
+    thinking_markers = [
+        "Thinking Process:",
+        "思考过程",
+        "1.  **Analyze",
+        "1. **分析",
+        "让我",
+        "我需要",
+        "第一步",
+        "首先",
+    ]
+
+    for marker in thinking_markers:
+        if marker in response:
+            analysis["detected_patterns"].append(f"包含标记: {marker}")
+            analysis["has_thinking_process"] = True
+            break
+
+    # 检测答案标记
+    answer_markers = [
+        "Answer:\n",
+        "Final Answer:\n",
+        "**Answer:**\n",
+        "**Final Answer:**\n",
+    ]
+
+    for marker in answer_markers:
+        if marker in response:
+            analysis["has_answer_marker"] = True
+            analysis["detected_patterns"].append(f"答案标记: {marker}")
+            break
+
+    # 检测 JSON 结构
+    json_start = response.find('{')
+    json_end = response.rfind('}')
+
+    if json_start != -1 and json_end != -1 and json_end > json_start:
+        analysis["has_json_structure"] = True
+        analysis["json_start_index"] = json_start
+
+        # 尝试计算思考部分长度
+        if json_start > 0:
+            analysis["thinking_length"] = json_start
+
+    # 检测响应结构特征
+    lines = response.split('\n')
+    analysis["total_lines"] = len(lines)
+
+    # 检测是否有明显的思考/答案分隔
+    for i, line in enumerate(lines):
+        if line.strip() in ["Answer:", "Final Answer:", "**Answer:**", "**Final Answer:**"]:
+            analysis["answer_line_number"] = i + 1
+            break
+
+    return analysis
+
+
+def print_comparison(result_thinking: dict, result_non_thinking: dict):
+    """打印两种模式的对比结果"""
+    print(f"\n{'='*70}")
+    print(" 对比分析结果")
+    print(f"{'='*70}")
+
+    # 基本信息对比
+    print("\n【基本信息对比】")
+    print(f"{'指标':<30} {'思考模式':>15} {'非思考模式':>15}")
+    print("-" * 70)
+    print(f"{'响应时间':<30} {result_thinking['elapsed_time']:>14.2f}s {result_non_thinking['elapsed_time']:>14.2f}s")
+    print(f"{'响应长度':<30} {result_thinking['response_length']:>14,} {result_non_thinking['response_length']:>14,}")
+
+    if result_thinking['success'] and result_non_thinking['success']:
+        time_diff = result_thinking['elapsed_time'] - result_non_thinking['elapsed_time']
+        length_diff = result_thinking['response_length'] - result_non_thinking['response_length']
+
+        print(f"{'时间差异':<30} {f'+{time_diff:.2f}s' if time_diff > 0 else f'{time_diff:.2f}s':>15}")
+        print(f"{'长度差异':<30} {f'+{length_diff:,}' if length_diff > 0 else f'{length_diff:,}':>15,}")
+
+        # 特征对比
+        print("\n【内容特征对比】")
+        analysis_t = result_thinking['analysis']
+        analysis_nt = result_non_thinking['analysis']
+
+        print(f"{'指标':<30} {'思考模式':>15} {'非思考模式':>15}")
+        print("-" * 70)
+        print(f"{'包含思考过程':<30} {'是' if analysis_t['has_thinking_process'] else '否':>15} {'是' if analysis_nt['has_thinking_process'] else '否':>15}")
+        print(f"{'包含答案标记':<30} {'是' if analysis_t['has_answer_marker'] else '否':>15} {'是' if analysis_nt['has_answer_marker'] else '否':>15}")
+        print(f"{'包含JSON结构':<30} {'是' if analysis_t['has_json_structure'] else '否':>15} {'是' if analysis_nt['has_json_structure'] else '否':>15}")
+        print(f"{'思考部分长度':<30} {analysis_t['thinking_length']:>14,} {analysis_nt['thinking_length']:>14,}")
+        print(f"{'总行数':<30} {analysis_t.get('total_lines', 0):>14,} {analysis_nt.get('total_lines', 0):>14,}")
+
+        # 检测到的模式
+        print("\n【思考模式 - 检测到的特征】")
+        for pattern in analysis_t.get('detected_patterns', []):
+            print(f"  - {pattern}")
+
+        print("\n【非思考模式 - 检测到的特征】")
+        for pattern in analysis_nt.get('detected_patterns', []):
+            print(f"  - {pattern}")
+
+        # 内容预览
+        print("\n【思考模式 - 响应预览】")
+        print(result_thinking.get('response_preview', 'N/A')[:300])
+
+        print("\n【非思考模式 - 响应预览】")
+        print(result_non_thinking.get('response_preview', 'N/A')[:300])
+
+    else:
+        print("\n⚠️ 部分测试失败，无法完成完整对比")
+
+
+def save_summary_report(result_thinking: dict, result_non_thinking: dict, output_dir: Path):
+    """保存汇总报告"""
+    report_file = output_dir / "comparison_report.md"
+
+    with open(report_file, 'w', encoding='utf-8') as f:
+        f.write("# Qwen3.5 思考模式 vs 非思考模式 测试报告\n\n")
+        f.write(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
+
+        f.write("## 测试配置\n\n")
+        f.write(f"- 模型: `shutian_qwen3_5_122b`\n")
+        f.write(f"- 测试场景: 目录完整性审查 (catalog_integrity_review)\n")
+        f.write(f"- Prompt: 允许思考，要求最终输出 JSON\n\n")
+
+        f.write("## 性能对比\n\n")
+        f.write("| 指标 | 思考模式 | 非思考模式 | 差异 |\n")
+        f.write("|------|----------|------------|------|\n")
+
+        if result_thinking['success'] and result_non_thinking['success']:
+            time_t = result_thinking['elapsed_time']
+            time_nt = result_non_thinking['elapsed_time']
+            len_t = result_thinking['response_length']
+            len_nt = result_non_thinking['response_length']
+
+            f.write(f"| 响应时间 | {time_t:.2f}s | {time_nt:.2f}s | {time_t - time_nt:+.2f}s |\n")
+            f.write(f"| 响应长度 | {len_t:,} | {len_nt:,} | {len_t - len_nt:+,} |\n")
+
+        f.write("\n## 内容特征对比\n\n")
+
+        if result_thinking['success'] and result_non_thinking['success']:
+            analysis_t = result_thinking['analysis']
+            analysis_nt = result_non_thinking['analysis']
+
+            f.write("| 特征 | 思考模式 | 非思考模式 |\n")
+            f.write("|------|----------|------------|\n")
+            f.write(f"| 包含思考过程 | {'✅ 是' if analysis_t['has_thinking_process'] else '❌ 否'} | {'✅ 是' if analysis_nt['has_thinking_process'] else '❌ 否'} |\n")
+            f.write(f"| 包含答案标记 | {'✅ 是' if analysis_t['has_answer_marker'] else '❌ 否'} | {'✅ 是' if analysis_nt['has_answer_marker'] else '❌ 否'} |\n")
+            f.write(f"| 包含JSON结构 | {'✅ 是' if analysis_t['has_json_structure'] else '❌ 否'} | {'✅ 是' if analysis_nt['has_json_structure'] else '❌ 否'} |\n")
+            f.write(f"| 思考部分长度 | {analysis_t['thinking_length']:,} | {analysis_nt['thinking_length']:,} |\n")
+
+            f.write("\n### 思考模式检测到的特征\n\n")
+            for pattern in analysis_t.get('detected_patterns', []):
+                f.write(f"- {pattern}\n")
+
+            f.write("\n### 非思考模式检测到的特征\n\n")
+            for pattern in analysis_nt.get('detected_patterns', []):
+                f.write(f"- {pattern}\n")
+
+        f.write("\n## 结论\n\n")
+
+        if result_thinking['success'] and result_non_thinking['success']:
+            analysis_t = result_thinking['analysis']
+
+            if analysis_t['has_thinking_process']:
+                f.write("✅ **思考模式生效**: 模型输出了明显的思考过程，然后通过 `Answer:` 标记分隔最终答案。\n\n")
+            else:
+                f.write("⚠️ **思考模式未生效**: 模型未输出明显的思考过程。\n\n")
+
+            f.write("### 建议\n\n")
+            f.write("- 思考模式下响应时间更长，但可能获得更好的推理质量\n")
+            f.write("- 非思考模式下响应更快，适合对延迟敏感的场景\n")
+            f.write("- 当前 `_extract_json` 增强逻辑可以正确处理思考模式的输出\n")
+        else:
+            f.write("⚠️ 部分测试失败，请检查日志。\n")
+
+    print(f"\n📄 汇总报告已保存: {report_file}")
+
+
+async def main():
+    """主函数"""
+    print("="*70)
+    print(" Qwen3.5 思考模式 vs 非思考模式 输出对比测试")
+    print("="*70)
+    print(f"\n测试场景: 目录完整性审查 (catalog_integrity_review)")
+    print(f"模型: shutian_qwen3_5_122b")
+    print(f"Prompt: 允许思考，最终输出 JSON")
+
+    # 创建输出目录
+    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+    output_dir = Path(project_root) / "utils_test" / "Model_Test" / "output" / f"thinking_test_{timestamp}"
+    output_dir.mkdir(parents=True, exist_ok=True)
+
+    print(f"\n输出目录: {output_dir}")
+
+    # 测试思考模式
+    result_thinking = await test_with_mode("thinking_enabled", True, output_dir)
+
+    # 等待一小段时间，避免 trace_id 冲突
+    await asyncio.sleep(1)
+
+    # 测试非思考模式
+    result_non_thinking = await test_with_mode("thinking_disabled", False, output_dir)
+
+    # 打印对比结果
+    print_comparison(result_thinking, result_non_thinking)
+
+    # 保存汇总报告
+    save_summary_report(result_thinking, result_non_thinking, output_dir)
+
+    print(f"\n{'='*70}")
+    print(" 测试完成")
+    print(f"{'='*70}")
+    print(f"\n所有结果已保存到: {output_dir}")
+
+    return result_thinking['success'] and result_non_thinking['success']
+
+
+if __name__ == "__main__":
+    try:
+        success = asyncio.run(main())
+        sys.exit(0 if success else 1)
+    except KeyboardInterrupt:
+        print("\n\n测试被用户中断")
+        sys.exit(1)
+    except Exception as e:
+        print(f"\n\n测试运行出错: {e}")
+        import traceback
+        traceback.print_exc()
+        sys.exit(1)