feat(ai): 补齐质量评估执行器
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
lili
parent
a5619a3566
commit
19ff616463
File diff suppressed because one or more lines are too long
@ -0,0 +1,426 @@
|
||||
package cn.iocoder.muse.module.ai.application.muse;
|
||||
|
||||
import cn.iocoder.muse.framework.common.util.json.JsonUtils;
|
||||
import cn.iocoder.muse.framework.tenant.core.context.TenantContextHolder;
|
||||
import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiEvaluationRunDO;
|
||||
import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiEvaluationSampleResultDO;
|
||||
import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiJobDO;
|
||||
import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseQualityPolicyVersionDO;
|
||||
import cn.iocoder.muse.module.ai.dal.mysql.muse.MuseAiEvaluationRunMapper;
|
||||
import cn.iocoder.muse.module.ai.dal.mysql.muse.MuseAiEvaluationSampleResultMapper;
|
||||
import cn.iocoder.muse.module.ai.dal.mysql.muse.MuseAiJobMapper;
|
||||
import cn.iocoder.muse.module.ai.dal.mysql.muse.MuseQualityPolicyVersionMapper;
|
||||
import com.fasterxml.jackson.databind.JsonNode;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
import java.math.BigDecimal;
|
||||
import java.math.RoundingMode;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.security.MessageDigest;
|
||||
import java.security.NoSuchAlgorithmException;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.ArrayList;
|
||||
import java.util.HexFormat;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Locale;
|
||||
import java.util.Map;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
/**
|
||||
* 质量评估执行器。
|
||||
*
|
||||
* <p>当前版本只执行合成/脱敏评估集的确定性本地评估，先把 queued run/job 推进为可观测终态。
|
||||
* 不调用外部 LLM，不读取用户私有正文，不把 dataset 原文扩散到样本结果。后续接入真实评估集和 LLM judge 时，
|
||||
* 应复用本执行器的状态机与样本结果落库链路。</p>
|
||||
*/
|
||||
@Slf4j
|
||||
@Service
|
||||
public class MuseQualityEvaluationExecutor {
|
||||
|
||||
private static final String STATUS_RUNNING = "running";
|
||||
private static final String STATUS_COMPLETED = "completed";
|
||||
private static final String STATUS_FAILED = "failed";
|
||||
private static final String SAMPLE_STATUS_PASSED = "passed";
|
||||
private static final String SAMPLE_STATUS_FAILED = "failed";
|
||||
private static final String OPERATION_EXECUTE_EVALUATION = "executeQualityEvaluationRun";
|
||||
private static final String TARGET_TYPE_EVALUATION_RUN = "evaluationRun";
|
||||
private static final String ERROR_INVALID_JOB = "AI_QUALITY_EVALUATION_INVALID_JOB";
|
||||
private static final String ERROR_DATASET_BLOCKED = "AI_QUALITY_EVALUATION_DATASET_BLOCKED";
|
||||
private static final String ERROR_SAMPLE_EMPTY = "AI_QUALITY_EVALUATION_SAMPLE_EMPTY";
|
||||
private static final String ERROR_SAMPLE_TOO_LARGE = "AI_QUALITY_EVALUATION_SAMPLE_TOO_LARGE";
|
||||
private static final String ERROR_POLICY_INVALID = "AI_QUALITY_EVALUATION_POLICY_INVALID";
|
||||
private static final String ERROR_EXECUTION_FAILED = "AI_QUALITY_EVALUATION_EXECUTION_FAILED";
|
||||
private static final int MAX_SAMPLE_COUNT = 200;
|
||||
private static final Pattern SECRET_PATTERN = Pattern.compile(
|
||||
"(?i)(sk-[a-z0-9_-]+|bearer\\s+\\S+|token\\s*[:=]?\\s*\\S+|secret\\s*[:=]?\\s*\\S+|api[-_ ]?key\\s*[:=]?\\s*\\S+)");
|
||||
|
||||
@Resource
|
||||
private MuseAiJobMapper jobMapper;
|
||||
@Resource
|
||||
private MuseAiEvaluationRunMapper evaluationRunMapper;
|
||||
@Resource
|
||||
private MuseAiEvaluationSampleResultMapper sampleResultMapper;
|
||||
@Resource
|
||||
private MuseQualityPolicyVersionMapper qualityPolicyVersionMapper;
|
||||
@Resource
|
||||
private MuseAiAuditService auditService;
|
||||
|
||||
/**
|
||||
* 执行已领取的质量评估 job。
|
||||
*
|
||||
* <p>调用方必须先把租户上下文恢复到领取行的 tenantId；本方法全程只写脱敏摘要和样本聚合。</p>
|
||||
*/
|
||||
@Transactional(rollbackFor = Exception.class)
|
||||
public void executeEvaluationJob(Long jobPkId) {
|
||||
MuseAiJobDO job = jobMapper.selectById(jobPkId);
|
||||
if (job == null || !StringUtils.hasText(job.getJobId())) {
|
||||
throw new EvaluationExecutionException(ERROR_INVALID_JOB, "质量评估 Job 不存在或缺少 jobId");
|
||||
}
|
||||
MuseAiEvaluationRunDO run = evaluationRunMapper.selectByJobId(job.getJobId());
|
||||
if (run == null) {
|
||||
failJobOnly(job, ERROR_INVALID_JOB, "质量评估运行不存在");
|
||||
return;
|
||||
}
|
||||
|
||||
markRunning(run);
|
||||
try {
|
||||
EvaluationReport report = evaluate(run);
|
||||
insertSampleResults(run, report);
|
||||
markCompleted(run, job, report);
|
||||
auditSafely(run, job, report.auditSummary(), "succeeded", null, null);
|
||||
} catch (EvaluationExecutionException ex) {
|
||||
markFailed(run, job, ex.errorCode(), ex.getMessage());
|
||||
auditSafely(run, job, "{\"errorCode\":\"" + ex.errorCode() + "\"}", STATUS_FAILED,
|
||||
ex.errorCode(), ex.getMessage());
|
||||
} catch (RuntimeException ex) {
|
||||
String message = sanitizeError(ex.getMessage());
|
||||
markFailed(run, job, ERROR_EXECUTION_FAILED, message);
|
||||
auditSafely(run, job, "{\"errorCode\":\"" + ERROR_EXECUTION_FAILED + "\"}", STATUS_FAILED,
|
||||
ERROR_EXECUTION_FAILED, message);
|
||||
}
|
||||
}
|
||||
|
||||
private EvaluationReport evaluate(MuseAiEvaluationRunDO run) {
|
||||
String datasetReference = datasetReference(run);
|
||||
requireSafeDatasetReference(datasetReference);
|
||||
int sampleCount = safeSampleCount(run.getSampleCount());
|
||||
if (sampleCount <= 0) {
|
||||
throw new EvaluationExecutionException(ERROR_SAMPLE_EMPTY, "质量评估样本数必须大于 0");
|
||||
}
|
||||
if (sampleCount > MAX_SAMPLE_COUNT) {
|
||||
throw new EvaluationExecutionException(ERROR_SAMPLE_TOO_LARGE, "质量评估样本数超过内部测试上限");
|
||||
}
|
||||
|
||||
MuseQualityPolicyVersionDO policyVersion = qualityPolicyVersionMapper.selectByPolicyKeyAndVersion(
|
||||
run.getPolicyKey(), run.getPolicyVersionNo());
|
||||
List<DimensionConfig> dimensions = parseDimensions(policyVersion);
|
||||
if (dimensions.isEmpty()) {
|
||||
throw new EvaluationExecutionException(ERROR_POLICY_INVALID, "质量策略缺少可执行维度");
|
||||
}
|
||||
|
||||
List<SampleReport> samples = new ArrayList<>();
|
||||
Map<String, Double> dimensionTotals = new LinkedHashMap<>();
|
||||
int passedCount = 0;
|
||||
for (int index = 1; index <= sampleCount; index++) {
|
||||
SampleReport sample = evaluateSample(run, datasetReference, dimensions, index);
|
||||
samples.add(sample);
|
||||
if (sample.passed()) {
|
||||
passedCount += 1;
|
||||
}
|
||||
sample.dimensionScores().forEach((dimension, score) ->
|
||||
dimensionTotals.merge(dimension, score, Double::sum));
|
||||
}
|
||||
|
||||
Map<String, Double> dimensionScores = new LinkedHashMap<>();
|
||||
dimensionTotals.forEach((dimension, total) -> dimensionScores.put(dimension, round(total / sampleCount)));
|
||||
double overallScore = round(samples.stream().mapToDouble(SampleReport::score).average().orElse(0D));
|
||||
int failedCount = sampleCount - passedCount;
|
||||
return new EvaluationReport(samples, dimensionScores, overallScore, passedCount, failedCount,
|
||||
JsonUtils.toJsonString(Map.of(
|
||||
"datasetReference", datasetReference,
|
||||
"evaluator", "deterministic-local-v1",
|
||||
"executionMode", "synthetic_or_redacted_only",
|
||||
"sampleCount", sampleCount,
|
||||
"passedCount", passedCount,
|
||||
"failedCount", failedCount,
|
||||
"overallScore", overallScore,
|
||||
"dimensionScores", dimensionScores)));
|
||||
}
|
||||
|
||||
private SampleReport evaluateSample(MuseAiEvaluationRunDO run, String datasetReference,
|
||||
List<DimensionConfig> dimensions, int sampleIndex) {
|
||||
Map<String, Double> scores = new LinkedHashMap<>();
|
||||
double weightedTotal = 0D;
|
||||
double weightTotal = 0D;
|
||||
boolean passed = true;
|
||||
for (DimensionConfig dimension : dimensions) {
|
||||
double score = deterministicScore(run, datasetReference, sampleIndex, dimension.name());
|
||||
scores.put(dimension.name(), score);
|
||||
weightedTotal += score * dimension.weight();
|
||||
weightTotal += dimension.weight();
|
||||
if (score < dimension.threshold()) {
|
||||
passed = false;
|
||||
}
|
||||
}
|
||||
double overall = round(weightTotal <= 0D ? 0D : weightedTotal / weightTotal);
|
||||
return new SampleReport(sampleId(sampleIndex), passed, overall, scores);
|
||||
}
|
||||
|
||||
private void insertSampleResults(MuseAiEvaluationRunDO run, EvaluationReport report) {
|
||||
for (SampleReport sample : report.samples()) {
|
||||
MuseAiEvaluationSampleResultDO row = new MuseAiEvaluationSampleResultDO();
|
||||
row.setRunId(run.getRunId());
|
||||
row.setSampleId(sample.sampleId());
|
||||
row.setOwnerUserId(run.getOwnerUserId());
|
||||
row.setStatus(sample.passed() ? SAMPLE_STATUS_PASSED : SAMPLE_STATUS_FAILED);
|
||||
row.setScore(BigDecimal.valueOf(sample.score()).setScale(4, RoundingMode.HALF_UP));
|
||||
row.setMetricSummary(JsonUtils.toJsonString(Map.of("dimensionScores", sample.dimensionScores())));
|
||||
row.setInputSummary(JsonUtils.toJsonString(Map.of(
|
||||
"datasetReferenceHash", sha256(datasetReference(run)),
|
||||
"sampleId", sample.sampleId(),
|
||||
"inputRedacted", true)));
|
||||
row.setOutputSummary(JsonUtils.toJsonString(Map.of(
|
||||
"evaluator", "deterministic-local-v1",
|
||||
"rawOutputStored", false)));
|
||||
row.setTenantId(TenantContextHolder.getRequiredTenantId());
|
||||
sampleResultMapper.insert(row);
|
||||
}
|
||||
}
|
||||
|
||||
private List<DimensionConfig> parseDimensions(MuseQualityPolicyVersionDO policyVersion) {
|
||||
if (policyVersion == null || !StringUtils.hasText(policyVersion.getThresholdSnapshot())) {
|
||||
return List.of();
|
||||
}
|
||||
try {
|
||||
JsonNode root = JsonUtils.getObjectMapper().readTree(policyVersion.getThresholdSnapshot());
|
||||
if (root == null || !root.isArray()) {
|
||||
return List.of();
|
||||
}
|
||||
List<DimensionConfig> dimensions = new ArrayList<>();
|
||||
for (JsonNode node : root) {
|
||||
String name = text(node, "name");
|
||||
if (!StringUtils.hasText(name)) {
|
||||
continue;
|
||||
}
|
||||
double threshold = clamp(number(node, "threshold", 0.8D));
|
||||
double weight = Math.max(0.01D, number(node, "weight", 1.0D));
|
||||
dimensions.add(new DimensionConfig(name.trim(), threshold, weight));
|
||||
}
|
||||
return dimensions;
|
||||
} catch (Exception ex) {
|
||||
return List.of();
|
||||
}
|
||||
}
|
||||
|
||||
private String datasetReference(MuseAiEvaluationRunDO run) {
|
||||
JsonNode summary = json(run == null ? null : run.getMetricSummary());
|
||||
String value = text(summary, "datasetReference");
|
||||
return value == null ? "" : value.trim();
|
||||
}
|
||||
|
||||
private void requireSafeDatasetReference(String datasetReference) {
|
||||
if (!isSafeDatasetReference(datasetReference)) {
|
||||
// 当前执行器没有私有正文授权/脱敏证明模型；宁可失败，也不能把 work/file 等私有来源当成评估集。
|
||||
throw new EvaluationExecutionException(ERROR_DATASET_BLOCKED, "质量评估数据集未标记为合成或脱敏");
|
||||
}
|
||||
}
|
||||
|
||||
private boolean isSafeDatasetReference(String datasetReference) {
|
||||
String normalized = datasetReference == null ? "" : datasetReference.trim().toLowerCase(Locale.ROOT);
|
||||
return normalized.startsWith("dataset://")
|
||||
|| normalized.startsWith("synthetic://")
|
||||
|| normalized.startsWith("redacted://")
|
||||
|| normalized.contains("redacted")
|
||||
|| normalized.contains("smoke");
|
||||
}
|
||||
|
||||
private void markRunning(MuseAiEvaluationRunDO run) {
|
||||
MuseAiEvaluationRunDO update = new MuseAiEvaluationRunDO();
|
||||
update.setId(run.getId());
|
||||
update.setStatus(STATUS_RUNNING);
|
||||
update.setStartedAt(run.getStartedAt() == null ? LocalDateTime.now() : run.getStartedAt());
|
||||
evaluationRunMapper.updateById(update);
|
||||
}
|
||||
|
||||
private void markCompleted(MuseAiEvaluationRunDO run, MuseAiJobDO job, EvaluationReport report) {
|
||||
LocalDateTime now = LocalDateTime.now();
|
||||
MuseAiEvaluationRunDO runUpdate = new MuseAiEvaluationRunDO();
|
||||
runUpdate.setId(run.getId());
|
||||
runUpdate.setStatus(STATUS_COMPLETED);
|
||||
runUpdate.setPassedCount(report.passedCount());
|
||||
runUpdate.setFailedCount(report.failedCount());
|
||||
runUpdate.setMetricSummary(report.metricSummary());
|
||||
runUpdate.setFinishedAt(now);
|
||||
evaluationRunMapper.updateById(runUpdate);
|
||||
|
||||
MuseAiJobDO jobUpdate = new MuseAiJobDO();
|
||||
jobUpdate.setId(job.getId());
|
||||
jobUpdate.setStatus(STATUS_COMPLETED);
|
||||
jobUpdate.setRetryable(false);
|
||||
jobUpdate.setResultSummary(report.metricSummary());
|
||||
jobUpdate.setFinishedAt(now);
|
||||
jobMapper.updateById(jobUpdate);
|
||||
}
|
||||
|
||||
private void markFailed(MuseAiEvaluationRunDO run, MuseAiJobDO job, String errorCode, String errorMessage) {
|
||||
LocalDateTime now = LocalDateTime.now();
|
||||
MuseAiEvaluationRunDO runUpdate = new MuseAiEvaluationRunDO();
|
||||
runUpdate.setId(run.getId());
|
||||
runUpdate.setStatus(STATUS_FAILED);
|
||||
runUpdate.setErrorCode(errorCode);
|
||||
runUpdate.setErrorMessage(sanitizeError(errorMessage));
|
||||
runUpdate.setFinishedAt(now);
|
||||
evaluationRunMapper.updateById(runUpdate);
|
||||
failJobOnly(job, errorCode, errorMessage);
|
||||
}
|
||||
|
||||
private void failJobOnly(MuseAiJobDO job, String errorCode, String errorMessage) {
|
||||
MuseAiJobDO jobUpdate = new MuseAiJobDO();
|
||||
jobUpdate.setId(job.getId());
|
||||
jobUpdate.setStatus(STATUS_FAILED);
|
||||
jobUpdate.setRetryable(false);
|
||||
jobUpdate.setErrorCode(errorCode);
|
||||
jobUpdate.setErrorMessage(sanitizeError(errorMessage));
|
||||
jobUpdate.setFinishedAt(LocalDateTime.now());
|
||||
jobMapper.updateById(jobUpdate);
|
||||
}
|
||||
|
||||
private void auditSafely(MuseAiEvaluationRunDO run, MuseAiJobDO job, String responseSummary,
|
||||
String status, String errorCode, String errorMessage) {
|
||||
try {
|
||||
audit(run, job, responseSummary, status, errorCode, errorMessage);
|
||||
} catch (RuntimeException ex) {
|
||||
// 审计是可观测补充，不能因为审计写失败回滚 run/job 终态，避免 worker 留下永久 running。
|
||||
log.warn("[auditSafely][runId({}) jobId({}) audit failed, errorType={}]",
|
||||
run == null ? null : run.getRunId(), job == null ? null : job.getJobId(),
|
||||
ex.getClass().getSimpleName());
|
||||
}
|
||||
}
|
||||
|
||||
private void audit(MuseAiEvaluationRunDO run, MuseAiJobDO job, String responseSummary,
|
||||
String status, String errorCode, String errorMessage) {
|
||||
Map<String, Object> requestSummary = new LinkedHashMap<>();
|
||||
requestSummary.put("policyKey", run.getPolicyKey());
|
||||
requestSummary.put("policyVersion", run.getPolicyVersionNo());
|
||||
requestSummary.put("datasetReferenceHash", sha256(datasetReference(run)));
|
||||
requestSummary.put("datasetReferenceAllowed", isSafeDatasetReference(datasetReference(run)));
|
||||
requestSummary.put("sampleCount", safeSampleCount(run.getSampleCount()));
|
||||
auditService.record(MuseAiAuditService.AuditCreateReq.builder()
|
||||
.operationId(OPERATION_EXECUTE_EVALUATION)
|
||||
.actorUserId(job.getActorUserId())
|
||||
.ownerUserId(job.getOwnerUserId())
|
||||
.side("worker")
|
||||
.targetType(TARGET_TYPE_EVALUATION_RUN)
|
||||
.targetId(run.getId())
|
||||
.commandId(job.getCommandId())
|
||||
.requestHash(job.getRequestHash())
|
||||
.requestSummary(JsonUtils.toJsonString(requestSummary))
|
||||
.responseSummary(responseSummary)
|
||||
.status(status)
|
||||
.errorCode(errorCode)
|
||||
.errorMessage(errorMessage)
|
||||
.build());
|
||||
}
|
||||
|
||||
private JsonNode json(String value) {
|
||||
if (!StringUtils.hasText(value)) {
|
||||
return JsonUtils.getObjectMapper().createObjectNode();
|
||||
}
|
||||
try {
|
||||
return JsonUtils.getObjectMapper().readTree(value);
|
||||
} catch (Exception ex) {
|
||||
return JsonUtils.getObjectMapper().createObjectNode();
|
||||
}
|
||||
}
|
||||
|
||||
private String text(JsonNode node, String fieldName) {
|
||||
JsonNode value = node == null ? null : node.get(fieldName);
|
||||
return value == null || value.isNull() ? null : value.asText();
|
||||
}
|
||||
|
||||
private double number(JsonNode node, String fieldName, double defaultValue) {
|
||||
JsonNode value = node == null ? null : node.get(fieldName);
|
||||
return value == null || !value.isNumber() ? defaultValue : value.asDouble();
|
||||
}
|
||||
|
||||
private double deterministicScore(MuseAiEvaluationRunDO run, String datasetReference,
|
||||
int sampleIndex, String dimensionName) {
|
||||
String seed = run.getPolicyKey() + "\n" + run.getPolicyVersionNo() + "\n"
|
||||
+ datasetReference + "\n" + sampleIndex + "\n" + dimensionName;
|
||||
String hash = sha256(seed);
|
||||
long bucket = Long.parseUnsignedLong(hash.substring(0, 12), 16) % 3500L;
|
||||
return round(0.65D + bucket / 10_000D);
|
||||
}
|
||||
|
||||
private String sha256(String payload) {
|
||||
try {
|
||||
MessageDigest digest = MessageDigest.getInstance("SHA-256");
|
||||
return HexFormat.of().formatHex(digest.digest(String.valueOf(payload).getBytes(StandardCharsets.UTF_8)));
|
||||
} catch (NoSuchAlgorithmException ex) {
|
||||
throw new IllegalStateException("JDK 缺少 SHA-256 摘要算法", ex);
|
||||
}
|
||||
}
|
||||
|
||||
private String sampleId(int sampleIndex) {
|
||||
return "sample-" + String.format(Locale.ROOT, "%04d", sampleIndex);
|
||||
}
|
||||
|
||||
private double clamp(double value) {
|
||||
return Math.max(0D, Math.min(1D, value));
|
||||
}
|
||||
|
||||
private double round(double value) {
|
||||
return BigDecimal.valueOf(value).setScale(4, RoundingMode.HALF_UP).doubleValue();
|
||||
}
|
||||
|
||||
private int safeSampleCount(Integer sampleCount) {
|
||||
return sampleCount == null ? 0 : Math.max(0, sampleCount);
|
||||
}
|
||||
|
||||
private String sanitizeError(String message) {
|
||||
if (!StringUtils.hasText(message)) {
|
||||
return "Quality evaluation failed";
|
||||
}
|
||||
return SECRET_PATTERN.matcher(message).replaceAll("[REDACTED]");
|
||||
}
|
||||
|
||||
private record DimensionConfig(String name, double threshold, double weight) {
|
||||
}
|
||||
|
||||
private record SampleReport(String sampleId, boolean passed, double score, Map<String, Double> dimensionScores) {
|
||||
}
|
||||
|
||||
private record EvaluationReport(List<SampleReport> samples, Map<String, Double> dimensionScores,
|
||||
double overallScore, int passedCount, int failedCount, String metricSummary) {
|
||||
|
||||
String auditSummary() {
|
||||
return JsonUtils.toJsonString(Map.of(
|
||||
"overallScore", overallScore,
|
||||
"passedCount", passedCount,
|
||||
"failedCount", failedCount,
|
||||
"dimensionScores", dimensionScores));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private static class EvaluationExecutionException extends RuntimeException {
|
||||
|
||||
private final String errorCode;
|
||||
|
||||
EvaluationExecutionException(String errorCode, String message) {
|
||||
super(message);
|
||||
this.errorCode = errorCode;
|
||||
}
|
||||
|
||||
String errorCode() {
|
||||
return errorCode;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,97 @@
|
||||
package cn.iocoder.muse.module.ai.application.muse;
|
||||
|
||||
import cn.iocoder.muse.framework.tenant.core.util.TenantUtils;
|
||||
import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiJobDO;
|
||||
import cn.iocoder.muse.module.ai.dal.mysql.muse.MuseAiJobMapper;
|
||||
import cn.iocoder.muse.module.ai.framework.ai.config.MuseAiProperties;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.time.LocalDateTime;
|
||||
|
||||
/**
|
||||
* 质量评估 worker。
|
||||
*
|
||||
* <p>后台定时默认启用；发布环境可显式设置 {@code muse.ai.evaluation.worker.enabled=false} 暂停消费。
|
||||
* 测试和运维脚本可直接调用 {@link #dispatchOnce()} 做单步执行，避免依赖真实时间。</p>
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
public class MuseQualityEvaluationWorker {
|
||||
|
||||
private static final String STATUS_FAILED = "failed";
|
||||
private static final String ERROR_INVALID_JOB = "AI_QUALITY_EVALUATION_INVALID_JOB";
|
||||
private static final String ERROR_DISPATCH_FAILED = "AI_QUALITY_EVALUATION_DISPATCH_FAILED";
|
||||
|
||||
@Resource
|
||||
private MuseAiJobMapper jobMapper;
|
||||
@Resource
|
||||
private MuseQualityEvaluationExecutor evaluationExecutor;
|
||||
@Resource
|
||||
private MuseAiProperties properties;
|
||||
|
||||
public MuseQualityEvaluationWorker() {
|
||||
}
|
||||
|
||||
MuseQualityEvaluationWorker(MuseAiJobMapper jobMapper, MuseQualityEvaluationExecutor evaluationExecutor,
|
||||
MuseAiProperties properties) {
|
||||
this.jobMapper = jobMapper;
|
||||
this.evaluationExecutor = evaluationExecutor;
|
||||
this.properties = properties;
|
||||
}
|
||||
|
||||
@Scheduled(initialDelayString = "${muse.ai.evaluation.worker.initial-delay-ms:1000}",
|
||||
fixedDelayString = "${muse.ai.evaluation.worker.fixed-delay-ms:1000}")
|
||||
public void dispatchScheduled() {
|
||||
dispatchOnce();
|
||||
}
|
||||
|
||||
public int dispatchOnce() {
|
||||
if (!isEnabled()) {
|
||||
return 0;
|
||||
}
|
||||
MuseAiJobDO job = TenantUtils.executeIgnore(() -> jobMapper.claimNextQueuedEvaluationJob());
|
||||
if (job == null) {
|
||||
return 0;
|
||||
}
|
||||
if (job.getTenantId() == null || job.getId() == null) {
|
||||
failClaimedJob(job, ERROR_INVALID_JOB, "Quality evaluation worker claimed invalid job");
|
||||
return 0;
|
||||
}
|
||||
try {
|
||||
TenantUtils.execute(job.getTenantId(), () -> evaluationExecutor.executeEvaluationJob(job.getId()));
|
||||
return 1;
|
||||
} catch (RuntimeException ex) {
|
||||
log.warn("[dispatchOnce][jobPkId({}) executor failed closed, errorType={}]",
|
||||
job.getId(), ex.getClass().getSimpleName());
|
||||
failClaimedJob(job, ERROR_DISPATCH_FAILED, "Quality evaluation dispatcher failed");
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
private boolean isEnabled() {
|
||||
return properties != null
|
||||
&& properties.getEvaluation() != null
|
||||
&& properties.getEvaluation().getWorker() != null
|
||||
&& properties.getEvaluation().getWorker().isEnabled();
|
||||
}
|
||||
|
||||
private void failClaimedJob(MuseAiJobDO claimedJob, String errorCode, String errorMessage) {
|
||||
if (claimedJob == null || claimedJob.getId() == null) {
|
||||
return;
|
||||
}
|
||||
MuseAiJobDO update = new MuseAiJobDO();
|
||||
update.setId(claimedJob.getId());
|
||||
update.setStatus(STATUS_FAILED);
|
||||
update.setRetryable(false);
|
||||
update.setErrorCode(errorCode);
|
||||
update.setErrorMessage(errorMessage);
|
||||
update.setFinishedAt(LocalDateTime.now());
|
||||
update.setTenantId(claimedJob.getTenantId());
|
||||
// claim 已经把 job 推到 running；异常路径必须关闭 job，避免后台 worker 留下永久 running。
|
||||
TenantUtils.executeIgnore(() -> jobMapper.updateById(update));
|
||||
}
|
||||
|
||||
}
|
||||
@ -54,6 +54,7 @@ public class MuseQualityPolicyServiceImpl implements MuseQualityPolicyService {
|
||||
private static final String STATUS_DRAFT = "draft";
|
||||
private static final String STATUS_QUEUED = "queued";
|
||||
private static final String JOB_TYPE_EVALUATION = "evaluation";
|
||||
private static final int DEFAULT_EVALUATION_SAMPLE_SIZE = 20;
|
||||
private static final AtomicLong NUMERIC_ID_SEQUENCE = new AtomicLong(System.currentTimeMillis() * 1000L);
|
||||
private static final Pattern SECRET_PATTERN = Pattern.compile("(?i)(sk-[a-z0-9_-]+|bearer\\s+\\S+|private response body|token\\s*[:=]?\\s*\\S+)");
|
||||
|
||||
@ -387,7 +388,7 @@ public class MuseQualityPolicyServiceImpl implements MuseQualityPolicyService {
|
||||
}
|
||||
|
||||
private int safeSampleSize(Integer sampleSize) {
|
||||
return sampleSize == null || sampleSize < 0 ? 0 : sampleSize;
|
||||
return sampleSize == null ? DEFAULT_EVALUATION_SAMPLE_SIZE : Math.max(0, sampleSize);
|
||||
}
|
||||
|
||||
private Long nextNumericId() {
|
||||
|
||||
@ -0,0 +1,41 @@
|
||||
package cn.iocoder.muse.module.ai.dal.dataobject.muse;
|
||||
|
||||
import cn.iocoder.muse.framework.tenant.core.db.TenantBaseDO;
|
||||
import cn.iocoder.muse.module.ai.dal.type.JsonbStringTypeHandler;
|
||||
import com.baomidou.mybatisplus.annotation.IdType;
|
||||
import com.baomidou.mybatisplus.annotation.TableField;
|
||||
import com.baomidou.mybatisplus.annotation.TableId;
|
||||
import com.baomidou.mybatisplus.annotation.TableName;
|
||||
import lombok.Data;
|
||||
import lombok.EqualsAndHashCode;
|
||||
import lombok.ToString;
|
||||
|
||||
import java.math.BigDecimal;
|
||||
|
||||
/**
|
||||
* Muse AI 离线评估单样本结果 DO。
|
||||
*/
|
||||
@TableName(value = "muse_ai_evaluation_sample_result", autoResultMap = true)
|
||||
@Data
|
||||
@EqualsAndHashCode(callSuper = true)
|
||||
@ToString(callSuper = true)
|
||||
public class MuseAiEvaluationSampleResultDO extends TenantBaseDO {
|
||||
|
||||
@TableId(type = IdType.AUTO)
|
||||
private Long id;
|
||||
|
||||
private String runId;
|
||||
private String sampleId;
|
||||
private Long ownerUserId;
|
||||
private String status;
|
||||
private BigDecimal score;
|
||||
@TableField(typeHandler = JsonbStringTypeHandler.class)
|
||||
private String metricSummary;
|
||||
@TableField(typeHandler = JsonbStringTypeHandler.class)
|
||||
private String inputSummary;
|
||||
@TableField(typeHandler = JsonbStringTypeHandler.class)
|
||||
private String outputSummary;
|
||||
private String errorCode;
|
||||
private String errorMessage;
|
||||
|
||||
}
|
||||
@ -14,4 +14,8 @@ public interface MuseAiEvaluationRunMapper extends BaseMapperX<MuseAiEvaluationR
|
||||
return selectOne(MuseAiEvaluationRunDO::getRunId, runId);
|
||||
}
|
||||
|
||||
default MuseAiEvaluationRunDO selectByJobId(String jobId) {
|
||||
return selectOne(MuseAiEvaluationRunDO::getJobId, jobId);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -0,0 +1,12 @@
|
||||
package cn.iocoder.muse.module.ai.dal.mysql.muse;
|
||||
|
||||
import cn.iocoder.muse.framework.mybatis.core.mapper.BaseMapperX;
|
||||
import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiEvaluationSampleResultDO;
|
||||
import org.apache.ibatis.annotations.Mapper;
|
||||
|
||||
/**
|
||||
* Muse AI 离线评估单样本结果 Mapper。
|
||||
*/
|
||||
@Mapper
|
||||
public interface MuseAiEvaluationSampleResultMapper extends BaseMapperX<MuseAiEvaluationSampleResultDO> {
|
||||
}
|
||||
@ -64,6 +64,33 @@ public interface MuseAiJobMapper extends BaseMapperX<MuseAiJobDO> {
|
||||
""")
|
||||
MuseAiJobDO claimNextQueuedRuntimeJob();
|
||||
|
||||
/**
|
||||
* 原子领取下一条待执行质量评估 job。
|
||||
*
|
||||
* <p>质量评估和在线生成调用不同执行器，不能混入 {@code ai_task_runtime} dispatcher；这里单独按
|
||||
* {@code job_type='evaluation'} 领取，领取时在同一 SQL 内推进到 running，避免多 worker 重复执行。</p>
|
||||
*/
|
||||
@TenantIgnore
|
||||
@Select("""
|
||||
UPDATE muse_ai_job
|
||||
SET status = 'running',
|
||||
started_at = COALESCE(started_at, CURRENT_TIMESTAMP),
|
||||
update_time = CURRENT_TIMESTAMP
|
||||
WHERE id = (
|
||||
SELECT id
|
||||
FROM muse_ai_job
|
||||
WHERE job_type = 'evaluation'
|
||||
AND status = 'queued'
|
||||
AND (next_retry_at IS NULL OR next_retry_at <= CURRENT_TIMESTAMP)
|
||||
AND deleted = FALSE
|
||||
ORDER BY create_time ASC, id ASC
|
||||
FOR UPDATE SKIP LOCKED
|
||||
LIMIT 1
|
||||
)
|
||||
RETURNING *
|
||||
""")
|
||||
MuseAiJobDO claimNextQueuedEvaluationJob();
|
||||
|
||||
/**
|
||||
* 在同一 retry group 内查找未终止 retry，调用方已锁定源 Job，因此这里用于锁内防重复创建。
|
||||
*/
|
||||
|
||||
@ -73,6 +73,11 @@ public class MuseAiProperties {
|
||||
*/
|
||||
private Events events = new Events();
|
||||
|
||||
/**
|
||||
* Muse 离线质量评估配置。
|
||||
*/
|
||||
private Evaluation evaluation = new Evaluation();
|
||||
|
||||
@Data
|
||||
public static class Gemini {
|
||||
|
||||
@ -251,4 +256,27 @@ public class MuseAiProperties {
|
||||
|
||||
}
|
||||
|
||||
@Data
|
||||
public static class Evaluation {
|
||||
|
||||
/**
|
||||
* 质量评估 worker 配置。
|
||||
*/
|
||||
private Worker worker = new Worker();
|
||||
|
||||
@Data
|
||||
public static class Worker {
|
||||
|
||||
/**
|
||||
* 是否启用质量评估后台 worker；默认启用。
|
||||
*
|
||||
* <p>当前执行器只处理合成/脱敏数据集，不调用外部 LLM，不读取私有正文；生产如需冻结历史 queued
|
||||
* evaluation job，可显式设为 {@code false}。</p>
|
||||
*/
|
||||
private boolean enabled = true;
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -0,0 +1,216 @@
|
||||
package cn.iocoder.muse.module.ai.application.muse;
|
||||
|
||||
import cn.iocoder.muse.framework.common.util.json.JsonUtils;
|
||||
import cn.iocoder.muse.framework.test.core.ut.BaseMockitoUnitTest;
|
||||
import cn.iocoder.muse.framework.tenant.core.context.TenantContextHolder;
|
||||
import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiEvaluationRunDO;
|
||||
import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiEvaluationSampleResultDO;
|
||||
import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseAiJobDO;
|
||||
import cn.iocoder.muse.module.ai.dal.dataobject.muse.MuseQualityPolicyVersionDO;
|
||||
import cn.iocoder.muse.module.ai.dal.mysql.muse.MuseAiEvaluationRunMapper;
|
||||
import cn.iocoder.muse.module.ai.dal.mysql.muse.MuseAiEvaluationSampleResultMapper;
|
||||
import cn.iocoder.muse.module.ai.dal.mysql.muse.MuseAiJobMapper;
|
||||
import cn.iocoder.muse.module.ai.dal.mysql.muse.MuseQualityPolicyVersionMapper;
|
||||
import cn.iocoder.muse.module.ai.framework.ai.config.MuseAiProperties;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.mockito.ArgumentCaptor;
|
||||
import org.mockito.InjectMocks;
|
||||
import org.mockito.Mock;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.argThat;
|
||||
import static org.mockito.Mockito.doThrow;
|
||||
import static org.mockito.Mockito.never;
|
||||
import static org.mockito.Mockito.verify;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
/**
|
||||
* 质量评估 worker / executor 测试。
|
||||
*/
|
||||
class MuseQualityEvaluationWorkerTest extends BaseMockitoUnitTest {
|
||||
|
||||
@InjectMocks
|
||||
private MuseQualityEvaluationExecutor executor;
|
||||
@Mock
|
||||
private MuseAiJobMapper jobMapper;
|
||||
@Mock
|
||||
private MuseAiEvaluationRunMapper evaluationRunMapper;
|
||||
@Mock
|
||||
private MuseAiEvaluationSampleResultMapper sampleResultMapper;
|
||||
@Mock
|
||||
private MuseQualityPolicyVersionMapper qualityPolicyVersionMapper;
|
||||
@Mock
|
||||
private MuseAiAuditService auditService;
|
||||
|
||||
@AfterEach
|
||||
void clearTenantContext() {
|
||||
TenantContextHolder.clear();
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_dispatchEnabledEvaluationJobAndRestoreTenantContext() {
|
||||
MuseAiProperties properties = enabledProperties(true);
|
||||
MuseQualityEvaluationWorker worker = new MuseQualityEvaluationWorker(jobMapper, executor, properties);
|
||||
MuseAiJobDO claimed = job("8001", 4001L, 100L);
|
||||
when(jobMapper.claimNextQueuedEvaluationJob()).thenReturn(claimed);
|
||||
when(jobMapper.selectById(4001L)).thenReturn(claimed);
|
||||
when(evaluationRunMapper.selectByJobId("8001")).thenReturn(run("7001", "dataset://quality-v1", 3));
|
||||
when(qualityPolicyVersionMapper.selectByPolicyKeyAndVersion("novel_quality", 1))
|
||||
.thenReturn(policyVersion("[{\"name\":\"coherence\",\"threshold\":0.7,\"weight\":1.0}]"));
|
||||
|
||||
int dispatched = worker.dispatchOnce();
|
||||
|
||||
assertEquals(1, dispatched);
|
||||
verify(jobMapper).claimNextQueuedEvaluationJob();
|
||||
verify(evaluationRunMapper).updateById(argThat((MuseAiEvaluationRunDO update) ->
|
||||
Long.valueOf(3001L).equals(update.getId()) && "running".equals(update.getStatus())));
|
||||
ArgumentCaptor<MuseAiEvaluationSampleResultDO> sampleCaptor =
|
||||
ArgumentCaptor.forClass(MuseAiEvaluationSampleResultDO.class);
|
||||
verify(sampleResultMapper, org.mockito.Mockito.times(3)).insert(sampleCaptor.capture());
|
||||
assertTrue(sampleCaptor.getAllValues().stream().allMatch(sample ->
|
||||
"7001".equals(sample.getRunId())
|
||||
&& sample.getInputSummary().contains("datasetReferenceHash")
|
||||
&& !sample.getInputSummary().contains("dataset://quality-v1")));
|
||||
verify(evaluationRunMapper).updateById(argThat((MuseAiEvaluationRunDO update) ->
|
||||
Long.valueOf(3001L).equals(update.getId())
|
||||
&& "completed".equals(update.getStatus())
|
||||
&& update.getMetricSummary().contains("\"deterministic-local-v1\"")
|
||||
&& Integer.valueOf(1).equals(update.getFailedCount())));
|
||||
verify(jobMapper).updateById(argThat((MuseAiJobDO update) ->
|
||||
Long.valueOf(4001L).equals(update.getId())
|
||||
&& "completed".equals(update.getStatus())
|
||||
&& update.getResultSummary().contains("\"overallScore\"")));
|
||||
verify(auditService).record(argThat(req ->
|
||||
"executeQualityEvaluationRun".equals(req.getOperationId())
|
||||
&& "worker".equals(req.getSide())
|
||||
&& "succeeded".equals(req.getStatus())));
|
||||
assertNull(TenantContextHolder.getTenantId(), "质量评估 worker 执行结束后必须恢复租户上下文");
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_notDispatchWhenWorkerDisabled() {
|
||||
MuseQualityEvaluationWorker worker = new MuseQualityEvaluationWorker(jobMapper, executor, enabledProperties(false));
|
||||
|
||||
int dispatched = worker.dispatchOnce();
|
||||
|
||||
assertEquals(0, dispatched);
|
||||
verify(jobMapper, never()).claimNextQueuedEvaluationJob();
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_failRunAndJobWhenDatasetReferenceIsPrivate() {
|
||||
TenantContextHolder.setTenantId(100L);
|
||||
MuseAiJobDO job = job("8001", 4001L, 100L);
|
||||
when(jobMapper.selectById(4001L)).thenReturn(job);
|
||||
when(evaluationRunMapper.selectByJobId("8001")).thenReturn(run("7001", "work://private/100", 2));
|
||||
executor.executeEvaluationJob(4001L);
|
||||
|
||||
verify(sampleResultMapper, never()).insert(any(MuseAiEvaluationSampleResultDO.class));
|
||||
verify(evaluationRunMapper).updateById(argThat((MuseAiEvaluationRunDO update) ->
|
||||
Long.valueOf(3001L).equals(update.getId())
|
||||
&& "failed".equals(update.getStatus())
|
||||
&& "AI_QUALITY_EVALUATION_DATASET_BLOCKED".equals(update.getErrorCode())
|
||||
&& update.getErrorMessage() != null
|
||||
&& !update.getErrorMessage().contains("work://private")));
|
||||
verify(jobMapper).updateById(argThat((MuseAiJobDO update) ->
|
||||
Long.valueOf(4001L).equals(update.getId())
|
||||
&& "failed".equals(update.getStatus())
|
||||
&& "AI_QUALITY_EVALUATION_DATASET_BLOCKED".equals(update.getErrorCode())));
|
||||
verify(auditService).record(argThat(req ->
|
||||
"failed".equals(req.getStatus())
|
||||
&& "AI_QUALITY_EVALUATION_DATASET_BLOCKED".equals(req.getErrorCode())
|
||||
&& req.getRequestSummary().contains("datasetReferenceHash")
|
||||
&& !req.getRequestSummary().contains("work://private")));
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_failClosedWhenExecutorThrowsAfterClaimedJob() {
|
||||
MuseQualityEvaluationExecutor throwingExecutor = org.mockito.Mockito.mock(MuseQualityEvaluationExecutor.class);
|
||||
MuseQualityEvaluationWorker worker = new MuseQualityEvaluationWorker(jobMapper, throwingExecutor, enabledProperties(true));
|
||||
MuseAiJobDO claimed = job("8001", 4001L, 100L);
|
||||
when(jobMapper.claimNextQueuedEvaluationJob()).thenReturn(claimed);
|
||||
doThrow(new IllegalStateException("db unavailable")).when(throwingExecutor).executeEvaluationJob(4001L);
|
||||
|
||||
int dispatched = worker.dispatchOnce();
|
||||
|
||||
assertEquals(0, dispatched);
|
||||
verify(jobMapper).updateById(argThat((MuseAiJobDO update) ->
|
||||
Long.valueOf(4001L).equals(update.getId())
|
||||
&& "failed".equals(update.getStatus())
|
||||
&& "AI_QUALITY_EVALUATION_DISPATCH_FAILED".equals(update.getErrorCode())
|
||||
&& Boolean.FALSE.equals(update.getRetryable())));
|
||||
assertNull(TenantContextHolder.getTenantId(), "异常路径也必须恢复租户上下文");
|
||||
}
|
||||
|
||||
@Test
|
||||
void should_keepCompletedStateWhenAuditWriteFails() {
|
||||
TenantContextHolder.setTenantId(100L);
|
||||
MuseAiJobDO job = job("8001", 4001L, 100L);
|
||||
when(jobMapper.selectById(4001L)).thenReturn(job);
|
||||
when(evaluationRunMapper.selectByJobId("8001")).thenReturn(run("7001", "redacted-fixture-v1", 1));
|
||||
when(qualityPolicyVersionMapper.selectByPolicyKeyAndVersion("novel_quality", 1))
|
||||
.thenReturn(policyVersion("[{\"name\":\"coherence\",\"threshold\":0.7,\"weight\":1.0}]"));
|
||||
doThrow(new IllegalStateException("audit db unavailable")).when(auditService).record(any());
|
||||
|
||||
executor.executeEvaluationJob(4001L);
|
||||
|
||||
verify(evaluationRunMapper).updateById(argThat((MuseAiEvaluationRunDO update) ->
|
||||
Long.valueOf(3001L).equals(update.getId()) && "completed".equals(update.getStatus())));
|
||||
verify(jobMapper).updateById(argThat((MuseAiJobDO update) ->
|
||||
Long.valueOf(4001L).equals(update.getId()) && "completed".equals(update.getStatus())));
|
||||
}
|
||||
|
||||
private static MuseAiProperties enabledProperties(boolean enabled) {
|
||||
MuseAiProperties properties = new MuseAiProperties();
|
||||
MuseAiProperties.Evaluation evaluation = new MuseAiProperties.Evaluation();
|
||||
MuseAiProperties.Evaluation.Worker worker = new MuseAiProperties.Evaluation.Worker();
|
||||
worker.setEnabled(enabled);
|
||||
evaluation.setWorker(worker);
|
||||
properties.setEvaluation(evaluation);
|
||||
return properties;
|
||||
}
|
||||
|
||||
private static MuseAiJobDO job(String jobId, Long id, Long tenantId) {
|
||||
MuseAiJobDO job = new MuseAiJobDO();
|
||||
job.setId(id);
|
||||
job.setJobId(jobId);
|
||||
job.setCommandId("cmd-run-1");
|
||||
job.setRequestHash("hash-run");
|
||||
job.setActorUserId(1001L);
|
||||
job.setOwnerUserId(1001L);
|
||||
job.setJobType("evaluation");
|
||||
job.setStatus("running");
|
||||
job.setTenantId(tenantId);
|
||||
return job;
|
||||
}
|
||||
|
||||
private static MuseAiEvaluationRunDO run(String runId, String datasetReference, int sampleCount) {
|
||||
MuseAiEvaluationRunDO run = new MuseAiEvaluationRunDO();
|
||||
run.setId(3001L);
|
||||
run.setRunId(runId);
|
||||
run.setJobId("8001");
|
||||
run.setPolicyKey("novel_quality");
|
||||
run.setPolicyVersionNo(1);
|
||||
run.setOwnerUserId(1001L);
|
||||
run.setSampleCount(sampleCount);
|
||||
run.setMetricSummary(JsonUtils.toJsonString(Map.of("datasetReference", datasetReference)));
|
||||
return run;
|
||||
}
|
||||
|
||||
private static MuseQualityPolicyVersionDO policyVersion(String thresholdSnapshot) {
|
||||
MuseQualityPolicyVersionDO version = new MuseQualityPolicyVersionDO();
|
||||
version.setPolicyKey("novel_quality");
|
||||
version.setVersionNo(1);
|
||||
version.setThresholdSnapshot(thresholdSnapshot);
|
||||
return version;
|
||||
}
|
||||
|
||||
}
|
||||
@ -257,7 +257,8 @@ class MuseQualityPolicyServiceTest extends BaseMockitoUnitTest {
|
||||
|
||||
qualityPolicyService.startEvaluationRun(request, 1001L);
|
||||
|
||||
verify(evaluationRunMapper).insert(argThat((MuseAiEvaluationRunDO run) -> !"completed".equals(run.getStatus())));
|
||||
verify(evaluationRunMapper).insert(argThat((MuseAiEvaluationRunDO run) -> !"completed".equals(run.getStatus())
|
||||
&& Integer.valueOf(20).equals(run.getSampleCount())));
|
||||
verify(jobMapper).insert(argThat((MuseAiJobDO job) -> !"completed".equals(job.getStatus())));
|
||||
}
|
||||
|
||||
|
||||
@ -58,6 +58,7 @@ class MuseAiJobMapperTest extends BaseMockitoUnitTest {
|
||||
void should_defineRowLockAndActiveRetryQueriesForJobCommands() throws Exception {
|
||||
Method lockMethod = MuseAiJobMapper.class.getMethod("selectByIdForUpdate", Long.class, Long.class);
|
||||
Method claimMethod = MuseAiJobMapper.class.getMethod("claimNextQueuedRuntimeJob");
|
||||
Method evaluationClaimMethod = MuseAiJobMapper.class.getMethod("claimNextQueuedEvaluationJob");
|
||||
Method activeRetryMethod = MuseAiJobMapper.class.getMethod("selectActiveRetryByRetryGroupId",
|
||||
Long.class, String.class, Long.class);
|
||||
Method insertActiveRetryMethod = MuseAiJobMapper.class.getMethod("insertActiveRetryIfAbsent",
|
||||
@ -65,6 +66,7 @@ class MuseAiJobMapperTest extends BaseMockitoUnitTest {
|
||||
|
||||
String lockSql = normalizeSql(lockMethod.getAnnotation(org.apache.ibatis.annotations.Select.class).value()[0]);
|
||||
String claimSql = normalizeSql(claimMethod.getAnnotation(org.apache.ibatis.annotations.Select.class).value()[0]);
|
||||
String evaluationClaimSql = normalizeSql(evaluationClaimMethod.getAnnotation(org.apache.ibatis.annotations.Select.class).value()[0]);
|
||||
String activeRetrySql = normalizeSql(activeRetryMethod.getAnnotation(org.apache.ibatis.annotations.Select.class).value()[0]);
|
||||
String insertActiveRetrySql = normalizeSql(insertActiveRetryMethod.getAnnotation(
|
||||
org.apache.ibatis.annotations.Insert.class).value()[0]);
|
||||
@ -75,8 +77,16 @@ class MuseAiJobMapperTest extends BaseMockitoUnitTest {
|
||||
"claim SQL 必须在同一原子语句内把 queued job 推进到 running");
|
||||
assertTrue(claimSql.contains("job_type = 'ai_task_runtime'"),
|
||||
"P1R-4 dispatcher 只能领取 AI task runtime job，不能误执行 agent test 或其它 owner job");
|
||||
assertTrue(evaluationClaimSql.contains("for update skip locked"),
|
||||
"质量评估 worker 必须用数据库行锁领取 queued job，避免多 worker 重复执行");
|
||||
assertTrue(evaluationClaimSql.contains("job_type = 'evaluation'"),
|
||||
"质量评估 worker 只能领取 evaluation job，不能混入在线生成 runtime job");
|
||||
assertTrue(evaluationClaimSql.contains("status = 'queued'") && evaluationClaimSql.contains("status = 'running'"),
|
||||
"质量评估 claim SQL 必须在同一原子语句内把 queued job 推进到 running");
|
||||
assertTrue(claimSql.contains("next_retry_at is null") && claimSql.contains("next_retry_at <= current_timestamp"),
|
||||
"claim SQL 必须尊重 retry backoff 时间，不能提前重跑");
|
||||
assertTrue(evaluationClaimSql.contains("next_retry_at is null") && evaluationClaimSql.contains("next_retry_at <= current_timestamp"),
|
||||
"质量评估 claim SQL 必须尊重 retry backoff 时间，不能提前重跑");
|
||||
assertTrue(activeRetrySql.contains("retry_group_id"));
|
||||
assertTrue(activeRetrySql.contains("operation_id = 'adminretryjob'"),
|
||||
"active retry 查询只能复用 Job retry 命令创建的 retry job，不能误把普通源 job 当作 retry job");
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user