2 Commits

Author SHA1 Message Date
caoxiaozhu
81c2f85c3a feat(data_process): 问答对数据评测体系与质量分雷达图
- 三层评测:规则层沿用原五维规则分,语义层用本地 BGE 向量算问答/来源
  相关性,评审层复用生成模型按 rubric 打分(忠实度/正确性/清晰度等,
  区分 standard/reasoning/dpo 输出类型),任一层失败自动降级
- 组合分 = 规则 35% + 语义 20% + 评审 45%,缺层自动重归一
- 新增 results/evaluate-batch 批量评测接口,镜像批量重生成的并发、
  乐观锁与部分成功语义;生成阶段不再展示质量分
- 详情页与结果编辑页新增"数据评测"按钮和批量进度;质量分列悬停弹出
  雷达图浮窗(评审 5 维 + 语义 2 维、三层分项、评审理由)
- 手动编辑/恢复后重算规则与语义层并丢弃过期评审分,雷达图不再展示
  失效数据
2026-08-19 14:22:27 +08:00
caoxiaozhu
f47611020a fix(data_process): 修复 Word 切分行号断档与预览标题缺失
- 正文抽取下钻 SDT 内容控件,目录等内容不再整段丢失
- 切片投影匹配剥离序列化插入的列表自动编号,新增行锚点兜底与游标防回退
- fixed/semantic 切分路径定位失败时保留切片,不再静默丢弃
- Word 预览按段落大纲级别识别标题,未套标题样式的小节正常渲染
- 本地嵌入模型抽为共享单例,供语义分块与质量评分共用
2026-08-19 14:22:27 +08:00
22 changed files with 2232 additions and 67 deletions

View File

@@ -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,

View File

@@ -0,0 +1,24 @@
"""数据处理算法 - 本地语义嵌入模型共享单例。"""
from __future__ import annotations
import os
from functools import lru_cache
from typing import Any
@lru_cache(maxsize=1)
def semantic_embedding_model() -> Any:
"""加载本地嵌入模型,供语义分块与语义质量评分共用。
模型可在部署环境覆盖;默认模型体积较小且适合中英文语义判断。
返回 LlamaIndex BaseEmbedding通过 ``get_text_embedding`` 使用。
"""
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
return HuggingFaceEmbedding(
model_name=os.getenv("DATA_PROCESS_EMBEDDING_MODEL", "BAAI/bge-small-zh-v1.5"),
device=os.getenv("DATA_PROCESS_EMBEDDING_DEVICE", "cpu"),
trust_remote_code=False,
)

View File

@@ -7,12 +7,13 @@ import re
import unicodedata
import zipfile
import xml.etree.ElementTree as ET
from collections.abc import Mapping, Sequence
from collections.abc import Iterator, Mapping, Sequence
from pathlib import PurePosixPath
from typing import Any
from urllib.parse import unquote, urlsplit
from docx import Document
from docx.oxml.ns import qn
from docx.oxml.table import CT_Tbl
from docx.oxml.text.paragraph import CT_P
from docx.table import Table
@@ -121,6 +122,21 @@ def _validate_office_archive(raw: bytes, file_format: TextFormat) -> None:
except zipfile.BadZipFile as exc:
raise ValueError(f"invalid {file_format.upper()} file: not an Office ZIP package") from exc
def iter_document_blocks(parent: Any) -> Iterator[Any]:
"""按文档顺序产出正文段落与表格,并下钻 SDT 内容控件。
Word 的目录、复选框等内容控件包在 ``w:sdt`` 元素里,只遍历 body
直接子级会把这些段落整段丢掉。
"""
for child in parent.iterchildren():
if isinstance(child, (CT_P, CT_Tbl)):
yield child
elif child.tag == qn("w:sdt"):
content = child.find(qn("w:sdtContent"))
if content is not None:
yield from iter_document_blocks(content)
def _extract_docx_text(raw: bytes) -> str:
_validate_office_archive(raw, "docx")
try:
@@ -130,7 +146,7 @@ def _extract_docx_text(raw: bytes) -> str:
parts: list[str] = []
total = 0
for child in document.element.body.iterchildren():
for child in iter_document_blocks(document.element.body):
if isinstance(child, CT_P):
total = _append_bounded_text(parts, Paragraph(child, document).text, total)
continue

View File

@@ -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)

View File

@@ -18,12 +18,19 @@ from llama_index.core.base.embeddings.base import BaseEmbedding
from llama_index.core.node_parser import SemanticSplitterNodeParser, SentenceSplitter
from app.modules.data_process.algorithms import normalize_text
from app.modules.data_process.algorithms.embedding import semantic_embedding_model
ChunkMethod = Literal["layout_hybrid", "semantic", "fixed"]
_PAGE_FURNITURE = re.compile(
r"(?m)^\s*(?:第\s*\d+\s*页\s*共\s*\d+\s*页|[-—–]?\s*\d+\s*[/]\s*\d+\s*[-—–]?)\s*$"
)
# Docling 的 markdown 序列化会给列表项补上自动编号,而 Word 的编号存放在
# numbering.xml 中python-docx 抽取的正文不含这些编号;紧凑匹配前剥掉
# 行首编号,否则带列表的切片会整体定位失败。
_LIST_MARKER_PREFIX = re.compile(
r"(?m)^[ \t>]*(?:(?:\d{1,3}[.)])+|\([a-zA-Z0-9]{1,3}\)|[a-zA-Z][.)]|[-*+•·])[ \t]+"
)
_COMPACT_CHARACTER = re.compile(r"[\w\u3400-\u4dbf\u4e00-\u9fff]", re.UNICODE)
_CONVERTER_LOCK = threading.Lock()
@@ -149,18 +156,6 @@ def chunk_fixed_text(
return _text_chunks(text, chunk_size=chunk_size, chunk_overlap=chunk_overlap)
@lru_cache(maxsize=1)
def _semantic_embedding_model() -> BaseEmbedding:
# 模型可在部署环境覆盖;默认模型体积较小且适合中英文语义边界判断。
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
return HuggingFaceEmbedding(
model_name=os.getenv("DATA_PROCESS_EMBEDDING_MODEL", "BAAI/bge-small-zh-v1.5"),
device=os.getenv("DATA_PROCESS_EMBEDDING_DEVICE", "cpu"),
trust_remote_code=False,
)
def chunk_semantic_text(
text: str,
*,
@@ -175,7 +170,7 @@ def chunk_semantic_text(
if not normalized:
return []
splitter = SemanticSplitterNodeParser.from_defaults(
embed_model=embed_model or _semantic_embedding_model(),
embed_model=embed_model or semantic_embedding_model(),
breakpoint_percentile_threshold=breakpoint_percentile_threshold,
buffer_size=1,
sentence_splitter=_sentence_chunks,
@@ -193,6 +188,7 @@ def chunk_semantic_text(
if start is None:
start = _locate_text(normalized, content, 0)
if start is None:
result.append(_unlocated_chunk(content))
continue
if len(_tokenizer().encode(content)) <= chunk_size:
result.append(_make_text_chunk(normalized, start, start + len(content)))
@@ -203,6 +199,7 @@ def chunk_semantic_text(
chunk_overlap=chunk_overlap,
):
if child.source_start is None or child.source_end is None:
result.append(_unlocated_chunk(child.original_content))
continue
result.append(
_make_text_chunk(
@@ -235,6 +232,7 @@ def _nodes_to_chunks(nodes: list[Any], source_text: str) -> list[DocumentChunk]:
if start is None:
start = _locate_text(source_text, content, 0)
if start is None:
chunks.append(_unlocated_chunk(content))
continue
end = start + len(content)
chunks.append(_make_text_chunk(source_text, start, end))
@@ -247,6 +245,20 @@ def _locate_text(source: str, content: str, start: int) -> int | None:
return position if position >= 0 else None
def _unlocated_chunk(content: str) -> DocumentChunk:
"""正文在源文本中定位失败时保底保留切片,只放弃行号信息。"""
return DocumentChunk(
original_content=content,
contextualized_content=content,
source_start=None,
source_end=None,
source_start_line=None,
source_end_line=None,
token_count=len(_tokenizer().encode(content)),
)
def _make_text_chunk(source: str, start: int, end: int) -> DocumentChunk:
content = source[start:end]
return DocumentChunk(
@@ -316,6 +328,14 @@ def _compact_with_offsets(value: str) -> tuple[str, list[int]]:
return "".join(compact), offsets
def _expand_to_line_boundaries(source_text: str, start: int, end: int) -> tuple[int, int]:
while start > 0 and source_text[start - 1] not in "\r\n":
start -= 1
while end < len(source_text) and source_text[end] not in "\r\n":
end += 1
return start, end
def _project_layout_span(
source_text: str,
content: str,
@@ -324,21 +344,79 @@ def _project_layout_span(
source_offsets: list[int],
compact_start: int,
) -> tuple[int | None, int | None, int]:
compact_content, _ = _compact_with_offsets(content)
if len(compact_content) < 4:
for candidate in (content, _LIST_MARKER_PREFIX.sub("", content)):
compact_content, _ = _compact_with_offsets(candidate)
if len(compact_content) < 4:
continue
position = compact_source.find(compact_content, compact_start)
if position < 0:
position = compact_source.find(compact_content)
if position < 0:
continue
start, end = _expand_to_line_boundaries(
source_text,
source_offsets[position],
source_offsets[position + len(compact_content) - 1] + 1,
)
# 重复内容回退匹配可能命中已消费的更早位置,游标只进不退,
# 避免后续切片跟着错位。
return start, end, max(compact_start, position + len(compact_content))
return _project_layout_span_by_anchors(
source_text,
content,
compact_source=compact_source,
source_offsets=source_offsets,
compact_start=compact_start,
)
def _project_layout_span_by_anchors(
source_text: str,
content: str,
*,
compact_source: str,
source_offsets: list[int],
compact_start: int,
) -> tuple[int | None, int | None, int]:
"""按行锚点顺序匹配,容忍切片里插入的重复表头等非连续内容。"""
segments = [
compact
for compact in (
_compact_with_offsets(line)[0]
for line in _LIST_MARKER_PREFIX.sub("", content).split("\n")
)
if len(compact) >= 6
]
if not segments:
return None, None, compact_start
position = compact_source.find(compact_content, compact_start)
if position < 0:
position = compact_source.find(compact_content)
if position < 0:
total = sum(len(segment) for segment in segments)
def match_from(cursor: int) -> tuple[list[tuple[int, int]], int]:
matched: list[tuple[int, int]] = []
position = cursor
for segment in segments:
found = compact_source.find(segment, position)
if found < 0:
continue
matched.append((found, found + len(segment)))
position = found + len(segment)
return matched, sum(end - start for start, end in matched)
matched, covered = match_from(compact_start)
if covered * 2 < total:
retried, retry_covered = match_from(0)
if retry_covered > covered:
matched, covered = retried, retry_covered
# 覆盖不足一半时宁可不定位,也不能给出错误的行号。
if not matched or covered * 2 < total:
return None, None, compact_start
start = source_offsets[position]
end = source_offsets[position + len(compact_content) - 1] + 1
while start > 0 and source_text[start - 1] not in "\r\n":
start -= 1
while end < len(source_text) and source_text[end] not in "\r\n":
end += 1
return start, end, position + len(compact_content)
start, end = _expand_to_line_boundaries(
source_text,
source_offsets[matched[0][0]],
source_offsets[matched[-1][1] - 1] + 1,
)
return start, end, max(compact_start, matched[-1][1])
def chunk_layout_document(

View File

@@ -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

View File

@@ -11,6 +11,7 @@ import re
from typing import Any
from docx import Document
from docx.oxml.ns import qn
from docx.oxml.table import CT_Tbl
from docx.oxml.text.paragraph import CT_P
from docx.table import Table
@@ -27,6 +28,7 @@ from app.modules.data_process.algorithms import (
_xlsx_sheet_merge_ranges,
normalize_text,
)
from app.modules.data_process.algorithms.parsers.office import iter_document_blocks
MAX_DOCX_PREVIEW_BLOCKS = 2_000
MAX_XLSX_PREVIEW_ROWS = 200
@@ -49,12 +51,21 @@ def _docx_alignment(paragraph: Paragraph) -> str:
def _docx_heading_level(paragraph: Paragraph) -> int | None:
style = paragraph.style
if style is None:
return None
style_name = str(style.name or "")
style_id = str(style.style_id or "")
style_name = str(style.name or "") if style is not None else ""
style_id = str(style.style_id or "") if style is not None else ""
match = re.search(r"(?:heading|标题)\s*([1-6])", f"{style_name} {style_id}", re.IGNORECASE)
return int(match.group(1)) if match else None
if match:
return int(match.group(1))
# Word 的目录和导航窗格依据大纲级别识别标题;未套标题样式但带
# outlineLvl 的段落(如手工排版的编号小节)同样是标题。
outline = paragraph._p.find(f"{qn('w:pPr')}/{qn('w:outlineLvl')}")
if outline is not None:
value = outline.get(qn("w:val"))
if value is not None and value.isdigit():
level = int(value)
if 0 <= level <= 5:
return level + 1
return None
def build_docx_preview(raw: bytes) -> dict[str, Any]:
@@ -84,7 +95,7 @@ def build_docx_preview(raw: bytes) -> dict[str, Any]:
has_source_content = True
return text, start, source_cursor
for child in document.element.body.iterchildren():
for child in iter_document_blocks(document.element.body):
if rendered_blocks >= MAX_DOCX_PREVIEW_BLOCKS:
truncated = True
break

View File

@@ -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")

View File

@@ -9,6 +9,8 @@ from decimal import Decimal
import pytest
from docx import Document
from docx.oxml import parse_xml
from docx.oxml.ns import nsdecls, qn
from openpyxl import Workbook
from pptx import Presentation
from pptx.util import Inches
@@ -38,6 +40,7 @@ from app.modules.data_process.algorithms import (
stable_split_assignments,
structured_json_dumps,
)
from app.modules.data_process.office_preview import build_docx_preview
def _pdf_page_texts(*texts: str) -> tuple[PdfPageText, ...]:
@@ -318,6 +321,74 @@ def test_parse_pdf_docx_xlsx_and_pptx() -> None:
assert parsed_pptx.records == ()
def _docx_with_sdt_bytes() -> bytes:
"""构造带 SDT 目录内容控件的 docx段落顺序为正文、SDT、正文。"""
document = Document()
document.add_paragraph("正文开头。")
sdt = parse_xml(
"<w:sdt %s><w:sdtPr><w:id w:val='1'/></w:sdtPr>"
"<w:sdtContent><w:p><w:r><w:t>目录条目 第一章 概述</w:t></w:r></w:p>"
"</w:sdtContent></w:sdt>" % nsdecls("w")
)
body = document.element.body
sect_pr = body.find(qn("w:sectPr"))
if sect_pr is not None:
sect_pr.addprevious(sdt)
else:
body.append(sdt)
document.add_paragraph("正文结尾。")
output = io.BytesIO()
document.save(output)
return output.getvalue()
def test_docx_extraction_and_preview_include_sdt_content() -> None:
raw = _docx_with_sdt_bytes()
parsed = parse_text_content(raw, filename="toc.docx")
assert "目录条目 第一章 概述" in parsed.text
assert (
parsed.text.index("正文开头。")
< parsed.text.index("目录条目 第一章 概述")
< parsed.text.index("正文结尾。")
)
preview = build_docx_preview(raw)
paragraph_texts = [
block["text"] for block in preview["blocks"] if block["type"] == "paragraph"
]
assert "目录条目 第一章 概述" in paragraph_texts
# 预览偏移必须与正文抽取规则一致,否则前端定位会错位。
sdt_block = next(
block
for block in preview["blocks"]
if block.get("text") == "目录条目 第一章 概述"
)
assert parsed.text[sdt_block["source_start"] : sdt_block["source_end"]] == (
"目录条目 第一章 概述"
)
def test_docx_preview_detects_outline_level_headings() -> None:
"""未套标题样式但设了大纲级别的段落Word 目录按此收录)也按标题渲染。"""
document = Document()
document.add_heading("一级标题", level=1)
plain = document.add_paragraph("4.2.1 数据管理")
p_pr = plain._p.get_or_add_pPr()
p_pr.append(parse_xml("<w:outlineLvl %s w:val='2'/>" % nsdecls("w")))
document.add_paragraph("普通正文段落。")
output = io.BytesIO()
document.save(output)
preview = build_docx_preview(output.getvalue())
blocks = {b["text"]: b for b in preview["blocks"] if b["type"] == "paragraph"}
assert blocks["一级标题"]["heading_level"] == 1
assert blocks["4.2.1 数据管理"]["heading_level"] == 3
assert blocks["普通正文段落。"]["heading_level"] is None
def test_xlsx_record_locators_distinguish_sheets_rows_and_duplicate_records() -> None:
workbook = Workbook()
first = workbook.active

View File

@@ -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:

View File

@@ -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

View File

@@ -9,6 +9,7 @@ from app.modules.data_process.document_chunking import (
DocumentChunk,
_compact_with_offsets,
_document_converter,
_nodes_to_chunks,
_project_layout_span,
chunk_fixed_text,
chunk_semantic_text,
@@ -103,6 +104,86 @@ def test_layout_projection_ignores_layout_whitespace_but_keeps_source_lines() ->
assert cursor > 0
def test_layout_projection_tolerates_list_numbers_inserted_by_serializer() -> None:
# Word 自动编号存放在 numbering.xmlpython-docx 抽取的正文没有编号,
# 而 Docling 序列化切片时会补上 "1. " 前缀,投影不能因此失败。
source = "接入方式说明\n结构化数据接入需要先配置连接地址。\n非结构化接入需要上传文档。"
compact_source, offsets = _compact_with_offsets(source)
start, end, cursor = _project_layout_span(
source,
"1. 结构化数据接入需要先配置连接地址。\n2. 非结构化接入需要上传文档。",
compact_source=compact_source,
source_offsets=offsets,
compact_start=0,
)
assert start is not None and end is not None
assert source[start:end] == "结构化数据接入需要先配置连接地址。\n非结构化接入需要上传文档。"
assert cursor > 0
def test_layout_projection_never_moves_cursor_backwards() -> None:
source = "重复段落内容。\n中间正文。\n重复段落内容。"
compact_source, offsets = _compact_with_offsets(source)
# 重复内容回退匹配命中已消费的更早位置时,游标必须保持不退。
_, _, cursor = _project_layout_span(
source,
"重复段落内容。",
compact_source=compact_source,
source_offsets=offsets,
compact_start=compact_source.index("中间正文"),
)
assert cursor >= compact_source.index("中间正文")
def test_layout_projection_falls_back_to_line_anchors_for_inserted_content() -> None:
# 表格跨切片时 Docling 会在续片中重复表头,正文不再是连续子串;
# 按行锚点匹配仍应定位到表头所在行到末行数据之间的连续区间。
source = "表头甲\t表头乙\n第一行数据\t说明一\n第二行数据\t说明二"
compact_source, offsets = _compact_with_offsets(source)
start, end, _ = _project_layout_span(
source,
"表头甲 表头乙\n第二行数据 说明二",
compact_source=compact_source,
source_offsets=offsets,
compact_start=0,
)
assert start is not None and end is not None
assert source[start:end] == (
"表头甲\t表头乙\n第一行数据\t说明一\n第二行数据\t说明二"
)
def test_layout_projection_refuses_low_coverage_anchor_match() -> None:
source = "完全无关的正文内容甲。\n完全无关的正文内容乙。"
compact_source, offsets = _compact_with_offsets(source)
start, end, cursor = _project_layout_span(
source,
"找不到的数据行内容\n另一条找不到的数据行内容",
compact_source=compact_source,
source_offsets=offsets,
compact_start=0,
)
assert start is None
assert end is None
assert cursor == 0
def test_text_splitter_keeps_chunks_that_cannot_be_located() -> None:
class FakeNode:
def get_content(self) -> str:
return "这段文本在源文本中不存在。"
chunks = _nodes_to_chunks([FakeNode()], "完全不同的源文本。")
assert len(chunks) == 1
assert chunks[0].original_content == "这段文本在源文本中不存在。"
assert chunks[0].source_start is None
assert chunks[0].source_start_line is None
def test_short_layout_chunk_merges_with_neighbor_and_keeps_page_provenance() -> None:
source = "短标题\n这是一段足够长的正文内容,用于测试相邻切片合并。"
chunks = [

View File

@@ -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<DataProcessResultBatchEvaluateResult>(
`/data-process/${encodeURIComponent(taskId)}/results/evaluate-batch`,
payload,
{ timeout: 240_000 },
)
export const publishDataProcess = (taskId: string | number, payload: DataProcessPublishPayload) =>
post<DataProcessPublishResult>(`/data-process/${encodeURIComponent(taskId)}/publish`, payload)

View File

@@ -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,
])

View File

@@ -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<string, number>
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

View File

@@ -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"
/>
</div>
</div>

View File

@@ -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(() => {
<el-option label="已修改" value="modified" />
<el-option label="无效" value="invalid" />
</el-select>
<el-button
v-if="canEvaluate"
type="primary"
plain
:loading="evaluationRunning"
:disabled="resultLoading"
@click="runResultEvaluation"
>
<i v-if="!evaluationRunning" class="fa fa-check-square-o" aria-hidden="true" /> 数据评测
</el-button>
<el-button :loading="resultLoading" @click="loadResults"><i class="fa fa-refresh" /></el-button>
</div>
</div>
<div v-if="evaluationProgress.visible" class="evaluation-progress">
<span>
数据评测 {{ evaluationProgress.completed }} / {{ evaluationProgress.total }}
· 成功 {{ evaluationProgress.succeeded }} · 失败 {{ evaluationProgress.failed }}
</span>
<el-progress
:percentage="evaluationPercentage"
:show-text="false"
:stroke-width="5"
:color="evaluationProgress.failed > 0 ? '#d97706' : '#5b50f2'"
/>
</div>
<el-table
v-if="results.length"
:data="results"
@@ -845,9 +958,26 @@ onBeforeUnmount(() => {
</template>
<el-table-column label="质量分" width="88" align="center">
<template #default="{ row }">
<el-tooltip :content="qualityFlagsLabel((row as DataProcessResult).quality_score)">
<span>{{ qualityScoreLabel((row as DataProcessResult).quality_score) }}</span>
</el-tooltip>
<el-popover
v-if="qualityScoreEvaluated((row as DataProcessResult).quality_score)"
placement="top"
:width="296"
trigger="hover"
:show-after="150"
popper-class="quality-radar-popper"
>
<template #reference>
<span
class="detail-quality-score"
:class="qualityScoreTone((row as DataProcessResult).quality_score)"
>{{ qualityScoreLabel((row as DataProcessResult).quality_score) }}</span>
</template>
<QualityRadarPopover
:quality="(row as DataProcessResult).quality_score!"
:score="Number((row as DataProcessResult).quality_score?.overall)"
/>
</el-popover>
<span v-else class="detail-quality-empty">{{ qualityScoreLabel((row as DataProcessResult).quality_score) }}</span>
</template>
</el-table-column>
<el-table-column label="状态" width="90" align="center">
@@ -1099,6 +1229,35 @@ onBeforeUnmount(() => {
.result-filters :deep(.el-select) { width: 120px; }
.result-section :deep(.el-table) { border-radius: 0; }
.evaluation-progress {
display: grid;
grid-template-columns: 1fr 220px;
align-items: center;
gap: 14px;
padding: 10px 18px;
color: #667085;
background: #f8f9fc;
font-size: 12px;
}
.detail-quality-score {
display: inline-block;
min-width: 44px;
padding: 2px 8px;
border-radius: 10px;
color: #475467;
background: #f2f4f7;
font-weight: 700;
font-variant-numeric: tabular-nums;
cursor: default;
&.is-success { color: #067647; background: #e6f4ee; }
&.is-warning { color: #b54708; background: #fef0c7; }
&.is-danger { color: #b42318; background: #fee4e2; }
}
.detail-quality-empty { color: #98a2b3; }
:global(.data-process-result-tooltip) {
box-sizing: border-box;
max-width: min(520px, calc(100vw - 32px));

View File

@@ -0,0 +1,252 @@
<script setup lang="ts">
import { computed } from 'vue'
import VChart from 'vue-echarts'
import '@/plugins/echarts'
import type { EChartsOption } from 'echarts'
import type { ResultQualityDetails } from './types'
const props = defineProps<{
quality: ResultQualityDetails
score?: number
}>()
// 轴标签按词意预置断行,避免长中文标签把雷达网格挤偏或被机械切词。
const JUDGE_DIMENSION_LABELS: Record<string, string> = {
faithfulness: '忠实度',
correctness: '正确性',
clarity: '问题\n清晰度',
completeness: '回答\n完整性',
alignment: '指令\n对齐',
reasoning_validity: '推理\n有效性',
chosen_quality: 'chosen\n质量',
rejected_quality: 'rejected\n质量',
preference_reasonableness: '偏好\n区分',
}
const SEMANTIC_DIMENSION_LABELS: Record<string, string> = {
question_answer: '问答\n相关',
answer_source: '来源\n覆盖',
}
interface RadarDimension {
name: string
value: number
}
const radarDimensions = computed<RadarDimension[]>(() => {
const dimensions: RadarDimension[] = []
for (const [key, value] of Object.entries(props.quality?.judge?.scores ?? {})) {
dimensions.push({
name: JUDGE_DIMENSION_LABELS[key] ?? key,
value: Math.round(value * 20),
})
}
for (const [key, value] of Object.entries(props.quality?.semantic ?? {})) {
if (key === 'overall' || typeof value !== 'number') continue
dimensions.push({
name: SEMANTIC_DIMENSION_LABELS[key] ?? key,
value: Math.round(value),
})
}
return dimensions
})
// 可用维度太少时雷达图失去意义,降级为分层分数展示。
const showRadar = computed(() => radarDimensions.value.length >= 3)
const radarOption = computed<EChartsOption>(() => ({
radar: {
indicator: radarDimensions.value.map((dimension) => ({
name: dimension.name,
max: 100,
})),
radius: '56%',
center: ['50%', '50%'],
splitNumber: 4,
axisName: {
color: '#667085',
fontSize: 10,
lineHeight: 13,
},
splitArea: { areaStyle: { color: ['#fbfbfd', '#f2f4f8'] } },
splitLine: { lineStyle: { color: '#e4e7ec' } },
axisLine: { lineStyle: { color: '#e4e7ec' } },
},
series: [{
type: 'radar',
symbol: 'circle',
symbolSize: 3,
data: [{
value: radarDimensions.value.map((dimension) => dimension.value),
name: '质量维度',
areaStyle: { color: 'rgba(91, 80, 242, 0.18)' },
lineStyle: { color: '#5b50f2', width: 1.5 },
itemStyle: { color: '#5b50f2' },
}],
}],
}))
const layerScores = computed(() => {
const layers = props.quality?.layers ?? {}
return [
{ label: '规则层', value: layers.rule },
{ label: '语义层', value: layers.semantic },
{ label: '评审层', value: layers.judge },
].filter((layer): layer is { label: string; value: number } => (
typeof layer.value === 'number'
))
})
const displayScore = computed(() => {
if (typeof props.score === 'number' && !isNaN(props.score)) {
return props.score.toFixed(1)
}
return null
})
const scoreTone = computed(() => {
const numScore = props.score ?? 0
return numScore >= 80 ? 'is-success' : numScore >= 60 ? 'is-warning' : 'is-danger'
})
</script>
<template>
<div class="quality-popover">
<div v-if="displayScore !== null" class="popover-header">
<strong>质量评测</strong>
<span class="popover-score" :class="scoreTone">{{ displayScore }}</span>
</div>
<VChart
v-if="showRadar"
class="quality-radar"
:option="radarOption"
autoresize
/>
<div v-else class="radar-fallback">
维度数据不足已评测维度少于 3 个时以分层分数为准
</div>
<div class="layer-scores">
<div v-for="layer in layerScores" :key="layer.label" class="layer-item">
<span>{{ layer.label }}</span>
<el-progress
:percentage="Math.round(layer.value)"
:stroke-width="6"
:show-text="false"
:color="layer.value >= 80 ? '#12b76a' : layer.value >= 60 ? '#f0b429' : '#d92d20'"
/>
<em>{{ layer.value.toFixed(0) }}</em>
</div>
</div>
<p v-if="quality?.judge?.reason" class="judge-reason">{{ quality.judge.reason }}</p>
<div v-if="quality?.judge?.issues?.length" class="judge-issues">
<span v-for="issue in quality.judge.issues" :key="issue" class="issue-tag">{{ issue }}</span>
</div>
<div v-if="quality?.judge?.model" class="judge-model">评审模型{{ quality.judge.model }}</div>
</div>
</template>
<style scoped lang="scss">
.quality-popover {
display: flex;
flex-direction: column;
gap: 10px;
width: 100%;
box-sizing: border-box;
}
.popover-header {
display: flex;
align-items: center;
justify-content: space-between;
padding-bottom: 6px;
border-bottom: 1px solid #f2f4f7;
strong {
color: #344054;
font-size: 13px;
}
}
.popover-score {
color: #344054;
font-size: 18px;
font-weight: 700;
font-variant-numeric: tabular-nums;
&.is-success { color: #12b76a; }
&.is-warning { color: #d99b0b; }
&.is-danger { color: #d92d20; }
}
.quality-radar {
width: 100%;
height: 220px;
box-sizing: border-box;
}
.radar-fallback {
padding: 18px 10px;
color: #98a2b3;
font-size: 12px;
text-align: center;
}
.layer-scores {
display: flex;
flex-direction: column;
gap: 6px;
}
.layer-item {
display: grid;
grid-template-columns: 44px 1fr 28px;
align-items: center;
gap: 8px;
color: #667085;
font-size: 11px;
em {
color: #344054;
font-style: normal;
font-weight: 600;
text-align: right;
font-variant-numeric: tabular-nums;
}
}
.judge-reason {
margin: 0;
color: #475467;
font-size: 12px;
line-height: 1.6;
}
.judge-issues {
display: flex;
flex-wrap: wrap;
gap: 4px;
}
.issue-tag {
padding: 2px 8px;
color: #b54708;
background: #fef0c7;
border-radius: 3px;
font-size: 11px;
}
.judge-model {
color: #98a2b3;
font-size: 11px;
}
</style>
<style lang="scss">
.quality-radar-popper {
padding: 14px 16px !important;
}
</style>

View File

@@ -1,6 +1,7 @@
<script setup lang="ts">
import { computed, ref } from 'vue'
import type { BulkResultRegenerationState, PreviewItem, ResultItem } from './types'
import QualityRadarPopover from './QualityRadarPopover.vue'
import type { BulkResultRegenerationState, PreviewItem, ResultEvaluationState, ResultItem } from './types'
import type { DataProcessOutputType } from '@/types/dataProcess'
const props = defineProps<{
@@ -9,6 +10,7 @@ const props = defineProps<{
selectedId: string | null
regeneratingResultId: string | null
bulkRegeneration: BulkResultRegenerationState
evaluation: ResultEvaluationState
outputType: DataProcessOutputType
}>()
@@ -17,6 +19,7 @@ const emit = defineEmits<{
'update:field': [id: string, field: 'instruction' | 'input' | 'output' | 'chosen' | 'rejected', value: string]
'regenerate:item': [id: string]
'regenerate:all': []
'evaluate:all': []
}>()
const search = ref('')
@@ -30,6 +33,15 @@ const bulkRegenerationActive = computed(() => props.bulkRegeneration.status ===
const bulkRegenerationVisible = computed(() => (
props.bulkRegeneration.status !== 'idle' && props.bulkRegeneration.total > 0
))
const evaluationActive = computed(() => props.evaluation.status === 'running')
const evaluationVisible = computed(() => (
props.evaluation.status !== 'idle' && props.evaluation.total > 0
))
const evaluationPercentage = computed(() => {
if (!props.evaluation.total) return 0
return Math.round((props.evaluation.completed / props.evaluation.total) * 100)
})
const evaluatedCount = computed(() => props.items.filter((item) => item.qualityDetails?.evaluated).length)
const bulkRegenerationPercentage = computed(() => (
props.bulkRegeneration.total > 0
? Math.round((props.bulkRegeneration.completed / props.bulkRegeneration.total) * 100)
@@ -98,21 +110,47 @@ function selectRelative(offset: number) {
<template>
<section class="result-step">
<div class="result-workspace">
<aside class="result-list-pane" :class="{ 'has-bulk-progress': bulkRegenerationVisible }">
<aside class="result-list-pane" :class="{ 'has-bulk-progress': bulkRegenerationVisible || evaluationVisible }">
<div class="pane-header result-list-header">
<div class="result-list-title"><strong>生成结果</strong><span> {{ items.length }} </span></div>
<el-button
v-if="invalidCount > 0"
size="small"
plain
type="primary"
:loading="bulkRegenerationActive"
:disabled="Boolean(regeneratingResultId) || bulkRegenerationActive"
@click="emit('regenerate:all')"
>
<i v-if="!bulkRegenerationActive" class="fa fa-refresh" style="margin-right: 4px;" />
{{ bulkRegenerationActive ? '重新生成中' : `全部重新生成(${invalidCount}` }}
</el-button>
<div class="result-list-title">
<strong>生成结果</strong><span> {{ items.length }} <template v-if="evaluatedCount"> · 已评测 {{ evaluatedCount }}</template></span>
</div>
<div class="result-list-actions">
<el-button
size="small"
plain
:loading="evaluationActive"
:disabled="!items.length || bulkRegenerationActive || Boolean(regeneratingResultId)"
@click="emit('evaluate:all')"
>
<i v-if="!evaluationActive" class="fa fa-check-square-o" style="margin-right: 4px;" />
数据评测
</el-button>
<el-button
v-if="invalidCount > 0"
size="small"
plain
type="primary"
:loading="bulkRegenerationActive"
:disabled="Boolean(regeneratingResultId) || bulkRegenerationActive || evaluationActive"
@click="emit('regenerate:all')"
>
<i v-if="!bulkRegenerationActive" class="fa fa-refresh" style="margin-right: 4px;" />
{{ bulkRegenerationActive ? '重新生成中' : `全部重新生成(${invalidCount}` }}
</el-button>
</div>
</div>
<div v-if="evaluationVisible" class="bulk-regeneration-progress">
<div>
<span>数据评测 {{ evaluation.completed }} / {{ evaluation.total }}</span>
<span>成功 {{ evaluation.succeeded }} · 失败 {{ evaluation.failed }}</span>
</div>
<el-progress
:percentage="evaluationPercentage"
:show-text="false"
:stroke-width="5"
:color="evaluation.failed > 0 ? '#d97706' : '#5b50f2'"
/>
</div>
<div v-if="bulkRegenerationVisible" class="bulk-regeneration-progress">
<div>
@@ -146,6 +184,23 @@ function selectRelative(offset: number) {
<strong>{{ item.instruction || '未填写指令' }}</strong>
<small>{{ outputType === 'dpo' ? (item.chosen || '未填写 Chosen') : (item.output || '未填写输出') }}</small>
</span>
<el-popover
v-if="item.qualityScore != null && item.qualityDetails"
placement="right"
:width="296"
trigger="hover"
:show-after="150"
popper-class="quality-radar-popper"
>
<template #reference>
<span
class="result-score"
:class="item.qualityScore >= 80 ? 'is-success' : item.qualityScore >= 60 ? 'is-warning' : 'is-danger'"
@click.stop
>{{ item.qualityScore.toFixed(0) }}</span>
</template>
<QualityRadarPopover :quality="item.qualityDetails" :score="item.qualityScore" />
</el-popover>
<i v-if="itemRegenerating(item.id)" class="css-spinner" />
<i
v-else
@@ -303,6 +358,42 @@ function selectRelative(offset: number) {
}
}
.result-list-actions {
display: flex;
flex: none;
align-items: center;
gap: 8px;
}
.result-score {
flex: none;
min-width: 34px;
padding: 2px 8px;
border-radius: 10px;
color: #475467;
background: #f2f4f7;
font-size: 12px;
font-weight: 700;
font-variant-numeric: tabular-nums;
text-align: center;
cursor: default;
&.is-success {
color: #067647;
background: #e6f4ee;
}
&.is-warning {
color: #b54708;
background: #fef0c7;
}
&.is-danger {
color: #b42318;
background: #fee4e2;
}
}
.pane-header {
display: flex;
align-items: center;

View File

@@ -1,6 +1,9 @@
import type {
DataProcessOutputType,
DataProcessPreviewFileStatus,
DataProcessQualityJudge,
DataProcessQualityLayers,
DataProcessQualitySemantic,
DataProcessReasoningDetail,
} from '@/types/dataProcess'
@@ -165,6 +168,14 @@ export interface GenerationState {
message: string
}
export interface ResultQualityDetails {
semantic?: DataProcessQualitySemantic | null
judge?: DataProcessQualityJudge | null
layers?: DataProcessQualityLayers | null
evaluated?: boolean
flags?: string[]
}
export interface ResultItem {
id: string
previewItemId: string | null
@@ -188,7 +199,7 @@ export interface ResultItem {
error?: string
split?: 'train' | 'validation' | 'test'
qualityScore?: number
qualityDetails?: Record<string, number>
qualityDetails?: ResultQualityDetails
updatedAt?: string
}
@@ -201,3 +212,11 @@ export interface BulkResultRegenerationState {
targetIds: string[]
failedIds: string[]
}
export interface ResultEvaluationState {
status: 'idle' | 'running' | 'completed' | 'partial' | 'failed'
total: number
completed: number
succeeded: number
failed: number
}

View File

@@ -0,0 +1,126 @@
import { computed, reactive, type Ref } from 'vue'
import { ElMessage } from 'element-plus'
import { evaluateDataProcessResults } from '@/api/modules/dataProcess'
import { mapResult } from './useDataProcessGeneration'
import type { ResultEvaluationState, ResultItem } from './types'
interface EvaluationBindings {
taskId: Ref<string | null>
results: Ref<ResultItem[]>
selectedResultId: Ref<string | null>
}
// 与批量重新生成一致的分块大小4 个后端 worker 消费三轮,
// 单条评测最长 60 秒12 条在批量接口 240 秒超时预算内。
const EVALUATION_CHUNK_SIZE = 12
function hasUnsavedChanges(item: ResultItem) {
return item.instruction !== item.savedInstruction
|| item.input !== item.savedInput
|| item.output !== item.savedOutput
|| item.chosen !== item.savedChosen
|| item.rejected !== item.savedRejected
}
export function useDataProcessEvaluation(bindings: EvaluationBindings) {
const evaluation = reactive<ResultEvaluationState>({
status: 'idle',
total: 0,
completed: 0,
succeeded: 0,
failed: 0,
})
const evaluationBusy = computed(() => evaluation.status === 'running')
function resetEvaluation() {
Object.assign(evaluation, {
status: 'idle',
total: 0,
completed: 0,
succeeded: 0,
failed: 0,
})
}
async function evaluateAllResults() {
const taskId = bindings.taskId.value
if (!taskId) return false
if (evaluationBusy.value) {
ElMessage.warning('请等待当前数据评测完成')
return false
}
const candidates = bindings.results.value.filter((item) => item.updatedAt)
if (!candidates.length) {
ElMessage.info('当前没有可评测的结果')
return false
}
const unsaved = candidates.find(hasUnsavedChanges)
if (unsaved) {
bindings.selectedResultId.value = unsaved.id
ElMessage.warning('存在未保存的修改,请先保存后再进行数据评测')
return false
}
Object.assign(evaluation, {
status: 'running',
total: candidates.length,
completed: 0,
succeeded: 0,
failed: 0,
})
let interrupted = false
try {
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, {
items: chunk.map((item) => ({
result_id: item.id,
expected_updated_at: item.updatedAt as string,
})),
})
for (const item of evaluated.items) {
const index = bindings.results.value.findIndex((entry) => entry.id === String(item.id))
if (index >= 0) bindings.results.value[index] = mapResult(item)
}
evaluation.completed += evaluated.total
evaluation.succeeded += evaluated.succeeded
evaluation.failed += evaluated.failed
} catch {
const remaining = candidates.slice(offset)
evaluation.completed = evaluation.total
evaluation.failed += remaining.length
interrupted = true
break
}
}
if (!interrupted && evaluation.failed === 0) {
evaluation.status = 'completed'
ElMessage.success(`数据评测完成:成功 ${evaluation.succeeded}`)
} else if (evaluation.succeeded > 0) {
evaluation.status = 'partial'
ElMessage.warning(
`数据评测完成:成功 ${evaluation.succeeded} 条,失败 ${evaluation.failed}`,
)
} else {
evaluation.status = 'failed'
ElMessage.error(`数据评测失败:${evaluation.failed} 条结果未完成评测`)
}
return evaluation.failed === 0
} catch {
evaluation.status = 'failed'
ElMessage.error('批量数据评测意外中断,已完成的评分保持不变')
return false
}
}
return {
evaluation,
evaluationBusy,
evaluateAllResults,
resetEvaluation,
}
}

View File

@@ -28,6 +28,7 @@ const POLL_INTERVAL_MS = 1500
const BULK_REGENERATION_CHUNK_SIZE = 12
function mapResult(item: DataProcessResult): ResultItem {
const quality = item.quality_score
return {
id: String(item.id),
previewItemId: item.preview_item_id == null ? null : String(item.preview_item_id),
@@ -50,11 +51,21 @@ function mapResult(item: DataProcessResult): ResultItem {
status: item.status,
error: item.error || undefined,
split: item.split || undefined,
qualityScore: item.quality_score?.overall,
// 生成阶段只有内部规则分,界面不展示;数据评测完成后才显示组合分。
qualityScore: quality?.evaluated ? quality.overall : undefined,
qualityDetails: {
semantic: quality?.semantic ?? null,
judge: quality?.judge ?? null,
layers: quality?.layers ?? null,
evaluated: Boolean(quality?.evaluated),
flags: quality?.flags ?? [],
},
updatedAt: item.updated_at,
}
}
export { mapResult }
export function useDataProcessGeneration(bindings: GenerationBindings) {
const results = ref<ResultItem[]>([])
const selectedResultId = ref<string | null>(null)