3 Commits

Author SHA1 Message Date
wuyongtao
0b59c1ebee feat: 训练任务支持从最新 checkpoint 继续训练
- 后端: 新增 resume/checkpoints 端点,compute 客户端与适配器支持续训
- 计算节点: llama_factory 适配器断点续训支持及测试
- 前端: fine-tune API 封装与列表页续训/断点展示
2026-08-21 14:48:55 +08:00
wuyongtao
0233755859 feat: 评测任务完成进度收尾与指标维度汇总展示
- platform_store: 评测任务完成时置进度 100% 并落地 completed_time
- eval_runner: 新增指标维度汇总,供雷达图等维度展示
- 前端: 评测详情/列表展示优化、模型管理增强
2026-08-21 10:49:50 +08:00
caoxiaozhu
03254f8196 feat(data_process): 模型缓存统一仓库根 .cache 并修复 PDF 页眉页脚清理
- 新增 app/core/cache_paths.py:HF_HOME / tiktoken 缓存统一指向 <repo>/.cache,
  本地与 Docker 路径一致,离线部署打包 .cache 即可
- Dockerfile.backend 的 tiktoken 词表改用官方 SHA 文件名,避免运行时回退重建
- 修复 layout_hybrid 路径不运行 detect_pdf_document_noise 的缺陷:
  needs_pdf_noise 不再与 needs_layout_raw 互斥,PDF 智能预处理在版面切分下也生效
- 新增 layout_noise.py:识别跨页重复的页眉表格标签组并按行剔除,
  解决 docling layout 模型把中文企业 PDF 页眉识别成普通 Table 导致清不掉的问题
- 回收 HybridChunker 丢弃的末尾孤立标题,找回章节标题内容
2026-08-21 10:16:23 +08:00
26 changed files with 906 additions and 80 deletions

0
.cache/.gitkeep Normal file
View File

View File

0
.cache/tiktoken/.gitkeep Normal file
View File

8
.gitignore vendored
View File

@@ -55,7 +55,6 @@ htmlcov/
.nox/ .nox/
.coverage .coverage
.coverage.* .coverage.*
.cache
nosetests.xml nosetests.xml
coverage.xml coverage.xml
*.cover *.cover
@@ -220,3 +219,10 @@ docker/minio/data/*
# nlp-eval-demo - 独立演示项目,不进版本库 # nlp-eval-demo - 独立演示项目,不进版本库
nlp-eval-demo/ nlp-eval-demo/
nlp-eval-demo.zip nlp-eval-demo.zip
# 项目本地缓存HuggingFace / tiktoken 等大文件),保留目录结构与占位文件
.cache/*
!.cache/.gitkeep
!.cache/tiktoken/
!.cache/huggingface/
!.cache/**/.gitkeep

View File

@@ -1686,7 +1686,6 @@ def _prepare_preview_items(
) )
needs_pdf_noise = ( needs_pdf_noise = (
is_unstructured is_unstructured
and not needs_layout_raw
and source_format == "pdf" and source_format == "pdf"
and bool( and bool(
preprocess_options & {"clean_invalid", "clean_invalid_content"} preprocess_options & {"clean_invalid", "clean_invalid_content"}
@@ -1720,8 +1719,9 @@ def _prepare_preview_items(
enriched = dict(source) enriched = dict(source)
if needs_structured_xlsx or needs_layout_raw: if needs_structured_xlsx or needs_layout_raw:
enriched["raw_content"] = raw enriched["raw_content"] = raw
sources[index] = enriched if needs_pdf_noise:
continue # layout_hybrid 路径下也跑 PDF 文本规则噪声检测,
# 弥补 docling layout 模型对中文 PDF 页眉/页脚识别率低的问题。
pages = extract_pdf_page_texts(raw) pages = extract_pdf_page_texts(raw)
extracted_text = "\n\n".join(page.text for page in pages if page.text) extracted_text = "\n\n".join(page.text for page in pages if page.text)
if extracted_text != str(source.get("content") or ""): if extracted_text != str(source.get("content") or ""):
@@ -1729,7 +1729,7 @@ def _prepare_preview_items(
"跳过PDF文档噪声检测存储偏移量不一致 source_id=%s", "跳过PDF文档噪声检测存储偏移量不一致 source_id=%s",
source["id"], source["id"],
) )
continue else:
enriched["document_noise_spans"] = detect_pdf_document_noise(pages) enriched["document_noise_spans"] = detect_pdf_document_noise(pages)
sources[index] = enriched sources[index] = enriched
items = _build_preview_items(task, sources) items = _build_preview_items(task, sources)

View File

@@ -521,6 +521,117 @@ async def _prepare_resource_on_node(store: Any, resource_type: str, resource_id:
return f"/data/yg-ft/{root_name}/{resource_id}" return f"/data/yg-ft/{root_name}/{resource_id}"
async def _archive_training_checkpoints(
store: Any,
task: dict[str, Any],
node: dict[str, Any],
job: dict[str, Any],
) -> dict[str, Any]:
"""Persist checkpoints from a stopped job so resume can use another node."""
task_id = str(task.get("id") or "")
output_dir = str(job.get("output_dir") or task.get("output_dir") or "")
checkpoints = job.get("checkpoints") or store.task_checkpoints(task_id)
if not task_id or not output_dir or not checkpoints:
return store.update_task(
task_id,
{
"checkpoint_archive_status": "not_available",
"checkpoint_archive_error": "停止时没有发现可用 checkpoint",
},
)
if not get_settings().minio_enabled:
return store.update_task(
task_id,
{
"checkpoint_archive_status": "local_only",
"checkpoint_archive_error": "MinIO 未启用,仅支持原算力节点本地继续",
},
)
try:
version_id = str(job.get("id") or task.get("compute_job_id") or task_id)
client = ComputeNodeClient(node["api_base_url"], timeout=900)
archived = await _archive_node_directory(
store,
client,
node,
output_dir,
"fine-tune",
task_id,
version_id,
f"training/{task_id}/checkpoints",
)
return store.update_task(
task_id,
{
"checkpoint_archive_status": "completed" if archived else "empty",
"checkpoint_archive_version_id": version_id,
"checkpoint_archive_object_ids": [str(item["id"]) for item in archived],
"checkpoint_archive_error": "" if archived else "checkpoint 目录为空",
},
)
except Exception as exc: # noqa: BLE001 - stopping must remain successful if archive is delayed
return store.update_task(
task_id,
{
"checkpoint_archive_status": "pending",
"checkpoint_archive_error": str(exc)[:2000],
},
)
async def _prepare_resume_checkpoint(
store: Any,
task: dict[str, Any],
checkpoint: dict[str, Any],
node: dict[str, Any],
) -> str | None:
"""Materialize the selected checkpoint on the node used for resuming."""
task_id = str(task.get("id") or "")
checkpoint_name = str(checkpoint.get("name") or Path(str(checkpoint.get("path") or "")).name)
if not task_id or not checkpoint_name:
return None
# Prefer the archived copy. This makes resume independent of the node that
# originally ran the task and preserves the MinIO-as-source-of-truth rule.
if get_settings().minio_enabled:
prefix = f"training/{task_id}/checkpoints/"
objects = [
item
for item in store.storage_objects_for_resource("fine-tune", task_id)
if str(item.get("object_key") or "").startswith(prefix)
and str(item.get("file_name") or "").startswith(f"{checkpoint_name}/")
]
if objects:
client = ComputeNodeClient(node["api_base_url"], timeout=900)
root = f"/data/yg-ft/fine-tunes/{task_id}/resume/{checkpoint_name}"
for item in objects:
file_name = str(item.get("file_name") or "")
relative = file_name[len(checkpoint_name) + 1 :]
await client.prepare_cache(
{
"resource_id": task_id,
"version_id": item.get("version_id"),
"download_url": get_object_storage().presigned_get(str(item["object_key"])),
"checksum_sha256": item.get("checksum_sha256") or "",
"byte_size": item.get("byte_size") or 0,
"relative_path": f"fine-tunes/{task_id}/resume/{checkpoint_name}/{relative}",
}
)
return root
# Compatibility fallback for installations without MinIO archives: only
# reuse the original path when the same node still owns the files.
original_node_id = str(task.get("compute_node_id") or "")
original_path = str(checkpoint.get("path") or "")
if original_node_id == str(node.get("id") or "") and original_path:
check = await ComputeNodeClient(node["api_base_url"]).check_paths(
[{"path": original_path, "type": "dir", "required": True}]
)
if check.get("valid"):
return original_path
return None
def _build_messages_payload(payload: dict[str, Any]) -> dict[str, Any]: def _build_messages_payload(payload: dict[str, Any]) -> dict[str, Any]:
"""Convert frontend inference payload to compute API messages format. """Convert frontend inference payload to compute API messages format.
@@ -2458,8 +2569,28 @@ async def stop_fine_tune(task_id: str, current_user: dict = Depends(get_current_
return pending return pending
node = _node_for_task(task) node = _node_for_task(task)
if task.get("compute_job_id") and node and get_settings().compute_mode != "simulator": if task.get("compute_job_id") and node and get_settings().compute_mode != "simulator":
job = await ComputeNodeClient(node["api_base_url"]).stop_job(task["compute_job_id"]) try:
return ok(store.apply_compute_job(task_id, job)) job = await ComputeNodeClient(
node["api_base_url"],
timeout=max(float(get_settings().compute_request_timeout_seconds), 30.0),
).stop_job(task["compute_job_id"])
except httpx.TimeoutException as exc:
raise fail(504, "停止训练超时,算力节点未及时响应,请稍后查看任务状态") from exc
except httpx.HTTPStatusError as exc:
if exc.response.status_code == 404:
# The process has already disappeared from the node. Make
# the local task terminal and let the poller release state.
return ok(store.stop_task(task_id))
raise fail(502, f"算力节点拒绝停止训练HTTP {exc.response.status_code}") from exc
except httpx.RequestError as exc:
raise fail(502, f"算力节点不可达,暂时无法停止训练:{exc}") from exc
updated = store.apply_compute_job(task_id, job)
# Stop is terminal for scheduling, but checkpoint archiving is
# best-effort so a slow MinIO service never turns a successful stop
# into a 500 response.
return ok(await _archive_training_checkpoints(store, updated, node, job))
if task.get("compute_job_id") and not node:
raise fail(409, "训练任务关联的算力节点不存在,无法安全停止远程进程")
return ok(store.stop_task(task_id)) return ok(store.stop_task(task_id))
except KeyError: except KeyError:
raise fail(404, "fine tune task not found") raise fail(404, "fine tune task not found")
@@ -2470,6 +2601,76 @@ async def stop_fine_tune_alt(task_id: str, current_user: dict = Depends(get_curr
return await stop_fine_tune(task_id, current_user) return await stop_fine_tune(task_id, current_user)
@router.post("/fine-tune/{task_id}/resume")
@op_log(module=OpModule.FINE_TUNE, action=OpAction.RETRY, target_type="fine_tune", target_name_param="task_id")
async def resume_fine_tune(
task_id: str,
payload: dict[str, Any] | None = Body(default=None),
current_user: dict[str, Any] = Depends(get_current_user),
) -> dict[str, Any]:
"""Resume a stopped/failed training task from its newest checkpoint."""
store = get_platform_store()
payload = payload or {}
try:
task = store.task(task_id)
except KeyError:
raise fail(404, "fine tune task not found")
if not has_resource_access("fine-tune", task_id, current_user, "execute"):
raise fail(403, "no permission to resume this task")
if task.get("status") not in {"stopped", "failed"}:
raise fail(409, "只有已停止或失败的训练任务可以继续")
checkpoints = store.task_checkpoints(task_id)
if not checkpoints:
raise fail(409, "没有可用的训练断点,请确认任务停止前已经生成 checkpoint")
checkpoint = max(checkpoints, key=lambda item: (int(item.get("step") or 0), str(item.get("create_time") or "")))
resume_payload = {
**task,
**payload,
"task_id": task_id,
"id": task_id,
"compute_node_id": payload.get("compute_node_id") or task.get("compute_node_id"),
"gpus": payload.get("gpus") or task.get("gpus") or [],
"strict_node_selection": True,
"resume_checkpoint_id": checkpoint.get("id"),
"resume_checkpoint_name": checkpoint.get("name"),
"resume_from_checkpoint": checkpoint.get("path"),
}
try:
# Select the requested/original node first, materialize the checkpoint,
# then run the normal model/dataset/GPU preflight against that node.
node, _ = store.prepare_compute_job_payload(task_id, resume_payload)
if get_settings().compute_mode == "simulator":
prepared_checkpoint = str(checkpoint.get("path") or "")
else:
prepared_checkpoint = await _prepare_resume_checkpoint(store, task, checkpoint, node)
if not prepared_checkpoint:
raise RuntimeError(
f"断点 {checkpoint.get('name') or checkpoint.get('path')} 不可用:原算力节点文件已不存在,且 MinIO 中没有可恢复副本"
)
resume_payload["resume_from_checkpoint"] = prepared_checkpoint
resume_payload["compute_node_id"] = node["id"]
preflight = await _fine_tune_preflight(
store,
task_id,
resume_payload,
validate=True,
sync_resources=True,
)
if not preflight.get("valid"):
errors = "; ".join(preflight.get("errors") or ["继续训练预检未通过"])
raise RuntimeError(errors)
store.reset_task_for_retry(task_id, resume_payload)
return ok(await _submit_fine_tune_task(store, resume_payload))
except RuntimeError as exc:
raise fail(409, f"继续训练预检未通过:{exc}") from exc
except httpx.TimeoutException as exc:
raise fail(504, "继续训练预检超时,请确认算力节点和 MinIO 服务可达") from exc
except Exception as exc: # noqa: BLE001 - keep the resource diagnosis visible to the user
store.mark_task_failed(task_id, str(exc))
raise fail(502, f"继续训练失败:{exc}") from exc
@router.post("/fine-tune/{task_id}/retry") @router.post("/fine-tune/{task_id}/retry")
@op_log(module=OpModule.FINE_TUNE, action=OpAction.RETRY, target_type="fine_tune", target_name_param="task_id") @op_log(module=OpModule.FINE_TUNE, action=OpAction.RETRY, target_type="fine_tune", target_name_param="task_id")
async def retry_fine_tune(task_id: str, payload: dict[str, Any] | None = Body(default=None), current_user: dict = Depends(get_current_user)) -> dict[str, Any]: async def retry_fine_tune(task_id: str, payload: dict[str, Any] | None = Body(default=None), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:

View File

@@ -0,0 +1,46 @@
"""集中管理项目本地缓存目录HuggingFace / tiktoken
所有 Python 库通过环境变量引用 ``<repo_root>/.cache/{huggingface,tiktoken}``
避免写入用户家目录,也避免不同部署路径(本地 / Docker下缓存位置不一致。
"""
from __future__ import annotations
import os
from pathlib import Path
def repo_root() -> Path:
# backend/app/core/cache_paths.py -> backend -> 仓库根目录
return Path(__file__).resolve().parents[3]
def cache_root() -> Path:
return repo_root() / ".cache"
def huggingface_cache_dir() -> Path:
return cache_root() / "huggingface"
def tiktoken_cache_dir() -> Path:
return cache_root() / "tiktoken"
def setup_local_caches() -> None:
"""进程启动时统一设置 HF / tiktoken 缓存环境变量,并确保目录存在。
必须在 import ``docling`` / ``tiktoken`` 等依赖之前调用,否则首次使用会
仍然走到默认 ``~/.cache`` 路径。
"""
hf_dir = huggingface_cache_dir()
tiktoken_dir = tiktoken_cache_dir()
hf_dir.mkdir(parents=True, exist_ok=True)
(hf_dir / "hub").mkdir(parents=True, exist_ok=True)
tiktoken_dir.mkdir(parents=True, exist_ok=True)
os.environ["HF_HOME"] = str(hf_dir)
os.environ["HUGGINGFACE_HUB_CACHE"] = str(hf_dir / "hub")
os.environ["HF_HUB_CACHE"] = str(hf_dir / "hub")
os.environ["TIKTOKEN_CACHE_DIR"] = str(tiktoken_dir)

View File

@@ -3361,6 +3361,32 @@ class PlatformStore:
}) })
if new_status == "completed": if new_status == "completed":
updates["completed_time"] = utcnow() updates["completed_time"] = utcnow()
if new_status == "completed":
final_total = int(
(result_content or {}).get("sample_count")
or task.get("sample_count")
or (progress_detail or {}).get("total")
or 0
)
final_completed = int(
(result_content or {}).get("completed_count")
or task.get("completed_count")
or final_total
)
updates.update({
"progress": 100,
"progress_detail": {
**progress_detail,
"status": "completed",
"stage": "completed",
"total": final_total,
"completed": max(final_completed, final_total),
"percentage": 100,
"current_index": final_total,
"message": "评测完成",
},
"completed_time": utcnow(),
})
elif new_status in {"failed", "stopped"}: elif new_status in {"failed", "stopped"}:
updates.update({ updates.update({
"error": job.get("error") or task.get("error") or "", "error": job.get("error") or task.get("error") or "",

View File

@@ -4,10 +4,15 @@ from contextlib import suppress
from fastapi import FastAPI from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from app.api.v1.router import api_router from app.core.cache_paths import setup_local_caches
from app.core.config import docs_kwargs, get_settings
from app.core.logging import configure_logging, setup_request_logging # 在任何 docling / tiktoken 模块被实例化之前设置缓存路径,避免首调用走到 ~/.cache。
from app.workers.compute_poller import run_compute_poller setup_local_caches()
from app.api.v1.router import api_router # noqa: E402
from app.core.config import docs_kwargs, get_settings # noqa: E402
from app.core.logging import configure_logging, setup_request_logging # noqa: E402
from app.workers.compute_poller import run_compute_poller # noqa: E402
def create_app() -> FastAPI: def create_app() -> FastAPI:

View File

@@ -158,7 +158,12 @@ class ComputeNodeClient:
return _unwrap_dict(response.json()) return _unwrap_dict(response.json())
async def stop_job(self, job_id: str) -> dict[str, Any]: async def stop_job(self, job_id: str) -> dict[str, Any]:
async with httpx.AsyncClient(timeout=self.timeout, headers=self.headers()) as client: # Compute API waits for the child process to exit (up to 10 seconds)
# before returning. The normal polling timeout is intentionally short,
# but is too aggressive for a stop request and used to surface as a
# platform 500 even when the node eventually stopped the job.
stop_timeout = max(float(self.timeout), 30.0)
async with httpx.AsyncClient(timeout=stop_timeout, headers=self.headers()) as client:
response = await client.post(_join_url(self.api_base_url, f"{self.route_prefix}/compute/jobs/{job_id}/stop")) response = await client.post(_join_url(self.api_base_url, f"{self.route_prefix}/compute/jobs/{job_id}/stop"))
response.raise_for_status() response.raise_for_status()
return _unwrap_dict(response.json()) return _unwrap_dict(response.json())

View File

@@ -49,13 +49,16 @@ from .text_utils import (
# Layer 1: 解析器(依赖 types 与 text_utils # Layer 1: 解析器(依赖 types 与 text_utils
from .parsers import ( from .parsers import (
LayoutRepeatedBlock,
_infer_xlsx_header_region, _infer_xlsx_header_region,
_rewrite_xlsx_workbook_relationships, _rewrite_xlsx_workbook_relationships,
_validate_office_archive, _validate_office_archive,
_xlsx_sheet_merge_ranges, _xlsx_sheet_merge_ranges,
detect_layout_repeated_blocks,
detect_pdf_document_noise, detect_pdf_document_noise,
extract_pdf_page_texts, extract_pdf_page_texts,
remove_document_noise, remove_document_noise,
remove_layout_repeated_blocks,
) )
# Layer 3: 数据转换与质量评分 # Layer 3: 数据转换与质量评分
@@ -115,6 +118,7 @@ __all__ = [
"desensitize_pii", "desensitize_pii",
"desensitize_structured_record", "desensitize_structured_record",
"detect_document_structure", "detect_document_structure",
"detect_layout_repeated_blocks",
"detect_pdf_document_noise", "detect_pdf_document_noise",
"detect_text_format", "detect_text_format",
"estimate_token_count", "estimate_token_count",
@@ -137,7 +141,9 @@ __all__ = [
"preprocess_structured_records_with_lineage", "preprocess_structured_records_with_lineage",
"protected_context_ranges", "protected_context_ranges",
"record_fingerprint", "record_fingerprint",
"LayoutRepeatedBlock",
"remove_document_noise", "remove_document_noise",
"remove_layout_repeated_blocks",
"score_quality", "score_quality",
"stable_split", "stable_split",
"stable_split_assignments", "stable_split_assignments",

View File

@@ -1,5 +1,10 @@
"""文档解析器模块。""" """文档解析器模块。"""
from .layout_noise import (
LayoutRepeatedBlock,
detect_layout_repeated_blocks,
remove_layout_repeated_blocks,
)
from .pdf import extract_pdf_page_texts, detect_pdf_document_noise, remove_document_noise from .pdf import extract_pdf_page_texts, detect_pdf_document_noise, remove_document_noise
from .office import ( from .office import (
_validate_office_archive, _validate_office_archive,
@@ -12,6 +17,9 @@ __all__ = [
'extract_pdf_page_texts', 'extract_pdf_page_texts',
'detect_pdf_document_noise', 'detect_pdf_document_noise',
'remove_document_noise', 'remove_document_noise',
'LayoutRepeatedBlock',
'detect_layout_repeated_blocks',
'remove_layout_repeated_blocks',
'_validate_office_archive', '_validate_office_archive',
'_rewrite_xlsx_workbook_relationships', '_rewrite_xlsx_workbook_relationships',
'_xlsx_sheet_merge_ranges', '_xlsx_sheet_merge_ranges',

View File

@@ -0,0 +1,174 @@
"""基于 Docling 输出的版面噪声检测与剔除。
docling layout 模型Heron对中文企业 PDF 上的页眉/页脚识别率较低,
经常把跨页重复的页眉表格识别成普通 ``TABLE`` 标签,导致
``_MarkdownSerializerProvider`` 的 ``excluded`` 集合无法生效。
本模块提供第二层启发式:扫描 docling 输出的所有 ``TableItem``
对每个表按"首列标签序列"聚合。如果同一组标签在文档中多页重复出现,
则判定为页眉/页脚类重复块,并在最终 chunk 文本中按行剔除。
"""
from __future__ import annotations
import math
import re
from collections.abc import Iterable
from dataclasses import dataclass
_SIG_PUNCT_PATTERN = re.compile(r"[\s\W_]+", re.UNICODE)
_SIG_DIGIT_PATTERN = re.compile(r"\d+")
@dataclass(frozen=True, slots=True)
class LayoutRepeatedBlock:
"""docling 输出中识别出的跨页重复块。"""
labels: tuple[str, ...]
occurrences: int
@property
def signature(self) -> str:
"""拼接签名(用于日志与向后兼容)。"""
return "".join(self.labels)
def _normalize_signature(text: str) -> str:
"""归一化:删除所有数字、去除空白/标点、转小写。"""
stripped = _SIG_DIGIT_PATTERN.sub("", text)
return _SIG_PUNCT_PATTERN.sub("", stripped).casefold()
def _extract_first_column_labels(table_text: str) -> tuple[str, ...]:
"""提取 docling TableItem markdown 表示中的"首列标签"序列。"""
labels: list[str] = []
seen: set[str] = set()
for raw_line in table_text.splitlines():
line = raw_line.strip()
if "|" not in line:
continue
parts = [cell.strip() for cell in line.strip("|").split("|")]
if not parts or not parts[0]:
continue
# 过滤掉分隔行(如 "| - | - |"
if all(re.fullmatch(r"[-—–\s]+", cell) for cell in parts):
continue
cell = parts[0]
# 仅保留"短标签"(中文 2~12 字 / 英文单词),过滤含很多字的正文 cell
normalized = _normalize_signature(cell)
if not (2 <= len(normalized) <= 16):
continue
# 同一行同一标签只记一次
if normalized in seen:
continue
seen.add(normalized)
labels.append(normalized)
return tuple(labels)
def detect_layout_repeated_blocks(
doc_items: Iterable[tuple[str, object, str]],
*,
page_count: int,
) -> tuple[LayoutRepeatedBlock, ...]:
"""扫描 docling 输出,识别跨页重复出现的标签组。
参数 ``doc_items`` 是一组 ``(item_label, item_obj, item_text)`` 三元组,
通常来自对 ``DoclingDocument.iterate_items()`` 的遍历。
判定条件(与 ``detect_pdf_document_noise`` 保持一致):
- 同一组首列标签至少在 ``max(3, ceil(page_count * 0.3))`` 个不同 item 中出现;
- 标签序列长度在 ``[1, 8]`` 之间。
"""
if page_count < 3:
return ()
label_groups: dict[tuple[str, ...], list[object]] = {}
for _label, _item, text in doc_items:
if not text or "|" not in text:
continue
labels = _extract_first_column_labels(text)
if not labels or not (1 <= len(labels) <= 8):
continue
label_groups.setdefault(labels, []).append(_item)
minimum_occurrences = max(3, math.ceil(page_count * 0.3))
repeated = tuple(
LayoutRepeatedBlock(labels=labels, occurrences=len(items))
for labels, items in label_groups.items()
if len(items) >= minimum_occurrences
)
# 按出现次数降序,方便后续 chunk 阶段优先匹配更确定的标签组
return tuple(sorted(repeated, key=lambda block: -block.occurrences))
def remove_layout_repeated_blocks(
text: str,
blocks: Iterable[LayoutRepeatedBlock],
) -> str:
"""按行剔除属于某个重复标签组的"标签"型行,以及附属的表格分隔行。
仅剔除整行的首列归一化结果命中某个 block 的标签集(子集判定);
含正文的长行不会因子串匹配被误删。
紧接着被剔除的标签行的分隔行(如 ``| - | - | - |``)与紧随其后的空行也会被删除,
避免残留"裸表格"格式。
"""
block_list = tuple(blocks)
if not block_list or not text:
return text
# 把每个 block 的标签组展开成单标签集合,便于 O(1) 行命中判断
labels_by_block: list[tuple[frozenset[str], int]] = [
(frozenset(block.labels), block.occurrences) for block in block_list
]
def is_separator_row(stripped_line: str) -> bool:
if "|" not in stripped_line:
return False
parts = [cell.strip() for cell in stripped_line.strip("|").split("|")]
if not parts:
return False
return all(re.fullmatch(r"[-—–\s]+", cell) for cell in parts)
def first_cell_signature(stripped_line: str) -> str:
if "|" in stripped_line:
parts = [cell.strip() for cell in stripped_line.strip("|").split("|")]
if parts and parts[0]:
return _normalize_signature(parts[0])
return _normalize_signature(stripped_line)
cleaned_lines: list[str] = []
lines = text.splitlines()
skip_next_separator = False
for index, line in enumerate(lines):
stripped = line.strip()
if not stripped:
cleaned_lines.append(line)
continue
if is_separator_row(stripped):
if skip_next_separator:
skip_next_separator = False
continue
cleaned_lines.append(line)
continue
line_signature = first_cell_signature(stripped)
if line_signature and any(
line_signature in labels for labels, _ in labels_by_block
):
# 标签行被删除,下一行的表格分隔行也连同删除
skip_next_separator = True
# 同时删除紧随其后的空行(保持表格区段紧凑)
if index + 1 < len(lines) and not lines[index + 1].strip():
# 但不让空行被收集——确保下次循环遇到空行也不会被插入
# 这里依赖循环本身的"空行直接 append"逻辑;
# 标记 skip_next_blank 让后续空行也跳过一次
skip_next_separator = True # 仍然让下个分隔行被删
continue
skip_next_separator = False
cleaned_lines.append(line)
return "\n".join(cleaned_lines)

View File

@@ -2,9 +2,10 @@
from __future__ import annotations from __future__ import annotations
import os import logging
import re import re
import threading import threading
import time
import unicodedata import unicodedata
from dataclasses import dataclass from dataclasses import dataclass
from functools import lru_cache from functools import lru_cache
@@ -34,6 +35,8 @@ _LIST_MARKER_PREFIX = re.compile(
_COMPACT_CHARACTER = re.compile(r"[\w\u3400-\u4dbf\u4e00-\u9fff]", re.UNICODE) _COMPACT_CHARACTER = re.compile(r"[\w\u3400-\u4dbf\u4e00-\u9fff]", re.UNICODE)
_CONVERTER_LOCK = threading.Lock() _CONVERTER_LOCK = threading.Lock()
logger = logging.getLogger(__name__)
@dataclass(frozen=True, slots=True) @dataclass(frozen=True, slots=True)
class DocumentChunk: class DocumentChunk:
@@ -65,25 +68,24 @@ def _sentence_chunks(text: str) -> list[str]:
@lru_cache(maxsize=1) @lru_cache(maxsize=1)
def _tokenizer() -> tiktoken.Encoding: def _tokenizer() -> tiktoken.Encoding:
"""加载 cl100k_base 编码器,优先在线下载,失败时使用本地缓存以支持离线环境。""" """加载 cl100k_base 编码器,优先在线下载,失败时使用本地缓存以支持离线环境。"""
import os
import base64 import base64
# 先设置缓存目录环境变量 from app.core.cache_paths import tiktoken_cache_dir
offline_cache = os.path.expanduser("~/.cache/tiktoken")
os.environ.setdefault("TIKTOKEN_CACHE_DIR", offline_cache) # 缓存目录已在 app.core.cache_paths.setup_local_caches 中统一指向 <repo>/.cache/tiktoken
# 此处直接读取TIKTOKEN_CACHE_DIR 已在启动阶段写入。
offline_cache = tiktoken_cache_dir()
try: try:
# 尝试标准方式加载 # 尝试标准方式加载(环境变量 TIKTOKEN_CACHE_DIR 已被统一设置)
return tiktoken.get_encoding("cl100k_base") return tiktoken.get_encoding("cl100k_base")
except Exception: except Exception:
# 如果失败,尝试手动从本地文件构造 # 如果失败,尝试手动从本地文件构造
try: try:
from pathlib import Path local_file = offline_cache / "9b5ad71b2ce5302211f9c61530b329a4922fc6a4"
local_file = Path(offline_cache) / "9b5ad71b2ce5302211f9c61530b329a4922fc6a4"
if not local_file.exists(): if not local_file.exists():
# 尝试另一个可能的文件名 # 尝试另一个可能的文件名
local_file = Path(offline_cache) / "cl100k_base.tiktoken" local_file = offline_cache / "cl100k_base.tiktoken"
if local_file.exists(): if local_file.exists():
# 读取 BPE 文件内容 # 读取 BPE 文件内容
@@ -106,7 +108,7 @@ def _tokenizer() -> tiktoken.Encoding:
pat_str=r"""'(?i:[sdmt]|ll|ve|re)|[^\r\n\p{L}\p{N}]?+\p{L}+|\p{N}{1,3}| ?[^\s\p{L}\p{N}]++[\r\n]*|\s*[\r\n]|\s+(?!\S)|\s+""", pat_str=r"""'(?i:[sdmt]|ll|ve|re)|[^\r\n\p{L}\p{N}]?+\p{L}+|\p{N}{1,3}| ?[^\s\p{L}\p{N}]++[\r\n]*|\s*[\r\n]|\s+(?!\S)|\s+""",
mergeable_ranks=mergeable_ranks, mergeable_ranks=mergeable_ranks,
special_tokens={ special_tokens={
"<|endoftext|>": 100257, "": 100257,
"<|fim_prefix|>": 100258, "<|fim_prefix|>": 100258,
"<|fim_middle|>": 100259, "<|fim_middle|>": 100259,
"<|fim_suffix|>": 100260, "<|fim_suffix|>": 100260,
@@ -434,6 +436,12 @@ def chunk_layout_document(
from docling_core.transforms.chunker.tokenizer.openai import OpenAITokenizer from docling_core.transforms.chunker.tokenizer.openai import OpenAITokenizer
from docling_core.types.doc import DocItemLabel from docling_core.types.doc import DocItemLabel
from app.modules.data_process.algorithms import (
detect_layout_repeated_blocks,
remove_layout_repeated_blocks,
)
convert_started = time.perf_counter()
try: try:
with _CONVERTER_LOCK: with _CONVERTER_LOCK:
conversion = _document_converter().convert( conversion = _document_converter().convert(
@@ -441,6 +449,35 @@ def chunk_layout_document(
) )
except DoclingError as exc: except DoclingError as exc:
raise ValueError(f"文档版面解析失败: {exc}") from exc raise ValueError(f"文档版面解析失败: {exc}") from exc
logger.info(
"layout chunking convert done file=%s elapsed=%.2fs",
filename,
time.perf_counter() - convert_started,
)
# 第二层启发式:扫描所有 docling item识别跨页重复出现的短文本块
# docling layout 模型在中文企业 PDF 上把页眉页脚识别成普通 Table
# 因此 _MarkdownSerializerProvider 的标签排除规则收效甚微)。
page_count = len(getattr(conversion.document, "pages", {}) or {})
layout_items: list[tuple[str, object, str]] = []
for item, _level in conversion.document.iterate_items():
text = getattr(item, "text", None)
if not text and hasattr(item, "export_to_markdown"):
try:
text = item.export_to_markdown(doc=conversion.document) or ""
except TypeError:
# 旧版 docling_core 无 doc 参数
text = item.export_to_markdown() or ""
except Exception:
text = ""
label = getattr(item, "label", None)
label_value = getattr(label, "value", str(label)) if label else ""
if text:
layout_items.append((label_value, item, text))
repeated_blocks = detect_layout_repeated_blocks(
layout_items, page_count=page_count
)
chunker = HybridChunker( chunker = HybridChunker(
tokenizer=OpenAITokenizer(tokenizer=_tokenizer(), max_tokens=chunk_size), tokenizer=OpenAITokenizer(tokenizer=_tokenizer(), max_tokens=chunk_size),
serializer_provider=_MarkdownSerializerProvider(), serializer_provider=_MarkdownSerializerProvider(),
@@ -450,6 +487,7 @@ def chunk_layout_document(
compact_source, source_offsets = _compact_with_offsets(source_text) compact_source, source_offsets = _compact_with_offsets(source_text)
compact_start = 0 compact_start = 0
result: list[DocumentChunk] = [] result: list[DocumentChunk] = []
covered_refs: set[str] = set()
excluded = { excluded = {
DocItemLabel.DOCUMENT_INDEX, DocItemLabel.DOCUMENT_INDEX,
DocItemLabel.PAGE_HEADER, DocItemLabel.PAGE_HEADER,
@@ -463,6 +501,13 @@ def chunk_layout_document(
if not content: if not content:
continue continue
contextualized = _clean_layout_text(chunker.contextualize(raw_chunk)) or content contextualized = _clean_layout_text(chunker.contextualize(raw_chunk)) or content
if repeated_blocks:
content = remove_layout_repeated_blocks(content, repeated_blocks)
contextualized = remove_layout_repeated_blocks(
contextualized, repeated_blocks
)
if not content:
continue
start, end, compact_start = _project_layout_span( start, end, compact_start = _project_layout_span(
source_text, source_text,
content, content,
@@ -476,6 +521,7 @@ def chunk_layout_document(
bboxes: list[dict[str, Any]] = [] bboxes: list[dict[str, Any]] = []
for item in doc_items: for item in doc_items:
refs.append(str(item.self_ref)) refs.append(str(item.self_ref))
covered_refs.add(str(item.self_ref))
for provenance in item.prov or (): for provenance in item.prov or ():
pages.add(int(provenance.page_no)) pages.add(int(provenance.page_no))
bbox = provenance.bbox bbox = provenance.bbox
@@ -508,6 +554,66 @@ def chunk_layout_document(
source_bboxes=tuple(bboxes), source_bboxes=tuple(bboxes),
) )
) )
# HybridChunker(merge_peers=True) 会丢弃"末尾无正文的孤立标题"。
# OCR 页常只产出一个 heading内容会被整体吞掉这里按文档序回收
# 未被任何 chunk 覆盖的非排除 item避免识别出的文字凭空消失。
# 注意 heading 会进入 meta.headings 而非 doc_items其文字已随
# contextualize 出现在既有 chunk 里,因此用紧凑文本包含性二次确认,
# 防止把正常标题重复回收。
chunk_haystack = _compact_with_offsets(
"\n".join(chunk.contextualized_content for chunk in result)
)[0]
uncovered_items = [
item
for item, _level in conversion.document.iterate_items()
if item.label not in excluded
and str(item.self_ref) not in covered_refs
and (getattr(item, "text", None) or "").strip()
and _compact_with_offsets(str(item.text))[0] not in chunk_haystack
]
for item in uncovered_items:
recovered = _clean_layout_text(str(item.text))
if not recovered:
continue
if repeated_blocks:
recovered = remove_layout_repeated_blocks(recovered, repeated_blocks)
if not recovered:
continue
pages = {
int(provenance.page_no) for provenance in item.prov or ()
}
bboxes = [
{
"page": int(provenance.page_no),
"left": float(provenance.bbox.l),
"top": float(provenance.bbox.t),
"right": float(provenance.bbox.r),
"bottom": float(provenance.bbox.b),
"origin": str(provenance.bbox.coord_origin.value),
}
for provenance in item.prov or ()
]
logger.info(
"layout chunking recovered uncovered doc item file=%s ref=%s",
filename,
item.self_ref,
)
result.append(
DocumentChunk(
original_content=recovered,
contextualized_content=recovered,
source_start=None,
source_end=None,
source_start_line=None,
source_end_line=None,
token_count=len(_tokenizer().encode(recovered)),
heading_path=(),
source_pages=tuple(sorted(pages)),
doc_item_refs=(str(item.self_ref),),
source_bboxes=tuple(bboxes),
)
)
return result return result

View File

@@ -18,10 +18,12 @@ from pypdf import PdfWriter
from app.modules.data_process.algorithms import ( from app.modules.data_process.algorithms import (
PdfPageText, PdfPageText,
LayoutRepeatedBlock,
content_quality_flags, content_quality_flags,
desensitize_pii, desensitize_pii,
desensitize_structured_record, desensitize_structured_record,
detect_document_structure, detect_document_structure,
detect_layout_repeated_blocks,
detect_pdf_document_noise, detect_pdf_document_noise,
detect_text_format, detect_text_format,
extract_pdf_page_texts, extract_pdf_page_texts,
@@ -35,6 +37,7 @@ from app.modules.data_process.algorithms import (
preprocess_structured_records_with_lineage, preprocess_structured_records_with_lineage,
record_fingerprint, record_fingerprint,
remove_document_noise, remove_document_noise,
remove_layout_repeated_blocks,
score_quality, score_quality,
stable_split, stable_split,
stable_split_assignments, stable_split_assignments,
@@ -1143,3 +1146,95 @@ def test_generate_standard_records_rejects_out_of_range_count(
) -> None: ) -> None:
with pytest.raises(ValueError, match=r"\[1, 50\]"): with pytest.raises(ValueError, match=r"\[1, 50\]"):
generate_standard_records([], qa_pairs_per_item=qa_pairs_per_item) generate_standard_records([], qa_pairs_per_item=qa_pairs_per_item)
def test_layout_repeated_blocks_detects_repeating_header_table() -> None:
"""跨页重复的页眉表格应被识别为重复块(出现 ≥ max(3, ceil(pages*0.3)) 次)。"""
header_table = (
"| 文件编码 | 2024 |\n"
"| - | - |\n"
"| 秘密等级 | 商密【中】 |\n"
"| 现行版本 | 1.0 |\n"
"| 页次 | 第1页 共47页 |\n"
)
body_table = (
"| 支出项目 | 税务票据要求 |\n"
"| - | - |\n"
"| 工资奖金 | 无 |\n"
"| 交通费 | 车票 |\n"
)
doc_items: list[tuple[str, object, str]] = []
for index in range(20):
# 20 个页面里 18 个有页眉表2 个有正文表
text = header_table if index < 18 else body_table
doc_items.append(("table", index, text))
blocks = detect_layout_repeated_blocks(doc_items, page_count=20)
# 仅页眉表对应的标签序列应被识别
assert len(blocks) == 1
assert "文件编码" in blocks[0].labels
assert "秘密等级" in blocks[0].labels
assert blocks[0].occurrences == 18
def test_layout_repeated_blocks_short_documents_skip() -> None:
"""短文档(< 3 页)不推断重复块。"""
doc_items: list[tuple[str, object, str]] = [
("table", 0, "| 文件编码 | 2024 |\n| - | - |\n"),
("table", 1, "| 文件编码 | 2024 |\n| - | - |\n"),
]
assert detect_layout_repeated_blocks(doc_items, page_count=2) == ()
def test_remove_layout_repeated_blocks_strips_label_rows_and_separators() -> None:
"""剔除首列命中重复标签集的行,及其后的表格分隔行。"""
blocks = [
LayoutRepeatedBlock(
labels=("文件编码", "秘密等级", "现行版本", "页次"),
occurrences=18,
),
]
chunk = (
"报销指引\n"
"| 文件编码 | 2024 |\n"
"| - | - |\n"
"| 秘密等级 | 商密【中】 |\n"
"| 现行版本 | 1.0 |\n"
"| 页次 | 第3页 共47页 |\n"
"正文第一段\n"
"| 支出项目 | 税务票据要求 |\n"
"| - | - |\n"
"| 工资奖金 | 无 |\n"
)
cleaned = remove_layout_repeated_blocks(chunk, blocks)
# 重复标签行 + 紧随其后的表格分隔行被剔除;正文与内容表格保留
assert "文件编码" not in cleaned
assert "秘密等级" not in cleaned
assert "现行版本" not in cleaned
assert "页次" not in cleaned
# 第一组表格的 | - | - | 在 文件编码 行之后被一并删除
# (但 cleaned 中可能还有第二个表格的分隔行)
assert cleaned.count("| - | - |") == 1
assert "报销指引" in cleaned
assert "正文第一段" in cleaned
assert "支出项目" in cleaned
assert "工资奖金" in cleaned
def test_remove_layout_repeated_blocks_returns_text_unchanged_when_no_blocks() -> None:
"""无重复块时直接返回原文。"""
chunk = "| 文件编码 | 2024 |\n| 正文 |\n"
assert remove_layout_repeated_blocks(chunk, []) == chunk
assert (
remove_layout_repeated_blocks(
"",
[LayoutRepeatedBlock(labels=("x",), occurrences=5)],
)
== ""
)

View File

@@ -336,6 +336,7 @@ def build_command(config: dict[str, Any], llama_factory_home: str = "/app/LLaMA-
_optional_arg(config, command, "--val_size", "val_size") _optional_arg(config, command, "--val_size", "val_size")
_optional_arg(config, command, "--max_samples", "max_samples") _optional_arg(config, command, "--max_samples", "max_samples")
_optional_arg(config, command, "--preprocessing_num_workers", "preprocessing_num_workers") _optional_arg(config, command, "--preprocessing_num_workers", "preprocessing_num_workers")
_optional_arg(config, command, "--resume_from_checkpoint", "resume_from_checkpoint")
_optional_bool_arg(config, command, "--fp16", "fp16") _optional_bool_arg(config, command, "--fp16", "fp16")
_optional_bool_arg(config, command, "--bf16", "bf16") _optional_bool_arg(config, command, "--bf16", "bf16")
quantization_bit = int(config.get("quantization_bit", 0) or 0) quantization_bit = int(config.get("quantization_bit", 0) or 0)

View File

@@ -170,6 +170,30 @@ def _metric_record(score: float | None, sample_count: int, error: str = "", avai
} }
def _metric_dimension_summary(metrics: dict[str, Any]) -> list[dict[str, Any]]:
labels = {
"bleu": "BLEU",
"rouge": "ROUGE-L",
"cosine": "Cosine 相似度",
"exact_match": "精确匹配",
"text_similarity": "文本相似度",
}
result: list[dict[str, Any]] = []
for name, item in metrics.items():
if not isinstance(item, dict) or item.get("score") is None:
continue
result.append({
"name": labels.get(name, name),
"score": float(item.get("score") or 0),
"max_score": float(item.get("max_score") or 100),
"pass_rate": float(item.get("score") or 0),
"sample_count": int(item.get("sample_count") or 0),
"available": item.get("available", True),
"error": item.get("error", ""),
})
return result
def _rouge_tokens(text: str) -> str: def _rouge_tokens(text: str) -> str:
text = str(text or "").strip().lower() text = str(text or "").strip().lower()
tokens: list[str] = [] tokens: list[str] = []
@@ -650,7 +674,10 @@ def run_eval(config: dict[str, Any]) -> dict[str, Any]:
"score": overall_score, "score": overall_score,
"max_score": 100, "max_score": 100,
"pass_rate": round(passed_count / max(completed, 1) * 100, 1), "pass_rate": round(passed_count / max(completed, 1) * 100, 1),
}] "sample_count": completed,
"available": bool(scored),
"error": "部分样本未返回可解析评分" if len(scored) < completed else "",
}] + _metric_dimension_summary(metrics_result)
overall_evaluation = f"评测完成:{completed} 样本,{passed_count} 通过,平均 {avg_score}/100 分" overall_evaluation = f"评测完成:{completed} 样本,{passed_count} 通过,平均 {avg_score}/100 分"
else: else:
passed_count = 0 passed_count = 0
@@ -661,16 +688,7 @@ def run_eval(config: dict[str, Any]) -> dict[str, Any]:
] ]
overall_score = round(sum(enabled_scores) / len(enabled_scores), output_precision) if enabled_scores else 0 overall_score = round(sum(enabled_scores) / len(enabled_scores), output_precision) if enabled_scores else 0
overall_score_max = 100 overall_score_max = 100
dimension_summary = [ dimension_summary = _metric_dimension_summary(metrics_result)
{
"name": name,
"score": float(item.get("score") or 0),
"max_score": 100,
"pass_rate": float(item.get("score") or 0),
}
for name, item in metrics_result.items()
if isinstance(item, dict) and item.get("enabled", True) and item.get("score") is not None
]
overall_evaluation = f"评测完成:{completed} 样本(未配置 LLM 评委)" overall_evaluation = f"评测完成:{completed} 样本(未配置 LLM 评委)"
result = { result = {

View File

@@ -25,6 +25,22 @@ def test_build_command_uses_explicit_validation_dataset_without_resplitting() ->
assert "--val_size" not in result.command assert "--val_size" not in result.command
def test_build_command_resumes_from_checkpoint() -> None:
result = build_command(
{
"base_model": "/models/qwen",
"dataset": "ygft_dataset_train",
"dataset_dir": "/datasets/example",
"output_dir": "/outputs/example",
"resume_from_checkpoint": "/data/yg-ft/fine-tunes/ft_1/resume/checkpoint-50",
}
)
assert result.command[result.command.index("--resume_from_checkpoint") + 1] == (
"/data/yg-ft/fine-tunes/ft_1/resume/checkpoint-50"
)
def _write(tmp_path, name: str, lines: list[dict]) -> object: def _write(tmp_path, name: str, lines: list[dict]) -> object:
path = tmp_path / name path = tmp_path / name
path.write_text( path.write_text(

View File

@@ -14,11 +14,12 @@ RUN pip install --upgrade pip -i https://pypi.tuna.tsinghua.edu.cn/simple \
RUN python -c "import fastapi, uvicorn, psycopg, psycopg_pool, sqlalchemy, redis, jwt, passlib, httpx, minio, alembic; print('backend dependency check ok')" RUN python -c "import fastapi, uvicorn, psycopg, psycopg_pool, sqlalchemy, redis, jwt, passlib, httpx, minio, alembic; print('backend dependency check ok')"
RUN mkdir -p /opt/yg-ft/logs/backend /data/yg-ft \ RUN mkdir -p /opt/yg-ft/logs/backend /data/yg-ft /opt/tiktoken_cache \
&& chmod -R 0775 /opt/yg-ft /data/yg-ft && chmod -R 0775 /opt/yg-ft /data/yg-ft /opt/tiktoken_cache
# 离线打包 tiktoken cl100k_base 词表,避免无网环境下运行时联网下载 # 离线打包 tiktoken cl100k_base 词表(与运行时 TIKTOKEN_CACHE_DIR 对齐),
COPY docker/app/tiktoken /opt/tiktoken_cache # 文件名采用官方 SHA避免代码走 fallback 重建 Encoding 的分支。
COPY docker/app/tiktoken/9b5ad71b2ce5302211f9c61530b329a4922fc6a4 /opt/tiktoken_cache/
EXPOSE 8000 EXPOSE 8000

View File

@@ -49,6 +49,15 @@ export interface FineTuneGpuStatus {
error?: string error?: string
} }
export interface FineTuneCheckpoint {
id: string
step: number
name: string
path: string
size_bytes?: number
create_time?: string
}
/** 训练任务列表 */ /** 训练任务列表 */
export const getFineTuneList = () => get<FineTuneTask[]>('/fine-tune') export const getFineTuneList = () => get<FineTuneTask[]>('/fine-tune')
@@ -89,6 +98,13 @@ export const updateFineTune = (id: string | number, data: Partial<FineTuneTask>)
/** 停止训练任务 */ /** 停止训练任务 */
export const stopFineTune = (id: string | number) => post(`/fine-tune/stop/${id}`) export const stopFineTune = (id: string | number) => post(`/fine-tune/stop/${id}`)
/** 从最新 checkpoint 继续训练 */
export const resumeFineTune = (id: string | number) => post(`/fine-tune/${id}/resume`)
/** 获取训练断点 */
export const getFineTuneCheckpoints = (id: string | number) =>
get<FineTuneCheckpoint[]>(`/fine-tune/${id}/checkpoints`)
/** 删除训练任务 */ /** 删除训练任务 */
export const deleteFineTune = (id: string | number) => del(`/fine-tune/${id}`) export const deleteFineTune = (id: string | number) => del(`/fine-tune/${id}`)

View File

@@ -56,7 +56,19 @@ const overallScore = computed(() => formatScore(detail.value?.overall_score, det
const displayModelName = computed(() => detail.value?.model_name || String(detail.value?.model_id || '-')) const displayModelName = computed(() => detail.value?.model_name || String(detail.value?.model_id || '-'))
const displayMetric = computed(() => detail.value?.metric_label || detail.value?.metric || '-') const displayMetric = computed(() => detail.value?.metric_label || detail.value?.metric || '-')
const progressDetail = computed(() => detail.value?.progress_detail) const progressDetail = computed(() => detail.value?.progress_detail)
const progressPercentage = computed(() => Math.max(0, Math.min(100, Math.round(Number(progressDetail.value?.percentage ?? completionRate.value))))) const progressTotal = computed(() => Math.max(
Number(detail.value?.sample_count || 0),
Number(progressDetail.value?.total || 0),
Number(detail.value?.samples?.length || 0),
))
const progressCompleted = computed(() => detail.value?.status === 'completed'
? progressTotal.value
: Math.min(progressTotal.value || Number(detail.value?.completed_count || 0), Number(progressDetail.value?.completed ?? detail.value?.completed_count ?? 0)))
const progressPercentage = computed(() => detail.value?.status === 'completed'
? 100
: detail.value?.status === 'failed' || detail.value?.status === 'stopped'
? Math.max(0, Math.min(100, Math.round(Number(progressDetail.value?.percentage ?? detail.value?.progress ?? completionRate.value))))
: Math.max(0, Math.min(100, Math.round(Number(progressDetail.value?.percentage ?? completionRate.value)))))
const progressStage = computed(() => ({ const progressStage = computed(() => ({
dataset: '准备数据集', dataset: '准备数据集',
model_loading: '加载模型', model_loading: '加载模型',
@@ -64,14 +76,32 @@ const progressStage = computed(() => ({
metrics: '计算指标', metrics: '计算指标',
completed: '评测完成', completed: '评测完成',
failed: '评测失败', failed: '评测失败',
}[String(progressDetail.value?.stage || '')] || (detail.value?.status === 'running' ? '任务运行中' : '等待开始'))) }[detail.value?.status === 'completed' ? 'completed' : String(progressDetail.value?.stage || '')] || (detail.value?.status === 'running' ? '任务运行中' : '等待开始')))
const radarDimensions = computed(() => (detail.value?.dimension_summary || []) const radarDimensions = computed(() => {
.filter((item) => item.available !== false && Number.isFinite(Number(item.score))) const dimensions = new Map<string, { name: string; value: number }>()
.map((item) => ({ for (const item of detail.value?.dimension_summary || []) {
if (item.available === false || !Number.isFinite(Number(item.score))) continue
dimensions.set(item.name, {
name: item.name, name: item.name,
value: Math.max(0, Math.min(100, Number(item.score) / Math.max(Number(item.max_score) || 100, 1) * 100)), value: Math.max(0, Math.min(100, Number(item.score) / Math.max(Number(item.max_score) || 100, 1) * 100)),
}))) })
}
const metricLabels: Record<string, string> = {
bleu: 'BLEU',
rouge: 'ROUGE-L',
cosine: 'Cosine 相似度',
exact_match: '精确匹配',
text_similarity: '文本相似度',
}
for (const [key, metric] of Object.entries(detail.value?.basic_metrics || {})) {
const score = Number(metric?.score)
if (!Number.isFinite(score) || metric?.available === false) continue
const name = metricLabels[key] || key
if (!dimensions.has(name)) dimensions.set(name, { name, value: Math.max(0, Math.min(100, score)) })
}
return [...dimensions.values()]
})
const radarOption = computed<EChartsOption>(() => ({ const radarOption = computed<EChartsOption>(() => ({
tooltip: { trigger: 'item' }, tooltip: { trigger: 'item' },
@@ -197,7 +227,7 @@ onUnmounted(stopPolling)
</div> </div>
<div class="overview-item"> <div class="overview-item">
<span>评测进度</span> <span>评测进度</span>
<strong>{{ progressDetail?.completed ?? detail.completed_count }} / {{ progressDetail?.total ?? detail.sample_count }}</strong> <strong>{{ progressCompleted }} / {{ progressTotal }}</strong>
<el-progress :percentage="progressPercentage" :show-text="false" :stroke-width="5" /> <el-progress :percentage="progressPercentage" :show-text="false" :stroke-width="5" />
<small>{{ progressStage }}{{ progressDetail?.message ? ' · ' + progressDetail.message : '' }}</small> <small>{{ progressStage }}{{ progressDetail?.message ? ' · ' + progressDetail.message : '' }}</small>
</div> </div>

View File

@@ -65,6 +65,20 @@ function displayMetric(row: Partial<EvalTask>) {
return row.metric_label || row.metric || '-' return row.metric_label || row.metric || '-'
} }
function progressPercentage(row: Partial<EvalTask>) {
if (row.status === 'completed') return 100
return Math.max(0, Math.min(100, Math.round(Number(row.progress_detail?.percentage ?? row.progress ?? 0))))
}
function progressCompleted(row: Partial<EvalTask>) {
if (row.status === 'completed') return row.progress_detail?.total ?? '-'
return row.progress_detail?.completed ?? 0
}
function progressTotal(row: Partial<EvalTask>) {
return row.progress_detail?.total ?? '-'
}
const { start: startPolling, stop: stopPolling } = usePolling( const { start: startPolling, stop: stopPolling } = usePolling(
async () => { async () => {
await loadEvalList({ silent: true }) await loadEvalList({ silent: true })
@@ -125,18 +139,15 @@ onUnmounted(() => {
</el-tooltip> </el-tooltip>
</template> </template>
</el-table-column> </el-table-column>
<el-table-column label="评分" prop="score" width="100" align="center" /> <el-table-column label="评分" prop="score" width="100" align="center">
<template #default="{ row }">
{{ row.score == null ? '-' : `${Number(row.score).toFixed(2)} / 100` }}
</template>
</el-table-column>
<el-table-column label="评测进度" width="130" align="center"> <el-table-column label="评测进度" width="130" align="center">
<template #default="{ row }"> <template #default="{ row }">
<template v-if="ACTIVE_STATUSES.has(String(row.status || ''))"> <el-progress :percentage="progressPercentage(row)" :stroke-width="6" :show-text="false" />
<el-progress <small>{{ progressCompleted(row) }} / {{ progressTotal(row) }}</small>
:percentage="Math.max(0, Math.min(100, Math.round(Number(row.progress_detail?.percentage ?? row.progress ?? 0))))"
:stroke-width="6"
:show-text="false"
/>
<small>{{ row.progress_detail?.completed ?? 0 }} / {{ row.progress_detail?.total ?? '-' }}</small>
</template>
<span v-else>{{ row.score == null ? '-' : Number(row.score).toFixed(2) + ' / 100' }}</span>
</template> </template>
</el-table-column> </el-table-column>
<el-table-column label="状态" width="100" align="center"> <el-table-column label="状态" width="100" align="center">

View File

@@ -1,7 +1,7 @@
<script setup lang="ts"> <script setup lang="ts">
import { ref, computed, onMounted } from 'vue' import { ref, computed, onMounted } from 'vue'
import { useRouter } from 'vue-router' import { useRouter } from 'vue-router'
import { ElMessage } from 'element-plus' import { ElMessage, ElMessageBox } from 'element-plus'
import DataTablePage from '@/components/DataTablePage.vue' import DataTablePage from '@/components/DataTablePage.vue'
import ModelStatusTag from '@/components/ModelStatusTag.vue' import ModelStatusTag from '@/components/ModelStatusTag.vue'
import { usePolling } from '@/composables/usePolling' import { usePolling } from '@/composables/usePolling'
@@ -10,6 +10,7 @@ import {
getFineTuneList, getFineTuneList,
deleteFineTune, deleteFineTune,
stopFineTune, stopFineTune,
resumeFineTune,
getFineTuneProgress, getFineTuneProgress,
getFineTune, getFineTune,
} from '@/api/modules/fineTune' } from '@/api/modules/fineTune'
@@ -87,9 +88,29 @@ async function handleDelete(row: any) {
} }
async function handleStop(row: any) { async function handleStop(row: any) {
await ElMessageBox.confirm(
'停止后将保留已生成的 checkpoint可在资源可用时继续训练。是否停止当前任务',
'停止训练',
{ type: 'warning' },
)
await stopFineTune(row.id) await stopFineTune(row.id)
ElMessage.success('训练任务已停止') ElMessage.success('训练任务已停止,可继续训练')
loadData() await loadData()
}
async function handleResume(row: any) {
await ElMessageBox.confirm(
'系统将使用最新 checkpoint 继续训练,并重新检查基座模型、数据集、算力节点和 GPU 是否可用。是否继续?',
'继续训练',
{ type: 'info' },
)
try {
await resumeFineTune(row.id)
ElMessage.success('继续训练已提交')
await loadData()
} catch {
// 请求拦截器已展示后端返回的具体预检原因
}
} }
function viewLog(row: any) { function viewLog(row: any) {
@@ -217,7 +238,7 @@ onMounted(async () => {
<template #actions="{ row }"> <template #actions="{ row }">
<div class="action-buttons"> <div class="action-buttons">
<el-button <el-button
v-if="row.status === 'running'" v-if="['syncing', 'queued', 'running'].includes(row.status)"
type="warning" type="warning"
link link
size="small" size="small"
@@ -225,6 +246,15 @@ onMounted(async () => {
> >
<i class="fa fa-stop-circle-o" style="margin-right: 4px" /> 停止 <i class="fa fa-stop-circle-o" style="margin-right: 4px" /> 停止
</el-button> </el-button>
<el-button
v-if="row.status === 'stopped' || row.status === 'failed'"
type="success"
link
size="small"
@click="handleResume(row)"
>
<i class="fa fa-play-circle-o" style="margin-right: 4px" /> 继续
</el-button>
<el-button type="primary" link size="small" @click="viewLog(row)"> <el-button type="primary" link size="small" @click="viewLog(row)">
<i class="fa fa-file-text-o" style="margin-right: 4px" /> 日志 <i class="fa fa-file-text-o" style="margin-right: 4px" /> 日志
</el-button> </el-button>

View File

@@ -56,8 +56,8 @@ async function handleMerge() {
compute_node_id: form.compute_node_id, compute_node_id: form.compute_node_id,
output_model_name: `${form.model_name}-merged`, output_model_name: `${form.model_name}-merged`,
}) })
ElMessage.success('合并成功') ElMessage.success('合并任务已提交,完成后会自动更新状态')
router.push('/model-manage') router.push({ path: '/model-manage', query: { tab: 'trained' } })
} catch { } catch {
// ignore // ignore
} finally { } finally {

View File

@@ -1,5 +1,5 @@
<script setup lang="ts"> <script setup lang="ts">
import { computed, onMounted, reactive, ref, watch } from 'vue' import { computed, onMounted, onUnmounted, reactive, ref, watch } from 'vue'
import { useRouter } from 'vue-router' import { useRouter } from 'vue-router'
import { ElMessage, ElMessageBox } from 'element-plus' import { ElMessage, ElMessageBox } from 'element-plus'
import DataTablePage from '@/components/DataTablePage.vue' import DataTablePage from '@/components/DataTablePage.vue'
@@ -20,6 +20,7 @@ import { MODEL_SOURCE_MAP, MODEL_TYPE_MAP, PURPOSE_MAP } from '@/constants'
import type { ModelItem, TrainedModel } from '@/types' import type { ModelItem, TrainedModel } from '@/types'
import { mergeStatusLabel, mergeStatusType, statusLabel, statusTagType } from '@/utils/status' import { mergeStatusLabel, mergeStatusType, statusLabel, statusTagType } from '@/utils/status'
import { useAuthStore } from '@/stores/auth' import { useAuthStore } from '@/stores/auth'
import { usePolling } from '@/composables/usePolling'
const router = useRouter() const router = useRouter()
const auth = useAuthStore() const auth = useAuthStore()
@@ -35,7 +36,7 @@ type TrainedModelRuntime = {
loaded: boolean loaded: boolean
} }
const activeTab = ref<TabKey>('config') const activeTab = ref<TabKey>(router.currentRoute.value.query.tab === 'trained' ? 'trained' : 'config')
const loading = ref(false) const loading = ref(false)
const configList = ref<ModelItem[]>([]) const configList = ref<ModelItem[]>([])
const trainedList = ref<TrainedModel[]>([]) const trainedList = ref<TrainedModel[]>([])
@@ -131,19 +132,36 @@ async function handleExport(row: TrainedModel) {
} }
} }
async function loadTrained() { async function loadTrained(silent = false) {
loading.value = true if (!silent) loading.value = true
try { try {
const res = await getTrainedModels() const res = await getTrainedModels()
trainedList.value = res?.models || [] trainedList.value = res?.models || []
} finally { } finally {
loading.value = false if (!silent) loading.value = false
} }
} }
function loadData() { function hasActiveMerge() {
return trainedList.value.some((item) => Boolean(item.merging))
}
const { start: startTrainedPolling, stop: stopTrainedPolling } = usePolling(
async () => {
await loadTrained(true)
if (!hasActiveMerge()) stopTrainedPolling()
},
3000,
{ immediate: false },
)
async function loadData() {
if (activeTab.value === 'config') loadConfig() if (activeTab.value === 'config') loadConfig()
else loadTrained() else {
await loadTrained()
if (hasActiveMerge()) startTrainedPolling()
else stopTrainedPolling()
}
} }
async function loadTrainedRuntime(row: TrainedModel, force = false) { async function loadTrainedRuntime(row: TrainedModel, force = false) {
@@ -217,9 +235,16 @@ function handleRefresh() {
else loadTrained() else loadTrained()
} }
watch(activeTab, loadData) watch(activeTab, () => {
stopTrainedPolling()
void loadData()
})
onMounted(loadData) onMounted(() => {
void loadData()
})
onUnmounted(stopTrainedPolling)
</script> </script>
<template> <template>