Merge branch 'dev' of http://192.168.0.3:3000/CRBC-MaaS-Platform-Project/LQAgentPlatform into dev

core/construction_review/component/ai_review_engine.py

@@ -463,9 +463,74 @@ class AIReviewEngine(BaseReviewer):
         Returns:
             Dict[str, Any]: 敏感信息检查结果
         """
+        from core.construction_review.component.reviewers.utils import (
+            check_sensitive_words_async,
+            format_check_results
+        )
+        from foundation.observability.logger.loggering import server_logger as logger
+        import time
+        
+        start_time = time.time()
         trace_id = "sensitive_check" + trace_id_idx
-        return await self.sensitive_review("sensitive_check", trace_id, review_content,
-                                          review_location_label, state, stage_name)
+        first_results = await check_sensitive_words_async(review_content) # 先使用关键词匹配式审查
+        
+        # 判断是否检测到敏感词
+        if first_results:
+            logger.info(f"检测到 {len(first_results)} 个敏感词，准备送入大模型二审")
+            # 有敏感词，拼接原文与敏感词列表，进入大模型二审
+            # 格式化敏感词列表
+            sensitive_words_info = []
+            for item in first_results:
+                sensitive_words_info.append(
+                    f"敏感词: {item['word']}, 位置: {item['position']}-{item['end_position']}, 来源: {item['source']}"
+                )
+            formatted_sensitive_words = "\n".join(sensitive_words_info)
+            
+            logger.info(f"格式化后的敏感词信息：\n{formatted_sensitive_words}")
+            
+            # 调用大模型得到敏感词审查结果
+            return await self.review("sensitive_check", trace_id, "basic", "sensitive_word_check", 
+                                   review_content, formatted_sensitive_words,
+                                   None, review_location_label, state, stage_name)
+        else:
+            # 没有检测到敏感词，构造返回体
+            logger.info("没有检测到敏感词，未进入二审")
+            from core.construction_review.component.reviewers.base_reviewer import ReviewResult
+            
+            execution_time = time.time() - start_time
+            result = ReviewResult(
+                success = True,
+                details = {"name": "sensitive_check", "response": "无明显问题"},
+                error_message = None,
+                execution_time = execution_time
+            )
+            
+            # 推送审查完成信息
+            if state and state.get("progress_manager"):
+                import asyncio
+                review_result_data = {
+                    'name': 'sensitive_check',
+                    'success': result.success,
+                    'details': result.details,
+                    'error_message': result.error_message,
+                    'execution_time': result.execution_time,
+                    'timestamp': time.time()
+                }
+                asyncio.create_task(
+                    state["progress_manager"].update_stage_progress(
+                        callback_task_id=state["callback_task_id"],
+                        stage_name=stage_name,
+                        current=None,
+                        status="processing",
+                        message=f"sensitive_check 审查完成，未检测到敏感词，耗时: {result.execution_time:.2f}s",
+                        issues=[review_result_data],
+                        event_type="processing"
+                    )
+                )
+            
+            logger.info(f"sensitive_check 审查完成，未检测到敏感词，耗时: {result.execution_time:.2f}s")
+            
+            return result
 
     async def check_non_parameter_compliance(self, trace_id_idx: str, review_content: str, review_references: str,
                                          reference_source: str, review_location_label: str, state: str, stage_name: str) -> Dict[str, Any]:

core/construction_review/component/reviewers/base_reviewer.py

@@ -156,12 +156,30 @@ class BaseReviewer(ABC):
 
             results = await check_sensitive_words_async(review_content)
             formatted = format_check_results(results, review_content)
-
+            
             logger.info(f"敏感词检查完成，发现 {formatted['count']} 个敏感词")
-
+            
             if not formatted['has_sensitive']:
                 response_str = "无明显问题"
             else:
+                ###################
+                # 在此处设计大模型上下文审核，防止过于敏感
+                ###################
+                prompt_kwargs = {} 
+                # 提示词模板参数填充物
+                prompt_kwargs["review_content"] = review_content
+                #处理一下 results
+                prompt_kwargs["review_references"] = results or ""
+                # 构造提示词
+                task_prompt_info = {
+                    "task_prompt": self.prompt_loader.get_prompt_template("basic", "sensitive_check", **prompt_kwargs),
+                    "task_name": name
+                } 
+                # 调用模型
+                model_response = await self.model_client.get_model_generate_invoke(
+                    trace_id=trace_id,
+                    task_prompt_info=task_prompt_info
+                )
                 issues = []
                 for item in formatted['details']:
                     context = item.get('context', review_content[max(0, item['position']-20):min(len(review_content), item['end_position']+20)])

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

@@ -3,38 +3,66 @@
 # 1.1 词句语法检查功能
 grammar_check:
   system_prompt: |
-    你是施工方案词句语法审查专家，负责检查文本中的错别字和重复字词等语法问题。
-
-    审查要求:
-    - 检查错别字、多字、少字、重复字词等语法错误
-    - 检查标点符号使用错误
-    - 检查明显的语法结构错误
-
-    注意事项:
-    1. 务必结合语境进行分析检查
-    2. 对于表格制表符、不需要检查
-    3. 对于术语概念不得曲解
-    4. 没有明显词句语法错误的内容不予检查，输出无明显问题
-    5. 已检查出的问题项仅输出一次检查结果，禁止对同一内容重复检查
-
-    风险等级分类:
-    - 高风险:影响施工安全、可能导致严重后果的词句错误
-    - 中风险:影响理解表达、可能导致一般性问题的词句错误
-    - 低风险:形式问题、不影响实质内容的词句错误
-
-    审查参考:
-    {review_references}
+    system
+    # role
+    你是词句语法审查专家
+
+    ## workflow
+    - 负责检查文本中的错别字和重复字词等语法问题。
+    - 检查待审查的词句语法标点格式是否符合规则。
+    - 检查错别字、多字、少字、重复字词等语法错误。
+    - 如果发现一个句子有错误，请正确的定位其错误位置，而不是想当然的去找问题。如句子末尾未加句号不要定位到句中去了。
+    - 给出了最终建议后，需要再次考虑如果按照建议去做了，原文是否通顺合理。
+    - 请减少对标点符号的严格审查，如果有问题请三思你所说的位置有没有符号，符号对不对。
+    - 仅检查句尾的标点问题，不要检查句中标点问题。
+    - 请**不要为文章新增标点符号**如句号、逗号，只需要审查标点符号是否正确即可。
+
+    ## example
+    1. 出现了明显错字，出现了的地得使用混乱。
+    2. 如果出现了错字，大概率是因为拼音拼写错误导致的错字，请根据其相似读音推测正确的字，如果拿不准请不给出修改为xx的建议。
+    3. 如“供气”错打为了“供器”，应当结合上下文推测“供qi”应该为什么。
+    4. 对于条款编号而言，'一）'这样的结构是正确的，符合中文规范
+
+    ## rule
+    - 不需要强求要输出问题，除非是非常明显的错误。小问题可以忽略。
+    - 仅检查句尾的标点问题，不要检查句中标点问题。
+    - 不要为长句切分、添加分句标点。
+    - 请着重对错别字的审查上，大量减少对标点符号的审查力度。
+    - 务必遵循<example>中的规则。
+    - 如果没有错误请不要添加新的issue。
+    - 遵循中文的语法规范。
+    - 遵循以下**强制**规范：
+        1. 务必结合语境进行分析检查。
+        2. 对于**表格制表符**、**表格内容**不需要检查。
+        3. 对于术语概念不得曲解。
+        4. 没有明显词句语法错误、标点错误的内容不予检查，输出无明显问题。
+        5. 已检查出的问题项仅输出一次检查结果，禁止对同一内容重复检查。
+        6. 统一解释：如果表格中出现了多列相同的表头标题，不是错误，而是解析时这几个是合并的表头。
+
+    ## output
+    ```json
+        {{
+            "issue_point": "问题标题描述",
+            "location": "当前问题对应的原始条款内容及位置,如六、验收标准 (页码: 85)，以及其语境上下文",
+            "suggestion": "具体的修改建议内容",
+            "reason": "问题的原因分析和依据说明",
+            "risk_level": ""
+        }}
+    ```
+    # 风险等级分类risk_level:
+    - 高风险:影响施工安全、可能导致严重后果的词句错误。
+    - 中风险:影响理解表达、可能导致一般性问题的词句错误。
+    - 低风险:形式问题、不影响实质内容的词句错误。
 
   user_prompt_template: |
-    请审查以下内容的词句语法错误，重点关注错别字、重复字词、标点符号和语法结构：
-
+    请审查以下内容的词句语法错误，重点关注错别字、重复字词和语法结构：
     【待检查文本】
     {review_content}
-
-    输出格式:务必须严格按照以下标准JSON格式输出审查结果:
-    如果未发现明显的词句语法错误，请输出:无明显问题
+    输出格式:务必须严格按照以下标准json格式输出审查结果:
+    如果未发现明显的词句语法错误，请输出:无明显问题。
     如果发现问题，请按以下格式输出:
-    location字段直接输出原字段内容，不得猜测
+    location字段直接输出原字段内容，不得猜测。
+    ## 示例
     ```json
     {{
       "issue_point": "问题标题描述",
@@ -208,23 +236,24 @@ sensitive_word_check:
 
     审查要求:
     - 重点关注政治敏感、商业机密、表述不当
-    - 简明扼要指出敏感词汇和处理建议
+    - 你只需要考虑初筛找到的敏感词与原文，不需要你自行去找敏感词
+    - 通过给出的敏感词初筛内容，并联系上下文，确定初筛的关键词匹配的敏感词是否合理，如果合理则给出issue，如果初筛的敏感词在原文中的语义并无恶意则直接输出:无明显问题
+    - 例如：原文“应禁止工人在宿舍中赌博”，初筛内容：“赌博”；解释：原文中“赌博”二字无恶意；结论：无明显问题
     - 风险等级分类:
       * 高风险:影响审查结论、可能导致法律问题或严重安全隐患
       * 中风险:影响专业表达、可能导致理解偏差或一般性问题
       * 低风险:形式问题、不影响实质内容和安全
 
-    审查参考:
-    {review_references}
-
   user_prompt_template: |
-    请审查以下内容的敏感词:
+    以下我初步查找到的敏感词内容：
+    {review_references}
 
+    请根据以下内容作为上下文，确定敏感词是否合理:
     {review_content}
 
     输出格式:务必须严格按照以下标准JSON格式输出审查结果:
-    如果未发现问题，请输出:无明显问题
-    如果发现问题，请按以下格式输出:
+    如果初筛敏感词在原文中并无恶意，请输出:无明显问题
+    如果发现初筛出的敏感词确实存在问题，请按以下格式输出:
     location字段直接输出原字段内容，不得猜测
     ```json
     {{

utils_test/Prompt_Test/test_prompt.py

@@ -0,0 +1,361 @@
+"""
+测试提示词脚本
+使用requests库测试语法审查API
+支持本地模型和在线模型两种调用方式
+"""
+import requests
+import json
+from openai import OpenAI
+
+
+class PromptTester:
+    """提示词测试类"""
+    
+    def __init__(self, api_url, api_key=None, is_local=True):
+        """
+        初始化测试器
+        
+        Args:
+            api_url: API地址
+            api_key: API密钥（在线模型需要）
+            is_local: 是否使用本地模型（True: 本地模型使用requests, False: 在线模型使用OpenAI SDK）
+        """
+        self.api_url = api_url
+        self.api_key = api_key
+        self.is_local = is_local
+        
+        # 如果是在线模型，初始化OpenAI客户端
+        if not is_local and api_key:
+            self.client = OpenAI(
+                base_url=api_url.replace('/chat/completions', ''),  # 移除末尾的路径
+                api_key=api_key
+            )
+        else:
+            self.client = None
+        self.system_prompt = """
+system
+# role
+你是词句语法审查专家
+
+## workflow
+- 负责检查文本中的错别字和重复字词等语法问题。
+- 检查待审查的词句语法标点格式是否符合规则。
+- 检查错别字、多字、少字、重复字词等语法错误。
+- 如果发现一个句子有错误，请正确的定位其错误位置，而不是想当然的去找问题。如句子末尾未加句号不要定位到句中去了。
+- 给出了最终建议后，需要再次考虑如果按照建议去做了，原文是否通顺合理。
+- 请减少对标点符号的严格审查，如果有问题请三思你所说的位置有没有符号，符号对不对。
+- 仅检查句尾的标点问题，不要检查句中标点问题。
+- 请**不要为文章新增标点符号**如句号、逗号，只需要审查标点符号是否正确即可。
+
+## example
+1. 出现了明显错字，出现了的地得使用混乱。
+2. 如果出现了错字，大概率是因为拼音拼写错误导致的错字，请根据其相似读音推测正确的字，如果拿不准请不给出修改为xx的建议。
+3. 如“供气”错打为了“供器”，应当结合上下文推测“供qi”应该为什么。
+4. 对于条款编号而言，'一）'这样的结构是正确的，符合中文规范
+
+## rule
+- 不需要强求要输出问题，除非是非常明显的错误。小问题可以忽略。
+- 仅检查句尾的标点问题，不要检查句中标点问题。
+- 不要为长句切分、添加分句标点。
+- 请着重对错别字的审查上，大量减少对标点符号的审查力度。
+- 务必遵循<example>中的规则。
+- 如果没有错误请不要添加新的issue。
+- 遵循中文的语法规范。
+- 遵循以下**强制**规范：
+    1. 务必结合语境进行分析检查。
+    2. 对于**表格制表符**、**表格内容**不需要检查。
+    3. 对于术语概念不得曲解。
+    4. 没有明显词句语法错误、标点错误的内容不予检查，输出无明显问题。
+    5. 已检查出的问题项仅输出一次检查结果，禁止对同一内容重复检查。
+    6. 统一解释：如果表格中出现了多列相同的表头标题，不是错误，而是解析时这几个是合并的表头。
+
+## output
+```json
+    {{
+        "issue_point": "问题标题描述",
+        "location": "当前问题对应的原始条款内容及位置,如六、验收标准 (页码: 85)，以及其语境上下文",
+        "suggestion": "具体的修改建议内容",
+        "reason": "问题的原因分析和依据说明",
+        "risk_level": ""
+    }}
+```
+# 风险等级分类risk_level:
+- 高风险:影响施工安全、可能导致严重后果的词句错误。
+- 中风险:影响理解表达、可能导致一般性问题的词句错误。
+- 低风险:形式问题、不影响实质内容的词句错误。
+"""
+        self.user_prompt_template = """
+请审查以下内容的词句语法错误，重点关注错别字、重复字词和语法结构：
+【待检查文本】
+{review_content}
+输出格式:务必须严格按照以下标准json格式输出审查结果:
+如果未发现明显的词句语法错误，请输出:无明显问题。
+如果发现问题，请按以下格式输出:
+location字段直接输出原字段内容，不得猜测。
+## 示例
+```json
+{{
+  "issue_point": "问题标题描述",
+  "location": "当前问题对应的原始条款内容及位置,如六、验收标准 (页码: 85)，以及其语境上下文",
+  "suggestion": "具体的修改建议内容",
+  "reason": "问题的原因分析和依据说明",
+  "risk_level": ""
+}}
+```
+/no_think
+"""
+
+    def test_prompt(self, review_content, model="Qwen3-8B", temperature=0.3, max_tokens=2000):
+        """
+        测试提示词
+        
+        Args:
+            review_content: 待审查的文本内容
+            model: 模型名称
+            temperature: 温度参数
+            max_tokens: 最大token数
+            
+        Returns:
+            API响应结果
+        """
+        # 填充用户提示词
+        user_prompt = self.user_prompt_template.format(review_content=review_content)
+        
+        # 打印请求信息
+        print("=" * 80)
+        print(f"调用方式: {'本地模型 (requests)' if self.is_local else '在线模型 (OpenAI SDK)'}")
+        print("发送请求到:", self.api_url)
+        print("=" * 80)
+        print("System Prompt:")
+        print("-" * 80)
+        print(self.system_prompt)
+        print("=" * 80)
+        print("User Prompt:")
+        print("-" * 80)
+        print(user_prompt)
+        print("=" * 80)
+        
+        # 根据is_local选择不同的调用方式
+        if self.is_local:
+            # 本地模型：使用requests
+            return self._call_with_requests(user_prompt, model, temperature, max_tokens)
+        else:
+            # 在线模型：使用OpenAI SDK
+            return self._call_with_openai(user_prompt, model, temperature, max_tokens)
+    
+    def _call_with_requests(self, user_prompt, model, temperature, max_tokens):
+        """使用requests库调用本地模型"""
+        # 构建请求数据
+        payload = {
+            "model": model,
+            "messages": [
+                {
+                    "role": "system",
+                    "content": self.system_prompt
+                },
+                {
+                    "role": "user",
+                    "content": user_prompt
+                }
+            ],
+            "temperature": temperature,
+            "max_tokens": max_tokens
+        }
+        
+        try:
+            # 构建请求头
+            headers = {"Content-Type": "application/json"}
+            if self.api_key:
+                headers["Authorization"] = f"Bearer {self.api_key}"
+            
+            # 发送POST请求
+            response = requests.post(
+                self.api_url,
+                headers=headers,
+                json=payload,
+                timeout=60
+            )
+            
+            # 检查响应状态
+            response.raise_for_status()
+            
+            # 解析响应
+            result = response.json()
+            
+            print("响应状态码:", response.status_code)
+            print("=" * 80)
+            print("API响应结果:")
+            print("-" * 80)
+            print(json.dumps(result, ensure_ascii=False, indent=2))
+            print("=" * 80)
+            
+            # 提取AI回复内容
+            if "choices" in result and len(result["choices"]) > 0:
+                ai_response = result["choices"][0].get("message", {}).get("content", "")
+                print("AI回复内容:")
+                print("-" * 80)
+                print(ai_response)
+                print("=" * 80)
+            
+            return result
+            
+        except requests.exceptions.RequestException as e:
+            print(f"请求失败: {e}")
+            return None
+        except json.JSONDecodeError as e:
+            print(f"JSON解析失败: {e}")
+            print("原始响应:", response.text)
+            return None
+    
+    def _call_with_openai(self, user_prompt, model, temperature, max_tokens):
+        """使用OpenAI SDK调用在线模型"""
+        if not self.client:
+            print("错误: OpenAI客户端未初始化")
+            return None
+        
+        try:
+            # 使用OpenAI SDK调用
+            extra_body = {
+                # enable thinking, set to False to disable test
+                "enable_thinking": False,
+                # use thinking_budget to contorl num of tokens used for thinking
+                # "thinking_budget": 4096
+            }
+            response = self.client.chat.completions.create(
+                model=model,
+                messages=[
+                    {
+                        "role": "system",
+                        "content": self.system_prompt
+                    },
+                    {
+                        "role": "user",
+                        "content": user_prompt
+                    }
+                ],
+                temperature=temperature,
+                max_tokens=max_tokens,
+                extra_body=extra_body
+            )
+            
+            # 转换为字典格式
+            result = {
+                "id": response.id,
+                "object": response.object,
+                "created": response.created,
+                "model": response.model,
+                "choices": [
+                    {
+                        "index": choice.index,
+                        "message": {
+                            "role": choice.message.role,
+                            "content": choice.message.content
+                        },
+                        "finish_reason": choice.finish_reason
+                    }
+                    for choice in response.choices
+                ],
+                "usage": {
+                    "prompt_tokens": response.usage.prompt_tokens,
+                    "completion_tokens": response.usage.completion_tokens,
+                    "total_tokens": response.usage.total_tokens
+                }
+            }
+            
+            print("=" * 80)
+            print("API响应结果:")
+            print("-" * 80)
+            print(json.dumps(result, ensure_ascii=False, indent=2))
+            print("=" * 80)
+            
+            # 提取AI回复内容
+            if result["choices"] and len(result["choices"]) > 0:
+                ai_response = result["choices"][0]["message"]["content"]
+                print("AI回复内容:")
+                print("-" * 80)
+                print(ai_response)
+                print("=" * 80)
+            
+            return result
+            
+        except Exception as e:
+            print(f"OpenAI SDK调用失败: {e}")
+            return None
+
+
+def main():
+    """主函数"""
+    
+    # ==================== 配置区域 ====================
+    # 是否使用本地模型（True: 本地模型, False: 在线模型）
+    is_local = True
+    
+    # 本地模型配置
+    local_config = {
+        "api_url": "http://192.168.91.253:9002/v1/chat/completions",
+        "api_key": None,  # 本地模型不需要API Key
+        "model": "Qwen3-8B",
+        "temperature": 0.3,
+        "max_tokens": 2000
+    }
+    
+    # 在线模型配置（ModelScope）
+    online_config = {
+        "api_url": "https://api-inference.modelscope.cn/v1",
+        "api_key": "ms-c0349a0a-8f15-466b-96be-4f96d001d8f2",  # ModelScope Token
+        "model": "Qwen/Qwen3-14B",
+        "temperature": 0.7,
+        "max_tokens": 2000
+    }
+    # ==================== 配置区域结束 ====================
+    
+    # 根据is_local选择配置
+    config = local_config if is_local else online_config
+    
+    print("=" * 80)
+    print(f"当前使用: {'本地模型' if is_local else '在线模型'}")
+    print(f"API地址: {config['api_url']}")
+    print(f"模型名称: {config['model']}")
+    print("=" * 80)
+    
+    # 创建测试器
+    tester = PromptTester(config["api_url"], config["api_key"], is_local)
+    
+    custom_content = """
+
+上下时不得手持物件；必须从指定的路线上下，不得在高空投掷材料或工具等物；不得将易滚易滑的工具、材料堆放在脚手架上；工作完毕应及时将工具、零星材料、零部件等一切易坠落物件清理干净，以防落下伤人，上下大型零件时，应采用可靠的起吊机具。
+12）要处处注意危险标志和危险地方。夜间作业，必须设置足够的照明设施，否则禁止施工。
+13）严禁上下同时垂直作业。若特殊情况必须垂直作业，应经有关领导批准，并在上下两层间设备专用的防护棚或者其他隔离设施。
+14）严禁坐在高空无遮栏处休息，防止坠落。
+15）卷扬机等各种升降材料的设备严禁上下载人。
+16）超过3米长的铺板不能同时站两人作业。
+17）遇六级以上大风时，禁止露天进行高空作业。
+（4）起重作业安全保障措施
+1）起重设备在现场初次使用时应进行试吊，试吊严格按照《起重机试验规范和程序》(GB/T 5905-2011）执行，试吊合格后方可使用。
+2）起重作业前，检查起重设备的空载运转、回转、起重、变幅等各种机构的制动器、安全限位、防护装置等，确认正常后方可作业。
+3）起重作业时，应依次逐级操作，严禁越档操作，在变换运转方向时，应将控制器转到零位，停止转动后，再转向另一方向，严禁急开急停。
+4）司机、起重信号司索工等人员应经专业培训，取得特种设备作业人员证后，方可上岗作业。
+5）起吊物件旋转或移动，应将起吊物提升至可能遇到的障碍物上方0.5m 以上。
+6）起重作业严格执行安全操作规程，严格遵守“十不吊”的要求。
+7）吊装人员必须戴安全帽、安全带；
+8）吊装工作区设有明显标志，并设专人警戒，与吊装无关人员严禁入内。起重机工作时，起重臂杆旋转半径范围内，严禁站人或通过。
+9）运输、吊装构件时，严禁在被运输、吊装的构件上指挥和放置材料、工具
+10）吊装时，必须用吊笼或钢丝绳、保险绳绑扎牢固后才能吊运和传递，不得随意抛掷材料物体、工具，防止滑脱伤人或意外事故。
+11）构件必须绑扎牢固，起吊点应通过构件的重心位置，吊升时应平稳，避免振动或摆动，以防构件脱落。
+12）起吊构件时，速度不应太快，不得在高空停留过久，严禁猛升猛降，以防构件脱落。
+13）构件就位后临时固定前，不得松钩、解开吊装索具。构件固定后，应检查连接牢固和稳定情况，当确定连接安全可靠，才可拆除临时固定工具和进行下一步吊装。
+14）严禁夜间施工作业。15）起重机行驶的道路必须平整、坚实、可靠，停放地点必须平坦。
+16）起重机不得停放在斜坡道上工作，不允许起重机两条履带或支腿停留部位一高一低或土质一硬一软。
+
+"""
+    tester.test_prompt(
+        custom_content,
+        model=config["model"],
+        temperature=config["temperature"],
+        max_tokens=config["max_tokens"]
+    )
+
+
+if __name__ == "__main__":
+    main()
+

utils_test/Sensitive_Test/README.md

@@ -0,0 +1,318 @@
+# 敏感词检查单元测试说明（完整版）
+
+## 📋 测试概述
+
+本测试模块针对 `AIReviewEngine.check_sensitive` 方法进行单元测试，**包含完整的大模型二审功能**，支持自定义文本检查。
+
+## 🎯 测试目标
+
+测试 `core/construction_review/component/ai_review_engine.py` 文件中第 450-460 行的 `check_sensitive` 方法。
+
+## ✨ 主要特性
+
+### 1. **完整的大模型二审功能**
+
+-   ✅ 关键词匹配检测（第一步）
+-   ✅ 大模 �� 二审（第二步，与原函数完全一致）
+-   ✅ 调用 `review()` 方法进行深度审查
+-   ✅ 返回详细的审查结果
+
+### 2. **自定义文本输入**
+
+-   ✅ 交互式文本输入
+-   ✅ 支持多行文本
+-   ✅ 实时检查反馈
+-   ❌ 移除了内置测试用例
+
+### 3. **详细日志输出**
+
+-   ✅ 每个步骤的详细日志
+-   ✅ 执行时间统计
+-   ✅ 敏感词详细信息
+-   ✅ 大模型审查结果
+
+## 🔧 功能实现
+
+### 核心函数：`check_sensitive_with_llm_review()`
+
+```python
+async def check_sensitive_with_llm_review(
+    trace_id_idx: str,           # 追踪ID索引
+    review_content: str,         # 审查内容
+    review_references: str,      # 审查参考信息
+    review_location_label: str,  # 审查位置标签
+    state: Dict,                 # 状态字典
+    stage_name: str              # 阶段名称
+) -> Any:
+```
+
+**执行流程**：
+
+```
+步骤1: 关键词匹配检测
+   ↓
+检测到敏感词？
+   ↓ 是
+步骤2: 格式化敏感词信息
+   ↓
+步骤3: 调用大模型二审
+   ↓
+   └─→ self.ai_review_engine.review(
+           "sensitive_check",
+           trace_id,
+           "basic",
+           "sensitive_word_check",
+           review_content,
+           formatted_sensitive_words,
+           ...
+       )
+   ↓
+返回大模型审查结果
+
+   ↓ 否（未检测到敏感词）
+步骤2: 构造成功返回体
+   ↓
+返回 ReviewResult(success=True)
+```
+
+## 🚀 使用方法
+
+### 方法 1：直接运行（推荐）
+
+```bash
+cd h:\UGit\LQAgentPlatform
+python utils_test/Sensitive_Test/test_sensitive_check_standalone.py
+```
+
+### 方法 2：使用批处理脚本
+
+```bash
+cd h:\UGit\LQAgentPlatform\utils_test\Sensitive_Test
+run_test.bat
+```
+
+## 📝 交互式输入说明
+
+运行程序后，会提示输入要检查的文本：
+
+```
+================================================================================
+请输��要检查的文本内容
+================================================================================
+提示：
+  1. 可以输入多行文本，输入完成后单独一行输入 'END' 结束
+  2. 直接按回车（输入空行）将跳过自定义输入
+================================================================================
+>>>
+```
+
+### 输入示例
+
+**示例 1：单行文本**
+
+```
+>>> 本工程为住宅楼建设项目，采用框架结构
+... END
+```
+
+**示例 2：多行文本**
+
+```
+>>> 本施工方案编制依据包括：
+... 1. 《建筑工程施工质量验收统一标准》GB50300-2013
+... 2. 《混凝土结构工程施工质量验收规范》GB50204-2015
+... 3. 施工图纸及相关设计文件
+... END
+```
+
+**示例 3：跳过输入**
+
+```
+>>> [直接按回车]
+```
+
+## 📊 日志输出示例
+
+### 1. 初始化阶段
+
+```
+[INFO] ================================================================================
+[INFO] 初始化敏感词检查测试类
+[INFO] ================================================================================
+[INFO] AI审查引擎初始化成功
+[INFO] 正在初始化敏感词检测器...
+[INFO] 敏感词检测器初始化成功: {'total_words': 12345, 'sources': 15}
+```
+
+### 2. 检测阶段（未检测到敏感词）
+
+```
+[INFO] ================================================================================
+[INFO] 开始执行敏感词检查 - trace_id: sensitive_check_custom_001
+[INFO] 阶段名称: 自定义测试
+[INFO] 审查位置: 自定义文本检查
+[INFO] 审查内容长度: 156 字符
+[INFO] 审查内容预览: 本工程为住宅楼建设项目，采用框架结构，建筑面积约5000平方米...
+[INFO] ================================================================================
+[INFO] 步骤1: 开始关键词匹配检测...
+[INFO] 步骤1: 关键词检测完成，耗时: 0.0023s
+[INFO] 步骤2: 未检测到敏感词
+[INFO] 步骤3: 敏感词检查完成（未检测到敏感词），总耗时: 0.0025s
+[INFO] ================================================================================
+```
+
+### 3. 检测阶段（检测到敏感词 + 大模型二审）
+
+```
+[INFO] ================================================================================
+[INFO] 开始执行敏感词检查 - trace_id: sensitive_check_custom_001
+[INFO] 阶段名称: 自定义测试
+[INFO] 审查位置: 自定义文本检查
+[INFO] 审查内容长度: 89 字符
+[INFO] 审查内容: 这是一段包含敏感内容的测试文本...
+[INFO] ================================================================================
+[INFO] 步骤1: 开始关键词匹配检测...
+[INFO] 步骤1: 关键词检测完成，耗时: 0.0018s
+[WARNING] 步骤2: 检测到 2 个敏感词，准备送入大模型二审
+[WARNING] --------------------------------------------------------------------------------
+[WARNING]   敏感词 #1: 敏感词: xxx, 位置: 15-18, 来源: 某词库.txt
+[WARNING]   敏感词 #2: 敏感词: yyy, 位置: 35-38, 来源: 某词库.txt
+[WARNING] --------------------------------------------------------------------------------
+[INFO] 步骤3: 调用大模型进行二审...
+[INFO] 步骤3: 大模型二审完成，总耗时: 2.3456s
+[INFO] ================================================================================
+```
+
+### 4. 结果显示
+
+```
+[INFO] --------------------------------------------------------------------------------
+[INFO] 检查结果:
+[INFO] --------------------------------------------------------------------------------
+[INFO]   返回类型: ReviewResult
+[INFO]   是否成功: False
+[INFO]   执行时间: 2.3456s
+[INFO]   详细信息: {'name': 'sensitive_check', 'response': '...大模型审查结果...'}
+[WARNING]
+[WARNING] 检测到敏感内容，请查看上方详细信息
+[INFO] --------------------------------------------------------------------------------
+```
+
+## 🔍 与原函数的对比
+
+| 功能项           | 原函数 | 测试函数 | 说明                           |
+| ---------------- | ------ | -------- | ------------------------------ |
+| 关键词匹配检测   | ✅     | ✅       | 完全保留                       |
+| 敏感词信息格式化 | ✅     | ✅       | 完全保留                       |
+| **大模型二审**   | ✅     | ✅       | **完全保留，调用 review 方法** |
+| 消息推送         | ✅     | ❌       | 测试中移除                     |
+| 结果封装         | ✅     | ✅       | 完全保留                       |
+| 执行时间统计     | ✅     | ✅       | 完全保留                       |
+| 日志记录         | ✅     | ✅       | 增强版                         |
+
+## ⚙️ 技术细节
+
+### 大模型二审调用
+
+```python
+# 与原函数完全一致的调用方式
+result = await self.ai_review_engine.review(
+    "sensitive_check",              # 审查名称
+    trace_id,                       # 追踪ID
+    "basic",                        # 审查器类型
+    "sensitive_word_check",         # 提示词名称
+    review_content,                 # 审查内容
+    formatted_sensitive_words,      # 格式化的敏感词信息
+    None,                           # reference_source
+    review_location_label,          # 审查位置标签
+    state,                          # 状态字典
+    stage_name                      # 阶段名称
+)
+```
+
+### 敏感词格式化
+
+```python
+# 格式化敏感词信息，传递给大模型
+sensitive_words_info = []
+for item in first_results:
+    sensitive_words_info.append(
+        f"敏感词: {item['word']}, "
+        f"位置: {item['position']}-{item['end_position']}, "
+        f"来源: {item['source']}"
+    )
+formatted_sensitive_words = "\n".join(sensitive_words_info)
+```
+
+## ⚠️ 注意事项
+
+1. **依赖要求**：
+
+    - 需要完整的项目环境
+    - 需要敏感词库文件
+    - 需要大模型 API 配置
+
+2. **运行环境**：
+
+    - 必须在项目根目录运行
+    - 需要正确的 Python 环境
+    - 需要网络连接（大模型 API 调用）
+
+3. **输入限制**：
+
+    - 文本长度无限制
+    - 支持中英文混合
+    - 支持特殊字符
+
+4. **性能考虑**：
+    - 关键词检测：毫秒级
+    - 大模型二审：秒级（取决于 API 响应）
+
+## 📈 测试建议
+
+### 推荐测试场景
+
+1. **正常文本**：施工方案、技术文档等
+2. **边界情况**：空文本、超长文本
+3. **特殊字符**：技术符号、数学公式
+4. **混合文本**：中英文混合、数字混合
+
+### 测试流程
+
+```
+1. 运行测试程序
+   ↓
+2. 输入测试文本
+   ↓
+3. 观察检测过程
+   ↓
+4. 查看检测结果
+   ↓
+5. 分析大模型审查意见
+```
+
+## 🆚 版本对比
+
+### v1.0（原版本）
+
+-   ❌ 无大模型二审
+-   ✅ 内置测试用例
+-   ✅ 批量测试
+
+### v2.0（当前版本）
+
+-   ✅ **完整大模型二审**
+-   ✅ **自定义文本输入**
+-   ✅ **交互式操作**
+-   ✅ **详细日志输出**
+-   ❌ 移除内置测试用例
+
+## 📞 问题反馈
+
+如有问题或建议，请联系开发团队。
+
+---
+
+**最后更新**: 2025-12-25
+**版本**: v2.0
+**作者**: AI Assistant

utils_test/Sensitive_Test/run_test.bat

@@ -0,0 +1,39 @@
+@echo off
+chcp 65001 >nul
+echo ================================================================================
+echo 敏感词检查单元测试 - 运行脚本
+echo ================================================================================
+echo.
+
+cd /d %~dp0\..\..
+
+echo [1/3] 检查Python环境...
+python --version
+if %errorlevel% neq 0 (
+    echo [错误] Python未安装或未添加到PATH
+    pause
+    exit /b 1
+)
+echo.
+
+echo [2/3] 检查项目路径...
+echo 当前目录: %cd%
+if not exist "core\construction_review\component\ai_review_engine.py" (
+    echo [错误] 未找到项目文件，请确认在正确的目录运行
+    pause
+    exit /b 1
+)
+echo.
+
+echo [3/3] 运行测试...
+echo ================================================================================
+echo.
+
+python utils_test\Sensitive_Test\test_sensitive_check_standalone.py
+
+echo.
+echo ================================================================================
+echo 测试完成
+echo ================================================================================
+pause
+

utils_test/Sensitive_Test/test_sensitive_check.py

@@ -0,0 +1,325 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+敏感词检查功能单元测试
+测试 AIReviewEngine.check_sensitive 方法的各种场景
+"""
+
+import asyncio
+import sys
+import os
+import time
+from pathlib import Path
+from typing import Dict, Any, List
+from dataclasses import dataclass
+
+# 添加项目根目录到Python路径
+current_dir = Path(__file__).parent.absolute()
+project_root = current_dir.parent.parent
+sys.path.insert(0, str(project_root))
+os.chdir(str(project_root))
+
+from core.construction_review.component.reviewers.base_reviewer import ReviewResult
+from foundation.observability.logger.loggering import server_logger as logger
+
+
+class TestSensitiveCheck:
+    """敏感词检查功能测试类"""
+
+    def __init__(self):
+        self.test_results = []
+        logger.info("=" * 80)
+        logger.info("初始化敏感词检查测试类")
+        logger.info("=" * 80)
+
+    async def check_sensitive_simplified(
+        self, 
+        trace_id_idx: str, 
+        review_content: str, 
+        review_references: str = "",
+        review_location_label: str = "",
+        state: Dict = None,
+        stage_name: str = "测试阶段"
+    ) -> Dict[str, Any]:
+        """
+        简化版的敏感信息检查函数（用于测试）
+        
+        功能说明：
+        1. 使用关键词匹配进行敏感词检测
+        2. 如果检测到敏感词，格式化敏感词信息并返回
+        3. 如果未检测到敏感词，返回成功结果
+        4. 移除了原函数中的消息推送和大模型二审功能
+        
+        Args:
+            trace_id_idx: 追踪ID索引
+            review_content: 审查内容
+            review_references: 审查参考信息
+            review_location_label: 审查位置标签
+            state: 状态字典（简化版不使用）
+            stage_name: 阶段名称
+            
+        Returns:
+            Dict[str, Any]: 敏感信息检查结果
+        """
+        from core.construction_review.component.reviewers.utils import (
+            check_sensitive_words_async,
+            format_check_results
+        )
+        
+        logger.info("=" * 80)
+        logger.info(f"开始执行敏感词检查 - trace_id: sensitive_check{trace_id_idx}")
+        logger.info(f"阶段名称: {stage_name}")
+        logger.info(f"审查位置: {review_location_label}")
+        logger.info(f"审查内容长度: {len(review_content)} 字符")
+        logger.info(f"审查内容预览: {review_content[:100]}..." if len(review_content) > 100 else f"审查内容: {review_content}")
+        logger.info("=" * 80)
+        
+        start_time = time.time()
+        trace_id = "sensitive_check" + trace_id_idx
+        
+        # 第一步：使用关键词匹配式审查
+        logger.info("步骤1: 开始关键词匹配检测...")
+        first_results = await check_sensitive_words_async(review_content)
+        detection_time = time.time() - start_time
+        logger.info(f"步骤1: 关键词检测完成，耗时: {detection_time:.4f}s")
+        
+        # 判断是否检测到敏感词
+        if first_results:
+            logger.warning(f"步骤2: 检测到 {len(first_results)} 个敏感词")
+            logger.warning("-" * 80)
+            
+            # 格式化敏感词列表
+            sensitive_words_info = []
+            for idx, item in enumerate(first_results, 1):
+                word_info = f"敏感词: {item['word']}, 位置: {item['position']}-{item['end_position']}, 来源: {item['source']}"
+                sensitive_words_info.append(word_info)
+                logger.warning(f"  敏感词 #{idx}: {word_info}")
+            
+            formatted_sensitive_words = "\n".join(sensitive_words_info)
+            logger.warning("-" * 80)
+            
+            # 简化版：直接返回检测结果，不调用大模型二审
+            execution_time = time.time() - start_time
+            
+            result = ReviewResult(
+                success=False,
+                details={
+                    "name": "sensitive_check",
+                    "response": f"检测到{len(first_results)}个敏感词",
+                    "sensitive_words": first_results,
+                    "formatted_info": formatted_sensitive_words
+                },
+                error_message=None,
+                execution_time=execution_time
+            )
+            
+            logger.warning(f"步骤3: 敏感词检查完成（检测到敏感词），总耗时: {execution_time:.4f}s")
+            logger.info("=" * 80)
+            
+            return result
+            
+        else:
+            # 没有检测到敏感词，构造返回体
+            logger.info("步骤2: 未检测到敏感词")
+            
+            execution_time = time.time() - start_time
+            result = ReviewResult(
+                success=True,
+                details={"name": "sensitive_check", "response": "无明显问题"},
+                error_message=None,
+                execution_time=execution_time
+            )
+            
+            # 简化版：移除消息推送功能
+            logger.info(f"步骤3: 敏感词检查完成（未检测到敏感词），总耗时: {execution_time:.4f}s")
+            logger.info("=" * 80)
+            
+            return result
+
+    def get_test_cases(self) -> List[Dict[str, Any]]:
+        """获取测试用例"""
+        return [
+            {
+                "name": "正常文本-无敏感词",
+                "trace_id_idx": "_test_001",
+                "review_content": "本工程为住宅楼建设项目，采用框架结构，建筑面积约5000平方米。施工过程中应严格按照国家规范执行，确保工程质量。",
+                "review_location_label": "第一章 工程概况",
+                "expected_success": True,
+                "description": "测试正常的施工方案文本，不包含任何敏感词"
+            },
+            {
+                "name": "空文本",
+                "trace_id_idx": "_test_002",
+                "review_content": "",
+                "review_location_label": "空内容测试",
+                "expected_success": True,
+                "description": "测试空文本的处理"
+            },
+            {
+                "name": "短文本-无敏感词",
+                "trace_id_idx": "_test_003",
+                "review_content": "施工安全第一",
+                "review_location_label": "安全标语",
+                "expected_success": True,
+                "description": "测试短文本的处理"
+            },
+            {
+                "name": "长文本-无敏感词",
+                "trace_id_idx": "_test_004",
+                "review_content": """
+                本施工方案编制依据包括：
+                1. 《建筑工程施工质量验收统一标准》GB50300-2013
+                2. 《混凝土结构工程施工质量验收规范》GB50204-2015
+                3. 《建筑地基基础工程施工质量验收规范》GB50202-2018
+                4. 施工图纸及相关设计文件
+                5. 现场实际情况及勘察报告
+                
+                工程概况：
+                本工程位于某市某区，为高层住宅建筑，地上30层，地下2层。
+                建筑高度99.8米，总建筑面积约28000平方米。
+                结构形式为框架剪力墙结构，抗震设防烈度为7度。
+                """,
+                "review_location_label": "第一章 编制依据与工程概况",
+                "expected_success": True,
+                "description": "测试包含多段落的长文本"
+            },
+            {
+                "name": "包含特殊字符",
+                "trace_id_idx": "_test_005",
+                "review_content": "施工现场温度：-5℃~35℃，相对湿度：≤85%，风力：≤5级。混凝土强度等级：C30、C35。钢筋规格：Φ12、Φ16、Φ20。",
+                "review_location_label": "第三章 施工条件",
+                "expected_success": True,
+                "description": "测试包含特殊符号和技术参数的文本"
+            }
+        ]
+
+    async def run_single_test(self, test_case: Dict[str, Any]) -> bool:
+        """运行单个测试用例"""
+        logger.info("\n" + "█" * 80)
+        logger.info(f"测试用例: {test_case['name']}")
+        logger.info(f"描述: {test_case['description']}")
+        logger.info("█" * 80)
+        
+        try:
+            # 执行测试
+            result = await self.check_sensitive_simplified(
+                trace_id_idx=test_case['trace_id_idx'],
+                review_content=test_case['review_content'],
+                review_location_label=test_case['review_location_label'],
+                stage_name="单元测试阶段"
+            )
+            
+            # 验证结果
+            logger.info("\n" + "-" * 80)
+            logger.info("测试结果验证:")
+            logger.info(f"  返回类型: {type(result)}")
+            logger.info(f"  是否成功: {result.success}")
+            logger.info(f"  预期成功: {test_case['expected_success']}")
+            logger.info(f"  执行时间: {result.execution_time:.4f}s")
+            logger.info(f"  详细信息: {result.details}")
+            
+            if result.error_message:
+                logger.error(f"  错误信息: {result.error_message}")
+            
+            # 判断测试是否通过
+            test_passed = result.success == test_case['expected_success']
+            
+            if test_passed:
+                logger.info("✓ 测试通过")
+            else:
+                logger.error("✗ 测试失败")
+                logger.error(f"  预期 success={test_case['expected_success']}, 实际 success={result.success}")
+            
+            logger.info("-" * 80)
+            
+            return test_passed
+            
+        except Exception as e:
+            logger.error(f"✗ 测试异常: {str(e)}")
+            logger.exception("详细异常信息:")
+            return False
+
+    async def run_all_tests(self):
+        """运行所有测试用例"""
+        logger.info("\n" + "█" * 80)
+        logger.info("开始执行敏感词检查单元测试")
+        logger.info("█" * 80)
+        
+        # 初始化敏感词检测器
+        try:
+            from core.construction_review.component.reviewers.utils import SensitiveWordChecker
+            logger.info("\n正在初始化敏感词检测器...")
+            stats = SensitiveWordChecker.initialize()
+            logger.info(f"敏感词检测器初始化成功: {stats}")
+        except Exception as e:
+            logger.error(f"敏感词检测器初始化失败: {str(e)}")
+            return False
+        
+        # 获取测试用例
+        test_cases = self.get_test_cases()
+        logger.info(f"\n共有 {len(test_cases)} 个测试用例")
+        
+        # 运行测试
+        results = []
+        for idx, test_case in enumerate(test_cases, 1):
+            logger.info(f"\n{'=' * 80}")
+            logger.info(f"执行测试 {idx}/{len(test_cases)}")
+            logger.info(f"{'=' * 80}")
+            
+            passed = await self.run_single_test(test_case)
+            results.append({
+                'name': test_case['name'],
+                'passed': passed
+            })
+        
+        # 统计结果
+        logger.info("\n" + "█" * 80)
+        logger.info("测试结果汇总")
+        logger.info("█" * 80)
+        
+        passed_count = sum(1 for r in results if r['passed'])
+        total_count = len(results)
+        
+        logger.info(f"\n总测试数: {total_count}")
+        logger.info(f"通过数量: {passed_count}")
+        logger.info(f"失败数量: {total_count - passed_count}")
+        logger.info(f"通过率: {passed_count/total_count*100:.2f}%")
+        
+        logger.info("\n详细结果:")
+        for idx, result in enumerate(results, 1):
+            status = "✓ 通过" if result['passed'] else "✗ 失败"
+            logger.info(f"  {idx}. {result['name']}: {status}")
+        
+        logger.info("\n" + "█" * 80)
+        
+        if passed_count == total_count:
+            logger.info("✓ 所有测试通过!")
+            logger.info("█" * 80)
+            return True
+        else:
+            logger.error(f"✗ 有 {total_count - passed_count} 个测试失败")
+            logger.info("█" * 80)
+            return False
+
+    def run_tests_sync(self):
+        """同步方式运行测试"""
+        return asyncio.run(self.run_all_tests())
+
+
+def main():
+    """主函数"""
+    logger.info("=" * 80)
+    logger.info("敏感词检查功能单元测试")
+    logger.info("测试文件: test_sensitive_check.py")
+    logger.info("=" * 80)
+    
+    tester = TestSensitiveCheck()
+    success = tester.run_tests_sync()
+    
+    sys.exit(0 if success else 1)
+
+
+if __name__ == "__main__":
+    main()
+

utils_test/Sensitive_Test/test_sensitive_check_standalone.py

@@ -0,0 +1,352 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+"""
+敏感词检查功能单元测试（完整版本）
+测试 AIReviewEngine.check_sensitive 方法
+包含大模型二审功能，支持自定义检查文本
+"""
+
+import asyncio
+import sys
+import os
+import time
+from pathlib import Path
+from typing import Dict, Any, Optional
+
+# 添加项目根目录到Python路径
+current_dir = Path(__file__).parent.absolute()
+project_root = current_dir.parent.parent
+sys.path.insert(0, str(project_root))
+os.chdir(str(project_root))
+
+
+class SimpleLogger:
+    """简单日志类"""
+    
+    @staticmethod
+    def info(msg):
+        print(f"[INFO] {msg}")
+    
+    @staticmethod
+    def warning(msg):
+        print(f"[WARNING] {msg}")
+    
+    @staticmethod
+    def error(msg):
+        print(f"[ERROR] {msg}")
+    
+    @staticmethod
+    def exception(msg):
+        import traceback
+        print(f"[EXCEPTION] {msg}")
+        traceback.print_exc()
+
+
+logger = SimpleLogger()
+
+
+class TestSensitiveCheck:
+    """敏感词检查功能测试类"""
+
+    def __init__(self):
+        self.test_results = []
+        self.ai_review_engine = None
+        logger.info("=" * 80)
+        logger.info("初始化敏感词检查测试类")
+        logger.info("=" * 80)
+
+    def initialize_engine(self):
+        """初始化AI审查引擎"""
+        try:
+            from core.construction_review.component.ai_review_engine import AIReviewEngine
+            from core.base.task_models import TaskFileInfo
+            
+            # 创建 mock 的 TaskFileInfo 对象
+            logger.info("创建 mock TaskFileInfo 对象...")
+            mock_task_info = TaskFileInfo(
+                file_id="test_file_001",
+                callback_task_id="test_task_001",
+                user_id="test_user",
+                review_config=["sensitive_check"],  # 只启用敏感词检查
+                project_plan_type="test_project",
+                tendency_review_role="test_role"
+            )
+            
+            logger.info("初始化 AI 审查引擎...")
+            self.ai_review_engine = AIReviewEngine(task_file_info=mock_task_info)
+            logger.info("AI审查引擎初始化成功")
+            return True
+        except Exception as e:
+            logger.error(f"AI审查引擎初始化失败: {e}")
+            logger.exception("详细错误:")
+            return False
+
+    async def check_sensitive_with_llm_review(
+        self, 
+        trace_id_idx: str, 
+        review_content: str, 
+        review_references: str = "",
+        review_location_label: str = "",
+        state: Dict = None,
+        stage_name: str = "测试阶段"
+    ) -> Any:
+        """
+        完整版的敏感信息检查函数（包含大模型二审）
+        
+        功能说明：
+        1. 使用关键词匹配进行敏感词检测
+        2. 如果检测到敏感词，调用大模型进行二审（与原函数一致）
+        3. 如果未检测到敏感词，返回成功结果
+        4. 移除了消息推送功能（简化测试）
+        
+        Args:
+            trace_id_idx: 追踪ID索引
+            review_content: 审查内容
+            review_references: 审查参考信息
+            review_location_label: 审查位置标签
+            state: 状态字典
+            stage_name: 阶段名称
+            
+        Returns:
+            审查结果对象
+        """
+        try:
+            from core.construction_review.component.reviewers.utils import check_sensitive_words_async
+            from core.construction_review.component.reviewers.base_reviewer import ReviewResult
+        except ImportError as e:
+            logger.error(f"导入模块失败: {e}")
+            return None
+        
+        logger.info("=" * 80)
+        logger.info(f"开始执行敏感词检查 - trace_id: sensitive_check{trace_id_idx}")
+        logger.info(f"阶段名称: {stage_name}")
+        logger.info(f"审查位置: {review_location_label}")
+        logger.info(f"审查内容长度: {len(review_content)} 字符")
+        if len(review_content) > 100:
+            logger.info(f"审查内容预览: {review_content[:100]}...")
+        else:
+            logger.info(f"审查内容: {review_content}")
+        logger.info("=" * 80)
+        
+        start_time = time.time()
+        trace_id = "sensitive_check" + trace_id_idx
+        
+        # 第一步：使用关键词匹配式审查
+        logger.info("步骤1: 开始关键词匹配检测...")
+        try:
+            first_results = await check_sensitive_words_async(review_content)
+        except Exception as e:
+            logger.error(f"关键词检测失败: {e}")
+            logger.exception("详细错误:")
+            return None
+        
+        detection_time = time.time() - start_time
+        logger.info(f"步骤1: 关键词检测完成，耗时: {detection_time:.4f}s")
+        
+        # 判断是否检测到敏感词
+        if first_results:
+            logger.warning(f"步骤2: 检测到 {len(first_results)} 个敏感词，准备送入大模型二审")
+            logger.warning("-" * 80)
+            
+            # 格式化敏感词列表
+            sensitive_words_info = []
+            for idx, item in enumerate(first_results, 1):
+                word_info = f"敏感词: {item['word']}, 位置: {item['position']}-{item['end_position']}, 来源: {item['source']}"
+                sensitive_words_info.append(word_info)
+                logger.warning(f"  敏感词 #{idx}: {word_info}")
+            
+            formatted_sensitive_words = "\n".join(sensitive_words_info)
+            logger.warning("-" * 80)
+            
+            # 调用大模型进行二审（与原函数保持一致）
+            logger.info("步骤3: 调用大模型进行二审...")
+            try:
+                if not self.ai_review_engine:
+                    logger.error("AI审查引擎未初始化")
+                    return None
+                
+                # 调用 review 方法进行大模型二审
+                result = await self.ai_review_engine.review(
+                    "sensitive_check", 
+                    trace_id, 
+                    "basic", 
+                    "sensitive_word_check",
+                    review_content, 
+                    formatted_sensitive_words,
+                    None, 
+                    review_location_label, 
+                    state, 
+                    stage_name
+                )
+                
+                llm_review_time = time.time() - start_time
+                logger.info(f"步骤3: 大模型二审完成，总耗时: {llm_review_time:.4f}s")
+                logger.info("=" * 80)
+                
+                return result
+                
+            except Exception as e:
+                logger.error(f"大模型二审失败: {e}")
+                logger.exception("详细错误:")
+                return None
+            
+        else:
+            # 没有检测到敏感词，构造返回体
+            logger.info("步骤2: 未检测到敏感词")
+            
+            execution_time = time.time() - start_time
+            result = ReviewResult(
+                success=True,
+                details={"name": "sensitive_check", "response": "无明显问题"},
+                error_message=None,
+                execution_time=execution_time
+            )
+            
+            logger.info(f"步骤3: 敏感词检查完成（未检测到敏感词），总耗时: {execution_time:.4f}s")
+            logger.info("=" * 80)
+            
+            return result
+
+    def get_custom_text_from_user(self) -> Optional[str]:
+        """从用户获取自定义检查文本"""
+        logger.info("\n" + "=" * 80)
+        logger.info("请输入要检查的文本内容")
+        logger.info("=" * 80)
+        logger.info("提示：")
+        logger.info("  1. 可以输入多行文本，输入完成后单独一行输入 'END' 结束")
+        logger.info("  2. 直接按回车（输入空行）将跳过自定义输入")
+        logger.info("=" * 80)
+        
+        lines = []
+        first_line = True
+        
+        while True:
+            try:
+                if first_line:
+                    line = input(">>> ")
+                    first_line = False
+                    
+                    # 如果第一行就是空行，跳过自定义输入
+                    if not line.strip():
+                        logger.info("跳过自定义文本输入")
+                        return None
+                else:
+                    line = input("... ")
+                
+                # 检查是否结束输入
+                if line.strip().upper() == 'END':
+                    break
+                    
+                lines.append(line)
+                
+            except EOFError:
+                break
+            except KeyboardInterrupt:
+                logger.info("\n用户取消输入")
+                return None
+        
+        custom_text = "\n".join(lines)
+        
+        if not custom_text.strip():
+            logger.info("未输入有效文本")
+            return None
+        
+        logger.info(f"\n已接收文本，长度: {len(custom_text)} 字符")
+        return custom_text
+
+    async def run_custom_text_check(self, custom_text: str):
+        """运行自定义文本检查"""
+        logger.info("\n" + "█" * 80)
+        logger.info("执行自定义文本敏感词检查")
+        logger.info("█" * 80)
+        
+        try:
+            # 执行检查
+            result = await self.check_sensitive_with_llm_review(
+                trace_id_idx="_custom_001",
+                review_content=custom_text,
+                review_location_label="自定义文本检查",
+                stage_name="自定义测试"
+            )
+            
+            # 显示结果
+            logger.info("\n" + "-" * 80)
+            logger.info("检查结果:")
+            logger.info("-" * 80)
+            
+            if result:
+                logger.info(f"  返回类型: {type(result).__name__}")
+                logger.info(f"  是否成功: {result.success}")
+                logger.info(f"  执行时间: {result.execution_time:.4f}s")
+                logger.info(f"  详细信息: {result.details}")
+                
+                if result.error_message:
+                    logger.error(f"  错误信息: {result.error_message}")
+                
+                if not result.success:
+                    logger.warning("\n检测到敏感内容，请查看上方详细信息")
+                else:
+                    logger.info("\n✓ 未检测到敏感内容")
+            else:
+                logger.error("检查失败，未返回结果")
+            
+            logger.info("-" * 80)
+            
+        except Exception as e:
+            logger.error(f"✗ 检查异常: {str(e)}")
+            logger.exception("详细异常信息:")
+
+    async def run_interactive_mode(self):
+        """运行交互模式"""
+        logger.info("\n" + "█" * 80)
+        logger.info("敏感词检查 - 交互模式")
+        logger.info("█" * 80)
+        
+        # 初始化AI审查引擎
+        if not self.initialize_engine():
+            logger.error("AI审查引擎初始化失败，无法继续")
+            return False
+        
+        # 初始化敏感词检测器
+        try:
+            from core.construction_review.component.reviewers.utils import SensitiveWordChecker
+            logger.info("\n正在初始化敏感词检测器...")
+            stats = SensitiveWordChecker.initialize()
+            logger.info(f"敏感词检测器初始化成功: {stats}")
+        except Exception as e:
+            logger.error(f"敏感词检测器初始化失败: {str(e)}")
+            logger.exception("详细错误:")
+            return False
+        
+        # 获取自定义文本
+        custom_text = self.get_custom_text_from_user()
+        
+        if custom_text:
+            # 执行自定义文本检查
+            await self.run_custom_text_check(custom_text)
+        else:
+            logger.info("\n未提供自定义文本，程序结束")
+        
+        return True
+
+    def run_interactive_sync(self):
+        """同步方式运行交互模式"""
+        return asyncio.run(self.run_interactive_mode())
+
+
+def main():
+    """主函数"""
+    print("=" * 80)
+    print("敏感词检查功能单元测试（完整版本）")
+    print("测试文件: test_sensitive_check_standalone.py")
+    print("=" * 80)
+    
+    tester = TestSensitiveCheck()
+    success = tester.run_interactive_sync()
+    
+    sys.exit(0 if success else 1)
+
+
+if __name__ == "__main__":
+    main()