|
|
@@ -19,6 +19,7 @@ import org.springframework.transaction.PlatformTransactionManager;
|
|
|
import org.springframework.transaction.support.TransactionTemplate;
|
|
|
import org.springframework.transaction.TransactionDefinition;
|
|
|
|
|
|
+import java.text.BreakIterator;
|
|
|
import java.util.*;
|
|
|
import java.util.concurrent.CompletableFuture;
|
|
|
import java.util.concurrent.TimeUnit;
|
|
|
@@ -98,12 +99,55 @@ public class AIEvaluationService {
|
|
|
if(engagement == null) throw MyException.create(HttpStatus.BAD_REQUEST, "作业参与不存在");
|
|
|
Assignment assignment = assignmentMapper.selectById(assignmentId);
|
|
|
if (assignment == null) throw MyException.create(HttpStatus.BAD_REQUEST, "作业不存在");
|
|
|
- Evaluation evaluationJudge = evaluationMapperServiceImpl.getByVersion(assignmentId, studentId, engagement.getVersion());
|
|
|
- if(evaluationJudge != null) throw MyException.create(HttpStatus.BAD_REQUEST, "作业已进行过AI分析");
|
|
|
- // 先在独立事务中初始化记录,确保其他线程能立即查询到
|
|
|
- initEvaluationRecord(assignmentId, studentId, engagement);
|
|
|
- try{
|
|
|
- // 发起LLM请求
|
|
|
+ /**
|
|
|
+ * AI 作业评估主流程。
|
|
|
+ *
|
|
|
+ * 并发控制说明:
|
|
|
+ * 1. 通过数据库唯一索引 (assignment_id, student_id, version) 保证同一作业版本只能存在一条 evaluation 记录。
|
|
|
+ * 2. 首先调用 initEvaluationRecord() 插入一条占位记录(overall=0),用于抢占执行权限。
|
|
|
+ * - 插入成功:当前线程获得评估执行权。
|
|
|
+ * - 插入失败(DuplicateKeyException):说明已有评估任务存在。
|
|
|
+ *
|
|
|
+ * 重复请求处理:
|
|
|
+ * - 若已有记录且 overall=0:
|
|
|
+ * 表示正在进行 AI 分析。
|
|
|
+ * 若时间超过设定阈值(如 10 分钟),视为任务卡死,
|
|
|
+ * 删除占位记录并提示用户重新发起请求。
|
|
|
+ * - 若已有记录且 overall=1:
|
|
|
+ * 表示已完成评估,拒绝重复分析。
|
|
|
+ *
|
|
|
+ * 异常处理:
|
|
|
+ * - 若 AI 请求过程中发生异常,
|
|
|
+ * 删除占位记录以释放执行权,避免残留脏数据导致后续无法重新发起。
|
|
|
+ *
|
|
|
+ * 设计目的:
|
|
|
+ * - 防止用户重复点击或并发请求导致重复执行。
|
|
|
+ * - 防止系统异常导致任务永久卡死。
|
|
|
+ * - 保证 evaluation 表始终只存在一条有效执行记录。
|
|
|
+ */
|
|
|
+ try {
|
|
|
+ initEvaluationRecord(assignmentId, studentId, engagement);
|
|
|
+ } catch (org.springframework.dao.DuplicateKeyException ex) {
|
|
|
+ Evaluation exist = evaluationMapperServiceImpl.getByVersion(assignmentId, studentId, engagement.getVersion());
|
|
|
+ if (exist == null) {
|
|
|
+ throw MyException.create(HttpStatus.BAD_REQUEST, "作业正在进行AI分析,请稍后重试");
|
|
|
+ }
|
|
|
+ if (exist.getOverall() == 0) {
|
|
|
+
|
|
|
+ long timeoutMs = 10L * 60 * 1000;
|
|
|
+ Date expireBefore = new Date(System.currentTimeMillis() - timeoutMs);
|
|
|
+
|
|
|
+ if (exist.getTime() != null && exist.getTime().before(expireBefore)){
|
|
|
+ evaluationMapperServiceImpl.deleteByVersion(assignmentId, studentId, engagement.getVersion());
|
|
|
+ throw MyException.create(HttpStatus.BAD_REQUEST, "之前的 AI 分析已超时,已为您清理,请重新发起");
|
|
|
+
|
|
|
+ } else {
|
|
|
+ throw MyException.create(HttpStatus.BAD_REQUEST, "作业正在进行AI分析,请勿重复发起请求");
|
|
|
+ }
|
|
|
+ }
|
|
|
+ throw MyException.create(HttpStatus.BAD_REQUEST, "作业已进行过AI分析");
|
|
|
+ }
|
|
|
+ try {
|
|
|
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();
|
|
|
@@ -114,10 +158,34 @@ public class AIEvaluationService {
|
|
|
|
|
|
log.info("evaluation cost: {}", sw.formatTime());
|
|
|
return new AIEntry(role, retContent);
|
|
|
+
|
|
|
} catch (Exception e) {
|
|
|
- log.error("evaluation requestAI error: {}", e.getMessage());
|
|
|
- throw new MyException(HttpStatus.INTERNAL_SERVER_ERROR.value(), HttpStatus.INTERNAL_SERVER_ERROR.getReasonPhrase()+":"+"请求失败");
|
|
|
+ log.error("evaluation requestAI error: {}", e.getMessage(), e);
|
|
|
+ evaluationMapperServiceImpl.deleteByVersion(assignmentId, studentId, engagement.getVersion());
|
|
|
+ throw new MyException(HttpStatus.INTERNAL_SERVER_ERROR.value(),HttpStatus.INTERNAL_SERVER_ERROR.getReasonPhrase() + ":" + "请求失败,请重新尝试"
|
|
|
+ );
|
|
|
}
|
|
|
+ // Evaluation evaluationJudge = evaluationMapperServiceImpl.getByVersion(assignmentId, studentId, engagement.getVersion()); //如果连点两次,可能会重复
|
|
|
+ // if(evaluationJudge != null) throw MyException.create(HttpStatus.BAD_REQUEST, "作业已进行过AI分析");
|
|
|
+ // 先在独立事务中初始化记录,确保其他线程能立即查询到
|
|
|
+ // initEvaluationRecord(assignmentId, studentId, engagement);
|
|
|
+ // 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("evaluation cost: {}", sw.formatTime());
|
|
|
+ // return new AIEntry(role, retContent);
|
|
|
+ // } catch (Exception e) {
|
|
|
+ // log.error("evaluation requestAI error: {}", e.getMessage());
|
|
|
+ // evaluationMapperServiceImpl.deleteByVersion(assignmentId,studentId,engagement.getVersion()); // 删除初始化的记录,避免脏数据
|
|
|
+ // throw new MyException(HttpStatus.INTERNAL_SERVER_ERROR.value(), HttpStatus.INTERNAL_SERVER_ERROR.getReasonPhrase()+":"+"请求失败,请重新尝试");
|
|
|
+ // }
|
|
|
|
|
|
}
|
|
|
|
|
|
@@ -183,9 +251,10 @@ public class AIEvaluationService {
|
|
|
// 没有智能评价记录时,进行重试,等待整体智能批改将记录写入数据库
|
|
|
// 使用编程式事务在新事务中查询,避免REPEATABLE READ隔离级别导致看不到其他事务提交的数据
|
|
|
int retryTime = 0;
|
|
|
+ Evaluation evaluationJudge = null;
|
|
|
while (retryTime < 3) {
|
|
|
final int currentVersion = engagement.getVersion();
|
|
|
- Evaluation evaluationJudge = requiresNewTransactionTemplate.execute(status ->
|
|
|
+ evaluationJudge = requiresNewTransactionTemplate.execute(status ->
|
|
|
evaluationMapperServiceImpl.getByVersion(assignmentId, studentId, currentVersion)
|
|
|
);
|
|
|
log.info("rewritePerSentence find evaluation record, assignmentId: {}, studentId: {}, version: {}", assignmentId, studentId, engagement.getVersion());
|
|
|
@@ -195,84 +264,179 @@ public class AIEvaluationService {
|
|
|
Thread.sleep(1500);
|
|
|
retryTime++;
|
|
|
} catch (InterruptedException e) {
|
|
|
- throw new RuntimeException(e);
|
|
|
+ Thread.currentThread().interrupt();
|
|
|
+ throw MyException.create(HttpStatus.BAD_REQUEST, "处理被中断: " + e.getMessage());
|
|
|
}
|
|
|
} else {
|
|
|
- if(evaluationJudge.getSentence() != 0){
|
|
|
- throw MyException.create(HttpStatus.BAD_REQUEST, "作业已进行过逐句智能评价");
|
|
|
- } else {
|
|
|
- break;
|
|
|
- }
|
|
|
+ break;
|
|
|
}
|
|
|
- }
|
|
|
+ }
|
|
|
if(retryTime >= 3) {
|
|
|
log.info("rewritePerSentence failed, assignmentId: {}, studentId: {}, reason: over max retry times(3)", assignmentId, studentId);
|
|
|
throw MyException.create(HttpStatus.BAD_REQUEST, "逐句批改失败,超过重试次数,请重新发起请求");
|
|
|
}
|
|
|
|
|
|
- String content = engagement.getTextContent();
|
|
|
- String[] sentences = getSentenceList(content);
|
|
|
- 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];
|
|
|
+ try {
|
|
|
+ String content = engagement.getTextContent();
|
|
|
+ String[] sentences = getSentenceList(content);
|
|
|
+
|
|
|
+ log.info("rewritePerSentence start, assignmentId: {}, studentId: {}, sentenceCount: {}, version: {}",
|
|
|
+ assignmentId, studentId, sentences.length, engagement.getVersion());
|
|
|
+
|
|
|
+ List<SentenceEvaluation> sentenceEvaluationsList = 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
|
|
|
+ );
|
|
|
+ }
|
|
|
|
|
|
- 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(120, TimeUnit.SECONDS);
|
|
|
|
|
|
- // 等待所有任务完成
|
|
|
- try {
|
|
|
- CompletableFuture.allOf(futures).get(60, TimeUnit.SECONDS);
|
|
|
- // 收集所有结果
|
|
|
- for (CompletableFuture<SentenceEvaluation> future : futures) {
|
|
|
- SentenceEvaluation result = future.get();
|
|
|
- if (result != null) {
|
|
|
- sentenceEvaluationsList.add(result);
|
|
|
+ for (CompletableFuture<SentenceEvaluation> future : futures) {
|
|
|
+ SentenceEvaluation result = future.get();
|
|
|
+ if (result != null) {
|
|
|
+ sentenceEvaluationsList.add(result);
|
|
|
+ }
|
|
|
}
|
|
|
+ } catch (TimeoutException e) {
|
|
|
+ log.warn("rewritePerSentence timeout, assignmentId: {}, studentId: {}", 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());
|
|
|
}
|
|
|
- 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));
|
|
|
- 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());
|
|
|
- }
|
|
|
|
|
|
- // 所有任务完成后批量插入数据库
|
|
|
- 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);
|
|
|
- }
|
|
|
+ // 3) 批量保存逐句结果
|
|
|
+ if (!sentenceEvaluationsList.isEmpty()) {
|
|
|
+ sentenceEvaluationMapperServiceImpl.saveBatch(sentenceEvaluationsList);
|
|
|
+ }
|
|
|
|
|
|
+ // 4) 完成态:2 -> 1(只允许从处理中改成完成,避免误覆盖)
|
|
|
+ evaluationMapperServiceImpl.update(new UpdateWrapper<Evaluation>()
|
|
|
+ .set("sentence", 1)
|
|
|
+ .eq("student_id", studentId)
|
|
|
+ .eq("assignment_id", assignmentId)
|
|
|
+ .eq("version", engagement.getVersion())
|
|
|
+ .eq("sentence", 2)
|
|
|
+ );
|
|
|
+
|
|
|
+ log.info("rewritePerSentence success, assignmentId: {}, studentId: {}, cost: {}",
|
|
|
+ assignmentId, studentId, sw.formatTime());
|
|
|
+
|
|
|
+ return sentenceEvaluationsList;
|
|
|
+
|
|
|
+ } catch (RuntimeException ex) {
|
|
|
+ // 5) 出现任何异常:释放处理中状态(2 -> 0),允许重试,避免永久卡死
|
|
|
evaluationMapperServiceImpl.update(new UpdateWrapper<Evaluation>()
|
|
|
- .set("sentence", 1)
|
|
|
+ .set("sentence", 0)
|
|
|
.eq("student_id", studentId)
|
|
|
.eq("assignment_id", assignmentId)
|
|
|
- .eq("version", engagement.getVersion()));
|
|
|
-
|
|
|
- log.info("rewritePerSentence cost: {}", sw.formatTime());
|
|
|
- return sentenceEvaluationsList;
|
|
|
+ .eq("version", engagement.getVersion())
|
|
|
+ .eq("sentence", 2)
|
|
|
+ );
|
|
|
+ throw ex;
|
|
|
}
|
|
|
-
|
|
|
+ // String content = engagement.getTextContent();
|
|
|
+ // String[] sentences = getSentenceList(content);
|
|
|
+ // 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);
|
|
|
+ // }
|
|
|
+
|
|
|
+ // // 等待所有任务完成
|
|
|
+ // 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("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));
|
|
|
+ // 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());
|
|
|
+ // }
|
|
|
+
|
|
|
+ // // 所有任务完成后批量插入数据库
|
|
|
+ // 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>()
|
|
|
+ // .set("sentence", 1)
|
|
|
+ // .eq("student_id", studentId)
|
|
|
+ // .eq("assignment_id", assignmentId)
|
|
|
+ // .eq("version", engagement.getVersion()));
|
|
|
+
|
|
|
+ // log.info("rewritePerSentence cost: {}", sw.formatTime());
|
|
|
+ // return sentenceEvaluationsList;
|
|
|
+
|
|
|
+ }
|
|
|
+ //原来的getSentenceList方法存在问题,无法正确处理多种标点符号和不同语言的句子分割,这里改用BreakIterator来实现更准确的句子分割
|
|
|
private static String[] getSentenceList(String content) {
|
|
|
- String[] sentences = content.split("\\.");
|
|
|
- return sentences;
|
|
|
+ if (content == null || content.isBlank()) {
|
|
|
+ return new String[0];
|
|
|
+ }
|
|
|
+
|
|
|
+ BreakIterator iterator = BreakIterator.getSentenceInstance(Locale.US);
|
|
|
+ iterator.setText(content);
|
|
|
+
|
|
|
+ List<String> sentences = new ArrayList<>();
|
|
|
+ int start = iterator.first();
|
|
|
+ int end;
|
|
|
+
|
|
|
+ while ((end = iterator.next()) != BreakIterator.DONE) {
|
|
|
+ String sentence = content.substring(start, end).trim();
|
|
|
+ if (!sentence.isEmpty()) {
|
|
|
+ sentences.add(sentence);
|
|
|
+ }
|
|
|
+ start = end;
|
|
|
+ }
|
|
|
+ return sentences.toArray(new String[0]);
|
|
|
}
|
|
|
+ // private static String[] getSentenceList(String content) {
|
|
|
+ // String[] sentences = content.split("\\.");
|
|
|
+ // return sentences;
|
|
|
+ // }
|
|
|
|
|
|
private SentenceEvaluation processSingleSentenceWithRetry(Long assignmentId, Long studentId, String content, String description, String sentence, int sentenceNo, int version) {
|
|
|
int retryTime = 0;
|