v0.0.3-sse进度推送与审查事件类型更新

core/base/progress_manager.py

@@ -1,8 +1,3 @@
-"""
-任务进度管理器
-负责任务进度的存储、更新和查询
-"""
-
 import json
 import asyncio
 from typing import Dict, Any, Optional
@@ -12,9 +7,9 @@ from foundation.logger.loggering import server_logger as logger
 from foundation.base.config import config_handler
 
 class SSECallbackManager:
-    """SSE回调管理器 - 单例模式管理全局SSE回调"""
+    """SSE回调管理器 - 单例模式"""
     _instance = None
-    _callbacks = {}  # {callback_task_id: callback_function}
+    _callbacks = {}
 
     def __new__(cls):
         if cls._instance is None:
@@ -22,27 +17,21 @@ class SSECallbackManager:
         return cls._instance
 
     def register_callback(self, callback_task_id: str, callback_func):
-        """注册SSE回调函数"""
         self._callbacks[callback_task_id] = callback_func
         logger.info(f"SSE回调注册, 当前注册数: {len(self._callbacks)}")
 
     def unregister_callback(self, callback_task_id: str):
-        """注销SSE回调函数"""
         if callback_task_id in self._callbacks:
             del self._callbacks[callback_task_id]
             logger.info(f"SSE回调注销, 剩余注册数: {len(self._callbacks)}")
 
     async def trigger_callback(self, callback_task_id: str, current_data: dict):
-        """触发SSE回调"""
         if callback_task_id in self._callbacks:
             try:
-                # 直接异步执行回调，保持trace上下文
                 await self._callbacks[callback_task_id](callback_task_id, current_data)
                 logger.debug(f"SSE回调执行成功: {callback_task_id}")
-
                 logger.debug(f"SSE回调已触发: {callback_task_id}, 当前注册回调数: {len(self._callbacks)}")
                 return True
-
             except Exception as e:
                 logger.error(f"SSE回调执行失败: {callback_task_id}, {e}")
                 return False
@@ -51,19 +40,16 @@ class SSECallbackManager:
             return False
 
     def get_callbacks_count(self):
-        """获取当前回调数量"""
         return len(self._callbacks)
 
     def clear_all_callbacks(self):
-        """清空所有回调"""
         self._callbacks.clear()
         logger.info("已清空所有SSE回调")
 
-# 全局SSE回调管理器实例
 sse_callback_manager = SSECallbackManager()
 
 class ProgressManager:
-    """任务进度管理器 - 增长型进度管理版本"""
+    """任务进度管理器"""
 
     def __init__(self):
         self.redis_client = None
@@ -71,7 +57,6 @@ class ProgressManager:
         self._init_redis()
 
     def _init_redis(self):
-        """初始化Redis连接"""
         try:
             import redis
 
@@ -80,7 +65,6 @@ class ProgressManager:
             redis_password = config_handler.get('redis', 'REDIS_PASSWORD', '')
             redis_db = config_handler.get('redis', 'REDIS_DB', '0')
 
-            # 构建Redis连接URL
             if redis_password:
                 redis_url = f"redis://:{redis_password}@{redis_host}:{redis_port}/{redis_db}"
             else:
@@ -88,10 +72,7 @@ class ProgressManager:
 
             logger.debug(f"ProgressManager连接Redis: {redis_url}")
 
-            # 连接Redis
             self.redis_client = redis.from_url(redis_url, decode_responses=True)
-
-            # 测试连接
             self.redis_client.ping()
             self.redis_connected = True
             logger.debug(f"ProgressManager Redis连接成功: {redis_host}:{redis_port}")
@@ -100,20 +81,13 @@ class ProgressManager:
             logger.error(f"ProgressManager Redis连接失败: {e}")
             self.redis_connected = False
             logger.warning("ProgressManager将使用内存存储作为备选方案")
-            self.current_data = {}  # 备选内存存储
+            self.current_data = {}
 
     async def _get_redis_key(self, callback_task_id: str) -> str:
-        """获取Redis键名"""
         return f"current:{callback_task_id}"
 
     async def initialize_progress(self, callback_task_id: str, user_id: str, stages: list):
-        """初始化进度记录"""
         try:
-
-            # 设置总量为100（百分比模式）
-            stage_name = stages[0]["stage_name"] if stages else ""
-            message = "任务开始"
-
             current_data = {
                 "user_id": user_id,
                 "current": 0,
@@ -125,24 +99,21 @@ class ProgressManager:
             }
 
             if self.redis_connected:
-                # 使用同步Redis操作避免异步任务销毁问题
                 try:
                     redis_key = await self._get_redis_key(callback_task_id)
                     self.redis_client.setex(
                         redis_key,
-                        3600,  # 1小时过期
+                        3600,
                         json.dumps(current_data)
                     )
                     logger.info(f"初始化任务进度列表")
                 except Exception as redis_e:
                     logger.warning(f"初始化进度到Redis失败: {callback_task_id}, {redis_e}")
-                    # 降级到内存存储
                     if not hasattr(self, 'current_data'):
                         self.current_data = {}
                     self.current_data[callback_task_id] = current_data
                     logger.info(f"降级使用内存存储: {callback_task_id}")
             else:
-                # 使用内存存储
                 if not hasattr(self, 'current_data'):
                     self.current_data = {}
                 self.current_data[callback_task_id] = current_data
@@ -152,8 +123,19 @@ class ProgressManager:
             logger.error(f"初始化进度失败: {str(e)}")
             raise
 
-    async def update_stage_progress(self, callback_task_id: str, stage_name: str, current: int, status: str, message: str = ""):
-        """更新阶段进度"""
+    async def update_stage_progress(self, callback_task_id: str, stage_name: str = None, current: int = None, status: str = None, message: str = None, issues=None, user_id: str = None, overall_task_status: str = None):
+        """更新阶段进度 - 除callback_task_id外，其他参数都可选
+
+        Args:
+            callback_task_id: 回调任务ID（必需）
+            stage_name: 阶段名称（可选）
+            current: 当前进度（可选）
+            status: 状态（可选）
+            message: 消息（可选）
+            issues: 问题列表（可选）
+            user_id: 用户ID（可选）
+            overall_task_status: 整体任务状态（可选）
+        """
         try:
             task_progress = None
 
@@ -174,15 +156,25 @@ class ProgressManager:
                     logger.warning(f"内存中未找到任务进度: {callback_task_id}")
                     return
 
-            # 更新进度数据
-            task_progress["current"] = current
-            task_progress["stage_name"] = stage_name
-            task_progress["status"] = status
-            task_progress["message"] = message
+            # 更新进度数据 - 只有非空参数才更新
+            if current is not None:
+                task_progress["current"] = current
+            if stage_name is not None:
+                task_progress["stage_name"] = stage_name
+            if status is not None:
+                task_progress["status"] = status
+            if message is not None:
+                task_progress["message"] = message
             task_progress["updated_at"] = datetime.now().isoformat()
-
-            # 保留overall_task_status字段，不要被普通进度更新覆盖
-            if "overall_task_status" not in task_progress:
+            if issues is not None:
+                task_progress["issues"] = issues
+            else:
+                task_progress["issues"] = []
+            if user_id is not None:
+                task_progress["user_id"] = user_id
+            if overall_task_status is not None:
+                task_progress["overall_task_status"] = overall_task_status
+            elif "overall_task_status" not in task_progress:
                 task_progress["overall_task_status"] = "processing"
 
             try:
@@ -215,7 +207,10 @@ class ProgressManager:
             # 触发SSE推送 - 使用全局回调管理器
             logger.debug(f"触发SSE推送: {callback_task_id}")
             updated_progress = await self.get_progress(callback_task_id)
-            if updated_progress:
+            issues = task_progress.get("issues")
+            if updated_progress and issues and len(issues) > 0 and issues[0] != 'clear':
+                await sse_callback_manager.trigger_callback(callback_task_id, updated_progress)
+            elif updated_progress and not issues:  # 空列表时也要推送
                 await sse_callback_manager.trigger_callback(callback_task_id, updated_progress)
 
         except Exception as e:
@@ -257,7 +252,8 @@ class ProgressManager:
             else:
                 updated_at_timestamp = int(updated_at.timestamp())
 
-            return {
+            # 构建返回数据
+            result = {
                 "callback_task_id": callback_task_id,
                 "user_id": task_progress["user_id"],
                 "current": task_progress["current"],
@@ -268,6 +264,12 @@ class ProgressManager:
                 "updated_at": updated_at_timestamp
             }
 
+            # 添加可选字段
+            if "issues" in task_progress:
+                result["issues"] = task_progress["issues"]
+
+            return result
+
         except Exception as e:
             logger.error(f"获取进度失败: {str(e)}")
             return None

core/base/workflow_manager.py

@@ -268,8 +268,3 @@ class WorkflowManager:
             if task_chain.callback_task_id in self.active_chains:
                 del self.active_chains[task_chain.callback_task_id]
 
-
-
-    async def update_task_status(self, callback_task_id: str) -> Optional[Dict]:
-        """更新任务状态"""
-        pass

core/construction_review/component/ai_review_engine.py

@@ -87,14 +87,23 @@ class AIReviewEngine(BaseReviewer):
 
         grammar_result, semantic_result, completeness_result = await asyncio.gather(*basic_tasks, return_exceptions=True)
 
-
-
-        if isinstance(grammar_result, Exception):
-            grammar_result = {"error": str(grammar_result), "success": False}
-        if isinstance(semantic_result, Exception):
-            semantic_result = {"error": str(semantic_result), "success": False}
-        if isinstance(completeness_result, Exception):
-            completeness_result = {"error": str(completeness_result), "success": False}
+        def process_result(result):
+            """处理审查结果，统一转换为字典格式"""
+            if isinstance(result, Exception):
+                return {"error": str(result), "success": False}
+            elif hasattr(result, '__dict__'):  # ReviewResult对象
+                return {
+                    "success": result.success if hasattr(result, 'success') else False,
+                    "details": result.details if hasattr(result, 'details') else {},
+                    "error_message": result.error_message if hasattr(result, 'error_message') else None,
+                    "execution_time": result.execution_time if hasattr(result, 'execution_time') else None
+                }
+            else:
+                return result  # 已经是字典
+
+        grammar_result = process_result(grammar_result)
+        semantic_result = process_result(semantic_result)
+        completeness_result = process_result(completeness_result)
 
         return {
             'grammar_check': grammar_result,
@@ -121,13 +130,23 @@ class AIReviewEngine(BaseReviewer):
 
         mandatory_result, design_value_result, technical_param_result = await asyncio.gather(*technical_tasks, return_exceptions=True)
 
-        # 处理异常结果
-        if isinstance(mandatory_result, Exception):
-            mandatory_result = {"error": str(mandatory_result), "success": False}
-        if isinstance(design_value_result, Exception):
-            design_value_result = {"error": str(design_value_result), "success": False}
-        if isinstance(technical_param_result, Exception):
-            technical_param_result = {"error": str(technical_param_result), "success": False}
+        def process_result(result):
+            """处理审查结果，统一转换为字典格式"""
+            if isinstance(result, Exception):
+                return {"error": str(result), "success": False}
+            elif hasattr(result, '__dict__'):  # ReviewResult对象
+                return {
+                    "success": result.success if hasattr(result, 'success') else False,
+                    "details": result.details if hasattr(result, 'details') else {},
+                    "error_message": result.error_message if hasattr(result, 'error_message') else None,
+                    "execution_time": result.execution_time if hasattr(result, 'execution_time') else None
+                }
+            else:
+                return result  # 已经是字典
+
+        mandatory_result = process_result(mandatory_result)
+        design_value_result = process_result(design_value_result)
+        technical_param_result = process_result(technical_param_result)
 
         return {
             'mandatory_standards': mandatory_result,
@@ -157,11 +176,11 @@ class AIReviewEngine(BaseReviewer):
 
     async def check_grammar(self, trace_id_idx: str, review_content: str = None, review_references: str = None,
                           stage_name: str = None, state: dict = None, current_progress: int = None) -> Dict[str, Any]:
-        """语法检查"""
+        """词句语法检查"""
         reviewer_type = Stage.BASIC.value['reviewer_type']
         prompt_name = Stage.BASIC.value['sensitive']
         trace_id = prompt_name+trace_id_idx
-        return await self.review("语法检查", trace_id, reviewer_type, prompt_name, review_content, review_references,
+        return await self.review("词句语法检查", trace_id, reviewer_type, prompt_name, review_content, review_references,
                                stage_name, state, current_progress)
 
     async def check_semantic_logic(self, trace_id_idx: str, review_content: str = None, review_references: str = None,

core/construction_review/component/document_processor.py

@@ -66,8 +66,8 @@ class DocumentProcessor:
             # 结构化内容
             structured_result = self.structure_content(result)
 
-            if progress_callback:
-                progress_callback(100, "文档处理完成")
+            # if progress_callback:
+            #     progress_callback(100, "文档处理完成")
 
             return structured_result
 

core/construction_review/component/reviewers/base_reviewer.py

@@ -62,19 +62,18 @@ class BaseReviewer(ABC):
         try:
             # 添加进度更新
             # 安全检查：确保所有必要参数都存在才执行进度更新
-            if state and state.get("progress_manager") and stage_name and current_progress is not None:
-                asyncio.create_task(
-                    state["progress_manager"].update_stage_progress(
-                        callback_task_id=state["callback_task_id"],
-                        stage_name=stage_name,
-                        current=current_progress,
-                        status="processing",
-                        message=f"开始进行：{name}"
-                    )
-                )
+            # if state and state.get("progress_manager") and stage_name and current_progress is not None:
+            #     asyncio.create_task(
+            #         state["progress_manager"].update_stage_progress(
+            #             callback_task_id=state["callback_task_id"],
+            #             stage_name=stage_name,
+            #             current=current_progress,
+            #             status="processing",
+            #             message=f"开始进行：{name}"
+            #         )
+            #     )
             logger.info(f"开始执行 {name} 审查，trace_id: {trace_id},内容长度: {len(review_content)}")
             prompt_kwargs = {}
-            prompt_kwargs["content"] = review_content
             prompt_kwargs["review_content"] = review_content
             prompt_kwargs["review_references"] = review_references or ""
             # 添加审查位置标签到提示词参数中
@@ -92,16 +91,27 @@ class BaseReviewer(ABC):
             )
 
             # 格式化结果
-            result = self.format_result(model_response)
+            result = self.format_result(model_response, name)
             result.execution_time = time.time() - start_time
 
+            # 将审查结果转换为字典格式，添加到issues中
+            review_result_data = {
+                'name': name,
+                '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=current_progress,
                     status="processing",
-                    message=f"{name} 审查完成， 耗时: {result.execution_time:.2f}s"
+                    message=f"{name} 审查完成， 耗时: {result.execution_time:.2f}s",
+                    issues=[review_result_data]
                 )
             )
             logger.info(f"{name} 审查完成， 耗时: {result.execution_time:.2f}s")
@@ -120,13 +130,17 @@ class BaseReviewer(ABC):
                 execution_time=execution_time
             )
 
-    def format_result(self, model_response: str) -> ReviewResult:
+    def format_result(self, model_response: str, name: str = None) -> ReviewResult:
         """
         格式化模型返回结果为ReviewResult对象
         """
+        details = {"response": model_response}
+        if name:
+            details["name"] = name
+
         return ReviewResult(
             success=True,
-            details={"response": model_response},
+            details=details,
             error_message=None
         )
 

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

@@ -5,85 +5,85 @@ professional_suggestion:
   system_prompt: |
     你是施工方案专业性优化建议专家，负责提供专业技术改进建议。
 
-    审查要求：
+    审查要求:
     - 重点分析技术方案的专业性和合理性
     - 提供具体的专业改进建议
     - 简明扼要指出优化方向和实施方法
-    - 标注建议优先级：高/中/低
+    - 标注建议优先级:高/中/低
 
   user_prompt_template: |
-    请为以下施工方案提供专业性优化建议：
+    请为以下施工方案提供专业性优化建议:
 
     {review_content}
 
-    输出格式：
-    专业问题：[具体专业问题]
-    优化建议：[具体改进措施]
-    实施方法：[具体实施步骤]
-    优先级：[高/中/低]
+    输出格式:
+    专业问题:[具体专业问题]
+    优化建议:[具体改进措施]
+    实施方法:[具体实施步骤]
+    优先级:[高/中/低]
 
 # 3.2 规范性优化建议
 standardization_suggestion:
   system_prompt: |
     你是施工方案规范性优化建议专家，负责提供标准化改进建议。
 
-    审查要求：
+    审查要求:
     - 重点分析方案的规范性和标准化程度
     - 提供具体的标准化改进建议
     - 简明扼要指出规范化要求和实施标准
-    - 标注改进紧迫性：高/中/低
+    - 标注改进紧迫性:高/中/低
 
   user_prompt_template: |
-    请为以下施工方案提供规范性优化建议：
+    请为以下施工方案提供规范性优化建议:
 
     {review_content}
 
-    输出格式：
-    规范问题：[具体规范性问题]
-    改进建议：[具体改进措施]
-    参考标准：[相关标准规范]
-    紧迫性：[高/中/低]
+    输出格式:
+    规范问题:[具体规范性问题]
+    改进建议:[具体改进措施]
+    参考标准:[相关标准规范]
+    紧迫性:[高/中/低]
 
 # 3.3 完整性优化建议
 completeness_suggestion:
   system_prompt: |
     你是施工方案完整性优化建议专家，负责提供内容完整性改进建议。
 
-    审查要求：
+    审查要求:
     - 重点分析方案内容的完整性和系统性
     - 提供具体的内容补充建议
     - 简明扼要指出缺失内容和补充要求
-    - 标注补充重要性：关键/重要/一般
+    - 标注补充重要性:关键/重要/一般
 
   user_prompt_template: |
-    请为以下施工方案提供完整性优化建议：
+    请为以下施工方案提供完整性优化建议:
 
     {review_content}
 
-    输出格式：
-    缺失内容：[具体缺失内容]
-    补充建议：[具体补充措施]
-    补充要求：[详细补充要求]
-    重要性：[关键/重要/一般]
+    输出格式:
+    缺失内容:[具体缺失内容]
+    补充建议:[具体补充措施]
+    补充要求:[详细补充要求]
+    重要性:[关键/重要/一般]
 
 # 3.4 可读性优化建议
 readability_suggestion:
   system_prompt: |
     你是施工方案可读性优化建议专家，负责提供文档可读性改进建议。
 
-    审查要求：
+    审查要求:
     - 重点分析方案的可读性和表达清晰度
     - 提供具体的可读性改进建议
     - 简明扼要指出表达问题和优化方法
-    - 标注改进效果：显著/一般/轻微
+    - 标注改进效果:显著/一般/轻微
 
   user_prompt_template: |
-    请为以下施工方案提供可读性优化建议：
+    请为以下施工方案提供可读性优化建议:
 
     {review_content}
 
-    输出格式：
-    表达问题：[具体表达问题]
-    优化建议：[具体改进措施]
-    优化方法：[具体优化方法]
-    改进效果：[显著/一般/轻微]
+    输出格式:
+    表达问题:[具体表达问题]
+    优化建议:[具体改进措施]
+    优化方法:[具体优化方法]
+    改进效果:[显著/一般/轻微]

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

@@ -5,157 +5,167 @@ semantic_logic_check:
   system_prompt: |
     你是施工方案语义逻辑审查专家，负责检查表述一致性和逻辑清晰度。
 
-    审查要求：
+    审查要求:
     - 重点关注语义矛盾、逻辑混乱、表述不清问题
     - 简明扼要指出问题位置和修改建议
-    - 风险等级分类：
-      * 高风险：影响审查结论、可能导致法律问题或严重安全隐患
-      * 中风险：影响专业表达、可能导致理解偏差或一般性问题
-      * 低风险：形式问题、不影响实质内容和安全
+    - 风险等级分类:
+      * 高风险:影响审查结论、可能导致法律问题或严重安全隐患
+      * 中风险:影响专业表达、可能导致理解偏差或一般性问题
+      * 低风险:形式问题、不影响实质内容和安全
 
-    审查参考：
+    审查参考:
     {review_references}
 
   user_prompt_template: |
-    请审查以下内容的语义逻辑：
+    请审查以下内容的语义逻辑:
 
     {review_content}
 
-    输出格式：务必须严格按照以下标准Markdown格式输出审查结果：
-    ## 问题：问题标题描述
+    输出格式:务必须严格按照以下标准Markdown格式输出审查结果:
+    如果未发现问题，请输出:无明显问题
+    如果发现问题，请按以下格式输出:
+    ## 问题:问题标题描述
 
-    **位置**：{review_location_label}
+    **位置**:{review_location_label}
 
-    **建议**：具体的修改建议内容
+    **建议**:具体的修改建议内容
 
-    **理由**：问题的原因分析和依据说明。
+    **理由**:问题的原因分析和依据说明。
 
-    **风险等级**：
+    **风险等级**:
 
 # 1.2 条文完整性检查功能
 completeness_check:
   system_prompt: |
     你是施工方案完整性审查专家，负责检查必备条文和技术要素完备性。
 
-    审查要求：
+    审查要求:
     - 重点关注法规条文、技术要素缺失
     - 简明扼要列出缺失内容和补充要求
-    - 风险等级分类：
-      * 高风险：影响审查结论、可能导致法律问题或严重安全隐患
-      * 中风险：影响专业表达、可能导致理解偏差或一般性问题
-      * 低风险：形式问题、不影响实质内容和安全
+    - 风险等级分类:
+      * 高风险:影响审查结论、可能导致法律问题或严重安全隐患
+      * 中风险:影响专业表达、可能导致理解偏差或一般性问题
+      * 低风险:形式问题、不影响实质内容和安全
 
-    审查参考：
+    审查参考:
     {review_references}
 
   user_prompt_template: |
-    请审查以下内容的条文完整性：
+    请审查以下内容的条文完整性:
 
     {review_content}
 
-    输出格式：务必须严格按照以下标准Markdown格式输出审查结果：
-    ## 问题：问题标题描述
+    输出格式:务必须严格按照以下标准Markdown格式输出审查结果:
+    如果未发现问题，请输出:无明显问题
+    如果发现问题，请按以下格式输出:
+    ## 问题:问题标题描述
 
-    **位置**：{review_location_label}
+    **位置**:{review_location_label}
 
-    **建议**：具体的修改建议内容
+    **建议**:具体的修改建议内容
 
-    **理由**：问题的原因分析和依据说明。
+    **理由**:问题的原因分析和依据说明。
 
-    **风险等级**：
+    **风险等级**:
 
 # 1.3 时效性检查功能
 timeliness_check:
   system_prompt: |
     你是施工方案时效性审查专家，负责检查引用标准和技术要求的现行有效性。
 
-    审查要求：
+    审查要求:
     - 重点关注标准版本是否过期、技术要求滞后，分为 现行、过期、废止
     - 简明扼要指出过时内容和更新要求
-    - 风险等级分类：
-      * 高风险：影响审查结论、可能导致法律问题或严重安全隐患
-      * 中风险：影响专业表达、可能导致理解偏差或一般性问题
-      * 低风险：形式问题、不影响实质内容和安全
+    - 风险等级分类:
+      * 高风险:影响审查结论、可能导致法律问题或严重安全隐患
+      * 中风险:影响专业表达、可能导致理解偏差或一般性问题
+      * 低风险:形式问题、不影响实质内容和安全
 
-    审查参考：
+    审查参考:
     {review_references}
 
   user_prompt_template: |
-    请审查以下内容的时效性：
+    请审查以下内容的时效性:
 
     {review_content}
 
-    输出格式：务必须严格按照以下标准Markdown格式输出审查结果：
-    ## 问题：问题标题描述
+    输出格式:务必须严格按照以下标准Markdown格式输出审查结果:
+    如果未发现问题，请输出:无明显问题
+    如果发现问题，请按以下格式输出:
+    ## 问题:问题标题描述
 
-    **位置**：{review_location_label}
+    **位置**:{review_location_label}
 
-    **建议**：具体的修改建议内容
+    **建议**:具体的修改建议内容
 
-    **理由**：问题的原因分析和依据说明。
+    **理由**:问题的原因分析和依据说明。
 
-    **风险等级**：
+    **风险等级**:
 
 # 1.4 引用规范检查功能
 reference_check:
   system_prompt: |
     你是施工方案引用规范审查专家，负责检查引用格式和内容准确性。
 
-    审查要求：
+    审查要求:
     - 重点关注引用格式错误、内容不准确
     - 简明扼要指出引用问题和修正方法
-    - 风险等级分类：
-      * 高风险：影响审查结论、可能导致法律问题或严重安全隐患
-      * 中风险：影响专业表达、可能导致理解偏差或一般性问题
-      * 低风险：形式问题、不影响实质内容和安全
+    - 风险等级分类:
+      * 高风险:影响审查结论、可能导致法律问题或严重安全隐患
+      * 中风险:影响专业表达、可能导致理解偏差或一般性问题
+      * 低风险:形式问题、不影响实质内容和安全
 
-    审查参考：
+    审查参考:
     {review_references}
 
   user_prompt_template: |
-    请审查以下内容的引用规范：
+    请审查以下内容的引用规范:
 
     {review_content}
 
-    输出格式：务必须严格按照以下标准Markdown格式输出审查结果：
-    ## 问题：问题标题描述
+    输出格式:务必须严格按照以下标准Markdown格式输出审查结果:
+    如果未发现问题，请输出:无明显问题
+    如果发现问题，请按以下格式输出:
+    ## 问题:问题标题描述
 
-    **位置**：{review_location_label}
+    **位置**:{review_location_label}
 
-    **建议**：具体的修改建议内容
+    **建议**:具体的修改建议内容
 
-    **理由**：问题的原因分析和依据说明。
+    **理由**:问题的原因分析和依据说明。
 
-    **风险等级**：
+    **风险等级**:
 
 # 1.5 敏感词检查功能
 sensitive_word_check:
   system_prompt: |
     你是施工方案敏感词审查专家，负责检查政治敏感和表述适宜性问题。
 
-    审查要求：
+    审查要求:
     - 重点关注政治敏感、商业机密、表述不当
     - 简明扼要指出敏感词汇和处理建议
-    - 风险等级分类：
-      * 高风险：影响审查结论、可能导致法律问题或严重安全隐患
-      * 中风险：影响专业表达、可能导致理解偏差或一般性问题
-      * 低风险：形式问题、不影响实质内容和安全
+    - 风险等级分类:
+      * 高风险:影响审查结论、可能导致法律问题或严重安全隐患
+      * 中风险:影响专业表达、可能导致理解偏差或一般性问题
+      * 低风险:形式问题、不影响实质内容和安全
 
-    审查参考：
+    审查参考:
     {review_references}
 
   user_prompt_template: |
-    请审查以下内容的敏感词：
+    请审查以下内容的敏感词:
 
     {review_content}
 
-    输出格式：务必须严格按照以下标准Markdown格式输出审查结果：
-    ## 问题：问题标题描述
+    输出格式:务必须严格按照以下标准Markdown格式输出审查结果:
+    如果未发现问题，请输出:无明显问题
+    如果发现问题，请按以下格式输出:
+    ## 问题:问题标题描述
 
-    **位置**：{review_location_label}
+    **位置**:{review_location_label}
 
-    **建议**：具体的修改建议内容
+    **建议**:具体的修改建议内容
 
-    **理由**：问题的原因分析和依据说明。
+    **理由**:问题的原因分析和依据说明。
 
-    **风险等级**：
+    **风险等级**:

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

@@ -5,96 +5,102 @@ mandatory_standards_check:
   system_prompt: |
     你是施工方案强制性标准审查专家，负责检查是否符合国家强制性标准。
 
-    审查参考：
+    审查参考:
     {review_references}
 
-    审查要求：
+    审查要求:
     - 重点检查强制性条文符合性
     - 识别违反强标的问题
     - 简明扼要指出违规内容和整改要求
-    - 风险等级分类：
-      * 高风险：影响审查结论、可能导致法律问题或严重安全隐患
-      * 中风险：影响专业表达、可能导致理解偏差或一般性问题
-      * 低风险：形式问题、不影响实质内容和安全
+    - 风险等级分类:
+      * 高风险:影响审查结论、可能导致法律问题或严重安全隐患
+      * 中风险:影响专业表达、可能导致理解偏差或一般性问题
+      * 低风险:形式问题、不影响实质内容和安全
 
   user_prompt_template: |
-    请审查以下内容的强制性标准符合性：
+    请审查以下内容的强制性标准符合性:
 
     {review_content}
 
-    输出格式：务必须严格按照以下标准Markdown格式输出审查结果：
-    ## 问题：问题标题描述
+    输出格式:务必须严格按照以下标准Markdown格式输出审查结果:
+    如果未发现问题，请输出:无明显问题
+    如果发现问题，请按以下格式输出:
+    ## 问题:问题标题描述
 
-    **位置**：{review_location_label}
+    **位置**:{review_location_label}
 
-    **建议**：具体的修改建议内容
+    **建议**:具体的修改建议内容
 
-    **理由**：问题的原因分析和依据说明。
+    **理由**:问题的原因分析和依据说明。
 
-    **风险等级**：
+    **风险等级**:
 
 # 2.2 技术参数精确检查功能
 technical_parameters_check:
   system_prompt: |
     你是施工方案技术参数审查专家，负责检查技术参数的精确性和合理性。
 
-    审查要求：
+    审查要求:
     - 重点检查技术参数的准确性和合理性
     - 识别参数错误或不合理设置
     - 简明扼要指出参数问题和修正建议
-    - 风险等级分类：
-      * 高风险：影响审查结论、可能导致法律问题或严重安全隐患
-      * 中风险：影响专业表达、可能导致理解偏差或一般性问题
-      * 低风险：形式问题、不影响实质内容和安全
+    - 风险等级分类:
+      * 高风险:影响审查结论、可能导致法律问题或严重安全隐患
+      * 中风险:影响专业表达、可能导致理解偏差或一般性问题
+      * 低风险:形式问题、不影响实质内容和安全
 
-    审查参考：
+    审查参考:
     {review_references}
 
   user_prompt_template: |
-    请审查以下内容的技术参数精确性：
+    请审查以下内容的技术参数精确性:
 
     {review_content}
 
-    输出格式：务必须严格按照以下标准Markdown格式输出审查结果：
-    ## 问题：问题标题描述
+    输出格式:务必须严格按照以下标准Markdown格式输出审查结果:
+    如果未发现问题，请输出:无明显问题
+    如果发现问题，请按以下格式输出:
+    ## 问题:问题标题描述
 
-    **位置**：{review_location_label}
+    **位置**:{review_location_label}
 
-    **建议**：具体的修改建议内容
+    **建议**:具体的修改建议内容
 
-    **理由**：问题的原因分析和依据说明。
+    **理由**:问题的原因分析和依据说明。
 
-    **风险等级**：
+    **风险等级**:
 
 # 2.3 设计值符合性检查功能
 design_values_check:
   system_prompt: |
     你是施工方案设计值审查专家，负责检查设计值的符合性和合理性。
 
-    审查要求：
+    审查要求:
     - 重点检查设计值与标准的符合性
     - 识别设计值超标或不合理问题
     - 简明扼要指出设计值问题和调整建议
-    - 风险等级分类：
-      * 高风险：影响审查结论、可能导致法律问题或严重安全隐患
-      * 中风险：影响专业表达、可能导致理解偏差或一般性问题
-      * 低风险：形式问题、不影响实质内容和安全
+    - 风险等级分类:
+      * 高风险:影响审查结论、可能导致法律问题或严重安全隐患
+      * 中风险:影响专业表达、可能导致理解偏差或一般性问题
+      * 低风险:形式问题、不影响实质内容和安全
 
-    审查参考：
+    审查参考:
     {review_references}
 
   user_prompt_template: |
-    请审查以下内容的设计值符合性：
+    请审查以下内容的设计值符合性:
 
     {review_content}
 
-    输出格式：务必须严格按照以下标准Markdown格式输出审查结果：
-    ## 问题：问题标题描述
+    输出格式:务必须严格按照以下标准Markdown格式输出审查结果:
+    如果未发现问题，请输出:无明显问题
+    如果发现问题，请按以下格式输出:
+    ## 问题:问题标题描述
 
-    **位置**：{review_location_label}
+    **位置**:{review_location_label}
 
-    **建议**：具体的修改建议内容
+    **建议**:具体的修改建议内容
 
-    **理由**：问题的原因分析和依据说明。
+    **理由**:问题的原因分析和依据说明。
 
-    **风险等级**：
+    **风险等级**:

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

@@ -137,7 +137,7 @@ class PromptLoader:
             # 返回默认提示词
             return {
                 'system_prompt': f"你是专业的施工方案审查专家，负责进行{prompt_name}审查。",
-                'user_prompt_template': "请审查：{content}"
+                'user_prompt_template': "请审查：{review_content}"
             }
 
     def get_prompt_template(self, reviewer_type: str, prompt_name: str, **kwargs) -> ChatPromptTemplate:
@@ -178,7 +178,7 @@ class PromptLoader:
             # 返回默认模板
             return ChatPromptTemplate.from_messages([
                 ("system", f"你是专业的施工方案审查专家，负责进行{prompt_name}审查。"),
-                ("user", "请审查：{content}")
+                ("user", "请审查：{review_content}")
             ])
   
     def manage_cache(self, reviewer_type: str = None, prompt_name: str = None, action: str = "clear"):

core/construction_review/workflows/ai_review_workflow.py

@@ -273,17 +273,6 @@ class AIReviewWorkflow:
                         # 方法内部进度计算（基于当前处理的单元）
                         current_progress = int((unit_index / total_units) * 100)
                         progress_message = f"正在处理第 {unit_index + 1}/{total_units} 个单元: {section_label}"
-
-                        if state["progress_manager"]:
-                            asyncio.create_task(
-                                state["progress_manager"].update_stage_progress(
-                                    callback_task_id=state["callback_task_id"],
-                                    stage_name=stage_name,
-                                    current=current_progress,
-                                    status="processing",
-                                    message=progress_message
-                                )
-                            )
                         
                         # 并发执行各种原子化审查方法
                         review_tasks = [
@@ -305,22 +294,64 @@ class AIReviewWorkflow:
                         # 计算总体风险等级
                         overall_risk = self._calculate_overall_risk(basic_result, technical_result, rag_result)
 
+                        issues = self._format_review_results_to_issues(
+                            state["callback_task_id"],
+                            unit_index,
+                            review_location_label,
+                            unit_content,
+                            basic_result,
+                            technical_result
+                        )
+                        logger.info(f"issues: {issues}")
+                        # 统计发现问题数量
+                        issues_count = sum(len(issue.get("review_lists", [])) for issue in issues)
+
                         # 更新进度
                         nonlocal completed_units
                         completed_units += 1
                         current = int((completed_units / total_units) * 100)
-                        message = f"已完成 {completed_units}/{total_units} 个审查单元"
+
+                        # 构建包含问题信息的消息
+                        if issues_count > 0:
+                            message = f"正在处理第 {unit_index + 1}/{total_units} 个单元: {section_label}（已发现{issues_count}个问题）"
+                        else:
+                            message = f"正在处理第 {unit_index + 1}/{total_units} 个单元: {section_label}"
+
                         logger.info(f"更新进度: {current}% {message}")
-                        # 更新ProgressManager进度
+
+                        # 更新ProgressManager进度，包含issues信息
                         if state["progress_manager"]:
-                            asyncio.create_task(
-                                state["progress_manager"].update_stage_progress(
-                                    callback_task_id=state["callback_task_id"],
-                                    stage_name="AI审查",
-                                    current=current,
-                                    status="processing",
-                                    message=message
+                            # 如果有issues，通过额外参数传递
+                            if issues:
+                                asyncio.create_task(
+                                    state["progress_manager"].update_stage_progress(
+                                        callback_task_id=state["callback_task_id"],
+                                        stage_name=stage_name,
+                                        current=current,
+                                        status="unit_review_update",
+                                        message=message,
+                                        issues=issues,
+                                        user_id=state.get("user_id", ""),
+                                        overall_task_status="processing"
+                                    )
+                                )
+
+                            else:
+                                asyncio.create_task(
+                                    state["progress_manager"].update_stage_progress(
+                                        callback_task_id=state["callback_task_id"],
+                                        stage_name=stage_name,
+                                        current=current,
+                                        status="processing",
+                                        message=message
+                                    )
                                 )
+
+                        
+                        # 清空当前issues
+                        await state["progress_manager"].update_stage_progress(
+                            callback_task_id=state["callback_task_id"],
+                            issues=['clear']
                             )
 
                         return ReviewResult(
@@ -488,4 +519,146 @@ class AIReviewWorkflow:
         """获取工作流状态"""
         if self.progress_manager:
             return await self.progress_manager.get_progress(self.callback_task_id)
-        return {}
+        return {}
+
+
+    def _format_review_results_to_issues(self,callback_task_id: str, unit_index: int, review_location_label: str,
+                                    unit_content: Dict[str, Any], basic_result: Dict[str, Any],
+                                    technical_result: Dict[str, Any]) -> List[Dict[str, Any]]:
+        """
+        将审查结果格式化为issues结构
+
+        Args:
+            callback_task_id: 回调任务ID
+            unit_index: 单元索引
+            review_location_label: 审查位置标签
+            unit_content: 单元内容
+            basic_result: 基础合规性审查结果
+            technical_result: 技术性审查结果
+
+        Returns:
+            List[Dict]: 格式化后的issues列表
+        """
+        issues = []
+        review_lists = []
+        risk_count = {"high": 0, "medium": 0, "low": 0}
+        max_risk_level = "low"
+
+        # 合并所有审查结果
+        all_results = {}
+        if basic_result:
+            all_results.update(basic_result)
+        if technical_result:
+            all_results.update(technical_result)
+
+        logger.info(f"开始格式化审查结果，合并后结果: {list(all_results.keys())}")
+
+        for check_key, check_result in all_results.items():
+            logger.info(f"处理检查项: {check_key}, 结果类型: {type(check_result)}")
+
+            if check_key == 'overall_score':  # 跳过分数字段
+                logger.info(f"跳过分数字段: {check_key}")
+                continue
+
+            logger.info(f"检查项 {check_key} 的结果: {check_result}")
+
+            if check_result and "details" in check_result and "response" in check_result["details"]:
+                response = check_result["details"]["response"]
+                check_name = check_result["details"].get("name", check_key)
+                logger.info(f"解析检查项 {check_name} 的响应，长度: {len(response)}")
+                check_issues = self._parse_ai_review_response(response, check_name)
+                review_lists.extend(check_issues)
+            else:
+                logger.warning(f"检查项 {check_key} 格式不符合要求，缺少details或response字段")
+                logger.warning(f"check_result内容: {check_result}")
+
+        # 统计风险等级
+        for issue in review_lists:
+            risk_level = issue.get("risk_info", {}).get("risk_level", "low")
+            if risk_level in risk_count:
+                risk_count[risk_level] += 1
+
+        # 确定最高风险等级
+        if risk_count["high"] > 0:
+            max_risk_level = "high"
+        elif risk_count["medium"] > 0:
+            max_risk_level = "medium"
+
+        # 如果有审查结果，创建issue
+        if review_lists:
+            issue = {
+                "issue_id": f"{callback_task_id}-{max_risk_level}-{unit_index}",
+                "metadata": {
+                    "review_location_label": review_location_label,
+                    "original_content": unit_content.get('content', '')
+                },
+                "risk_summary": {
+                    "max_risk_level": max_risk_level,
+                    "risk_count": risk_count
+                },
+                "review_lists": review_lists
+            }
+            issues.append(issue)
+
+        return issues
+
+
+    def _parse_ai_review_response(self,response: str, check_name: str) -> List[Dict[str, Any]]:
+        """
+        解析AI审查的Markdown格式响应
+
+        Args:
+            response: AI审查响应
+            check_name: 检查项名称（如"词句语法检查"）
+
+        Returns:
+            List[Dict]: 解析后的审查结果列表
+        """
+        review_lists = []
+
+        try:
+            if "无明显问题" in response or "无问题" in response or "符合要求" in response:
+                review_lists.append({
+                    "check_item": check_name,  
+                    "check_result": "无明显问题",
+                    "exist_issue": False,
+                    "risk_info": {"risk_level": "low"}
+                })
+                return review_lists
+
+            lines = response.split('\n')
+
+            # 初始化默认风险等级
+            risk_level = "medium"  # 默认中等风险
+
+            # 先扫描一遍提取风险等级
+            for line in lines:
+                line = line.strip()
+                if line.startswith('**风险等级**：'):
+                    extracted_risk = line[7:].strip().lower()
+                    if "高" in extracted_risk:
+                        risk_level = "high"
+                    elif "中" in extracted_risk:
+                        risk_level = "medium"
+                    else:
+                        risk_level = "low"
+                    break
+
+            # 创建简化的结果对象，直接存储完整响应
+            review_lists.append({
+                "check_item": check_name,
+                "check_result": response,  # 直接存储完整的AI响应
+                "exist_issue": True,
+                "risk_info": {"risk_level": risk_level}
+            })
+
+        except Exception as e:
+            logger.error(f"解析AI审查响应失败: {str(e)}")
+            review_lists.append({
+                "check_item": check_name,
+                "check_result": response,
+                "exist_issue": True,  
+                "risk_info": {"risk_level": "low"}
+            })
+
+        return review_lists

core/construction_review/workflows/document_workflow.py

@@ -56,7 +56,7 @@ class DocumentWorkflow:
                 callback_task_id=self.callback_task_id,
                 stage_name="文档解析",
                 current=100,
-                status="completed",
+                status="docu_ans_completed",
                 message="文档解析完成"
             )
 

views/construction_review/launch_review.py

@@ -16,7 +16,7 @@ from fastapi.responses import StreamingResponse
 from core.base.redis_duplicate_checker import RedisDuplicateChecker
 from foundation.logger.loggering import server_logger as logger
 from foundation.trace.trace_context import TraceContext, auto_trace
-from foundation.utils.redis_utils import get_file_info, delete_file_info
+from foundation.utils.redis_utils import get_file_info
 from core.base.workflow_manager import WorkflowManager
 from core.base.progress_manager import ProgressManager, sse_callback_manager
 from views.construction_review.file_upload import validate_upload_parameters
@@ -66,12 +66,17 @@ class SimpleSSEManager:
         """发送进度更新 - 将进度数据放入队列推送给客户端"""
         queue = self.connections.get(callback_task_id)
         if queue:
+            # 根据数据状态决定事件类型
+            event_type = "unit_review_update" if current_data.get("status") == "unit_review_update" else "progress_update"
+
             await queue.put({
-                "type": "progress_update",
+                "type": event_type,
                 "data": current_data,
                 "timestamp": datetime.now().isoformat()
             })
-            logger.debug(f"SSE进度已推送: {callback_task_id}")
+            logger.debug(f"SSE进度已推送: {callback_task_id}, 事件类型: {event_type}")
+        else:
+            logger.warning(f"SSE连接已断开，跳过进度推送: {callback_task_id} - AI审查任务继续执行")
 
 sse_manager = SimpleSSEManager()
 
@@ -177,6 +182,7 @@ async def launch_review_sse(request_data: LaunchReviewRequest):
             # 发送连接确认
             connected_data = json.dumps({
                 "callback_task_id": callback_task_id,
+                "stage": "startup",
                 "message": "启动审查SSE连接已建立，正在处理请求...",
                 "timestamp": datetime.now().isoformat()
             }, ensure_ascii=False)
@@ -184,7 +190,6 @@ async def launch_review_sse(request_data: LaunchReviewRequest):
 
             # 处理启动审查逻辑
             try:
-                from foundation.utils.redis_utils import get_file_info
 
                 # 从callback_task_id中提取file_id (格式: file_id-timestamp)
                 file_id = callback_task_id.rsplit('-', 1)[0] if '-' in callback_task_id else callback_task_id
@@ -209,15 +214,6 @@ async def launch_review_sse(request_data: LaunchReviewRequest):
                 # 标记任务为已使用
                 await duplicatechecker.mark_task_as_used(callback_task_id)
 
-                # 获取文件信息
-                status_data = json.dumps({
-                    "callback_task_id": callback_task_id,
-                    "stage": "loading",
-                    "message": "正在加载文件信息...",
-                    "timestamp": datetime.now().isoformat()
-                }, ensure_ascii=False)
-                yield format_sse_event("processing", status_data)
-
                 file_info = await get_file_info(file_id, include_content=True)
 
                 if not file_info:
@@ -242,17 +238,8 @@ async def launch_review_sse(request_data: LaunchReviewRequest):
 
 
 
-                # 发送提交任务状态
-                status_data = json.dumps({
-                    "callback_task_id": callback_task_id,
-                    "stage": "submitting",
-                    "message": "正在提交AI审查任务...",
-                    "timestamp": datetime.now().isoformat()
-                }, ensure_ascii=False)
-                yield format_sse_event("processing", status_data)
-
                 # 提交处理任务到工作流管理器
-                task_id = await workflow_manager.submit_task_processing(file_info)
+                await workflow_manager.submit_task_processing(file_info)
 
                 # 发送成功启动状态
                 success_data = json.dumps({
@@ -278,19 +265,26 @@ async def launch_review_sse(request_data: LaunchReviewRequest):
                             if current_data:
                                 progress_json = json.dumps(current_data, ensure_ascii=False)
                                 yield format_sse_event("progress", progress_json)
-
-                                overall_task_status = current_data.get("overall_task_status")
-                                if overall_task_status in ["completed", "failed"]:
-                                    completion_data = {
-                                        "callback_task_id": callback_task_id,
-                                        "task_status": overall_task_status,
-                                        "overall_progress": current_data.get("current", 100),
-                                        "timestamp": datetime.now().isoformat(),
-                                        "message": "审查任务处理完成！"
-                                    }
-                                    completion_json = json.dumps(completion_data, ensure_ascii=False)
-                                    yield format_sse_event("completed", completion_json)
-                                    break
+                        elif message.get("type") == "unit_review_update":
+                            current_data = message.get("data")
+                            if current_data:
+                                unit_review_json = json.dumps(current_data, ensure_ascii=False)
+                                yield format_sse_event("unit_review", unit_review_json)
+
+                        # 统一检查任务完成状态
+                        if 'current_data' in locals() and current_data:
+                            overall_task_status = current_data.get("overall_task_status")
+                            if overall_task_status in ["completed", "failed"]:
+                                completion_data = {
+                                    "callback_task_id": callback_task_id,
+                                    "task_status": overall_task_status,
+                                    "overall_progress": current_data.get("current", 100),
+                                    "timestamp": datetime.now().isoformat(),
+                                    "message": "审查任务处理完成！"
+                                }
+                                completion_json = json.dumps(completion_data, ensure_ascii=False)
+                                yield format_sse_event("completed", completion_json)
+                                break
 
                     except Exception as e:
                         logger.error(f"队列消息处理异常: {callback_task_id}")
@@ -357,12 +351,3 @@ async def launch_review_sse(request_data: LaunchReviewRequest):
 
 
 
-@launch_review_router.get("/sse/launch_review_status")
-async def get_launch_review_sse_status():
-    """获取启动审查SSE连接状态 - 返回当前活跃的启动审查SSE连接信息"""
-    return {
-        "active_connections": len(sse_manager.connections),
-        "connections": list(sse_manager.connections.keys()),
-        "timestamp": datetime.now().isoformat(),
-        "service": "launch_review_sse"
-    }