Merge branch 'feat_PiGai' of FanYanPeng/EAI-Backend into refactor

src/main/java/com/njuzr/eaibackend/service/AIEvaluationService.java

@@ -212,22 +212,33 @@ public class AIEvaluationService {
 
         String content = engagement.getTextContent();
         String[] sentences = getSentenceList(content);
-        List<SentenceEvaluation> sentenceEvaluationsList = new ArrayList<>();
-        CompletableFuture<?>[] futures = new CompletableFuture[sentences.length];
+        log.info("rewritePerSentence 开始处理, assignmentId: {}, studentId: {}, 句子数量: {}", assignmentId, studentId, sentences.length);
+        
+        // 使用线程安全的集合，避免多线程并发写入导致数据丢失
+        List<SentenceEvaluation> sentenceEvaluationsList = Collections.synchronizedList(new ArrayList<>());
+        @SuppressWarnings("unchecked")
+        CompletableFuture<SentenceEvaluation>[] futures = new CompletableFuture[sentences.length];
 
         for (int i = 0; i < sentences.length; i++) {
             final int sentenceNo = i;
             final String sentence = sentences[i] + ".";
             futures[i] = CompletableFuture.supplyAsync(() ->
                     processSingleSentenceWithRetry(assignmentId, studentId, content, assignment.getDescription(), sentence, sentenceNo, engagement.getVersion()),
-                    aiPerSentenceExecutor)
-                    .thenAccept(sentenceEvaluationsList::add);
+                    aiPerSentenceExecutor);
         }
 
         // 等待所有任务完成
         try {
             CompletableFuture.allOf(futures).get(60, TimeUnit.SECONDS);
-            log.info("rewritePerSentence allOf success, assignmentId: {}, studentId: {}, version: {}", assignmentId, studentId, engagement.getVersion());
+            // 收集所有结果
+            for (CompletableFuture<SentenceEvaluation> future : futures) {
+                SentenceEvaluation result = future.get();
+                if (result != null) {
+                    sentenceEvaluationsList.add(result);
+                }
+            }
+            log.info("rewritePerSentence allOf success, assignmentId: {}, studentId: {}, version: {}, sentenceCount: {}, resultCount: {}", 
+                    assignmentId, studentId, engagement.getVersion(), sentences.length, sentenceEvaluationsList.size());
         } catch (TimeoutException e) {
             log.warn("rewritePerSentence failed, assignmentId: {}, studentId: {}, reason: timeout", assignmentId, studentId);
             Arrays.stream(futures).forEach(f -> f.cancel(true));
@@ -240,8 +251,12 @@ public class AIEvaluationService {
         }
 
         // 所有任务完成后批量插入数据库
+        log.info("rewritePerSentence准备保存, listSize: {}, list: {}", sentenceEvaluationsList.size(), sentenceEvaluationsList);
         if(!sentenceEvaluationsList.isEmpty()) {
             sentenceEvaluationMapperServiceImpl.saveBatch(sentenceEvaluationsList);
+            log.info("rewritePerSentence saveBatch完成, assignmentId: {}, studentId: {}", assignmentId, studentId);
+        } else {
+            log.warn("rewritePerSentence 列表为空，跳过保存! assignmentId: {}, studentId: {}", assignmentId, studentId);
         }
 
         evaluationMapperServiceImpl.update(new UpdateWrapper<Evaluation>()
@@ -291,15 +306,38 @@ public class AIEvaluationService {
             Integer version, int retryCount
     ) {
         AIRequestService.WorkflowRunResponse response = aiRequestService.requestPreSentenceWorkflow(content, assignmentDesc, sentence);
+        
+        // 详细日志：打印完整响应用于排查
+        log.info("processSingleSentence response, assignmentId: {}, sentenceNo: {}, response: {}", assignmentId, sentenceNo, response);
+        
         if(response == null || response.getData() == null){
             log.error("processSingleSentence failed, assignmentId: {}, studentId: {}, sentence: {}, response: {}, retryCount: {}/{}", assignmentId, studentId, sentence, response, retryCount+1, RETRY_COUNT);
             return null;
         }
+        
+        // 检查 workflow 执行状态
+        if(!"succeeded".equals(response.getData().getStatus())) {
+            log.error("processSingleSentence workflow failed, assignmentId: {}, studentId: {}, sentenceNo: {}, status: {}, error: {}", 
+                    assignmentId, studentId, sentenceNo, response.getData().getStatus(), response.getData().getError());
+            return null;
+        }
+        
+        // 检查 outputs 是否为空
+        if(response.getData().getOutputs() == null) {
+            log.error("processSingleSentence outputs is null, assignmentId: {}, studentId: {}, sentenceNo: {}, data: {}", assignmentId, studentId, sentenceNo, response.getData());
+            return null;
+        }
+        
+        String resultContent = response.getData().getOutputs().getResult_content();
+        String resultType = response.getData().getOutputs().getResult_type();
+        String resultCategory = response.getData().getOutputs().getResult_category();
+        
+        log.info("processSingleSentence outputs, sentenceNo: {}, type: {}, category: {}, content: {}", sentenceNo, resultType, resultCategory, resultContent);
 
         return new SentenceEvaluation()
-                .setEvaluation(response.getData().getOutputs().getResult_content())
-                .setType(response.getData().getOutputs().getResult_type())
-                .setCategory(response.getData().getOutputs().getResult_category())
+                .setEvaluation(resultContent)
+                .setType(resultType)
+                .setCategory(resultCategory)
                 .setAssignmentId(assignmentId)
                 .setStudentId(studentId)
                 .setContent(content)
@@ -391,16 +429,15 @@ public class AIEvaluationService {
     /**
      * 双智评：同时触发整体智评和逐句智评
      * 1. 创建智评记录
-     * 2. 同步执行整体智评（等待结果）
-     * 3. 异步执行逐句智评（不等待结果）
+     * 2. 异步执行整体智评
+     * 3. 异步执行逐句智评
      *
      * @param assignmentId 作业ID
      * @param studentId 学生ID
      * @return 整体智评结果的CompletableFuture
      */
-    @Transactional(propagation = Propagation.REQUIRED, rollbackFor = Exception.class)
     public CompletableFuture<AIEntry> doubleEvaluationAsync(Long assignmentId, Long studentId) {
-        // 1. 基础校验和记录初始化
+        // 1. 基础校验
         Engagement engagement = studentAssignmentMapper.findEngagementByStudentIdAndAssignmentId(studentId, assignmentId);
         if (engagement == null) throw MyException.create(HttpStatus.BAD_REQUEST, "作业参与不存在");
         Assignment assignment = assignmentMapper.selectById(assignmentId);
@@ -409,23 +446,24 @@ public class AIEvaluationService {
             throw MyException.create(HttpStatus.BAD_REQUEST, "该版本作业已进行过AI分析");
         }
 
-        // 2. 创建智评记录（原子操作）
+        // 2. 创建智评记录（在独立事务中，确保立即可见）
         initEvaluationRecord(assignmentId, studentId, engagement);
         log.info("双智评记录已创建，开始并行处理");
 
-        // 3. 同步执行整体智评（阻塞等待结果）
+        // 3. 异步执行整体智评（使用核心方法，跳过检查和初始化）
         CompletableFuture<AIEntry> overallFuture = CompletableFuture.supplyAsync(() -> {
             try {
-                return evaluation(assignmentId, studentId);
+                return doEvaluationCore(assignmentId, studentId, assignment, engagement);
             } catch (Exception e) {
-                throw MyException.create(HttpStatus.BAD_REQUEST, "整体智评失败");
+                log.error("整体智评失败: assignmentId={}, studentId={}, error={}", assignmentId, studentId, e.getMessage(), e);
+                throw MyException.create(HttpStatus.BAD_REQUEST, "整体智评失败: " + e.getMessage());
             }
         }, aiTaskExecutor);
 
-        // 4. 异步执行逐句智评（不阻塞主线程）
+        // 4. 异步执行逐句智评（使用核心方法，跳过检查和等待，使用编程式事务）
         CompletableFuture.runAsync(() -> {
             try {
-                rewritePerSentence(assignmentId, studentId);
+                doRewritePerSentenceCore(assignmentId, studentId, assignment, engagement);
             } catch (Exception e) {
                 log.error("逐句智评异步执行失败: assignmentId={}, studentId={}, error={}",
                         assignmentId, studentId, e.getMessage(), e);
@@ -435,9 +473,93 @@ public class AIEvaluationService {
         return overallFuture
                 .exceptionally(ex -> {
                     log.error("双智评整体处理失败", ex);
-                    throw new MyException(HttpStatus.INTERNAL_SERVER_ERROR.value(), "双智评执行异常");
+                    throw new MyException(HttpStatus.INTERNAL_SERVER_ERROR.value(), "双智评执行异常: " + ex.getMessage());
                 })
-                .orTimeout(120, TimeUnit.SECONDS); // 整体智评超时控制
+                .orTimeout(120, TimeUnit.SECONDS);
+    }
+
+    /**
+     * 整体智评核心逻辑（不包含检查和记录初始化）
+     */
+    private AIEntry doEvaluationCore(Long assignmentId, Long studentId, Assignment assignment, Engagement engagement) {
+        StopWatch sw = StopWatch.createStarted();
+        try {
+            // 发起LLM请求
+            String prompt = String.format(PromptConstant.REWRITE_PROMPT_TEMPLATE, assignment.getDescription(), engagement.getTextContent());
+            AIRequestService.AIResponse response = requestAIWithRetry(prompt, 0);
+            String retContent = response.getChoices().get(0).getMessage().getContent();
+            String role = response.getChoices().get(0).getMessage().getRole();
+
+            // 保存评估结果
+            saveOverallEvaluationResult(assignmentId, studentId, engagement, retContent);
+
+            log.info("doEvaluationCore cost: {}", sw.formatTime());
+            return new AIEntry(role, retContent);
+        } catch (Exception e) {
+            log.error("doEvaluationCore requestAI error: {}", e.getMessage());
+            throw new MyException(HttpStatus.INTERNAL_SERVER_ERROR.value(), "整体智评请求失败: " + e.getMessage());
+        }
+    }
+
+    /**
+     * 逐句智评核心逻辑（不包含检查和等待，使用编程式事务）
+     */
+    private void doRewritePerSentenceCore(Long assignmentId, Long studentId, Assignment assignment, Engagement engagement) {
+        StopWatch sw = StopWatch.createStarted();
+
+        String content = engagement.getTextContent();
+        String[] sentences = getSentenceList(content);
+        log.info("doRewritePerSentenceCore 开始处理, assignmentId: {}, studentId: {}, 句子数量: {}", assignmentId, studentId, sentences.length);
+        
+        List<SentenceEvaluation> sentenceEvaluationsList = Collections.synchronizedList(new ArrayList<>());
+        @SuppressWarnings("unchecked")
+        CompletableFuture<SentenceEvaluation>[] futures = new CompletableFuture[sentences.length];
+
+        for (int i = 0; i < sentences.length; i++) {
+            final int sentenceNo = i;
+            final String sentence = sentences[i] + ".";
+            futures[i] = CompletableFuture.supplyAsync(() ->
+                    processSingleSentenceWithRetry(assignmentId, studentId, content, assignment.getDescription(), sentence, sentenceNo, engagement.getVersion()),
+                    aiPerSentenceExecutor);
+        }
+
+        // 等待所有任务完成
+        try {
+            CompletableFuture.allOf(futures).get(60, TimeUnit.SECONDS);
+            // 收集所有结果
+            for (CompletableFuture<SentenceEvaluation> future : futures) {
+                SentenceEvaluation result = future.get();
+                if (result != null) {
+                    sentenceEvaluationsList.add(result);
+                }
+            }
+            log.info("doRewritePerSentenceCore allOf success, assignmentId: {}, studentId: {}, version: {}, sentenceCount: {}, resultCount: {}", 
+                    assignmentId, studentId, engagement.getVersion(), sentences.length, sentenceEvaluationsList.size());
+        } catch (TimeoutException e) {
+            log.warn("doRewritePerSentenceCore failed, assignmentId: {}, studentId: {}, reason: timeout", assignmentId, studentId);
+            Arrays.stream(futures).forEach(f -> f.cancel(true));
+            throw MyException.create(HttpStatus.BAD_REQUEST, "逐句智评处理超时");
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw MyException.create(HttpStatus.BAD_REQUEST, "逐句智评处理被中断: " + e.getMessage());
+        } catch (Exception e) {
+            throw MyException.create(HttpStatus.BAD_REQUEST, "逐句智评处理失败: " + e.getMessage());
+        }
+
+        // 使用编程式事务保存结果（确保异步线程中事务生效）
+        requiresNewTransactionTemplate.executeWithoutResult(status -> {
+            if (!sentenceEvaluationsList.isEmpty()) {
+                sentenceEvaluationMapperServiceImpl.saveBatch(sentenceEvaluationsList);
+            }
+
+            evaluationMapperServiceImpl.update(new UpdateWrapper<Evaluation>()
+                    .set("sentence", 1)
+                    .eq("student_id", studentId)
+                    .eq("assignment_id", assignmentId)
+                    .eq("version", engagement.getVersion()));
+        });
+
+        log.info("doRewritePerSentenceCore cost: {}", sw.formatTime());
     }
 
 }

src/main/java/com/njuzr/eaibackend/service/impl/TextAnalysisServiceImpl.java

@@ -42,6 +42,9 @@ public class TextAnalysisServiceImpl implements TextAnalysisService {
     private final ObjectMapper objectMapper;
     private final RestTemplate restTemplate;
 
+    // Stanford CoreNLP 同步锁（Simple API 内部 Pipeline 非线程安全）
+    private static final Object NLP_LOCK = new Object();
+
     // 外部API配置
     private static final String WORD_LEVEL_API_URL = "https://laurenceanthony.net/software/wordfamilyfinder/get_result.php";
     private static final String DATABASE = "basewords_130.db";
@@ -82,8 +85,12 @@ public class TextAnalysisServiceImpl implements TextAnalysisService {
                     textContent.length(), studentId, assignmentId);
 
             // 2. 使用Stanford CoreNLP进行分句和分词
-            Document doc = new Document(textContent);
-            List<Sentence> allSentences = new ArrayList<>(doc.sentences()); // 一次性加载所有句子到内存
+            // 注意：Stanford CoreNLP Simple API 的内部 Pipeline 是全局共享的，非线程安全，需要同步
+            List<Sentence> allSentences;
+            synchronized (NLP_LOCK) {
+                Document doc = new Document(textContent);
+                allSentences = new ArrayList<>(doc.sentences()); // 一次性加载所有句子到内存
+            }
 
             // 后续操作都使用allSentences集合，避免频繁调用doc.sentences()方法
             List<String> sentences = allSentences.stream()
@@ -93,10 +100,14 @@ public class TextAnalysisServiceImpl implements TextAnalysisService {
             log.info("分句完成，共{}个句子 - studentId: {}, assignmentId: {}",
                     sentences.size(), studentId, assignmentId);
 
-            // 3. 分词并获取词汇级别
+            // 3. 分词并获取词汇级别（words() 也需要同步保护）
             List<TextAnalysis.WordLevel> wordLevels = new ArrayList<>();
             for (Sentence sentence : allSentences) {
-                for (String word : sentence.words()) {
+                List<String> words;
+                synchronized (NLP_LOCK) {
+                    words = sentence.words();
+                }
+                for (String word : words) {
                     TextAnalysis.WordLevel wordLevel = new TextAnalysis.WordLevel();
                     wordLevel.setWord(word);