v0.0.3-AI审查sse进度推送优化v1

config/config.ini

@@ -29,7 +29,7 @@ QWEN_API_KEY=ms-9ad4a379-d592-4acd-b92c-8bac08a4a045
 
 [ai_review]
 # 调试模式配置
-MAX_REVIEW_UNITS=1
+MAX_REVIEW_UNITS=3
 REVIEW_MODE=random
 # REVIEW_MODE=all/random/first
 

core/base/progress_manager.py

@@ -297,7 +297,7 @@ class ProgressManager:
 
             task_progress["status"] = "completed"
             task_progress["overall_task_status"] = "completed"
-            task_progress["message"] = "任务已全部完成"
+            task_progress["message"] = "施工方案审查任务已完成!"
             task_progress["updated_at"] = datetime.now().isoformat()
 
 

core/base/workflow_manager.py

@@ -200,24 +200,24 @@ class WorkflowManager:
 
             task_chain.results['ai_review'] = ai_result
 
-            # 阶段3：报告生成（串行）
-            task_chain.current_stage = "report_generation"
-
-            report_workflow = ReportWorkflow(
-                file_id=task_chain.file_id,
-                callback_task_id=task_chain.callback_task_id,
-                user_id=task_chain.user_id,
-                ai_review_results=ai_result,
-                progress_manager=self.progress_manager
-            )
-
-            # 同步执行报告生成
-            loop = asyncio.new_event_loop()
-            asyncio.set_event_loop(loop)
-            report_result = loop.run_until_complete(report_workflow.execute())
-            loop.close()
-
-            task_chain.results['report'] = report_result
+            # # 阶段3：报告生成（串行）
+            # task_chain.current_stage = "report_generation"
+
+            # report_workflow = ReportWorkflow(
+            #     file_id=task_chain.file_id,
+            #     callback_task_id=task_chain.callback_task_id,
+            #     user_id=task_chain.user_id,
+            #     ai_review_results=ai_result,
+            #     progress_manager=self.progress_manager
+            # )
+
+            # # 同步执行报告生成
+            # loop = asyncio.new_event_loop()
+            # asyncio.set_event_loop(loop)
+            # report_result = loop.run_until_complete(report_workflow.execute())
+            # loop.close()
+
+            # task_chain.results['report'] = report_result
 
             # 完成任务链
             task_chain.status = "completed"

core/construction_review/component/document_processor.py

@@ -40,7 +40,8 @@ class DocumentProcessor:
             logger.warning(f"LLM分类器初始化失败，将使用基础处理模式: {str(e)}")
 
     async def process_document(self, file_content: bytes, file_type: str,
-                             progress_callback: Optional[Callable[[int, str], None]] = None) -> Dict[str, Any]:
+                            #  progress_callback: Optional[Callable[[int, str], None]] = None
+                             ) -> Dict[str, Any]:
         """
         处理文档
 
@@ -54,7 +55,8 @@ class DocumentProcessor:
         """
         try:
             logger.info(f"开始处理文档，类型: {file_type}")
-
+            # if progress_callback:
+                #  progress_callback(20, "开始文档处理")
             # 简化处理：直接解析
             if file_type.lower() == 'pdf':
                 result = await self.parse_pdf_content(file_content)
@@ -66,8 +68,7 @@ class DocumentProcessor:
             # 结构化内容
             structured_result = self.structure_content(result)
 
-            # if progress_callback:
-            #     progress_callback(100, "文档处理完成")
+
 
             return structured_result
 

core/construction_review/workflows/ai_review_workflow.py

@@ -221,15 +221,15 @@ class AIReviewWorkflow:
     async def _ai_review_node(self, state: AIReviewState) -> AIReviewState:
         """AI审查节点"""
         try:
-            logger.info(f"正在执行: {state['file_id']}")
-            if state["progress_manager"]:
-                await state["progress_manager"].update_stage_progress(
-                    callback_task_id=state["callback_task_id"],
-                    stage_name="AI审查",
-                    current=0,
-                    status="processing",
-                    message=f"正在执行: {state['file_id']}"
-                )
+            # logger.info(f"正在执行: {state['file_id']}")
+            # if state["progress_manager"]:
+            #     await state["progress_manager"].update_stage_progress(
+            #         callback_task_id=state["callback_task_id"],
+            #         stage_name="AI审查",
+            #         current=0,
+            #         status="processing",
+            #         message=f"正在执行: {state['file_id']}"
+            #     )
             state["current_stage"] = "ai_review"
 
             # 筛选要审查的单元

core/construction_review/workflows/document_workflow.py

@@ -32,35 +32,20 @@ class DocumentWorkflow:
             if not existing_progress:
                 logger.warning(f"文档处理工作流未找到进度数据: {self.callback_task_id}")
 
-
-            # 4. 执行文档处理
-            def progress_callback(current: int, message: str):
-                asyncio.create_task(
-                    self.progress_manager.update_stage_progress(
-                        callback_task_id=self.callback_task_id,
-                        stage_name="文档解析",
-                        current=current,
-                        status="processing",
-                        message=message
-                    )
-                )
-
-            structured_content = await self.document_processor.process_document(
-                file_content=file_content,
-                file_type=file_type,
-                progress_callback=progress_callback
-            )
-
-            # 5. 更新完成状态
             await self.progress_manager.update_stage_progress(
                 callback_task_id=self.callback_task_id,
                 stage_name="文档解析",
                 current=100,
                 status="docu_ans_completed",
-                message="文档解析完成"
+                message="开始文档解析"
+            )
+
+            structured_content = await self.document_processor.process_document(
+                file_content=file_content,
+                file_type=file_type,
+                # progress_callback=progress_callback
             )
 
-            # 6. 保存处理结果
             result = {
                 'file_id': self.file_id,
                 'structured_content': structured_content,

views/construction_review/launch_review.py

@@ -249,7 +249,7 @@ async def launch_review_sse(request_data: LaunchReviewRequest):
                     "project_plan_type": project_plan_type,
                     "status": "submitted",
                     "submitted_at": file_info['launched_at'],
-                    "message": "AI审查任务已成功启动",
+                    "message": "施工方案审查任务启动成功，请耐心等待结果...",
                     "timestamp": datetime.now().isoformat()
                 }, ensure_ascii=False)
                 yield format_sse_event("submitted", success_data)
@@ -271,20 +271,20 @@ async def launch_review_sse(request_data: LaunchReviewRequest):
                                 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
+                        # # 统一检查任务完成状态
+                        # 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}")