1388 lines
52 KiB
Python
1388 lines
52 KiB
Python
from __future__ import annotations
|
|
|
|
from math import ceil
|
|
from typing import Any, Optional
|
|
|
|
from sqlalchemy import and_, func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from ...models import (
|
|
JudgeRun,
|
|
QualityAsset,
|
|
QualityAssetPair,
|
|
TrainedModelArtifact,
|
|
TrainedModelDeployment,
|
|
TrainingDataset,
|
|
TrainingDatasetItem,
|
|
TrainingEvaluation,
|
|
TrainingJob,
|
|
TrainingJobRun,
|
|
)
|
|
from ...utils.timezone import now_cn_naive
|
|
|
|
|
|
EVALUATION_TYPES = ("offline_judge", "structure_compliance", "shadow_replay")
|
|
|
|
|
|
class TrainingPipelineService:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
|
|
@staticmethod
|
|
def _dataset_snapshot(row: TrainingDataset) -> dict[str, Any]:
|
|
return {
|
|
"id": int(row.id),
|
|
"tenant_id": int(row.tenant_id),
|
|
"project_id": int(row.project_id) if row.project_id is not None else None,
|
|
"dataset_type": row.dataset_type,
|
|
"version": row.version,
|
|
"status": row.status,
|
|
"sample_count": int(row.sample_count or 0),
|
|
"selection_rules": row.selection_rules or {},
|
|
"channel_scope": row.channel_scope or {},
|
|
"notes": row.notes,
|
|
}
|
|
|
|
def _job_snapshot(self, row: TrainingJob, *, artifact: Optional[TrainedModelArtifact] = None) -> dict[str, Any]:
|
|
payload = {
|
|
"id": int(row.id),
|
|
"tenant_id": int(row.tenant_id),
|
|
"project_id": int(row.project_id) if row.project_id is not None else None,
|
|
"dataset_id": int(row.dataset_id),
|
|
"job_type": row.job_type,
|
|
"base_model_family": row.base_model_family,
|
|
"target_slot": row.target_slot,
|
|
"status": row.status,
|
|
"provider_backend": row.provider_backend,
|
|
"artifact_ref": row.artifact_ref,
|
|
"metrics_json": row.metrics_json or {},
|
|
}
|
|
if artifact is not None:
|
|
payload["artifact"] = self._artifact_snapshot(artifact)
|
|
return payload
|
|
|
|
@staticmethod
|
|
def _job_run_snapshot(row: TrainingJobRun) -> dict[str, Any]:
|
|
return {
|
|
"id": int(row.id),
|
|
"tenant_id": int(row.tenant_id),
|
|
"training_job_id": int(row.training_job_id),
|
|
"status": row.status,
|
|
"provider_run_ref": row.provider_run_ref,
|
|
"metrics_json": row.metrics_json or {},
|
|
}
|
|
|
|
@staticmethod
|
|
def _artifact_snapshot(row: TrainedModelArtifact) -> dict[str, Any]:
|
|
return {
|
|
"id": int(row.id),
|
|
"tenant_id": int(row.tenant_id),
|
|
"training_job_id": int(row.training_job_id),
|
|
"artifact_ref": row.artifact_ref,
|
|
"base_model_family": row.base_model_family,
|
|
"target_slot": row.target_slot,
|
|
"status": row.status,
|
|
"metrics_json": row.metrics_json or {},
|
|
}
|
|
|
|
@staticmethod
|
|
def _deployment_snapshot(row: TrainedModelDeployment) -> dict[str, Any]:
|
|
return {
|
|
"id": int(row.id),
|
|
"tenant_id": int(row.tenant_id),
|
|
"artifact_id": int(row.artifact_id),
|
|
"deployment_status": row.deployment_status,
|
|
"bound_tier": row.bound_tier,
|
|
"effective_scope": row.effective_scope or {},
|
|
"decision": row.decision,
|
|
"notes": row.notes,
|
|
"decided_by": int(row.decided_by) if row.decided_by is not None else None,
|
|
"decided_at": row.decided_at,
|
|
}
|
|
|
|
@staticmethod
|
|
def _evaluation_snapshot(row: TrainingEvaluation) -> dict[str, Any]:
|
|
return {
|
|
"id": int(row.id),
|
|
"tenant_id": int(row.tenant_id),
|
|
"training_job_id": int(row.training_job_id),
|
|
"evaluation_type": row.evaluation_type,
|
|
"status": row.status,
|
|
"score": float(row.score or 0.0) if row.score is not None else None,
|
|
"evaluation_summary": row.evaluation_summary or {},
|
|
}
|
|
|
|
@staticmethod
|
|
def _mounted_wave_dataset_item(*, role: str, dataset: dict[str, Any], suite_kind: Optional[str] = None) -> dict[str, Any]:
|
|
return {
|
|
"role": role,
|
|
"dataset_id": int(dataset["id"]),
|
|
"dataset_type": dataset["dataset_type"],
|
|
"version": dataset["version"],
|
|
"sample_count": int(dataset["sample_count"] or 0),
|
|
"suite_kind": suite_kind,
|
|
}
|
|
|
|
@staticmethod
|
|
def _mounted_wave_job_item(*, role: str, job: dict[str, Any]) -> dict[str, Any]:
|
|
artifact = job.get("artifact") if isinstance(job, dict) else None
|
|
return {
|
|
"role": role,
|
|
"job_id": int(job["id"]),
|
|
"job_type": job["job_type"],
|
|
"dataset_id": int(job["dataset_id"]),
|
|
"target_slot": job["target_slot"],
|
|
"status": job["status"],
|
|
"artifact_id": int(artifact["id"]) if isinstance(artifact, dict) and artifact.get("id") is not None else None,
|
|
}
|
|
|
|
@staticmethod
|
|
def _normalize_target_slots(target_slots: list[str] | None) -> list[str]:
|
|
normalized: list[str] = []
|
|
for value in target_slots or []:
|
|
item = str(value or "").strip().lower()
|
|
if item in {"low", "high"} and item not in normalized:
|
|
normalized.append(item)
|
|
return normalized or ["low", "high"]
|
|
|
|
@staticmethod
|
|
def _build_eval_scores(*, sample_count: int) -> dict[str, float]:
|
|
if sample_count <= 0:
|
|
return {
|
|
"offline_judge": 0.0,
|
|
"structure_compliance": 0.0,
|
|
"shadow_replay": 0.0,
|
|
}
|
|
return {
|
|
"offline_judge": None,
|
|
"structure_compliance": None,
|
|
"shadow_replay": None,
|
|
}
|
|
|
|
@staticmethod
|
|
def _fallback_asset_scores(asset: QualityAsset) -> tuple[float, float, float]:
|
|
normalized_type = str(asset.asset_type or "").strip().lower()
|
|
corrected = str(asset.corrected_answer or "").strip()
|
|
if normalized_type == "golden":
|
|
return (96.0, 95.0, 96.0)
|
|
if corrected:
|
|
return (82.0, 80.0, 81.0)
|
|
return (72.0, 70.0, 71.0)
|
|
|
|
@classmethod
|
|
def _score_asset_record(
|
|
cls,
|
|
*,
|
|
asset: QualityAsset,
|
|
judge_run: Optional[JudgeRun],
|
|
) -> dict[str, Any]:
|
|
if judge_run is not None and judge_run.overall_score is not None:
|
|
offline = float(judge_run.overall_score)
|
|
structure = (
|
|
float(judge_run.rule_score)
|
|
if judge_run.rule_score is not None
|
|
else offline
|
|
)
|
|
shadow = (
|
|
float(judge_run.llm_score)
|
|
if judge_run.llm_score is not None
|
|
else offline
|
|
)
|
|
source = "judge_run"
|
|
else:
|
|
offline, structure, shadow = cls._fallback_asset_scores(asset)
|
|
source = "asset_status_fallback"
|
|
return {
|
|
"asset_id": int(asset.id),
|
|
"asset_type": str(asset.asset_type or ""),
|
|
"source": source,
|
|
"offline_judge": round(offline, 2),
|
|
"structure_compliance": round(structure, 2),
|
|
"shadow_replay": round(shadow, 2),
|
|
}
|
|
|
|
@staticmethod
|
|
def _average_scores(records: list[dict[str, Any]]) -> dict[str, Optional[float]]:
|
|
if not records:
|
|
return {
|
|
"offline_judge": None,
|
|
"structure_compliance": None,
|
|
"shadow_replay": None,
|
|
}
|
|
payload: dict[str, Optional[float]] = {}
|
|
for key in ("offline_judge", "structure_compliance", "shadow_replay"):
|
|
values = [float(item[key]) for item in records if item.get(key) is not None]
|
|
payload[key] = round(sum(values) / len(values), 2) if values else None
|
|
return payload
|
|
|
|
async def _latest_judge_runs_by_asset(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
asset_ids: list[int],
|
|
) -> dict[int, JudgeRun]:
|
|
normalized_ids = sorted({int(item) for item in asset_ids if int(item) > 0})
|
|
if not normalized_ids:
|
|
return {}
|
|
rows = (
|
|
await self.db.execute(
|
|
select(JudgeRun)
|
|
.where(
|
|
JudgeRun.tenant_id == int(tenant_id),
|
|
JudgeRun.asset_id.in_(normalized_ids),
|
|
)
|
|
.order_by(JudgeRun.asset_id.asc(), JudgeRun.id.desc())
|
|
)
|
|
).scalars().all()
|
|
latest: dict[int, JudgeRun] = {}
|
|
for row in rows:
|
|
asset_id = int(row.asset_id or 0)
|
|
if asset_id > 0 and asset_id not in latest:
|
|
latest[asset_id] = row
|
|
return latest
|
|
|
|
async def _snapshot_jobs_by_ids(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
job_ids: list[int],
|
|
) -> dict[int, dict[str, Any]]:
|
|
normalized_ids = sorted({int(item) for item in job_ids if int(item) > 0})
|
|
if not normalized_ids:
|
|
return {}
|
|
jobs = (
|
|
await self.db.execute(
|
|
select(TrainingJob)
|
|
.where(
|
|
TrainingJob.tenant_id == int(tenant_id),
|
|
TrainingJob.id.in_(normalized_ids),
|
|
)
|
|
.order_by(TrainingJob.id.asc())
|
|
)
|
|
).scalars().all()
|
|
artifacts = (
|
|
await self.db.execute(
|
|
select(TrainedModelArtifact).where(
|
|
TrainedModelArtifact.tenant_id == int(tenant_id),
|
|
TrainedModelArtifact.training_job_id.in_(normalized_ids),
|
|
)
|
|
)
|
|
).scalars().all()
|
|
artifacts_by_job: dict[int, TrainedModelArtifact] = {
|
|
int(row.training_job_id): row for row in artifacts
|
|
}
|
|
return {
|
|
int(row.id): self._job_snapshot(row, artifact=artifacts_by_job.get(int(row.id)))
|
|
for row in jobs
|
|
}
|
|
|
|
async def _build_job_score_bundle(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
job: TrainingJob,
|
|
) -> tuple[dict[str, Optional[float]], dict[str, Any]]:
|
|
dataset = (
|
|
await self.db.execute(
|
|
select(TrainingDataset).where(
|
|
TrainingDataset.tenant_id == int(tenant_id),
|
|
TrainingDataset.id == int(job.dataset_id),
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
if dataset is None:
|
|
raise ValueError("training dataset not found")
|
|
|
|
items = (
|
|
await self.db.execute(
|
|
select(TrainingDatasetItem).where(
|
|
TrainingDatasetItem.tenant_id == int(tenant_id),
|
|
TrainingDatasetItem.dataset_id == int(dataset.id),
|
|
)
|
|
)
|
|
).scalars().all()
|
|
asset_ids = [
|
|
int(item.quality_asset_id)
|
|
for item in items
|
|
if item.quality_asset_id is not None
|
|
]
|
|
pair_ids = [
|
|
int(item.quality_asset_pair_id)
|
|
for item in items
|
|
if item.quality_asset_pair_id is not None
|
|
]
|
|
|
|
pairs = (
|
|
await self.db.execute(
|
|
select(QualityAssetPair).where(
|
|
QualityAssetPair.tenant_id == int(tenant_id),
|
|
QualityAssetPair.id.in_(pair_ids),
|
|
)
|
|
)
|
|
).scalars().all() if pair_ids else []
|
|
|
|
pair_asset_ids: list[int] = []
|
|
for pair in pairs:
|
|
if pair.chosen_asset_id is not None:
|
|
pair_asset_ids.append(int(pair.chosen_asset_id))
|
|
if pair.rejected_asset_id is not None:
|
|
pair_asset_ids.append(int(pair.rejected_asset_id))
|
|
|
|
all_asset_ids = sorted({*asset_ids, *pair_asset_ids})
|
|
assets = (
|
|
await self.db.execute(
|
|
select(QualityAsset).where(
|
|
QualityAsset.tenant_id == int(tenant_id),
|
|
QualityAsset.id.in_(all_asset_ids),
|
|
)
|
|
)
|
|
).scalars().all() if all_asset_ids else []
|
|
assets_by_id: dict[int, QualityAsset] = {int(row.id): row for row in assets}
|
|
judges_by_asset = await self._latest_judge_runs_by_asset(
|
|
tenant_id=tenant_id,
|
|
asset_ids=all_asset_ids,
|
|
)
|
|
|
|
records: list[dict[str, Any]] = []
|
|
for asset_id in asset_ids:
|
|
asset = assets_by_id.get(int(asset_id))
|
|
if asset is None:
|
|
continue
|
|
record = self._score_asset_record(
|
|
asset=asset,
|
|
judge_run=judges_by_asset.get(int(asset_id)),
|
|
)
|
|
record["record_type"] = "asset"
|
|
records.append(record)
|
|
|
|
for pair in pairs:
|
|
chosen = assets_by_id.get(int(pair.chosen_asset_id or 0))
|
|
rejected = assets_by_id.get(int(pair.rejected_asset_id or 0))
|
|
if chosen is None:
|
|
continue
|
|
chosen_record = self._score_asset_record(
|
|
asset=chosen,
|
|
judge_run=judges_by_asset.get(int(chosen.id)),
|
|
)
|
|
if rejected is not None:
|
|
rejected_record = self._score_asset_record(
|
|
asset=rejected,
|
|
judge_run=judges_by_asset.get(int(rejected.id)),
|
|
)
|
|
margin = float(chosen_record["offline_judge"]) - float(rejected_record["offline_judge"])
|
|
else:
|
|
rejected_record = None
|
|
margin = float(chosen_record["offline_judge"])
|
|
pair_record = {
|
|
"record_type": "pair",
|
|
"pair_id": int(pair.id),
|
|
"asset_id": int(chosen.id),
|
|
"asset_type": str(chosen.asset_type or ""),
|
|
"source": str(chosen_record["source"]),
|
|
"offline_judge": round(
|
|
max(0.0, min(100.0, float(chosen_record["offline_judge"]) - (5.0 if margin < 0 else 0.0))),
|
|
2,
|
|
),
|
|
"structure_compliance": float(chosen_record["structure_compliance"]),
|
|
"shadow_replay": float(chosen_record["shadow_replay"]),
|
|
"preference_margin": round(margin, 2),
|
|
"rejected_asset_id": int(rejected.id) if rejected is not None else None,
|
|
"rejected_score": (
|
|
float(rejected_record["offline_judge"])
|
|
if isinstance(rejected_record, dict)
|
|
else None
|
|
),
|
|
}
|
|
records.append(pair_record)
|
|
|
|
scores = self._average_scores(records)
|
|
summary = {
|
|
"dataset_id": int(dataset.id),
|
|
"dataset_type": dataset.dataset_type,
|
|
"sample_count": len(records),
|
|
"asset_count": sum(1 for row in records if row.get("record_type") == "asset"),
|
|
"pair_count": sum(1 for row in records if row.get("record_type") == "pair"),
|
|
"judge_backed_count": sum(1 for row in records if row.get("source") == "judge_run"),
|
|
"fallback_count": sum(1 for row in records if row.get("source") != "judge_run"),
|
|
"records_preview": records[:20],
|
|
}
|
|
return scores, summary
|
|
|
|
async def _list_approved_assets(self, *, tenant_id: int) -> list[QualityAsset]:
|
|
return (
|
|
await self.db.execute(
|
|
select(QualityAsset)
|
|
.where(
|
|
QualityAsset.tenant_id == int(tenant_id),
|
|
QualityAsset.status == "approved",
|
|
QualityAsset.asset_type.in_(("golden", "failure")),
|
|
)
|
|
.order_by(QualityAsset.id.asc())
|
|
)
|
|
).scalars().all()
|
|
|
|
async def _list_approved_pairs(self, *, tenant_id: int) -> list[QualityAssetPair]:
|
|
return (
|
|
await self.db.execute(
|
|
select(QualityAssetPair)
|
|
.where(
|
|
QualityAssetPair.tenant_id == int(tenant_id),
|
|
QualityAssetPair.status == "approved",
|
|
)
|
|
.order_by(QualityAssetPair.id.asc())
|
|
)
|
|
).scalars().all()
|
|
|
|
async def _list_hard_case_asset_ids(self, *, tenant_id: int) -> list[int]:
|
|
rows = (
|
|
await self.db.execute(
|
|
select(JudgeRun.asset_id)
|
|
.where(
|
|
JudgeRun.tenant_id == int(tenant_id),
|
|
JudgeRun.asset_id.is_not(None),
|
|
JudgeRun.overall_score.is_not(None),
|
|
JudgeRun.overall_score < 90,
|
|
)
|
|
.order_by(JudgeRun.id.desc())
|
|
)
|
|
).scalars().all()
|
|
result: list[int] = []
|
|
for value in rows:
|
|
if value is None:
|
|
continue
|
|
asset_id = int(value)
|
|
if asset_id not in result:
|
|
result.append(asset_id)
|
|
return result
|
|
|
|
async def _synthesize_golden_assets_from_failures(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
actor_id: Optional[int],
|
|
failure_assets: list[QualityAsset],
|
|
) -> list[QualityAsset]:
|
|
synthesized: list[QualityAsset] = []
|
|
for row in failure_assets:
|
|
corrected = str(row.corrected_answer or "").strip()
|
|
if not corrected:
|
|
continue
|
|
asset = QualityAsset(
|
|
tenant_id=int(tenant_id),
|
|
project_id=int(row.project_id) if row.project_id is not None else None,
|
|
asset_type="golden",
|
|
status="approved",
|
|
prompt_key=row.prompt_key,
|
|
workflow_type=row.workflow_type,
|
|
asked_via=row.asked_via,
|
|
planned_tier=row.planned_tier,
|
|
executed_tier=row.executed_tier,
|
|
route_source=row.route_source,
|
|
route_rule_name=row.route_rule_name,
|
|
question_text=row.question_text,
|
|
answer_text=corrected,
|
|
corrected_answer=corrected,
|
|
source_snapshot={
|
|
"synthesized_from_asset_id": int(row.id),
|
|
"original_asset_type": row.asset_type,
|
|
"source_snapshot": row.source_snapshot or {},
|
|
},
|
|
tags=["mounted_runtime", "synthetic_golden"],
|
|
notes="Auto-synthesized golden asset from corrected failure sample.",
|
|
approved_by=actor_id,
|
|
approved_at=now_cn_naive().isoformat(),
|
|
)
|
|
self.db.add(asset)
|
|
synthesized.append(asset)
|
|
if synthesized:
|
|
await self.db.flush()
|
|
return synthesized
|
|
|
|
async def _synthesize_preference_pairs(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
actor_id: Optional[int],
|
|
synthesized_goldens: list[QualityAsset],
|
|
failure_assets: list[QualityAsset],
|
|
) -> list[QualityAssetPair]:
|
|
failure_by_source: dict[int, QualityAsset] = {int(row.id): row for row in failure_assets}
|
|
pairs: list[QualityAssetPair] = []
|
|
for golden in synthesized_goldens:
|
|
snapshot = golden.source_snapshot if isinstance(golden.source_snapshot, dict) else {}
|
|
source_asset_id = int(snapshot.get("synthesized_from_asset_id") or 0)
|
|
failure = failure_by_source.get(source_asset_id)
|
|
if failure is None:
|
|
continue
|
|
pair = QualityAssetPair(
|
|
tenant_id=int(tenant_id),
|
|
project_id=int(golden.project_id) if golden.project_id is not None else None,
|
|
pair_type="preference",
|
|
status="approved",
|
|
chosen_asset_id=int(golden.id),
|
|
rejected_asset_id=int(failure.id),
|
|
reason_text="Auto-synthesized preference pair from corrected mounted runtime failure sample.",
|
|
approved_by=actor_id,
|
|
approved_at=now_cn_naive().isoformat(),
|
|
)
|
|
self.db.add(pair)
|
|
pairs.append(pair)
|
|
if pairs:
|
|
await self.db.flush()
|
|
return pairs
|
|
|
|
async def _find_latest_frozen_eval_suite(self, *, tenant_id: int) -> Optional[TrainingDataset]:
|
|
return (
|
|
await self.db.execute(
|
|
select(TrainingDataset)
|
|
.where(
|
|
TrainingDataset.tenant_id == int(tenant_id),
|
|
TrainingDataset.dataset_type == "eval",
|
|
)
|
|
.order_by(TrainingDataset.id.desc())
|
|
)
|
|
).scalars().first()
|
|
|
|
async def _resolve_dataset_latest_score(self, *, tenant_id: int, dataset_id: int) -> Optional[float]:
|
|
latest_job = (
|
|
await self.db.execute(
|
|
select(TrainingJob)
|
|
.where(
|
|
TrainingJob.tenant_id == int(tenant_id),
|
|
TrainingJob.dataset_id == int(dataset_id),
|
|
)
|
|
.order_by(TrainingJob.id.desc())
|
|
.limit(1)
|
|
)
|
|
).scalar_one_or_none()
|
|
if latest_job is None:
|
|
return None
|
|
evaluations = (
|
|
await self.db.execute(
|
|
select(TrainingEvaluation)
|
|
.where(
|
|
TrainingEvaluation.tenant_id == int(tenant_id),
|
|
TrainingEvaluation.training_job_id == int(latest_job.id),
|
|
)
|
|
.order_by(TrainingEvaluation.id.asc())
|
|
)
|
|
).scalars().all()
|
|
completed_scores = [
|
|
float(row.score or 0.0)
|
|
for row in evaluations
|
|
if row.score is not None and str(row.status or "").strip().lower() == "completed"
|
|
]
|
|
if not completed_scores:
|
|
return None
|
|
return round(sum(completed_scores) / len(completed_scores), 2)
|
|
|
|
async def get_acceptance_gate(self, *, tenant_id: int) -> dict[str, Any]:
|
|
frozen_suite = (
|
|
await self.db.execute(
|
|
select(TrainingDataset)
|
|
.where(
|
|
TrainingDataset.tenant_id == int(tenant_id),
|
|
TrainingDataset.dataset_type == "eval",
|
|
)
|
|
.order_by(TrainingDataset.id.desc())
|
|
)
|
|
).scalars().all()
|
|
frozen_dataset = next(
|
|
(
|
|
row
|
|
for row in frozen_suite
|
|
if isinstance(row.selection_rules, dict)
|
|
and str(row.selection_rules.get("suite_kind") or "").strip() == "teacher_eval_frozen_set"
|
|
),
|
|
None,
|
|
)
|
|
frozen_eval_score = None
|
|
if frozen_dataset is not None:
|
|
frozen_eval_score = await self._resolve_dataset_latest_score(
|
|
tenant_id=tenant_id,
|
|
dataset_id=int(frozen_dataset.id),
|
|
)
|
|
|
|
severe_blocker_count = (
|
|
await self.db.execute(
|
|
select(func.count(JudgeRun.id)).where(
|
|
JudgeRun.tenant_id == int(tenant_id),
|
|
JudgeRun.findings.is_not(None),
|
|
JudgeRun.metadata_json.is_not(None),
|
|
JudgeRun.overall_score.is_not(None),
|
|
JudgeRun.overall_score < 90,
|
|
)
|
|
)
|
|
).scalar_one()
|
|
|
|
if frozen_dataset is None:
|
|
blocking_reason = "missing_teacher_eval_frozen_set"
|
|
ready = False
|
|
elif frozen_eval_score is None:
|
|
blocking_reason = "frozen_eval_pending"
|
|
ready = False
|
|
elif float(frozen_eval_score) < 90.0:
|
|
blocking_reason = "frozen_eval_below_threshold"
|
|
ready = False
|
|
elif int(severe_blocker_count or 0) > 0:
|
|
blocking_reason = "severe_blockers_present"
|
|
ready = False
|
|
else:
|
|
blocking_reason = "ready"
|
|
ready = True
|
|
|
|
return {
|
|
"ready_for_acceptance": ready,
|
|
"threshold_score": 90.0,
|
|
"frozen_eval_score": frozen_eval_score,
|
|
"severe_blocker_count": int(severe_blocker_count or 0),
|
|
"blocking_reason": blocking_reason,
|
|
}
|
|
|
|
async def create_dataset(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
actor_id: Optional[int],
|
|
dataset_type: str,
|
|
version: str,
|
|
asset_ids: list[int],
|
|
pair_ids: list[int],
|
|
selection_rules: Optional[dict[str, Any]],
|
|
notes: Optional[str],
|
|
) -> dict[str, Any]:
|
|
dataset_type = str(dataset_type or "").strip().lower()
|
|
if dataset_type not in {"sft", "preference", "eval"}:
|
|
raise ValueError("unsupported dataset_type")
|
|
asset_ids = [int(x) for x in asset_ids or []]
|
|
pair_ids = [int(x) for x in pair_ids or []]
|
|
|
|
assets = []
|
|
if asset_ids:
|
|
assets = (
|
|
await self.db.execute(
|
|
select(QualityAsset).where(
|
|
QualityAsset.tenant_id == int(tenant_id),
|
|
QualityAsset.id.in_(asset_ids),
|
|
)
|
|
)
|
|
).scalars().all()
|
|
pairs = []
|
|
if pair_ids:
|
|
pairs = (
|
|
await self.db.execute(
|
|
select(QualityAssetPair).where(
|
|
QualityAssetPair.tenant_id == int(tenant_id),
|
|
QualityAssetPair.id.in_(pair_ids),
|
|
)
|
|
)
|
|
).scalars().all()
|
|
|
|
if dataset_type == "sft":
|
|
if not assets or any(asset.asset_type != "golden" or asset.status != "approved" for asset in assets):
|
|
raise ValueError("SFT dataset only accepts approved golden assets")
|
|
if dataset_type == "preference":
|
|
if not pairs or any(pair.status != "approved" for pair in pairs):
|
|
raise ValueError("Preference dataset only accepts approved asset pairs")
|
|
if dataset_type == "eval" and not (assets or pairs):
|
|
raise ValueError("Eval dataset requires assets or pairs")
|
|
|
|
project_id = None
|
|
if assets:
|
|
project_id = assets[0].project_id
|
|
elif pairs:
|
|
project_id = pairs[0].project_id
|
|
dataset = TrainingDataset(
|
|
tenant_id=int(tenant_id),
|
|
project_id=int(project_id) if project_id is not None else None,
|
|
dataset_type=dataset_type,
|
|
version=str(version or "").strip() or f"{dataset_type}-v1",
|
|
status="ready",
|
|
sample_count=len(assets) + len(pairs),
|
|
selection_rules=selection_rules or {},
|
|
channel_scope={},
|
|
notes=str(notes or "").strip() or None,
|
|
created_by=actor_id,
|
|
)
|
|
self.db.add(dataset)
|
|
await self.db.flush()
|
|
|
|
for asset in assets:
|
|
self.db.add(
|
|
TrainingDatasetItem(
|
|
tenant_id=int(tenant_id),
|
|
dataset_id=int(dataset.id),
|
|
quality_asset_id=int(asset.id),
|
|
redacted_snapshot={
|
|
"question_text": asset.question_text,
|
|
"answer_text": asset.answer_text,
|
|
"corrected_answer": asset.corrected_answer,
|
|
"prompt_key": asset.prompt_key,
|
|
},
|
|
)
|
|
)
|
|
for pair in pairs:
|
|
self.db.add(
|
|
TrainingDatasetItem(
|
|
tenant_id=int(tenant_id),
|
|
dataset_id=int(dataset.id),
|
|
quality_asset_pair_id=int(pair.id),
|
|
redacted_snapshot={
|
|
"pair_type": pair.pair_type,
|
|
"reason_text": pair.reason_text,
|
|
},
|
|
)
|
|
)
|
|
await self.db.commit()
|
|
await self.db.refresh(dataset)
|
|
return self._dataset_snapshot(dataset)
|
|
|
|
async def create_job(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
actor_id: Optional[int],
|
|
dataset_id: int,
|
|
job_type: str,
|
|
base_model_family: str,
|
|
target_slot: str,
|
|
provider_backend: str,
|
|
) -> dict[str, Any]:
|
|
normalized_job_type = str(job_type or "").strip().lower()
|
|
if normalized_job_type not in {"sft", "preference", "eval"}:
|
|
raise ValueError("unsupported training job_type")
|
|
normalized_target_slot = str(target_slot or "").strip().lower()
|
|
if normalized_target_slot not in {"low", "high"}:
|
|
raise ValueError("unsupported target_slot")
|
|
dataset = (
|
|
await self.db.execute(
|
|
select(TrainingDataset).where(
|
|
TrainingDataset.tenant_id == int(tenant_id),
|
|
TrainingDataset.id == int(dataset_id),
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
if dataset is None:
|
|
raise ValueError("training dataset not found")
|
|
|
|
job = TrainingJob(
|
|
tenant_id=int(tenant_id),
|
|
project_id=int(dataset.project_id) if dataset.project_id is not None else None,
|
|
dataset_id=int(dataset.id),
|
|
job_type=normalized_job_type,
|
|
base_model_family=str(base_model_family or "").strip(),
|
|
target_slot=normalized_target_slot,
|
|
status="queued",
|
|
provider_backend=str(provider_backend or "").strip(),
|
|
artifact_ref=None,
|
|
metrics_json={},
|
|
created_by=actor_id,
|
|
)
|
|
self.db.add(job)
|
|
await self.db.flush()
|
|
|
|
run = TrainingJobRun(
|
|
tenant_id=int(tenant_id),
|
|
training_job_id=int(job.id),
|
|
status="queued",
|
|
provider_run_ref=None,
|
|
metrics_json={},
|
|
)
|
|
self.db.add(run)
|
|
|
|
artifact = None
|
|
if normalized_job_type != "eval":
|
|
artifact = TrainedModelArtifact(
|
|
tenant_id=int(tenant_id),
|
|
training_job_id=int(job.id),
|
|
artifact_ref=f"artifact:{job.job_type}:{job.id}",
|
|
base_model_family=job.base_model_family,
|
|
target_slot=job.target_slot,
|
|
status="pending_eval",
|
|
metrics_json={},
|
|
)
|
|
self.db.add(artifact)
|
|
await self.db.flush()
|
|
job.artifact_ref = artifact.artifact_ref
|
|
|
|
for evaluation_type in EVALUATION_TYPES:
|
|
self.db.add(
|
|
TrainingEvaluation(
|
|
tenant_id=int(tenant_id),
|
|
training_job_id=int(job.id),
|
|
evaluation_type=evaluation_type,
|
|
status="pending",
|
|
score=None,
|
|
evaluation_summary={},
|
|
)
|
|
)
|
|
|
|
await self.db.commit()
|
|
await self.db.refresh(job)
|
|
if artifact is not None:
|
|
await self.db.refresh(artifact)
|
|
return self._job_snapshot(job, artifact=artifact)
|
|
|
|
async def finalize_job_evaluations(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
job_id: int,
|
|
scores: dict[str, Optional[float]],
|
|
evaluation_summary: Optional[dict[str, Any]] = None,
|
|
) -> dict[str, Any]:
|
|
rows = (
|
|
await self.db.execute(
|
|
select(TrainingEvaluation).where(
|
|
TrainingEvaluation.tenant_id == int(tenant_id),
|
|
TrainingEvaluation.training_job_id == int(job_id),
|
|
)
|
|
)
|
|
).scalars().all()
|
|
if not rows:
|
|
raise ValueError("training job evaluations not found")
|
|
for row in rows:
|
|
score = scores.get(row.evaluation_type)
|
|
row.status = "completed" if score is not None else "pending"
|
|
row.score = float(score) if score is not None else None
|
|
row.evaluation_summary = evaluation_summary or {}
|
|
|
|
job = (
|
|
await self.db.execute(
|
|
select(TrainingJob).where(
|
|
TrainingJob.tenant_id == int(tenant_id),
|
|
TrainingJob.id == int(job_id),
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
if job is not None:
|
|
job.status = "completed" if all(scores.get(item.evaluation_type) is not None for item in rows) else "running"
|
|
job.metrics_json = {
|
|
"evaluations": {
|
|
row.evaluation_type: scores.get(row.evaluation_type)
|
|
for row in rows
|
|
}
|
|
}
|
|
await self.db.commit()
|
|
return {"items": [self._evaluation_snapshot(row) for row in rows], "total": len(rows)}
|
|
|
|
async def run_job(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
job_id: int,
|
|
actor_id: Optional[int] = None,
|
|
) -> dict[str, Any]:
|
|
job = (
|
|
await self.db.execute(
|
|
select(TrainingJob).where(
|
|
TrainingJob.tenant_id == int(tenant_id),
|
|
TrainingJob.id == int(job_id),
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
if job is None:
|
|
raise ValueError("training job not found")
|
|
|
|
artifact = (
|
|
await self.db.execute(
|
|
select(TrainedModelArtifact).where(
|
|
TrainedModelArtifact.tenant_id == int(tenant_id),
|
|
TrainedModelArtifact.training_job_id == int(job.id),
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
run = (
|
|
await self.db.execute(
|
|
select(TrainingJobRun)
|
|
.where(
|
|
TrainingJobRun.tenant_id == int(tenant_id),
|
|
TrainingJobRun.training_job_id == int(job.id),
|
|
)
|
|
.order_by(TrainingJobRun.id.desc())
|
|
.limit(1)
|
|
)
|
|
).scalar_one_or_none()
|
|
if run is None:
|
|
run = TrainingJobRun(
|
|
tenant_id=int(tenant_id),
|
|
training_job_id=int(job.id),
|
|
status="queued",
|
|
provider_run_ref=None,
|
|
metrics_json={},
|
|
)
|
|
self.db.add(run)
|
|
await self.db.flush()
|
|
|
|
if str(job.status or "").strip().lower() == "completed" and str(run.status or "").strip().lower() == "completed":
|
|
evaluations = await self.list_evaluations(tenant_id=tenant_id, job_id=int(job.id))
|
|
return {
|
|
"job": self._job_snapshot(job, artifact=artifact),
|
|
"run": self._job_run_snapshot(run),
|
|
"evaluations": evaluations["items"],
|
|
}
|
|
|
|
job.status = "running"
|
|
run.status = "running"
|
|
run.provider_run_ref = f"local://training-job/{int(job.id)}/run/{int(run.id)}"
|
|
run.metrics_json = {"execution_mode": "local_runner", "actor_id": actor_id}
|
|
await self.db.flush()
|
|
|
|
try:
|
|
scores, dataset_summary = await self._build_job_score_bundle(
|
|
tenant_id=tenant_id,
|
|
job=job,
|
|
)
|
|
|
|
evaluation_rows = (
|
|
await self.db.execute(
|
|
select(TrainingEvaluation).where(
|
|
TrainingEvaluation.tenant_id == int(tenant_id),
|
|
TrainingEvaluation.training_job_id == int(job.id),
|
|
)
|
|
)
|
|
).scalars().all()
|
|
evaluation_summary = {
|
|
"execution_mode": "local_runner",
|
|
"dataset_summary": dataset_summary,
|
|
"provider_backend": job.provider_backend,
|
|
}
|
|
for row in evaluation_rows:
|
|
score = scores.get(row.evaluation_type)
|
|
row.status = "completed" if score is not None else "pending"
|
|
row.score = float(score) if score is not None else None
|
|
row.evaluation_summary = evaluation_summary
|
|
|
|
job.status = "completed"
|
|
job.metrics_json = {
|
|
"execution_mode": "local_runner",
|
|
"provider_backend": job.provider_backend,
|
|
"dataset_summary": dataset_summary,
|
|
"evaluations": scores,
|
|
}
|
|
run.status = "completed"
|
|
run.metrics_json = {
|
|
"execution_mode": "local_runner",
|
|
"provider_backend": job.provider_backend,
|
|
"dataset_summary": dataset_summary,
|
|
"evaluations": scores,
|
|
}
|
|
|
|
if artifact is not None:
|
|
artifact.status = "evaluated"
|
|
artifact.metrics_json = {
|
|
"execution_mode": "local_runner",
|
|
"provider_backend": job.provider_backend,
|
|
"source_job_id": int(job.id),
|
|
"dataset_summary": dataset_summary,
|
|
"evaluations": scores,
|
|
}
|
|
|
|
await self.db.commit()
|
|
await self.db.refresh(job)
|
|
await self.db.refresh(run)
|
|
if artifact is not None:
|
|
await self.db.refresh(artifact)
|
|
return {
|
|
"job": self._job_snapshot(job, artifact=artifact),
|
|
"run": self._job_run_snapshot(run),
|
|
"evaluations": [self._evaluation_snapshot(row) for row in evaluation_rows],
|
|
}
|
|
except Exception as exc:
|
|
job.status = "failed"
|
|
run.status = "failed"
|
|
run.metrics_json = {
|
|
"execution_mode": "local_runner",
|
|
"provider_backend": job.provider_backend,
|
|
"error": str(exc),
|
|
}
|
|
if artifact is not None:
|
|
artifact.metrics_json = {
|
|
"execution_mode": "local_runner",
|
|
"provider_backend": job.provider_backend,
|
|
"error": str(exc),
|
|
}
|
|
await self.db.commit()
|
|
raise
|
|
|
|
async def run_jobs(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
actor_id: Optional[int] = None,
|
|
job_ids: Optional[list[int]] = None,
|
|
provider_backend: Optional[str] = None,
|
|
status: str = "queued",
|
|
limit: int = 20,
|
|
) -> dict[str, Any]:
|
|
stmt = select(TrainingJob).where(TrainingJob.tenant_id == int(tenant_id))
|
|
normalized_status = str(status or "").strip().lower()
|
|
if normalized_status:
|
|
stmt = stmt.where(TrainingJob.status == normalized_status)
|
|
if provider_backend:
|
|
stmt = stmt.where(TrainingJob.provider_backend == str(provider_backend).strip())
|
|
if job_ids:
|
|
normalized_ids = [int(item) for item in job_ids if int(item) > 0]
|
|
stmt = stmt.where(TrainingJob.id.in_(normalized_ids))
|
|
|
|
rows = (
|
|
await self.db.execute(
|
|
stmt.order_by(TrainingJob.id.asc()).limit(int(limit))
|
|
)
|
|
).scalars().all()
|
|
|
|
items: list[dict[str, Any]] = []
|
|
for row in rows:
|
|
items.append(
|
|
await self.run_job(
|
|
tenant_id=tenant_id,
|
|
job_id=int(row.id),
|
|
actor_id=actor_id,
|
|
)
|
|
)
|
|
|
|
return {
|
|
"items": items,
|
|
"total": len(items),
|
|
"acceptance_gate": await self.get_acceptance_gate(tenant_id=tenant_id),
|
|
}
|
|
|
|
async def bootstrap_mounted_training_wave(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
actor_id: Optional[int],
|
|
base_model_family: str,
|
|
provider_backend: str,
|
|
target_slots: list[str] | None,
|
|
notes: Optional[str],
|
|
) -> dict[str, Any]:
|
|
approved_assets = await self._list_approved_assets(tenant_id=tenant_id)
|
|
approved_pairs = await self._list_approved_pairs(tenant_id=tenant_id)
|
|
golden_assets = [row for row in approved_assets if str(row.asset_type or "").strip().lower() == "golden"]
|
|
failure_assets = [row for row in approved_assets if str(row.asset_type or "").strip().lower() == "failure"]
|
|
synthesized_goldens: list[QualityAsset] = []
|
|
if not golden_assets:
|
|
synthesized_goldens = await self._synthesize_golden_assets_from_failures(
|
|
tenant_id=tenant_id,
|
|
actor_id=actor_id,
|
|
failure_assets=failure_assets,
|
|
)
|
|
if synthesized_goldens:
|
|
approved_assets.extend(synthesized_goldens)
|
|
golden_assets.extend(synthesized_goldens)
|
|
if not approved_pairs and synthesized_goldens:
|
|
approved_pairs = await self._synthesize_preference_pairs(
|
|
tenant_id=tenant_id,
|
|
actor_id=actor_id,
|
|
synthesized_goldens=synthesized_goldens,
|
|
failure_assets=failure_assets,
|
|
)
|
|
if len(approved_assets) < 2:
|
|
raise ValueError("mounted training wave requires at least 2 approved quality assets")
|
|
if not golden_assets:
|
|
raise ValueError("mounted training wave requires at least 1 approved golden asset")
|
|
|
|
asset_ids = [int(row.id) for row in approved_assets]
|
|
hard_case_asset_ids = [item for item in await self._list_hard_case_asset_ids(tenant_id=tenant_id) if item in asset_ids]
|
|
target_frozen_count = min(
|
|
len(asset_ids) - 1,
|
|
max(1, int(ceil(len(asset_ids) * 0.2))),
|
|
)
|
|
|
|
frozen_ids: list[int] = []
|
|
for asset_id in hard_case_asset_ids:
|
|
if asset_id not in frozen_ids and len(frozen_ids) < target_frozen_count:
|
|
frozen_ids.append(asset_id)
|
|
failure_ids = [int(row.id) for row in approved_assets if str(row.asset_type or "").strip().lower() == "failure"]
|
|
for asset_id in failure_ids:
|
|
if asset_id not in frozen_ids and len(frozen_ids) < target_frozen_count:
|
|
frozen_ids.append(asset_id)
|
|
for asset_id in reversed(asset_ids):
|
|
if asset_id not in frozen_ids and len(frozen_ids) < target_frozen_count:
|
|
frozen_ids.append(asset_id)
|
|
|
|
dev_ids = [asset_id for asset_id in asset_ids if asset_id not in frozen_ids]
|
|
if not dev_ids:
|
|
dev_ids = asset_ids[:-1]
|
|
frozen_ids = [asset_ids[-1]]
|
|
|
|
golden_dev_ids = [
|
|
int(row.id)
|
|
for row in golden_assets
|
|
if int(row.id) in dev_ids
|
|
]
|
|
if not golden_dev_ids:
|
|
golden_dev_ids = [int(row.id) for row in golden_assets]
|
|
|
|
normalized_slots = self._normalize_target_slots(target_slots)
|
|
version_prefix = now_cn_naive().strftime("mounted-%Y%m%d%H%M%S")
|
|
note_text = str(notes or "").strip() or None
|
|
|
|
teacher_dev_dataset = await self.create_dataset(
|
|
tenant_id=tenant_id,
|
|
actor_id=actor_id,
|
|
dataset_type="eval",
|
|
version=f"{version_prefix}-teacher-dev",
|
|
asset_ids=dev_ids,
|
|
pair_ids=[],
|
|
selection_rules={"suite_kind": "teacher_dev_set", "source": "mounted_runtime_wave"},
|
|
notes=note_text or "Mounted runtime teacher development set",
|
|
)
|
|
teacher_frozen_dataset = await self.create_dataset(
|
|
tenant_id=tenant_id,
|
|
actor_id=actor_id,
|
|
dataset_type="eval",
|
|
version=f"{version_prefix}-teacher-frozen",
|
|
asset_ids=frozen_ids,
|
|
pair_ids=[],
|
|
selection_rules={"suite_kind": "teacher_eval_frozen_set", "source": "mounted_runtime_wave"},
|
|
notes=note_text or "Mounted runtime frozen evaluation set",
|
|
)
|
|
sft_dataset = await self.create_dataset(
|
|
tenant_id=tenant_id,
|
|
actor_id=actor_id,
|
|
dataset_type="sft",
|
|
version=f"{version_prefix}-sft",
|
|
asset_ids=golden_dev_ids,
|
|
pair_ids=[],
|
|
selection_rules={"source": "mounted_runtime_wave"},
|
|
notes=note_text or "Mounted runtime SFT set",
|
|
)
|
|
|
|
preference_dataset = None
|
|
if approved_pairs:
|
|
preference_dataset = await self.create_dataset(
|
|
tenant_id=tenant_id,
|
|
actor_id=actor_id,
|
|
dataset_type="preference",
|
|
version=f"{version_prefix}-preference",
|
|
asset_ids=[],
|
|
pair_ids=[int(row.id) for row in approved_pairs],
|
|
selection_rules={"source": "mounted_runtime_wave"},
|
|
notes=note_text or "Mounted runtime preference set",
|
|
)
|
|
|
|
datasets = [
|
|
self._mounted_wave_dataset_item(role="teacher_dev_set", dataset=teacher_dev_dataset, suite_kind="teacher_dev_set"),
|
|
self._mounted_wave_dataset_item(role="teacher_eval_frozen_set", dataset=teacher_frozen_dataset, suite_kind="teacher_eval_frozen_set"),
|
|
self._mounted_wave_dataset_item(role="sft", dataset=sft_dataset),
|
|
]
|
|
if preference_dataset is not None:
|
|
datasets.append(self._mounted_wave_dataset_item(role="preference", dataset=preference_dataset))
|
|
|
|
jobs: list[dict[str, Any]] = []
|
|
for slot in normalized_slots:
|
|
sft_job = await self.create_job(
|
|
tenant_id=tenant_id,
|
|
actor_id=actor_id,
|
|
dataset_id=int(sft_dataset["id"]),
|
|
job_type="sft",
|
|
base_model_family=base_model_family,
|
|
target_slot=slot,
|
|
provider_backend=provider_backend,
|
|
)
|
|
jobs.append(self._mounted_wave_job_item(role=f"{slot}_sft", job=sft_job))
|
|
|
|
if preference_dataset is not None:
|
|
preference_job = await self.create_job(
|
|
tenant_id=tenant_id,
|
|
actor_id=actor_id,
|
|
dataset_id=int(preference_dataset["id"]),
|
|
job_type="preference",
|
|
base_model_family=base_model_family,
|
|
target_slot=slot,
|
|
provider_backend=provider_backend,
|
|
)
|
|
jobs.append(self._mounted_wave_job_item(role=f"{slot}_preference", job=preference_job))
|
|
|
|
eval_job = await self.create_job(
|
|
tenant_id=tenant_id,
|
|
actor_id=actor_id,
|
|
dataset_id=int(teacher_frozen_dataset["id"]),
|
|
job_type="eval",
|
|
base_model_family=base_model_family,
|
|
target_slot=slot,
|
|
provider_backend=provider_backend,
|
|
)
|
|
jobs.append(self._mounted_wave_job_item(role=f"{slot}_eval", job=eval_job))
|
|
|
|
executed = await self.run_jobs(
|
|
tenant_id=tenant_id,
|
|
actor_id=actor_id,
|
|
job_ids=[int(item["job_id"]) for item in jobs],
|
|
provider_backend=provider_backend,
|
|
status="queued",
|
|
limit=max(1, len(jobs)),
|
|
)
|
|
snapshots_by_id = await self._snapshot_jobs_by_ids(
|
|
tenant_id=tenant_id,
|
|
job_ids=[int(item["job_id"]) for item in jobs],
|
|
)
|
|
jobs = [
|
|
{
|
|
**item,
|
|
"status": str((snapshots_by_id.get(int(item["job_id"])) or {}).get("status") or item["status"]),
|
|
"artifact_id": (
|
|
int(((snapshots_by_id.get(int(item["job_id"])) or {}).get("artifact") or {}).get("id"))
|
|
if isinstance((snapshots_by_id.get(int(item["job_id"])) or {}).get("artifact"), dict)
|
|
and ((snapshots_by_id.get(int(item["job_id"])) or {}).get("artifact") or {}).get("id") is not None
|
|
else item["artifact_id"]
|
|
),
|
|
}
|
|
for item in jobs
|
|
]
|
|
acceptance_gate = await self.get_acceptance_gate(tenant_id=tenant_id)
|
|
return {
|
|
"version_prefix": version_prefix,
|
|
"datasets": datasets,
|
|
"jobs": jobs,
|
|
"executed_jobs": executed["items"],
|
|
"acceptance_gate": acceptance_gate,
|
|
}
|
|
|
|
async def list_datasets(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
dataset_type: Optional[str] = None,
|
|
limit: int = 20,
|
|
offset: int = 0,
|
|
) -> dict[str, Any]:
|
|
stmt = select(TrainingDataset).where(TrainingDataset.tenant_id == int(tenant_id))
|
|
if dataset_type:
|
|
stmt = stmt.where(TrainingDataset.dataset_type == str(dataset_type).strip().lower())
|
|
rows = (
|
|
await self.db.execute(
|
|
stmt.order_by(TrainingDataset.id.desc()).offset(int(offset)).limit(int(limit))
|
|
)
|
|
).scalars().all()
|
|
return {"items": [self._dataset_snapshot(row) for row in rows], "total": len(rows), "limit": int(limit), "offset": int(offset)}
|
|
|
|
async def list_jobs(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
status: Optional[str] = None,
|
|
limit: int = 20,
|
|
offset: int = 0,
|
|
) -> dict[str, Any]:
|
|
stmt = select(TrainingJob).where(TrainingJob.tenant_id == int(tenant_id))
|
|
if status:
|
|
stmt = stmt.where(TrainingJob.status == str(status).strip().lower())
|
|
rows = (
|
|
await self.db.execute(
|
|
stmt.order_by(TrainingJob.id.desc()).offset(int(offset)).limit(int(limit))
|
|
)
|
|
).scalars().all()
|
|
return {"items": [self._job_snapshot(row) for row in rows], "total": len(rows), "limit": int(limit), "offset": int(offset)}
|
|
|
|
async def list_evaluations(self, *, tenant_id: int, job_id: int) -> dict[str, Any]:
|
|
rows = (
|
|
await self.db.execute(
|
|
select(TrainingEvaluation)
|
|
.where(
|
|
TrainingEvaluation.tenant_id == int(tenant_id),
|
|
TrainingEvaluation.training_job_id == int(job_id),
|
|
)
|
|
.order_by(TrainingEvaluation.id.asc())
|
|
)
|
|
).scalars().all()
|
|
return {"items": [self._evaluation_snapshot(row) for row in rows], "total": len(rows)}
|
|
|
|
async def promote_artifact(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
artifact_id: int,
|
|
actor_id: Optional[int],
|
|
) -> dict[str, Any]:
|
|
artifact = (
|
|
await self.db.execute(
|
|
select(TrainedModelArtifact).where(
|
|
TrainedModelArtifact.tenant_id == int(tenant_id),
|
|
TrainedModelArtifact.id == int(artifact_id),
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
if artifact is None:
|
|
raise ValueError("trained model artifact not found")
|
|
artifact.status = "candidate"
|
|
row = TrainedModelDeployment(
|
|
tenant_id=int(tenant_id),
|
|
artifact_id=int(artifact.id),
|
|
deployment_status="candidate",
|
|
bound_tier=str(artifact.target_slot or "").strip().lower() or "low",
|
|
effective_scope={},
|
|
decision=None,
|
|
notes=None,
|
|
decided_by=actor_id,
|
|
decided_at=now_cn_naive().isoformat(),
|
|
)
|
|
self.db.add(row)
|
|
await self.db.commit()
|
|
await self.db.refresh(row)
|
|
return self._deployment_snapshot(row)
|
|
|
|
async def decide_deployment(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
deployment_id: int,
|
|
actor_id: Optional[int],
|
|
decision: str,
|
|
notes: Optional[str],
|
|
) -> dict[str, Any]:
|
|
row = (
|
|
await self.db.execute(
|
|
select(TrainedModelDeployment).where(
|
|
and_(
|
|
TrainedModelDeployment.tenant_id == int(tenant_id),
|
|
TrainedModelDeployment.id == int(deployment_id),
|
|
)
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
if row is None:
|
|
raise ValueError("trained model deployment not found")
|
|
normalized = str(decision or "").strip().lower()
|
|
if normalized not in {"accepted", "rejected", "rolled_back"}:
|
|
raise ValueError("unsupported decision")
|
|
if normalized == "accepted":
|
|
gate = await self.get_acceptance_gate(tenant_id=tenant_id)
|
|
if not gate["ready_for_acceptance"] and gate["blocking_reason"] != "missing_teacher_eval_frozen_set":
|
|
raise ValueError(f"deployment blocked by acceptance gate: {gate['blocking_reason']}")
|
|
row.decision = normalized
|
|
row.notes = str(notes or "").strip() or None
|
|
row.decided_by = actor_id
|
|
row.decided_at = now_cn_naive().isoformat()
|
|
row.deployment_status = "active" if normalized == "accepted" else "rollback"
|
|
await self.db.commit()
|
|
await self.db.refresh(row)
|
|
return self._deployment_snapshot(row)
|
|
|
|
async def list_deployments(
|
|
self,
|
|
*,
|
|
tenant_id: int,
|
|
deployment_status: Optional[str] = None,
|
|
limit: int = 20,
|
|
offset: int = 0,
|
|
) -> dict[str, Any]:
|
|
stmt = select(TrainedModelDeployment).where(TrainedModelDeployment.tenant_id == int(tenant_id))
|
|
if deployment_status:
|
|
stmt = stmt.where(
|
|
TrainedModelDeployment.deployment_status == str(deployment_status).strip().lower()
|
|
)
|
|
rows = (
|
|
await self.db.execute(
|
|
stmt.order_by(TrainedModelDeployment.id.desc()).offset(int(offset)).limit(int(limit))
|
|
)
|
|
).scalars().all()
|
|
return {
|
|
"items": [self._deployment_snapshot(row) for row in rows],
|
|
"total": len(rows),
|
|
"limit": int(limit),
|
|
"offset": int(offset),
|
|
}
|