YG-Soft/YG_FT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge: 合并远程 ft_wyt 分支，解决冲突

Browse Source
This commit is contained in:
wangjiming
2026-08-19 17:39:18 +08:00
parent 78680a5c5c 080ef6ab00
commit 8455d99d49
64 changed files with 4616 additions and 375 deletions
Download Patch File Download Diff File Expand all files Collapse all files

301
backend/app/api/v1/endpoints/data_process.py
View File

@@ -35,7 +35,12 @@ from fastapi.responses import StreamingResponse
from psycopg.rows import dict_row
from app.core.auth import filter_accessible_resource_ids, get_current_user, is_admin
from app.core.logging import get_structured_logger
from app.core.config import get_settings
from app.db.platform_store import get_platform_store
from app.modules.data_process.algorithms import (
ParsedText,
canonical_record_json,
@@ -61,6 +66,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,
@@ -82,6 +91,8 @@ from app.modules.data_process.store import (
new_id,
repeat_task_id,
)
from app.modules.storage.minio_store import get_object_storage
from app.modules.storage.policy import should_store_in_minio
from app.schemas.data_process import (
DataProcessRegenerateRequest,
DataProcessRepeatRequest,
@@ -97,6 +108,7 @@ from app.schemas.data_process import (
PreviewItemUpdate,
ProcessType,
PublishRequest,
ResultBatchEvaluateRequest,
ResultBatchRegenerateRequest,
ResultRegenerateRequest,
ResultUpdate,
@@ -221,9 +233,38 @@ def _commit_source_batch(
staged: list[StagedSourceObject],
) -> list[dict[str, Any]]:
storage.publish(staged)
storage_object_ids: list[str] = []
try:
# The source reference remains in the task schema for compatibility,
# while storage_objects provides the authoritative MinIO index.
if get_settings().minio_enabled:
for item in prepared:
reference = str(item.get("storage_object_id") or "")
if not reference.startswith("minio://"):
continue
object_key = storage.object_key(reference)
metadata = get_object_storage().stat(object_key)
object_row = get_platform_store().create_storage_object({
"resource_type": "data_process_source",
"resource_id": task_id,
"version_id": str(item.get("id") or new_id("dpsf")),
"bucket": get_object_storage().bucket,
"object_key": object_key,
"file_name": item.get("name"),
"content_type": (item.get("metadata") or {}).get("content_type", "application/octet-stream"),
"byte_size": metadata.get("byte_size") or item.get("raw_size") or 0,
"checksum_sha256": item.get("checksum_sha256"),
"status": "available",
"created_by": item.get("created_by"),
})
storage_object_ids.append(str(object_row["id"]))
return store.add_source_files(task_id, prepared)
except Exception:
for object_id in storage_object_ids:
try:
get_platform_store().update_storage_object(object_id, {"status": "deleted"})
except Exception:
pass
for item in staged:
try:
storage.delete(item.reference)
@@ -771,6 +812,25 @@ def _run_generation(
duplicate_count=duplicate_count,
error_count=error_count,
)
result_bytes = "".join(
structured_json_dumps(item) + "\n" for item in accepted
).encode("utf-8")
if should_store_in_minio(len(result_bytes)):
result_key = f"data-process/{task_id}/results/{generation_run_id}.jsonl"
uploaded = get_object_storage().put_bytes(result_key, result_bytes, "application/jsonl")
get_platform_store().create_storage_object({
"resource_type": "data_process_result",
"resource_id": task_id,
"version_id": generation_run_id,
"bucket": uploaded["bucket"],
"object_key": result_key,
"file_name": f"{generation_run_id}.jsonl",
"content_type": "application/jsonl",
"byte_size": len(result_bytes),
"checksum_sha256": hashlib.sha256(result_bytes).hexdigest(),
"status": "available",
"created_by": (store.get_task(task_id) or {}).get("created_by"),
})
logger.info(
"data process generation completed task_id=%s generation_run_id=%s "
"output_count=%s filtered_count=%s duplicate_count=%s error_count=%s "
@@ -930,7 +990,7 @@ def _repeat_file_copies(
source = store.get_source_file(source_task_id, old_file_id, include_content=True)
new_file_id = new_id("dpsf")
old_reference = str(source.get("storage_object_id") or "")
if old_reference.startswith("local://data-process/"):
if old_reference.startswith(("local://data-process/", "minio://data-process/")):
staged_object = storage.stage_copy(
batch_id=batch_id,
source_reference=old_reference,
@@ -2044,12 +2104,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,
@@ -2094,11 +2155,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,
@@ -2108,7 +2164,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"),
},
)
@@ -2271,6 +2332,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,
@@ -2444,6 +2545,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,

447
backend/app/api/v1/endpoints/platform.py
View File

@@ -23,8 +23,9 @@ from app.core.audit import audit_log, AuditActions
from app.core.op_log import op_log, OpModule, OpAction
from app.db.platform_store import get_platform_store
from app.modules.compute_gateway.client import ComputeNodeClient
from app.modules.compute_gateway.sync import fetch_eval_result_content, poll_compute_jobs_once
from app.modules.compute_gateway.sync import _archive_node_directory, fetch_eval_result_content, poll_compute_jobs_once
from app.modules.storage.minio_store import ObjectStorageError, get_object_storage
from app.modules.storage.policy import should_store_in_minio
router = APIRouter()
_LOGIN_FAILURES: dict[str, list[float]] = {}
@@ -57,6 +58,33 @@ def _select_first_online_node(store: Any) -> dict[str, Any] | None:
return None
def _normalize_gpu_indices(payload: dict[str, Any], *, allow_primary: bool = True) -> list[int]:
"""Normalize all frontend GPU selection shapes to sorted integer indexes."""
raw = payload.get("gpu_indices")
if raw is None:
raw = payload.get("gpus")
if raw is None and allow_primary and payload.get("gpu_id") is not None:
raw = [payload.get("gpu_id")]
if raw is None or raw == "":
return []
if isinstance(raw, str):
raw = [item.strip() for item in raw.split(",") if item.strip()]
if not isinstance(raw, (list, tuple, set)):
raw = [raw]
result: set[int] = set()
for item in raw:
if isinstance(item, str) and ":" in item:
item = item.rsplit(":", 1)[-1]
try:
index = int(item)
except (TypeError, ValueError) as exc:
raise ValueError(f"invalid GPU index: {item}") from exc
if index < 0:
raise ValueError("GPU index must be non-negative")
result.add(index)
return sorted(result)
async def _wait_for_object_storage() -> None:
"""Wait for MinIO before starting a resource task."""
settings = get_settings()
@@ -73,6 +101,106 @@ async def _wait_for_object_storage() -> None:
await asyncio.sleep(max(1, settings.storage_check_interval_seconds))
def _store_json_snapshot(
store: Any,
resource_type: str,
resource_id: str,
version_id: str,
object_key: str,
payload: dict[str, Any],
created_by: str | None = None,
) -> dict[str, Any]:
"""Persist non-secret task/model parameters as an auditable MinIO snapshot."""
sanitized = {
key: value
for key, value in payload.items()
if key not in {"api_key", "secret_key", "password", "token", "access_token"}
}
raw = json.dumps(sanitized, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8")
# Task/evaluation payloads are already persisted in PostgreSQL. Avoid an
# extra MinIO round trip for small non-secret parameter snapshots.
if not should_store_in_minio(len(raw), content_type="application/json", file_format="json"):
return {}
uploaded = get_object_storage().put_bytes(object_key, raw, "application/json")
return store.create_storage_object({
"resource_type": resource_type,
"resource_id": resource_id,
"version_id": version_id,
"bucket": uploaded["bucket"],
"object_key": object_key,
"file_name": Path(object_key).name,
"content_type": "application/json",
"byte_size": len(raw),
"checksum_sha256": hashlib.sha256(raw).hexdigest(),
"status": "available",
"created_by": created_by,
})
def _dataset_file_bytes(store: Any, file_id: str) -> bytes:
"""Return the canonical dataset bytes, lazily indexing legacy DB content."""
row = store.dataset_file(file_id)
if not get_settings().minio_enabled:
return str(row.get("content") or "").encode("utf-8")
storage_object = None
object_id = str(row.get("storage_object_id") or "")
if object_id:
try:
storage_object = store.storage_object(object_id)
except KeyError:
storage_object = None
if storage_object and storage_object.get("status") == "available":
try:
return get_object_storage().get_bytes(storage_object["object_key"])
except Exception as exc: # noqa: BLE001 - expose storage outage to callers
raise RuntimeError(f"dataset object is unavailable in MinIO: {exc}") from exc
# Compatibility migration for files created before MinIO was enabled.
raw = str(row.get("content") or "").encode("utf-8")
if not raw:
raise RuntimeError(f"dataset file has no MinIO object or legacy content: {file_id}")
# Small files intentionally remain database-backed. They can still be
# copied to a compute node directly when a task needs them.
if not should_store_in_minio(len(raw), file_format=row.get("file_format")):
return raw
object_key = (
f"datasets/{row['dataset_id']}/versions/"
f"{row.get('active_version_id') or row['id']}/{Path(str(row.get('name') or row['id'])).name}"
)
uploaded = get_object_storage().put_bytes(object_key, raw, "application/octet-stream")
created = store.create_storage_object({
"resource_type": "dataset",
"resource_id": str(row["dataset_id"]),
"version_id": str(row.get("active_version_id") or row["id"]),
"bucket": uploaded["bucket"],
"object_key": object_key,
"file_name": row.get("name"),
"content_type": "application/octet-stream",
"byte_size": len(raw),
"checksum_sha256": hashlib.sha256(raw).hexdigest(),
"status": "available",
})
store.link_dataset_file_storage_object(str(row["id"]), created["id"])
return raw
def _dataset_version_bytes(store: Any, file_id: str, version_id: str) -> bytes:
row = store.dataset_file(file_id)
try:
version = next(item for item in store.file_versions(file_id)["versions"] if item["id"] == version_id)
except StopIteration as exc:
raise KeyError(version_id) from exc
object_id = str(version.get("storage_object_id") or "")
if get_settings().minio_enabled and object_id:
try:
obj = store.storage_object(object_id)
return get_object_storage().get_bytes(obj["object_key"])
except KeyError:
pass
return _dataset_file_bytes(store, file_id)
def _select_eval_node(store: Any, preferred_node_id: str | None = None) -> dict[str, Any] | None:
"""Select the compute node for an eval job.
@@ -103,18 +231,63 @@ async def _prepare_resource_on_node(store: Any, resource_type: str, resource_id:
if not get_settings().minio_enabled or not resource_id:
return None
objects = store.storage_objects_for_resource(resource_type, resource_id)
if not objects and resource_type in {"model", "trained_model"}:
resource = None
if resource_type == "model":
try:
resource = store.model(resource_id)
except KeyError:
try:
resource = store.model_by_name(resource_id)
except KeyError:
resource = next((item for item in store.models() if item.get("path") == resource_id), None)
else:
resource = next(
(item for item in store.trained_models() if item.get("id") == resource_id or item.get("name") == resource_id),
None,
)
source_path = str(
(resource or {}).get("path")
or (resource or {}).get("merged_path")
or (resource or {}).get("artifact_dir")
or ""
)
resolved_id = str((resource or {}).get("id") or resource_id)
if source_path and resolved_id:
client = ComputeNodeClient(node["api_base_url"], timeout=900)
await _archive_node_directory(
store,
client,
node,
source_path,
resource_type,
resolved_id,
"legacy-import",
f"models/{resolved_id}" if resource_type == "model" else f"trained_models/{resolved_id}",
)
resource_id = resolved_id
objects = store.storage_objects_for_resource(resource_type, resource_id)
if not objects:
return None
client = ComputeNodeClient(node["api_base_url"], timeout=900)
root_name = "trained_models" if resource_type in {"trained_model", "model_artifact"} else f"{resource_type}s"
for obj in objects:
object_key = str(obj["object_key"])
marker = f"{root_name}/{resource_id}/versions/"
relative_name = Path(str(obj.get("file_name") or object_key)).name
if marker in object_key:
suffix = object_key.split(marker, 1)[1]
if "/" in suffix:
suffix = suffix.split("/", 1)[1]
if suffix:
relative_name = suffix
await client.prepare_cache({
"resource_id": resource_id,
"version_id": obj["version_id"],
"download_url": get_object_storage().presigned_get(obj["object_key"]),
"download_url": get_object_storage().presigned_get(object_key),
"checksum_sha256": obj.get("checksum_sha256") or "",
"byte_size": obj.get("byte_size") or 0,
"relative_path": f"{root_name}/{resource_id}/{Path(str(obj.get('file_name') or obj['object_key'])).name}",
"relative_path": f"{root_name}/{resource_id}/{relative_name}",
})
return f"/data/yg-ft/{root_name}/{resource_id}"
@@ -370,8 +543,23 @@ async def _submit_fine_tune_task(store: Any, payload: dict[str, Any]) -> dict[st
if not preflight["valid"]:
errors = "; ".join(preflight.get("errors") or ["preflight failed"])
raise RuntimeError(f"preflight failed: {errors}")
payload = {**payload, "compute_node_id": preflight["node"]["id"]}
prepared_job_payload = preflight.get("job_payload") or {}
payload = {
**payload,
"compute_node_id": preflight["node"]["id"],
"prepared_base_model_path": prepared_job_payload.get("model_name_or_path") or payload.get("prepared_base_model_path"),
}
task = store.start_task(payload)
if get_settings().minio_enabled:
_store_json_snapshot(
store,
"fine_tune",
str(task["id"]),
str(task["id"]),
f"training/{task['id']}/versions/{task['id']}/training-config.json",
task,
task.get("created_by"),
)
if get_settings().compute_mode == "simulator":
return task
node, job_payload = store.build_compute_job_payload(task["id"])
@@ -427,6 +615,20 @@ async def _fine_tune_preflight_with_job_payload(
if get_settings().minio_enabled and get_settings().compute_mode != "simulator":
try:
await _wait_for_object_storage()
base_model_path = str(job_payload.get("model_name_or_path") or job_payload.get("base_model") or "")
base_model_id = str(job_payload.get("base_model_id") or job_payload.get("model_id") or "")
base_model = None
if base_model_id:
try:
base_model = store.model(base_model_id)
except KeyError:
base_model = None
if base_model is None:
base_model = next((item for item in store.models() if item.get("path") == base_model_path), None)
if base_model:
prepared_model = await _prepare_resource_on_node(store, "model", str(base_model["id"]), node)
if prepared_model:
job_payload = {**job_payload, "base_model": prepared_model, "model_name_or_path": prepared_model}
except Exception as exc: # noqa: BLE001 - preflight exposes node storage failure
sync_errors.append(f"shared storage health check failed: {exc}")
if get_settings().compute_mode == "simulator":
@@ -1076,8 +1278,9 @@ async def merge_model(payload: dict[str, Any] = Body(...), current_user: dict =
@router.get("/dataset-manage/preview/{file_id}")
async def dataset_preview(file_id: str) -> dict[str, Any]:
try:
row = get_platform_store().dataset_file(file_id)
return ok({"content": row["content"]})
store = get_platform_store()
content = _dataset_file_bytes(store, file_id).decode("utf-8", errors="replace")
return ok({"content": content})
except KeyError:
raise fail(404, "dataset file not found")
@@ -1106,7 +1309,8 @@ async def dataset_version_content(file_id: str, version_id: str) -> dict[str, An
version = next((item for item in versions if item["id"] == version_id), None)
if not version:
raise KeyError(version_id)
return ok({"version": version, "content": row["content"]})
content = _dataset_version_bytes(get_platform_store(), file_id, version_id)
return ok({"version": version, "content": content.decode("utf-8", errors="replace")})
except KeyError:
raise fail(404, "dataset version not found")
@@ -1197,7 +1401,7 @@ async def _sync_training_dataset_to_compute_node(
) -> list[dict[str, Any]]:
if get_settings().minio_enabled:
files = store.training_dataset_files(dataset_id)
object_by_resource_name: dict[tuple[str, str], dict[str, Any]] = {}
object_by_resource_name: dict[tuple[str, str, str], dict[str, Any]] = {}
resource_ids = {str(dataset_id)} | {
str(item.get("dataset_id"))
for item in files
@@ -1206,35 +1410,58 @@ async def _sync_training_dataset_to_compute_node(
for resource_id in resource_ids:
for obj in store.storage_objects_for_resource("dataset", resource_id):
file_name = Path(str(obj.get("file_name") or obj.get("object_key") or "")).name
object_by_resource_name[(resource_id, file_name)] = obj
object_by_resource_name[(resource_id, file_name, str(obj.get("version_id") or ""))] = obj
results: list[dict[str, Any]] = []
client = ComputeNodeClient(node["api_base_url"])
for item in files:
target_name = Path(str(item.get("name") or f"{item['id']}.jsonl")).name
item_dataset_id = str(item.get("dataset_id") or dataset_id)
obj = object_by_resource_name.get((item_dataset_id, target_name))
version_id = str(item.get("active_version_id") or item["id"])
obj = object_by_resource_name.get((item_dataset_id, target_name, version_id))
if not obj and item.get("content"):
# 兼容 MinIO 接入前已经发布的数据处理数据集：
# 预检时用数据库正文补建对象，避免要求用户重新处理数据集。
# 兼容 MinIO 接入前已经发布的数据处理数据集。大文件补建
# MinIO 对象，小文件直接从数据库正文同步到目标节点。
raw = str(item.get("content") or "").encode("utf-8")
version_id = str(item.get("active_version_id") or item["id"])
object_key = f"datasets/{item_dataset_id}/versions/{version_id}/{target_name}"
uploaded = get_object_storage().put_bytes(object_key, raw, "application/jsonl")
obj = store.create_storage_object({
"resource_type": "dataset",
"resource_id": item_dataset_id,
"version_id": version_id,
"bucket": uploaded["bucket"],
"object_key": object_key,
"file_name": target_name,
"content_type": "application/jsonl",
"byte_size": len(raw),
"checksum_sha256": hashlib.sha256(raw).hexdigest(),
"status": "available",
})
store.link_dataset_file_storage_object(str(item["id"]), obj["id"])
if should_store_in_minio(len(raw)):
object_key = f"datasets/{item_dataset_id}/versions/{version_id}/{target_name}"
uploaded = get_object_storage().put_bytes(object_key, raw, "application/jsonl")
obj = store.create_storage_object({
"resource_type": "dataset",
"resource_id": item_dataset_id,
"version_id": version_id,
"bucket": uploaded["bucket"],
"object_key": object_key,
"file_name": target_name,
"content_type": "application/jsonl",
"byte_size": len(raw),
"checksum_sha256": hashlib.sha256(raw).hexdigest(),
"status": "available",
})
store.link_dataset_file_storage_object(str(item["id"]), obj["id"])
else:
result = await client.upload_file(
target_name,
raw,
f"datasets/{dataset_id}/{target_name}",
resource_type="dataset",
resource_id=dataset_id,
)
store.upsert_resource_replica(
node["id"], "dataset", dataset_id, str(result.get("local_path") or "")
)
results.append({
"node_id": node["id"],
"node_code": node.get("code"),
"file_id": item.get("id"),
"name": target_name,
"local_path": result.get("local_path"),
"byte_size": result.get("byte_size"),
"checksum_sha256": result.get("checksum_sha256"),
"storage_backend": "database",
})
continue
if not obj:
raise RuntimeError(f"dataset file is not available in MinIO: {target_name}")
raise RuntimeError(f"dataset file is not available: {target_name}")
url = get_object_storage().presigned_get(obj["object_key"])
result = await client.prepare_cache({
"resource_id": dataset_id,
@@ -1250,7 +1477,13 @@ async def _sync_training_dataset_to_compute_node(
dataset_id,
str(result.get("local_path") or ""),
)
results.append({**result, "file_id": item.get("id"), "name": target_name, "node_id": node["id"]})
results.append({
**result,
"file_id": item.get("id"),
"name": target_name,
"node_id": node["id"],
"storage_backend": "minio",
})
return results
if not dataset_id:
raise RuntimeError("train_dataset_id is required")
@@ -1316,7 +1549,11 @@ async def upload_dataset_files(
created_file = store.add_dataset_file(conn, dataset_id, file.filename or "upload.jsonl", content)
created.append(created_file)
pending_sync.append((created_file["id"], created_file["name"], raw))
if get_settings().minio_enabled:
if should_store_in_minio(
len(raw),
content_type=file.content_type,
file_format=Path(created_file["name"]).suffix,
):
object_key = f"datasets/{dataset_id}/versions/{created_file.get('active_version_id') or created_file['id']}/{Path(created_file['name']).name}"
uploaded = get_object_storage().put_bytes(object_key, raw, file.content_type or "application/octet-stream")
storage_object = get_platform_store().create_storage_object({
@@ -1360,7 +1597,10 @@ async def download_dataset(dataset_id: str, current_user: dict = Depends(get_cur
full_file = store.dataset_file(str(item["id"]))
except KeyError:
continue
files.append({**item, "content": full_file.get("content") or ""})
files.append({
**item,
"content": _dataset_file_bytes(store, str(item["id"])).decode("utf-8", errors="replace"),
})
if not files:
raise fail(404, "dataset has no downloadable files")
@@ -1401,8 +1641,12 @@ async def download_dataset(dataset_id: str, current_user: dict = Depends(get_cur
@router.get("/dataset-manage/download/{dataset_id}/{file_id}")
async def download_dataset_file(dataset_id: str, file_id: str, version_id: str | None = Query(default=None)) -> PlainTextResponse:
row = get_platform_store().dataset_file(file_id)
return PlainTextResponse(row["content"], media_type="text/plain")
store = get_platform_store()
row = store.dataset_file(file_id)
if str(row.get("dataset_id")) != str(dataset_id):
raise fail(404, "dataset file not found")
content = _dataset_version_bytes(store, file_id, version_id) if version_id else _dataset_file_bytes(store, file_id)
return PlainTextResponse(content.decode("utf-8", errors="replace"), media_type="text/plain")
@router.get("/dataset-manage")
@@ -1545,10 +1789,10 @@ async def start_fine_tune(
if node_id and gpu_indices:
if not store.check_gpu_access(current_user["id"], node_id, gpu_indices):
raise fail(403, "无权使用所选 GPU，请联系管理员分配")
# 记录创建者
if node_id and not gpu_indices:
payload["allowed_gpu_indices"] = store.assigned_gpu_indexes(current_user["id"], node_id)
payload["strict_node_selection"] = bool(node_id)
# 页面明确选择节点时，调度器必须保持节点约束；否则可能落到其它节点。
payload["strict_node_selection"] = bool(payload.get("compute_node_id") or payload.get("node_id"))
payload.setdefault("created_by", current_user.get("id"))
try:
return ok(await _submit_fine_tune_task(store, payload))
@@ -1679,6 +1923,42 @@ async def fine_tune_diagnostics(task_id: str) -> dict[str, Any]:
)
@router.get("/fine-tune/{task_id}/gpu-status")
async def fine_tune_gpu_status(task_id: str, current_user: dict[str, Any] = Depends(get_current_user)) -> dict[str, Any]:
"""Return live GPU metrics for the task's selected node and cards."""
store = get_platform_store()
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, "read"):
raise fail(403, "no permission to access this task")
node = _node_for_task(task)
selected = set(_normalize_gpu_indices({"gpus": task.get("gpus") or []}, allow_primary=False))
if not node:
return ok({"source": "unavailable", "items": [], "selected_gpus": sorted(selected)})
try:
if get_settings().compute_mode == "simulator":
live_items = store.gpus()
else:
live_items = await ComputeNodeClient(node["api_base_url"]).gpu_resources()
items = []
for item in live_items:
index = int(item.get("gpu_index", item.get("id", -1)))
if selected and index not in selected:
continue
items.append({
**item,
"id": index,
"node_id": node["id"],
"node_code": node.get("code"),
"node_name": node.get("name"),
})
return ok({"source": "compute", "items": items, "selected_gpus": sorted(selected)})
except Exception as exc: # noqa: BLE001 - let UI retain last good snapshot
return ok({"source": "unavailable", "items": [], "selected_gpus": sorted(selected), "error": str(exc)})
@router.put("/fine-tune/{task_id}")
@op_log(module=OpModule.FINE_TUNE, action=OpAction.UPDATE, target_type="fine_tune", target_name_param="task_id")
async def update_fine_tune(task_id: str, payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
@@ -1828,15 +2108,35 @@ async def model_eval_detail(task_id: str, current_user: dict = Depends(get_curre
async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
"""Start an evaluation task: submit eval job to compute node."""
store = get_platform_store()
try:
gpu_indices = _normalize_gpu_indices(payload)
except ValueError as exc:
raise fail(400, str(exc))
if not gpu_indices:
raise fail(400, "请选择至少一张 GPU")
# 1. Create eval task record
payload.setdefault("created_by", current_user.get("id"))
task = store.create_eval_task({**payload, "status": "pending"})
if get_settings().minio_enabled:
_store_json_snapshot(
store,
"eval",
str(task["id"]),
str(task["id"]),
f"evaluations/{task['id']}/versions/{task['id']}/evaluation-config.json",
payload,
current_user.get("id"),
)
# 2. Resolve model path (supports both regular models and trained models)
model_id = str(payload.get("model_id", ""))
model_path = ""
adapter_path = payload.get("adapter_path", "")
model_node_id = ""
ds_files: list[dict[str, Any]] = []
model_resource_type = "model"
model_resource_id = model_id
adapter_resource_id = ""
try:
db_model = store.model(model_id)
model_path = db_model.get("path", "")
@@ -1845,6 +2145,9 @@ async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: di
# Try trained_models table (IDs prefixed with tm_)
trained = next((m for m in store.trained_models() if m["id"] == model_id), None)
if trained:
model_resource_type = "trained_model" if trained.get("merged") else "model"
model_resource_id = trained.get("id") or model_id
adapter_resource_id = trained.get("id") or ""
model_node_id = trained.get("compute_node_id") or ""
merged_path = trained.get("merged_path", "")
base_path = trained.get("base_model_path", "")
@@ -1858,6 +2161,9 @@ async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: di
adapter_path = merged_path
else:
model_path = merged_path or base_path
if model_resource_type == "model":
base_model = next((item for item in store.models() if item.get("path") == base_path), None)
model_resource_id = str((base_model or {}).get("id") or base_path)
if not model_path:
store.update_eval_task(task["id"], {"status": "failed", "error": "model not found or no path"})
return ok({"task_id": task["id"], "status": "failed", "error": "model not found or no path"})
@@ -1930,6 +2236,33 @@ async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: di
store.update_eval_task(task["id"], {"status": "failed", "error": message})
return ok({"task_id": task["id"], "status": "failed", "error": message})
if get_settings().minio_enabled and get_settings().compute_mode != "simulator":
try:
prepared_model = await _prepare_resource_on_node(store, model_resource_type, model_resource_id, node)
if prepared_model:
model_path = prepared_model
if adapter_resource_id and model_resource_type == "model":
prepared_adapter = await _prepare_resource_on_node(store, "trained_model", adapter_resource_id, node)
if prepared_adapter:
adapter_path = prepared_adapter
dataset_sync = await _sync_training_dataset_to_compute_node(store, node, dataset_id)
if dataset_sync:
dataset_path = str(dataset_sync[0].get("local_path") or dataset_path)
except Exception as exc:
store.update_eval_task(task["id"], {"status": "failed", "error": f"MinIO resource preparation failed: {exc}"})
return ok({"task_id": task["id"], "status": "failed", "error": str(exc)})
node_gpus = {
int(item.get("id", item.get("gpu_index", -1))): item
for item in store.gpus()
if item.get("node_id") == node["id"]
}
unavailable = [index for index in gpu_indices if node_gpus.get(index, {}).get("status") != "idle"]
if unavailable:
message = f"selected GPU is not idle on compute node {node.get('code')}: {unavailable}"
store.update_eval_task(task["id"], {"status": "failed", "error": message})
return ok({"task_id": task["id"], "status": "failed", "error": message})
# 6. Build eval job payload
output_dir = f"/data/yg-ft/outputs/{task['id']}"
job_payload = {
@@ -1943,7 +2276,9 @@ async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: di
"output_dir": output_dir,
"basic_metrics": payload.get("basic_metrics", {}),
"dimension": dimension_cfg,
"gpus": [int(payload.get("gpu_id", 0))],
"gpu_id": gpu_indices[0],
"gpu_indices": gpu_indices,
"gpus": gpu_indices,
"temperature": payload.get("temperature", 0.1),
"max_new_tokens": payload.get("max_new_tokens", 512),
"compute_node_id": node["id"],
@@ -1991,7 +2326,10 @@ async def model_eval_delete(task_id: str, current_user: dict = Depends(get_curre
pending = _require_approval_or_admin("eval", task_id, current_user, f"删除评测任务 {task_id}")
if pending:
return pending
get_platform_store().delete_eval_task(task_id)
try:
get_platform_store().delete_eval_task(task_id)
except KeyError:
raise fail(404, "eval task not found")
return ok({"deleted": task_id})
@@ -2247,6 +2585,15 @@ async def model_compare_load(task_id: str, current_user: dict = Depends(get_curr
"model_name_or_path": model_path,
"template": item.get("template", "qwen"),
}
try:
item_gpu_indices = _normalize_gpu_indices(item)
except ValueError as exc:
loaded_models.append({**item, "status": "error", "error": str(exc)})
continue
if not item_gpu_indices:
loaded_models.append({**item, "status": "error", "error": "no GPU selected"})
continue
load_payload["gpu_indices"] = item_gpu_indices
if item.get("adapter_path"):
load_payload["adapter_name_or_path"] = item["adapter_path"]
if get_settings().compute_mode == "simulator":
@@ -2255,14 +2602,28 @@ async def model_compare_load(task_id: str, current_user: dict = Depends(get_curr
# 只派发：HTTP 响应成功即视为已接受（节点会异步加载），loaded 字段忽略
item_dispatched = False
errors = []
for node in _candidate_online_nodes(store, preferred_node_id):
candidate_nodes = _candidate_online_nodes(store, preferred_node_id)
if preferred_node_id:
candidate_nodes = candidate_nodes[:1]
for node in candidate_nodes:
try:
node_gpu_map = {
int(gpu.get("id", gpu.get("gpu_index", -1))): gpu
for gpu in store.gpus()
if gpu.get("node_id") == node["id"]
}
unavailable = [
index for index in item_gpu_indices
if node_gpu_map.get(index, {}).get("status") != "idle"
]
if unavailable:
raise RuntimeError(f"selected GPU is not idle on compute node {node.get('code')}: {unavailable}")
if get_settings().minio_enabled:
await _wait_for_object_storage()
client = ComputeNodeClient(node["api_base_url"])
await client.inference_load(load_payload)
store.mark_inference_loaded(node["id"])
loaded_models.append({**item, "status": "starting", "node_id": node["id"], "node_name": node.get("name")})
store.mark_inference_loaded(node["id"], item_gpu_indices)
loaded_models.append({**item, "gpu_indices": item_gpu_indices, "gpus": item_gpu_indices, "status": "starting", "node_id": node["id"], "node_name": node.get("name")})
item_dispatched = True
break
except Exception as exc: # noqa: BLE001 - try next candidate node
@@ -2365,7 +2726,7 @@ async def model_chat_local_preload(payload: dict[str, Any] = Body(...)) -> dict[
# 计算节点现在异步加载：HTTP 接受（loading/ready）即视为派发成功
result = await client.inference_load(payload)
if result.get("loaded") or result.get("status") in {"loading", "ready"}:
store.mark_inference_loaded(node["id"])
store.mark_inference_loaded(node["id"], _normalize_gpu_indices(payload))
return ok(result)
except Exception as exc:
return ok({"loaded": False, "error": str(exc)})
@@ -2445,7 +2806,7 @@ async def model_chat_trained_preload(payload: dict[str, Any] = Body(...), curren
# 计算节点现在异步加载：HTTP 接受（loading/ready）即视为派发成功
result = await client.inference_load({**payload, "compute_node_id": node["id"]})
if result.get("loaded") or result.get("status") in {"loading", "ready"}:
store.mark_inference_loaded(node["id"])
store.mark_inference_loaded(node["id"], _normalize_gpu_indices(payload))
return ok(result)
except Exception as exc:
return ok({"loaded": False, "error": str(exc)})