From 81c2f85c3adca5b087988dc396c660d5e628e5aa Mon Sep 17 00:00:00 2001 From: caoxiaozhu Date: Wed, 19 Aug 2026 14:21:54 +0800 Subject: [PATCH] =?UTF-8?q?feat(data=5Fprocess):=20=E9=97=AE=E7=AD=94?= =?UTF-8?q?=E5=AF=B9=E6=95=B0=E6=8D=AE=E8=AF=84=E6=B5=8B=E4=BD=93=E7=B3=BB?= =?UTF-8?q?=E4=B8=8E=E8=B4=A8=E9=87=8F=E5=88=86=E9=9B=B7=E8=BE=BE=E5=9B=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 三层评测：规则层沿用原五维规则分，语义层用本地 BGE 向量算问答/来源 相关性，评审层复用生成模型按 rubric 打分（忠实度/正确性/清晰度等， 区分 standard/reasoning/dpo 输出类型），任一层失败自动降级 - 组合分 = 规则 35% + 语义 20% + 评审 45%，缺层自动重归一 - 新增 results/evaluate-batch 批量评测接口，镜像批量重生成的并发、 乐观锁与部分成功语义；生成阶段不再展示质量分 - 详情页与结果编辑页新增"数据评测"按钮和批量进度；质量分列悬停弹出 雷达图浮窗（评审 5 维 + 语义 2 维、三层分项、评审理由） - 手动编辑/恢复后重算规则与语义层并丢弃过期评审分，雷达图不再展示 失效数据 --- backend/app/api/v1/endpoints/data_process.py | 244 +++++++++++++- .../data_process/algorithms/quality.py | 85 +++++ .../app/modules/data_process/evaluation.py | 313 ++++++++++++++++++ backend/app/schemas/data_process.py | 20 ++ backend/tests/test_data_process_api.py | 217 ++++++++++++ backend/tests/test_data_process_evaluation.py | 284 ++++++++++++++++ frontend/src/api/modules/dataProcess.ts | 11 + frontend/src/plugins/echarts.ts | 5 +- frontend/src/types/dataProcess.ts | 51 +++ .../data-process/DataProcessCreateView.vue | 16 + .../data-process/DataProcessDetailView.vue | 171 +++++++++- .../create/QualityRadarPopover.vue | 252 ++++++++++++++ .../data-process/create/ResultEditorStep.vue | 121 ++++++- .../src/views/data-process/create/types.ts | 21 +- .../create/useDataProcessEvaluation.ts | 126 +++++++ .../create/useDataProcessGeneration.ts | 13 +- 16 files changed, 1917 insertions(+), 33 deletions(-) create mode 100644 backend/app/modules/data_process/evaluation.py create mode 100644 backend/tests/test_data_process_evaluation.py create mode 100644 frontend/src/views/data-process/create/QualityRadarPopover.vue create mode 100644 frontend/src/views/data-process/create/useDataProcessEvaluation.ts diff --git a/backend/app/api/v1/endpoints/data_process.py b/backend/app/api/v1/endpoints/data_process.py index b706779..2a23cde 100644 --- a/backend/app/api/v1/endpoints/data_process.py +++ b/backend/app/api/v1/endpoints/data_process.py @@ -60,6 +60,10 @@ from app.modules.data_process.document_chunking import ( chunk_semantic_text, merge_short_chunks, ) +from app.modules.data_process.evaluation import ( + evaluate_result_record, + reevaluate_edited_record, +) from app.modules.data_process.generation import generate_model_records from app.modules.data_process.office_preview import ( MAX_XLSX_PREVIEW_ROWS, @@ -96,6 +100,7 @@ from app.schemas.data_process import ( PreviewItemUpdate, ProcessType, PublishRequest, + ResultBatchEvaluateRequest, ResultBatchRegenerateRequest, ResultRegenerateRequest, ResultUpdate, @@ -2039,12 +2044,13 @@ def update_result( or 20 ), ) - quality = score_quality( + # 编辑后内容已变化：重算规则与语义层，旧的评审分不再可信直接丢弃。 + update["quality_score"] = reevaluate_edited_record( merged, - min_output_length=minimum, source_content=source_content, + previous_quality=current.get("quality_score"), + min_output_length=minimum, ) - update["quality_score"] = asdict(quality) result = store.update_result( task_id, result_id, @@ -2089,11 +2095,6 @@ def restore_result( or 20 ), ) - quality = score_quality( - restored, - min_output_length=minimum, - source_content=source_content, - ) restored = store.update_result( task_id, result_id, @@ -2103,7 +2104,12 @@ def restore_result( "output": restored["output"], "chosen": restored["chosen"], "rejected": restored["rejected"], - "quality_score": asdict(quality), + "quality_score": reevaluate_edited_record( + restored, + source_content=source_content, + previous_quality=current.get("quality_score"), + min_output_length=minimum, + ), "expected_updated_at": current.get("updated_at"), }, ) @@ -2266,6 +2272,46 @@ def _safe_regeneration_error(exc: Exception) -> str: return re.sub(r"\s+", " ", str(exc)).strip()[:500] or "result regeneration failed" +def _evaluate_result_in_place( + task_id: str, + current: dict[str, Any], + source_content: str, + config: dict[str, Any], + evaluation_model: dict[str, Any] | None, + store: DataProcessStore, + *, + expected_updated_at: str, + model_client: httpx.Client | None = None, + minimum: int = 20, +) -> dict[str, Any]: + """评测单条结果并落库；复用逐结果互斥锁避免与重生成并发写冲突。""" + + result_id = str(current["id"]) + with _claim_result_regeneration(task_id, result_id): + quality = evaluate_result_record( + { + "instruction": current.get("instruction"), + "input": current.get("input"), + "output": current.get("output"), + "chosen": current.get("chosen"), + "rejected": current.get("rejected"), + }, + source_content=source_content, + model=evaluation_model, + config=config, + client=model_client, + min_output_length=minimum, + ) + return store.update_result( + task_id, + result_id, + { + "quality_score": quality, + "expected_updated_at": expected_updated_at, + }, + ) + + @router.post("/{task_id}/results/regenerate-batch") def regenerate_results_batch( task_id: str, @@ -2439,6 +2485,186 @@ def regenerate_results_batch( ) +@router.post("/{task_id}/results/evaluate-batch") +def evaluate_results_batch( + task_id: str, + payload: ResultBatchEvaluateRequest, + store: DataProcessStore = Depends(get_data_process_store), +) -> dict[str, Any]: + """对一批结果执行三层质量评测（规则+语义+评审），允许部分成功。""" + + started_at = time.perf_counter() + batch_id = new_id("dpeb") + with api_errors(): + task = store.get_task(task_id) + if task.get("status") == "running": + raise ConflictError("data process task is running") + if task.get("output_dataset_id"): + raise InvalidStateError("published results cannot be evaluated") + config = task.get("config") or {} + evaluation_model: dict[str, Any] | None = None + model_id = _value(config, "generation_model_id", "generationModelId", None) + if model_id: + try: + evaluation_model = store.get_generation_model(str(model_id)) + except NotFoundError: + logger.warning( + "data process evaluation model unavailable, judge layer " + "skipped task_id=%s model_id=%s", + task_id, + model_id, + ) + evaluation_config = { + **config, + "output_type": str( + _value(config, "output_type", "outputType", "standard") + ).strip().lower(), + } + minimum = max( + 1, + int(_value(config, "min_output_length", "minOutputLength", 20) or 20), + ) + + prepared: list[tuple[int, dict[str, Any], str, str]] = [] + failures: list[tuple[int, dict[str, str]]] = [] + for index, requested in enumerate(payload.items): + try: + current = store.get_result(task_id, requested.result_id) + if requested.expected_updated_at != str(current.get("updated_at") or ""): + raise ConflictError("data process result was modified by another request") + source_content = "" + preview_id = current.get("preview_item_id") + if preview_id: + preview = store.get_preview_item(task_id, str(preview_id)) + source_content = str( + preview.get("edited_content") + or preview.get("original_content") + or "" + ) + prepared.append( + (index, current, source_content, requested.expected_updated_at) + ) + except ConflictError as exc: + failures.append((index, { + "result_id": requested.result_id, + "code": "conflict", + "message": _safe_regeneration_error(exc), + })) + except (NotFoundError, InvalidStateError) as exc: + failures.append((index, { + "result_id": requested.result_id, + "code": "skipped", + "message": _safe_regeneration_error(exc), + })) + + logger.info( + "data process result batch evaluation started batch_id=%s task_id=%s " + "requested=%s prepared=%s judge_enabled=%s", + batch_id, + task_id, + len(payload.items), + len(prepared), + evaluation_model is not None, + ) + successes: list[tuple[int, dict[str, Any]]] = [] + if prepared: + try: + from app.modules.data_process.algorithms.embedding import ( + semantic_embedding_model, + ) + + semantic_embedding_model() + except Exception: + logger.warning( + "data process semantic embedding unavailable, semantic layer " + "will be skipped batch_id=%s", + batch_id, + ) + request_timeout = _result_regeneration_timeout(config) + model_timeout = httpx.Timeout( + request_timeout, + connect=min(10.0, request_timeout), + ) + model_limits = httpx.Limits( + max_connections=RESULT_REGENERATION_CONCURRENCY, + max_keepalive_connections=RESULT_REGENERATION_CONCURRENCY, + ) + with httpx.Client(timeout=model_timeout, limits=model_limits) as model_client, \ + ThreadPoolExecutor( + max_workers=min(RESULT_REGENERATION_CONCURRENCY, len(prepared)), + thread_name_prefix="data-result-evaluation", + ) as executor: + futures = { + executor.submit( + _evaluate_result_in_place, + task_id, + current, + source_content, + evaluation_config, + evaluation_model, + store, + expected_updated_at=expected_updated_at, + model_client=model_client if evaluation_model else None, + minimum=minimum, + ): (index, str(current["id"]), time.perf_counter()) + for index, current, source_content, expected_updated_at in prepared + } + for future in as_completed(futures): + index, result_id, item_started_at = futures[future] + try: + evaluated = future.result() + successes.append((index, evaluated)) + outcome = "succeeded" + except ConflictError as exc: + outcome = "conflict" + failures.append((index, { + "result_id": result_id, + "code": outcome, + "message": _safe_regeneration_error(exc), + })) + except Exception as exc: + outcome = "evaluation_failed" + failures.append((index, { + "result_id": result_id, + "code": outcome, + "message": _safe_regeneration_error(exc), + })) + logger.info( + "data process result batch evaluation item finished " + "batch_id=%s task_id=%s result_id=%s outcome=%s duration_ms=%.2f", + batch_id, + task_id, + result_id, + outcome, + (time.perf_counter() - item_started_at) * 1000, + ) + + success_items = [item for _, item in sorted(successes, key=lambda pair: pair[0])] + failure_items = [item for _, item in sorted(failures, key=lambda pair: pair[0])] + duration_ms = (time.perf_counter() - started_at) * 1000 + logger.info( + "data process result batch evaluation completed batch_id=%s task_id=%s " + "succeeded=%s failed=%s duration_ms=%.2f", + batch_id, + task_id, + len(success_items), + len(failure_items), + duration_ms, + ) + return ok( + { + "batch_id": batch_id, + "total": len(payload.items), + "succeeded": len(success_items), + "failed": len(failure_items), + "duration_ms": round(duration_ms, 2), + "items": success_items, + "failures": failure_items, + }, + "data process results evaluated", + ) + + @router.post("/{task_id}/results/{result_id}/regenerate") def regenerate_result( task_id: str, diff --git a/backend/app/modules/data_process/algorithms/quality.py b/backend/app/modules/data_process/algorithms/quality.py index 78b2fca..308c244 100644 --- a/backend/app/modules/data_process/algorithms/quality.py +++ b/backend/app/modules/data_process/algorithms/quality.py @@ -4,6 +4,7 @@ from __future__ import annotations import hashlib import json +import math import re import unicodedata from collections import Counter @@ -341,3 +342,87 @@ def score_quality( flags=tuple(flags), fingerprint=fingerprint, ) + + +def _cosine_similarity(left: Sequence[float], right: Sequence[float]) -> float: + if not left or not right or len(left) != len(right): + return 0.0 + dot = math.fsum(a * b for a, b in zip(left, right)) + norm_left = math.sqrt(math.fsum(a * a for a in left)) + norm_right = math.sqrt(math.fsum(b * b for b in right)) + if not norm_left or not norm_right: + return 0.0 + return dot / (norm_left * norm_right) + + +def semantic_quality_scores( + record: Mapping[str, Any], + *, + source_content: str = "", + embed_model: Any = None, +) -> dict[str, Any] | None: + """用本地嵌入向量计算语义相关性（0-100）。 + + 返回 ``question_answer``（问题↔答案）、``answer_source``（答案↔来源， + 无来源时缺省）与 ``overall``；嵌入模型不可用时返回 None 降级，不阻断流程。 + """ + + try: + if embed_model is None: + from .embedding import semantic_embedding_model + + embed_model = semantic_embedding_model() + if embed_model is None: + return None + + question = normalize_text( + " ".join( + str(record.get(field) or "") + for field in ("instruction", "input") + ) + ) + answer = normalize_text( + str(record.get("output") or "") or str(record.get("chosen") or "") + ) + source = normalize_text(source_content) + texts = [text for text in {question, answer, source} if text] + if not texts: + return None + vectors = {text: embed_model.get_text_embedding(text) for text in texts} + except Exception: + return None + + scores: dict[str, Any] = {} + if question and answer: + scores["question_answer"] = round( + 100 * max(0.0, _cosine_similarity(vectors[question], vectors[answer])), 2 + ) + if answer and source: + scores["answer_source"] = round( + 100 * max(0.0, _cosine_similarity(vectors[answer], vectors[source])), 2 + ) + if not scores: + return None + scores["overall"] = round(sum(scores.values()) / len(scores), 2) + return scores + + +def composite_overall( + *, + rule: float | None, + semantic: float | None = None, + judge: float | None = None, +) -> float: + """三层加权组合：规则 35% + 语义 20% + 评审 45%，缺失层自动重归一。""" + + if rule is None: + rule = 0.0 + if judge is not None and semantic is not None: + overall = rule * 0.35 + semantic * 0.20 + judge * 0.45 + elif semantic is not None: + overall = rule * 0.60 + semantic * 0.40 + elif judge is not None: + overall = rule * 0.55 + judge * 0.45 + else: + overall = rule + return round(max(0.0, min(100.0, overall)), 2) diff --git a/backend/app/modules/data_process/evaluation.py b/backend/app/modules/data_process/evaluation.py new file mode 100644 index 0000000..c9051f3 --- /dev/null +++ b/backend/app/modules/data_process/evaluation.py @@ -0,0 +1,313 @@ +"""数据处理 - 生成结果的多层质量评测。 + +三层体系：规则层（确定性规则分）+ 语义层（本地嵌入向量）+ 评审层 +（复用生成模型按 rubric 打分的 LLM-as-judge）。任一层失败自动降级， +评测永远返回可用结果，不阻断调用方流程。 +""" + +from __future__ import annotations + +import logging +from collections.abc import Mapping +from dataclasses import asdict +from datetime import UTC, datetime +from typing import Any + +import httpx + +from .algorithms import normalize_text, score_quality +from .algorithms.quality import composite_overall, semantic_quality_scores +from .generation import ( + ModelGenerationError, + _is_retryable_generation_error, + _json_payload, + _message_content, + chat_completions_url, +) + +logger = logging.getLogger(__name__) + +# 送入评审提示词的来源正文上限，避免超长切片挤占评分输出空间。 +_MAX_JUDGE_SOURCE_CHARS = 6000 + +_JUDGE_DIMENSIONS: dict[str, tuple[str, ...]] = { + "standard": ( + "faithfulness", + "correctness", + "clarity", + "completeness", + "alignment", + ), + "reasoning": ( + "faithfulness", + "correctness", + "clarity", + "completeness", + "alignment", + "reasoning_validity", + ), + "dpo": ( + "clarity", + "chosen_quality", + "rejected_quality", + "preference_reasonableness", + "faithfulness", + ), +} + +_DIMENSION_LABELS: dict[str, str] = { + "faithfulness": "忠实度", + "correctness": "正确性", + "clarity": "问题清晰度", + "completeness": "回答完整性", + "alignment": "指令对齐", + "reasoning_validity": "推理有效性", + "chosen_quality": "chosen 回答质量", + "rejected_quality": "rejected 回答质量", + "preference_reasonableness": "偏好区分合理性", +} + +_DIMENSION_RULES: dict[str, str] = { + "faithfulness": "忠实度：答案的全部陈述是否被参考资料支持，没有编造、没有引入资料之外的信息；未提供参考资料时按答案内部自洽性评估", + "correctness": "正确性：答案中的事实、概念与计算是否正确", + "clarity": "问题清晰度：问题是否清晰、自包含、无歧义，脱离上下文也能理解", + "completeness": "回答完整性：答案是否充分、直接地回应了问题的全部要点", + "alignment": "指令对齐：答案的形式与范围是否符合问题的要求（如格式、语言、范围限定）", + "reasoning_validity": "推理有效性：思维链步骤是否逻辑连贯、无跳步或循环论证，结论是否由推理过程自然得出", + "chosen_quality": "chosen 回答质量：更优回答的正确性、完整性与表述质量", + "rejected_quality": "rejected 回答质量：较差回答是否仍具备基本可读性，使对比训练有意义", + "preference_reasonableness": "偏好区分合理性：chosen 是否明显优于 rejected，且优劣差异与问题直接相关", +} + + +def _judge_system_prompt(output_type: str) -> str: + dimensions = _JUDGE_DIMENSIONS[output_type] + rules = "\n".join(f"- {_DIMENSION_RULES[name]}" for name in dimensions) + scores_schema = ", ".join(f'"{name}": 1-5' for name in dimensions) + return ( + "你是大模型训练数据质量评审员。严格依据用户消息中的【参考资料】评审这条训练数据，逐维度按 1-5 分打分：\n" + f"{rules}\n" + "评分锚点：5 分=完全符合维度描述；3 分=基本符合但有明显不足；1 分=严重不符合。\n" + "忠实度只依据参考资料与公认常识判断，无法得到支持的陈述必须扣分；不要因为答案冗长而加分。\n" + "只输出一个 JSON 对象，不要输出 JSON 之外的任何文字。\n" + '输出格式：{"scores": {' + scores_schema + '}, "reason": "一句话总评", "issues": ["具体问题，没有则为空数组"]}' + ) + + +def _judge_user_prompt(record: Mapping[str, Any], source_content: str) -> str: + source = normalize_text(source_content)[:_MAX_JUDGE_SOURCE_CHARS] or "（无参考资料）" + instruction = normalize_text(str(record.get("instruction") or "")) or "（空）" + input_text = normalize_text(str(record.get("input") or "")) + sections = [f"【参考资料】\n{source}", f"【问题】\n{instruction}"] + if input_text: + sections.append(f"【输入】\n{input_text}") + if record.get("chosen") or record.get("rejected"): + sections.append(f"【更优回答 chosen】\n{normalize_text(str(record.get('chosen') or '')) or '（空）'}") + sections.append(f"【较差回答 rejected】\n{normalize_text(str(record.get('rejected') or '')) or '（空）'}") + else: + output = normalize_text(str(record.get("output") or "")) + sections.append(f"【回答】\n{output or '（空）'}") + return "\n\n".join(sections) + + +def _validated_judge_payload(payload: Any, output_type: str) -> dict[str, Any]: + if not isinstance(payload, Mapping): + raise ModelGenerationError("评审响应不是 JSON 对象") + raw_scores = payload.get("scores") + if not isinstance(raw_scores, Mapping): + raise ModelGenerationError("评审响应缺少 scores 对象") + expected = _JUDGE_DIMENSIONS[output_type] + scores: dict[str, float] = {} + for name in expected: + value = raw_scores.get(name) + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ModelGenerationError(f"评审响应缺少维度 {name} 的有效分数") + scores[name] = round(max(1.0, min(5.0, float(value))), 1) + issues = payload.get("issues") + if not isinstance(issues, list): + issues = [] + issues = [str(item)[:200] for item in issues if str(item).strip()][:8] + reason = normalize_text(str(payload.get("reason") or ""))[:300] + return { + "scores": scores, + "overall": round(sum(scores.values()) / len(scores) * 20, 2), + "reason": reason, + "issues": issues, + } + + +def _judge_record( + record: Mapping[str, Any], + source_content: str, + *, + model: Mapping[str, Any], + config: Mapping[str, Any], + client: httpx.Client | None, +) -> dict[str, Any] | None: + output_type = str(config.get("output_type") or "standard").strip().lower() + if output_type not in _JUDGE_DIMENSIONS: + output_type = "standard" + endpoint = chat_completions_url(str(model.get("api_url") or "")) + model_name = str(model.get("online_model_name") or model.get("name") or "").strip() + if not model_name: + raise ModelGenerationError("generation model name is required") + temperature = 0.1 + max_tokens = max(256, min(2048, int(config.get("max_tokens", 1024) or 1024))) + timeout = max(1.0, min(120.0, float(config.get("request_timeout_seconds", 60) or 60))) + retries = max(0, min(5, int(config.get("generation_retries", 2) or 2))) + headers = {"Content-Type": "application/json"} + api_key = str(model.get("api_key") or "").strip() + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + + request_payload: dict[str, Any] = { + "model": model_name, + "messages": [ + {"role": "system", "content": _judge_system_prompt(output_type)}, + {"role": "user", "content": _judge_user_prompt(record, source_content)}, + ], + "temperature": temperature, + "max_tokens": max_tokens, + } + if bool(config.get("json_mode", False)): + request_payload["response_format"] = {"type": "json_object"} + + owns_client = client is None + http_client = client or httpx.Client(timeout=timeout) + try: + last_error: Exception | None = None + for _ in range(retries + 1): + try: + response = http_client.post(endpoint, headers=headers, json=request_payload) + response.raise_for_status() + body = response.json() + if not isinstance(body, Mapping): + raise ModelGenerationError("model response body must be a JSON object") + judged = _validated_judge_payload( + _json_payload(_message_content(body)), + output_type, + ) + judged["model"] = model_name + judged["output_type"] = output_type + return judged + except Exception as exc: + last_error = exc + if not _is_retryable_generation_error(exc): + break + raise ModelGenerationError(f"质量评审调用失败: {last_error}") + finally: + if owns_client: + http_client.close() + + +def evaluate_result_record( + record: Mapping[str, Any], + *, + source_content: str = "", + model: Mapping[str, Any] | None = None, + config: Mapping[str, Any] | None = None, + client: httpx.Client | None = None, + embed_model: Any = None, + min_output_length: int = 20, +) -> dict[str, Any]: + """对一条生成结果执行三层评测，返回可直接落库的 quality_score 字典。 + + 规则层字段保持原样平铺（向后兼容既有读取方）；新增 ``semantic``、 + ``judge``、``layers``、``evaluated`` 与组合 ``overall``。 + """ + + config_dict = dict(config or {}) + rule = score_quality( + record, + min_output_length=min_output_length, + source_content=source_content, + ) + quality: dict[str, Any] = asdict(rule) + + semantic = semantic_quality_scores( + record, + source_content=source_content, + embed_model=embed_model, + ) + judge: dict[str, Any] | None = None + if model is not None: + try: + judge = _judge_record( + record, + source_content, + model=model, + config=config_dict, + client=client, + ) + except Exception as exc: + logger.warning( + "data process judge evaluation degraded: %s", + exc, + ) + + layers = { + "rule": rule.overall, + "semantic": semantic.get("overall") if semantic else None, + "judge": judge.get("overall") if judge else None, + } + quality.update( + semantic=semantic, + judge=judge, + layers=layers, + evaluated=True, + evaluated_at=datetime.now(UTC).isoformat(), + overall=composite_overall( + rule=layers["rule"], + semantic=layers["semantic"], + judge=layers["judge"], + ), + ) + return quality + + +def reevaluate_edited_record( + record: Mapping[str, Any], + *, + source_content: str = "", + previous_quality: Mapping[str, Any] | None = None, + embed_model: Any = None, + min_output_length: int = 20, +) -> dict[str, Any]: + """手动编辑/恢复后重算规则与语义层，丢弃已过期的评审层。 + + 编辑会改变内容，旧的评审分不再可信；规则与语义层本地重算零成本。 + ``evaluated`` 标记沿用原值，保证已评测过的结果编辑后仍有可用分数。 + """ + + rule = score_quality( + record, + min_output_length=min_output_length, + source_content=source_content, + ) + quality: dict[str, Any] = asdict(rule) + semantic = semantic_quality_scores( + record, + source_content=source_content, + embed_model=embed_model, + ) + previous = dict(previous_quality or {}) + evaluated = bool(previous.get("evaluated")) + layers = { + "rule": rule.overall, + "semantic": semantic.get("overall") if semantic else None, + "judge": None, + } + quality.update( + semantic=semantic, + judge=None, + layers=layers, + evaluated=evaluated, + evaluated_at=( + datetime.now(UTC).isoformat() if evaluated else None + ), + overall=composite_overall( + rule=layers["rule"], + semantic=layers["semantic"], + ), + ) + return quality diff --git a/backend/app/schemas/data_process.py b/backend/app/schemas/data_process.py index aa3671f..4778ba0 100644 --- a/backend/app/schemas/data_process.py +++ b/backend/app/schemas/data_process.py @@ -384,6 +384,26 @@ class ResultBatchRegenerateRequest(BaseModel): return self +class ResultBatchEvaluateItem(BaseModel): + model_config = ConfigDict(extra="forbid") + + result_id: str = Field(min_length=1, max_length=100) + expected_updated_at: str = Field(min_length=1, max_length=100) + + +class ResultBatchEvaluateRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + + items: list[ResultBatchEvaluateItem] = Field(min_length=1, max_length=50) + + @model_validator(mode="after") + def validate_unique_results(self) -> ResultBatchEvaluateRequest: + result_ids = [item.result_id for item in self.items] + if len(result_ids) != len(set(result_ids)): + raise ValueError("result_id values must be unique") + return self + + class DatasetSplit(BaseModel): model_config = ConfigDict(extra="forbid") diff --git a/backend/tests/test_data_process_api.py b/backend/tests/test_data_process_api.py index 2170935..1e354ac 100644 --- a/backend/tests/test_data_process_api.py +++ b/backend/tests/test_data_process_api.py @@ -1691,6 +1691,223 @@ def test_batch_result_regeneration_rejects_locked_tasks_before_model_call( assert model_calls == 0 +def _prepare_evaluation_task( + client: TestClient, + store: Any, + tmp_path: Path, + *, + config: dict[str, Any] | None = None, +) -> str: + task_id = client.post( + "/modelTF/data-process", + json={ + "name": "数据评测", + "process_type": "structured", + "config": config or {"generation_model_id": "model-1", "output_type": "standard"}, + }, + ).json()["data"]["id"] + store.tasks[task_id].update( + status="completed", + progress=100, + workflow_step="results", + results_confirmed=False, + ) + store.models["model-1"] = { + "id": "model-1", + "online_model_name": "test-model", + "api_url": "https://model.example/v1", + "api_key": "secret", + } + store.previews[task_id] = [ + { + "id": "preview-1", + "status": "original", + "original_content": "申请编号用于唯一标识一笔报销申请。", + "edited_content": "申请编号用于唯一标识一笔报销申请。", + }, + { + "id": "preview-2", + "status": "original", + "original_content": "联系电话用于联系申请人。", + "edited_content": "联系电话用于联系申请人。", + }, + ] + store.results[task_id] = [ + { + "id": "result-1", + "preview_item_id": "preview-1", + "instruction": "申请编号有什么作用？", + "input": "", + "output": "申请编号用于唯一标识一笔报销申请。", + "original_instruction": "申请编号有什么作用？", + "original_input": "", + "original_output": "申请编号用于唯一标识一笔报销申请。", + "status": "valid", + "error": None, + "split": "train", + "quality_score": {}, + "updated_at": "2026-08-19T09:00:00Z", + }, + { + "id": "result-2", + "preview_item_id": "preview-2", + "instruction": "联系电话有什么作用？", + "input": "", + "output": "联系电话用于联系申请人。", + "original_instruction": "联系电话有什么作用？", + "original_input": "", + "original_output": "联系电话用于联系申请人。", + "status": "valid", + "error": None, + "split": "train", + "quality_score": {}, + "updated_at": "2026-08-19T09:00:01Z", + }, + ] + return task_id + + +def test_results_can_be_evaluated_in_batch_with_partial_success( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + client, store, _ = make_client(tmp_path) + task_id = _prepare_evaluation_task(client, store, tmp_path) + evaluation_calls: list[dict[str, Any]] = [] + + def fake_evaluate(record: dict[str, Any], **kwargs: Any) -> dict[str, Any]: + evaluation_calls.append({"record": deepcopy(record), "kwargs": {k: v for k, v in kwargs.items() if k != "client"}}) + return { + "overall": 88.0, + "completeness": 100.0, + "length": 100.0, + "readability": 100.0, + "relevance": 90.0, + "duplicate": 100.0, + "is_valid": True, + "flags": [], + "fingerprint": "fp", + "semantic": {"question_answer": 80.0, "answer_source": 90.0, "overall": 85.0}, + "judge": {"scores": {"faithfulness": 5}, "overall": 90.0}, + "layers": {"rule": 92.0, "semantic": 85.0, "judge": 90.0}, + "evaluated": True, + } + + monkeypatch.setattr(data_process_endpoint, "evaluate_result_record", fake_evaluate) + response = client.post( + f"/modelTF/data-process/{task_id}/results/evaluate-batch", + json={ + "items": [ + {"result_id": "result-1", "expected_updated_at": "2026-08-19T09:00:00Z"}, + # 乐观锁版本不匹配：该条应按冲突失败，另一条仍成功。 + {"result_id": "result-2", "expected_updated_at": "2026-08-18T00:00:00Z"}, + ], + }, + ) + + assert response.status_code == 200 + data = response.json()["data"] + assert data["total"] == 2 + assert data["succeeded"] == 1 + assert data["failed"] == 1 + assert [item["id"] for item in data["items"]] == ["result-1"] + assert data["failures"][0]["result_id"] == "result-2" + assert data["failures"][0]["code"] == "conflict" + + assert len(evaluation_calls) == 1 + assert evaluation_calls[0]["record"]["instruction"] == "申请编号有什么作用？" + assert evaluation_calls[0]["kwargs"]["model"]["online_model_name"] == "test-model" + assert evaluation_calls[0]["kwargs"]["source_content"] == "申请编号用于唯一标识一笔报销申请。" + + stored = store.results[task_id][0]["quality_score"] + assert stored["evaluated"] is True + assert stored["layers"]["judge"] == 90.0 + assert store.results[task_id][1]["quality_score"] == {} + + +def test_evaluation_without_generation_model_skips_judge_layer( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + client, store, _ = make_client(tmp_path) + task_id = _prepare_evaluation_task(client, store, tmp_path, config={"output_type": "standard"}) + seen_models: list[Any] = [] + + def fake_evaluate(record: dict[str, Any], **kwargs: Any) -> dict[str, Any]: + seen_models.append(kwargs.get("model")) + return { + "overall": 70.0, "is_valid": True, "flags": [], + "semantic": None, "judge": None, + "layers": {"rule": 70.0, "semantic": None, "judge": None}, + "evaluated": True, + } + + monkeypatch.setattr(data_process_endpoint, "evaluate_result_record", fake_evaluate) + response = client.post( + f"/modelTF/data-process/{task_id}/results/evaluate-batch", + json={"items": [{"result_id": "result-1", "expected_updated_at": "2026-08-19T09:00:00Z"}]}, + ) + + assert response.status_code == 200 + assert response.json()["data"]["succeeded"] == 1 + # 任务未配置生成模型时，评审层收到的 model 必须是 None。 + assert seen_models == [None] + + +def test_evaluation_rejects_running_task( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + client, store, _ = make_client(tmp_path) + task_id = _prepare_evaluation_task(client, store, tmp_path) + store.tasks[task_id]["status"] = "running" + evaluation_calls = 0 + + def fake_evaluate(*args: Any, **kwargs: Any) -> dict[str, Any]: + nonlocal evaluation_calls + evaluation_calls += 1 + return {"overall": 0, "is_valid": True, "flags": []} + + monkeypatch.setattr(data_process_endpoint, "evaluate_result_record", fake_evaluate) + response = client.post( + f"/modelTF/data-process/{task_id}/results/evaluate-batch", + json={"items": [{"result_id": "result-1", "expected_updated_at": "2026-08-19T09:00:00Z"}]}, + ) + + assert response.status_code == 409 + assert evaluation_calls == 0 + + +def test_result_update_preserves_evaluation_layers_and_drops_stale_judge( + tmp_path: Path, +) -> None: + client, store, _ = make_client(tmp_path) + task_id = _prepare_evaluation_task(client, store, tmp_path) + store.results[task_id][0]["quality_score"] = { + "overall": 90.0, + "is_valid": True, + "flags": [], + "semantic": {"overall": 85.0}, + "judge": {"overall": 92.0}, + "layers": {"rule": 90.0, "semantic": 85.0, "judge": 92.0}, + "evaluated": True, + } + + response = client.put( + f"/modelTF/data-process/{task_id}/results/result-1", + json={"output": "人工修正后的答案：申请编号唯一标识一笔报销申请。"}, + ) + + assert response.status_code == 200 + stored = store.results[task_id][0]["quality_score"] + # 手动编辑后：规则+语义重算，评审分丢弃，evaluated 标记保留。 + assert stored["evaluated"] is True + assert stored["judge"] is None + assert stored["layers"]["judge"] is None + assert stored["layers"]["rule"] is not None + assert stored["overall"] >= 0 + + def test_preview_build_replaces_only_selected_files_and_reports_file_counts( tmp_path: Path, ) -> None: diff --git a/backend/tests/test_data_process_evaluation.py b/backend/tests/test_data_process_evaluation.py new file mode 100644 index 0000000..eb974bf --- /dev/null +++ b/backend/tests/test_data_process_evaluation.py @@ -0,0 +1,284 @@ +"""数据评测模块（三层质量评分）的单元测试。""" + +from __future__ import annotations + +import json +from typing import Any + +import httpx +import pytest + +from app.modules.data_process.algorithms.quality import ( + composite_overall, + semantic_quality_scores, +) +from app.modules.data_process.evaluation import ( + _JUDGE_DIMENSIONS, + _judge_system_prompt, + _validated_judge_payload, + evaluate_result_record, + reevaluate_edited_record, +) +from app.modules.data_process.generation import ModelGenerationError + +RECORD = { + "instruction": "申请编号有什么作用？", + "input": "", + "output": "申请编号用于唯一标识一笔报销申请，便于跟踪审批状态。", +} +SOURCE = "报销系统中，申请编号用于唯一标识一笔报销申请，并支持跟踪审批状态。" + + +class _FakeEmbedModel: + """按关键词返回固定向量，模拟语义嵌入。""" + + def get_text_embedding(self, text: str) -> list[float]: + if "作用" in text or "编号" in text and "？" in text: + return [0.9, 0.1, 0.0] + if "申请编号" in text: + return [0.85, 0.2, 0.0] + return [0.0, 0.1, 0.9] + + +class _FailingEmbedModel: + def get_text_embedding(self, text: str) -> list[float]: + raise RuntimeError("embedding unavailable") + + +class _FakeResponse: + def __init__(self, payload: dict[str, Any]): + self._payload = payload + + def raise_for_status(self) -> None: + return None + + def json(self) -> dict[str, Any]: + return self._payload + + +class _FakeClient: + def __init__(self, content: str): + self._content = content + self.calls: list[dict[str, Any]] = [] + + def post(self, endpoint: str, headers: Any = None, json: Any = None) -> _FakeResponse: + self.calls.append({"endpoint": endpoint, "payload": json}) + return _FakeResponse({ + "choices": [{"message": {"content": self._content}, "finish_reason": "stop"}], + }) + + def close(self) -> None: + return None + + +class _RaisingClient: + def post(self, endpoint: str, headers: Any = None, json: Any = None) -> _FakeResponse: + raise httpx.ConnectError("model endpoint unreachable") + + def close(self) -> None: + return None + + +def _judge_content(scores: dict[str, float], **extra: Any) -> str: + return json.dumps({"scores": scores, "reason": "总体可靠", "issues": [], **extra}) + + +def test_judge_system_prompt_covers_rubric_dimensions() -> None: + standard = _judge_system_prompt("standard") + for name in _JUDGE_DIMENSIONS["standard"]: + assert name in standard + assert "1-5" in standard + + dpo = _judge_system_prompt("dpo") + assert "chosen_quality" in dpo + assert "preference_reasonableness" in dpo + + reasoning = _judge_system_prompt("reasoning") + assert "reasoning_validity" in reasoning + + +def test_validated_judge_payload_converts_scores_to_overall() -> None: + judged = _validated_judge_payload( + { + "scores": { + "faithfulness": 5, + "correctness": 4, + "clarity": 4, + "completeness": 3, + "alignment": 4, + }, + "reason": "答案可靠", + "issues": ["回答略冗长"], + }, + "standard", + ) + + assert judged["overall"] == round((5 + 4 + 4 + 3 + 4) / 5 * 20, 2) + assert judged["issues"] == ["回答略冗长"] + assert judged["reason"] == "答案可靠" + + +def test_validated_judge_payload_clamps_out_of_range_scores() -> None: + judged = _validated_judge_payload( + { + "scores": { + "faithfulness": 9, + "correctness": 4, + "clarity": 4, + "completeness": 0, + "alignment": 4, + }, + }, + "standard", + ) + + assert judged["scores"]["faithfulness"] == 5.0 + assert judged["scores"]["completeness"] == 1.0 + + +@pytest.mark.parametrize( + "scores", + [ + {"faithfulness": 5, "correctness": 4, "clarity": 4, "completeness": 3}, + { + "faithfulness": 5, + "correctness": 4, + "clarity": "high", + "completeness": 3, + "alignment": 4, + }, + ], +) +def test_validated_judge_payload_rejects_incomplete_scores(scores: dict[str, Any]) -> None: + with pytest.raises(ModelGenerationError): + _validated_judge_payload({"scores": scores}, "standard") + + +def test_semantic_quality_scores_uses_cosine_similarity() -> None: + scores = semantic_quality_scores( + RECORD, + source_content=SOURCE, + embed_model=_FakeEmbedModel(), + ) + + assert scores is not None + assert 0 < scores["question_answer"] <= 100 + assert 0 < scores["answer_source"] <= 100 + assert scores["overall"] == round((scores["question_answer"] + scores["answer_source"]) / 2, 2) + + +def test_semantic_quality_scores_degrades_to_none_on_failure() -> None: + assert ( + semantic_quality_scores( + RECORD, + source_content=SOURCE, + embed_model=_FailingEmbedModel(), + ) + is None + ) + + +def test_composite_overall_weights_available_layers() -> None: + assert composite_overall(rule=80, semantic=90, judge=70) == round(80 * 0.35 + 90 * 0.20 + 70 * 0.45, 2) + assert composite_overall(rule=80, semantic=90) == round(80 * 0.6 + 90 * 0.4, 2) + assert composite_overall(rule=80) == 80.0 + assert composite_overall(rule=None, judge=100) == 45.0 + + +def test_evaluate_result_record_combines_three_layers() -> None: + client = _FakeClient( + _judge_content({ + "faithfulness": 5, + "correctness": 4, + "clarity": 5, + "completeness": 4, + "alignment": 5, + }) + ) + quality = evaluate_result_record( + RECORD, + source_content=SOURCE, + model={"api_url": "https://model.example", "online_model_name": "judge-model"}, + config={"output_type": "standard", "generation_retries": 0}, + client=client, + embed_model=_FakeEmbedModel(), + ) + + assert quality["evaluated"] is True + assert quality["judge"] is not None + assert quality["judge"]["model"] == "judge-model" + assert quality["semantic"] is not None + assert quality["layers"]["judge"] == quality["judge"]["overall"] + assert quality["overall"] == composite_overall( + rule=quality["layers"]["rule"], + semantic=quality["layers"]["semantic"], + judge=quality["layers"]["judge"], + ) + # 评审提示词必须携带来源原文作为评分锚点（正文经 NFKC 归一化）。 + user_message = client.calls[0]["payload"]["messages"][1]["content"] + assert "申请编号用于唯一标识一笔报销" in user_message + + +def test_evaluate_result_record_degrades_when_model_fails() -> None: + quality = evaluate_result_record( + RECORD, + source_content=SOURCE, + model={"api_url": "https://model.example", "online_model_name": "judge-model"}, + config={"output_type": "standard", "generation_retries": 0}, + client=_RaisingClient(), + embed_model=_FakeEmbedModel(), + ) + + assert quality["judge"] is None + assert quality["layers"]["judge"] is None + assert quality["semantic"] is not None + assert quality["overall"] == composite_overall( + rule=quality["layers"]["rule"], + semantic=quality["layers"]["semantic"], + ) + + +def test_evaluate_result_record_without_model_runs_two_layers() -> None: + quality = evaluate_result_record( + RECORD, + source_content=SOURCE, + model=None, + embed_model=_FakeEmbedModel(), + ) + + assert quality["judge"] is None + assert quality["evaluated"] is True + assert quality["overall"] == composite_overall( + rule=quality["layers"]["rule"], + semantic=quality["layers"]["semantic"], + ) + + +def test_reevaluate_edited_record_drops_stale_judge() -> None: + previous = { + "evaluated": True, + "judge": {"overall": 90.0}, + } + quality = reevaluate_edited_record( + {**RECORD, "output": "编辑后的新答案内容，用于验证重评逻辑。"}, + source_content=SOURCE, + previous_quality=previous, + embed_model=_FakeEmbedModel(), + ) + + assert quality["evaluated"] is True + assert quality["judge"] is None + assert quality["layers"]["judge"] is None + assert quality["semantic"] is not None + + +def test_reevaluate_edited_record_keeps_unevaluated_state() -> None: + quality = reevaluate_edited_record( + RECORD, + source_content=SOURCE, + previous_quality={}, + embed_model=_FakeEmbedModel(), + ) + + assert quality["evaluated"] is False + assert quality["evaluated_at"] is None diff --git a/frontend/src/api/modules/dataProcess.ts b/frontend/src/api/modules/dataProcess.ts index a9c35e0..71faaff 100644 --- a/frontend/src/api/modules/dataProcess.ts +++ b/frontend/src/api/modules/dataProcess.ts @@ -20,6 +20,8 @@ import type { DataProcessPublishResult, DataProcessQualityScore, DataProcessResult, + DataProcessResultBatchEvaluatePayload, + DataProcessResultBatchEvaluateResult, DataProcessResultBatchRegeneratePayload, DataProcessResultBatchRegenerateResult, DataProcessResultRegeneratePayload, @@ -320,5 +322,14 @@ export const regenerateDataProcessResults = ( { timeout: 240_000 }, ) +export const evaluateDataProcessResults = ( + taskId: string | number, + payload: DataProcessResultBatchEvaluatePayload, +) => post( + `/data-process/${encodeURIComponent(taskId)}/results/evaluate-batch`, + payload, + { timeout: 240_000 }, +) + export const publishDataProcess = (taskId: string | number, payload: DataProcessPublishPayload) => post(`/data-process/${encodeURIComponent(taskId)}/publish`, payload) diff --git a/frontend/src/plugins/echarts.ts b/frontend/src/plugins/echarts.ts index ce1737c..3508f4b 100644 --- a/frontend/src/plugins/echarts.ts +++ b/frontend/src/plugins/echarts.ts @@ -3,18 +3,21 @@ */ import { use } from 'echarts/core' import { CanvasRenderer } from 'echarts/renderers' -import { BarChart, PieChart } from 'echarts/charts' +import { BarChart, PieChart, RadarChart } from 'echarts/charts' import { GridComponent, TooltipComponent, LegendComponent, + RadarComponent, } from 'echarts/components' use([ CanvasRenderer, BarChart, PieChart, + RadarChart, GridComponent, TooltipComponent, LegendComponent, + RadarComponent, ]) diff --git a/frontend/src/types/dataProcess.ts b/frontend/src/types/dataProcess.ts index 921387f..63e86b9 100644 --- a/frontend/src/types/dataProcess.ts +++ b/frontend/src/types/dataProcess.ts @@ -398,6 +398,52 @@ export interface DataProcessResultBatchRegenerateResult { failures: DataProcessResultBatchRegenerateFailure[] } +export interface DataProcessResultBatchEvaluateItem { + result_id: string + expected_updated_at: string +} + +export interface DataProcessResultBatchEvaluatePayload { + items: DataProcessResultBatchEvaluateItem[] +} + +export interface DataProcessResultBatchEvaluateFailure { + result_id: string + code: 'conflict' | 'skipped' | 'evaluation_failed' | 'internal_error' + message: string +} + +export interface DataProcessResultBatchEvaluateResult { + batch_id: string + total: number + succeeded: number + failed: number + duration_ms: number + items: DataProcessResult[] + failures: DataProcessResultBatchEvaluateFailure[] +} + +export interface DataProcessQualitySemantic { + question_answer?: number + answer_source?: number + overall?: number +} + +export interface DataProcessQualityJudge { + scores?: Record + overall?: number + reason?: string + issues?: string[] + model?: string + output_type?: string +} + +export interface DataProcessQualityLayers { + rule?: number | null + semantic?: number | null + judge?: number | null +} + export interface DataProcessQualityScore { overall?: number completeness?: number @@ -408,6 +454,11 @@ export interface DataProcessQualityScore { is_valid?: boolean flags?: string[] fingerprint?: string + semantic?: DataProcessQualitySemantic | null + judge?: DataProcessQualityJudge | null + layers?: DataProcessQualityLayers | null + evaluated?: boolean + evaluated_at?: string | null source_pages?: number[] heading_path?: string[] source_locator?: DataProcessSourceLocator diff --git a/frontend/src/views/data-process/DataProcessCreateView.vue b/frontend/src/views/data-process/DataProcessCreateView.vue index 21951fc..a449f69 100644 --- a/frontend/src/views/data-process/DataProcessCreateView.vue +++ b/frontend/src/views/data-process/DataProcessCreateView.vue @@ -18,6 +18,7 @@ import { previewAffectingOptionsFor, } from './create/dataProcessCreateState' import { useDataProcessGeneration } from './create/useDataProcessGeneration' +import { useDataProcessEvaluation } from './create/useDataProcessEvaluation' import { useDataProcessPreviewBuild } from './create/useDataProcessPreviewBuild' import { useDataProcessRegeneration } from './create/useDataProcessRegeneration' import { createDefaultExternalSource, externalSourcePayload, restoreExternalSourceConfig, sourceConfigForBackend } from './create/externalSourceConfig' @@ -126,6 +127,19 @@ const { outputType: activeOutputType, beforeGenerate: beforeStartGeneration, }) +const { + evaluation, + evaluateAllResults, + resetEvaluation, +} = useDataProcessEvaluation({ + taskId, + results, + selectedResultId, +}) +// 生成结果被重置（重新切分/上传/重新生成配置）时同步清空评测进度。 +watch(results, (items) => { + if (!items.length) resetEvaluation() +}) const { enqueueSourceUpload, sourceUploading } = useDataProcessSourceUpload({ taskId, uploadedFiles, @@ -1156,10 +1170,12 @@ onMounted(() => { :preview-items="previewItems" :regenerating-result-id="regeneratingResultId" :bulk-regeneration="bulkRegeneration" + :evaluation="evaluation" :output-type="activeOutputType" @update:field="updateResultField" @regenerate:all="regenerateAllResults" @regenerate:item="regenerateResult" + @evaluate:all="evaluateAllResults" /> diff --git a/frontend/src/views/data-process/DataProcessDetailView.vue b/frontend/src/views/data-process/DataProcessDetailView.vue index ef0919b..d2bbef3 100644 --- a/frontend/src/views/data-process/DataProcessDetailView.vue +++ b/frontend/src/views/data-process/DataProcessDetailView.vue @@ -5,6 +5,7 @@ import { ElMessage, ElMessageBox } from 'element-plus' import PageCard from '@/components/PageCard.vue' import { usePolling } from '@/composables/usePolling' import { + evaluateDataProcessResults, getDataProcessProgress, getDataProcessResults, getDataProcessTask, @@ -13,6 +14,7 @@ import { restoreDataProcessResult, updateDataProcessResult, } from '@/api/modules/dataProcess' +import QualityRadarPopover from './create/QualityRadarPopover.vue' import type { DataProcessDatasetSplit, DataProcessPublishPayload, @@ -456,13 +458,101 @@ function resultStatusType(status: DataProcessResultStatus) { } function qualityScoreLabel(value: DataProcessResult['quality_score']) { - if (value == null) return '-' + if (value == null || !value.evaluated) return '-' const score = value.overall return Number.isFinite(score) ? Number(score).toFixed(1) : '-' } -function qualityFlagsLabel(value: DataProcessResult['quality_score']) { - return value?.flags?.length ? value.flags.join('、') : '未命中质量规则' +function qualityScoreTone(value: DataProcessResult['quality_score']) { + const score = Number(value?.overall) + if (!value?.evaluated || !Number.isFinite(score)) return '' + return score >= 80 ? 'is-success' : score >= 60 ? 'is-warning' : 'is-danger' +} + +function qualityScoreEvaluated(value: DataProcessResult['quality_score']) { + return Boolean(value?.evaluated && Number.isFinite(Number(value?.overall))) +} + +const evaluationRunning = ref(false) +const evaluationProgress = reactive({ + visible: false, + total: 0, + completed: 0, + succeeded: 0, + failed: 0, +}) +// 与批量重生成一致的分块大小，单批在接口 240 秒超时预算内。 +const EVALUATION_CHUNK_SIZE = 12 +const canEvaluate = computed(() => ( + detail.value?.status === 'completed' && !hasCurrentPublishedDataset.value +)) +const evaluationPercentage = computed(() => ( + evaluationProgress.total + ? Math.round((evaluationProgress.completed / evaluationProgress.total) * 100) + : 0 +)) + +async function loadAllResultIds() { + const first = await getDataProcessResults(taskId.value, { page: 1, page_size: 500 }) + const items = [...first.items] + const pages = Math.ceil(first.total / first.page_size) + for (let page = 2; page <= pages; page += 1) { + const next = await getDataProcessResults(taskId.value, { page, page_size: 500 }) + items.push(...next.items) + } + return items +} + +async function runResultEvaluation() { + if (evaluationRunning.value || !canEvaluate.value) return + evaluationRunning.value = true + Object.assign(evaluationProgress, { + visible: true, + total: 0, + completed: 0, + succeeded: 0, + failed: 0, + }) + try { + const candidates = (await loadAllResultIds()).filter((item) => item.updated_at) + if (!candidates.length) { + ElMessage.info('当前没有可评测的结果') + return + } + evaluationProgress.total = candidates.length + for (let offset = 0; offset < candidates.length; offset += EVALUATION_CHUNK_SIZE) { + const chunk = candidates.slice(offset, offset + EVALUATION_CHUNK_SIZE) + try { + const evaluated = await evaluateDataProcessResults(taskId.value, { + items: chunk.map((item) => ({ + result_id: String(item.id), + expected_updated_at: item.updated_at as string, + })), + }) + evaluationProgress.completed += evaluated.total + evaluationProgress.succeeded += evaluated.succeeded + evaluationProgress.failed += evaluated.failed + } catch { + evaluationProgress.completed = evaluationProgress.total + evaluationProgress.failed += candidates.length - offset + break + } + } + await loadResults() + if (evaluationProgress.failed === 0) { + ElMessage.success(`数据评测完成：成功 ${evaluationProgress.succeeded} 条`) + } else if (evaluationProgress.succeeded > 0) { + ElMessage.warning( + `数据评测完成：成功 ${evaluationProgress.succeeded} 条，失败 ${evaluationProgress.failed} 条`, + ) + } else { + ElMessage.error(`数据评测失败：${evaluationProgress.failed} 条结果未完成评测`) + } + } catch { + ElMessage.error('数据评测中断，已完成的评分保持不变') + } finally { + evaluationRunning.value = false + } } function replaceResult(updated: DataProcessResult) { @@ -824,10 +914,33 @@ onBeforeUnmount(() => { + +