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

feat: 模型评测端到端闭环 — EvalRunner引擎 + 算力节点Job执行 + 结果回写

Browse Source 
算力节点 (compute):
- 新建 eval_runner.py: 评测执行引擎，作为subprocess运行
  - 加载模型 + JSONL数据集 + 逐样本推理
  - BLEU/ROUGE/Cosine基础指标计算
  - LLM Judge评分(OpenAI兼容API调用)
  - 结果写入eval_results.json
- adapter.py: build_command新增engine=eval分支
- main.py: 新增/json模块导入，新增/compute/files/read端点，eval job校验

后端:
- platform.py: 重写startEval提交eval job到算力节点
  - 支持models表和trained_models表查找
  - 已合并模型不传adapter路径
- platform_store.py: 新增update_eval_task/running_eval_tasks/apply_eval_job_result
- sync.py: poller新增eval job同步，异步读取eval_results.json回写结果

前端:
- EvalCreateView/DimensionCreateView: eval模型过滤扩展(API类型+api_url)
- EvalCreateView: GPU过滤在线节点空闲GPU
- EvalTaskSetupStep: GPU value从数组index改为gpu.id
- BasicMetricSetupStep: ROUGE方法名修正(rouge_1→rouge1)
- EvalView: 新增5秒轮询刷新

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
wuyongtao
2026-07-28 19:34:41 +08:00
parent c7c9ed925b
commit 0c39f2f5b9
12 changed files with 903 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1000,15 +1000,175 @@ async def model_eval_list() -> dict[str, Any]:
@router.get("/model-eval/{task_id}")
async def model_eval_detail(task_id: str) -> dict[str, Any]:
try:
return ok(get_platform_store().eval_task(task_id))
store = get_platform_store()
task = store.eval_task(task_id)
# If the eval job completed on a compute node, try to load results
if task.get("result_artifact_path"):
node = next(
(n for n in store.compute_nodes() if n["id"] == task.get("compute_node_id")),
None,
)
if node:
try:
client = ComputeNodeClient(node["api_base_url"])
job = await client.get_job(task["compute_job_id"])
artifacts = job.get("artifacts") or []
for art in artifacts:
if art.get("name") == "eval_results.json":
task["_result_artifact"] = art
break
except Exception:
pass
return ok(task)
except KeyError:
raise fail(404, "eval task not found")
@router.post("/model-eval/start")
async def model_eval_start(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
task = get_platform_store().create_eval_task(payload)
return ok({"task_id": task["id"], **task})
"""Start an evaluation task: submit eval job to compute node."""
store = get_platform_store()
# 1. Create eval task record
task = store.create_eval_task({**payload, "status": "pending"})
# 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", "")
try:
db_model = store.model(model_id)
model_path = db_model.get("path", "")
except KeyError:
# 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:
merged_path = trained.get("merged_path", "")
base_path = trained.get("base_model_path", "")
if trained.get("merged") and merged_path:
# Merged model: use merged_path as model, no adapter needed
model_path = merged_path
elif base_path:
# Unmerged: use base model + adapter checkpoint
model_path = base_path
if merged_path:
adapter_path = merged_path
else:
model_path = merged_path 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"})
# 3. Resolve dataset file
dataset_id = str(payload.get("dataset_id", ""))
dataset_path = ""
try:
ds_files = store.training_dataset_files(dataset_id)
if ds_files:
dataset_path = ds_files[0].get("local_path") or ds_files[0].get("name", "")
except Exception:
pass
if not dataset_path:
# Try to get file content and sync to compute
try:
ds = store.dataset(dataset_id)
for f in ds.get("files", []):
if f.get("content"):
dataset_path = f.get("name", f"dataset_{dataset_id}.jsonl")
break
except KeyError:
pass
if not dataset_path:
store.update_eval_task(task["id"], {"status": "failed", "error": "dataset not found or no files"})
return ok({"task_id": task["id"], "status": "failed", "error": "dataset not found or no files"})
# 4. Resolve dimension config
dimension_id = str(payload.get("dimension_id", ""))
dimension_cfg: dict[str, Any] = {}
if dimension_id:
try:
dim = store.dimension(dimension_id)
# Resolve eval model API config
eval_model_name = dim.get("eval_model", "")
api_url = ""
api_key = ""
if eval_model_name:
try:
eval_model = store.model(eval_model_name) if eval_model_name.startswith("m_") else store.model_by_name(eval_model_name)
api_url = eval_model.get("api_url", "")
api_key = eval_model.get("api_key", "")
except (KeyError, Exception):
pass
dimension_cfg = {
"type": dim.get("type", ""),
"eval_model": eval_model_name,
"eval_method": dim.get("eval_method", ""),
"eval_prompt": dim.get("eval_prompt", ""),
"api_url": api_url,
"api_key": api_key,
"score_min": dim.get("score_min", 0),
"score_max": dim.get("score_max", 5),
"pass_threshold": dim.get("pass_threshold", 3),
}
except KeyError:
pass
# 5. Select compute node
node = _select_first_online_node(store)
if not node:
store.update_eval_task(task["id"], {"status": "failed", "error": "no online compute node"})
return ok({"task_id": task["id"], "status": "failed", "error": "no online compute node"})
# 6. Build eval job payload
output_dir = f"/data/yg-ft/outputs/{task['id']}"
job_payload = {
"id": f"eval_{task['id']}",
"name": task.get("eval_task_name", task["id"]),
"engine": "eval",
"model_name_or_path": model_path,
"adapter_name_or_path": adapter_path,
"template": payload.get("template", "qwen"),
"dataset_path": dataset_path,
"output_dir": output_dir,
"basic_metrics": payload.get("basic_metrics", {}),
"dimension": dimension_cfg,
"gpus": [int(payload.get("gpu_id", 0))],
"temperature": payload.get("temperature", 0.1),
"max_new_tokens": payload.get("max_new_tokens", 512),
"compute_node_id": node["id"],
}
# 7. Submit to compute node via create_job (uses engine="eval" path)
try:
client = ComputeNodeClient(node["api_base_url"])
# Sync dataset file to compute node if needed
if not dataset_path.startswith("/"):
try:
ds_files = store.training_dataset_files(dataset_id)
if ds_files and ds_files[0].get("content"):
upload_result = await client.upload_file(
ds_files[0].get("name", "eval_data.jsonl"),
ds_files[0]["content"].encode("utf-8"),
f"datasets/{dataset_id}/{ds_files[0].get('name', 'eval_data.jsonl')}",
resource_type="dataset",
resource_id=dataset_id,
)
job_payload["dataset_path"] = upload_result.get("local_path", dataset_path)
except Exception:
pass
job = await client.create_job(job_payload)
store.update_eval_task(task["id"], {
"status": "running",
"compute_job_id": job.get("id"),
"compute_node_id": node["id"],
"output_dir": output_dir,
})
if job.get("status") in {"queued", "running"}:
store.mark_inference_loaded(node["id"])
return ok({"task_id": task["id"], "status": "running", "job": job})
except Exception as exc:
store.update_eval_task(task["id"], {"status": "failed", "error": str(exc)})
return ok({"task_id": task["id"], "status": "failed", "error": str(exc)})
@router.delete("/model-eval/{task_id}")

45
backend/app/db/platform_store.py
View File

@@ -2076,10 +2076,55 @@ class PlatformStore:
)
return self.eval_task(task_id)
def update_eval_task(self, task_id: str, updates: dict[str, Any]) -> dict[str, Any]:
"""Update fields in an eval task's payload without replacing the whole record."""
task = self.eval_task(task_id)
merged = {**task, **updates}
with self.connect() as conn:
conn.execute(
"UPDATE eval_tasks SET payload=?, status=? WHERE id=?",
(json_dumps(merged), merged.get("status", task.get("status", "pending")), task_id),
)
return self.eval_task(task_id)
def delete_eval_task(self, task_id: str) -> None:
with self.connect() as conn:
conn.execute("DELETE FROM eval_tasks WHERE id=?", (task_id,))
def running_eval_tasks(self) -> list[dict[str, Any]]:
"""Return eval tasks that have been submitted to a compute node and are still running."""
return [
task for task in self.eval_tasks()
if task.get("compute_job_id") and task.get("status") in {"queued", "running"}
]
def apply_eval_job_result(self, task_id: str, job: dict[str, Any], result_content: dict[str, Any] | None = None) -> dict[str, Any]:
"""Sync a compute job status/result back to an eval task."""
task = self.eval_task(task_id)
job_status = str(job.get("status", ""))
status_map = {"queued": "running", "running": "running", "completed": "completed",
"failed": "failed", "stopped": "stopped"}
new_status = status_map.get(job_status, job_status or task.get("status", "pending"))
updates: dict[str, Any] = {
"status": new_status,
"progress": int(job.get("progress", 0)),
"output_dir": job.get("output_dir", task.get("output_dir", "")),
}
# On completion, populate results from eval_results.json content
if new_status == "completed" and result_content:
updates.update({
"overall_score": result_content.get("overall_score", 0),
"overall_score_max": result_content.get("overall_score_max", 100),
"overall_evaluation": result_content.get("overall_evaluation", ""),
"improvement_suggestions": result_content.get("improvement_suggestions", []),
"dimension_summary": result_content.get("dimension_summary", []),
"samples": result_content.get("samples", []),
"sample_count": result_content.get("sample_count", 0),
"completed_count": result_content.get("completed_count", 0),
"passed_count": result_content.get("passed_count", 0),
})
return self.update_eval_task(task_id, updates)
def dimensions(self) -> list[dict[str, Any]]:
with self.connect() as conn:
rows = conn.execute("SELECT * FROM eval_dimensions ORDER BY create_time DESC").fetchall()

43
backend/app/modules/compute_gateway/sync.py
View File

@@ -48,4 +48,45 @@ async def poll_compute_jobs_once() -> dict[str, Any]:
standalone_synced.append(store.sync_model_merge_job(record["id"], job))
except Exception as exc: # noqa: BLE001 - keep polling other jobs
failed.append({"job_id": record["id"], "error": str(exc)})
return {"synced": len(synced) + len(standalone_synced), "failed": failed, "items": synced, "standalone": standalone_synced}
# ── Eval job sync ────────────────────────────────────────────────
eval_synced = 0
for eval_task in store.running_eval_tasks():
node = next(
(item for item in store.compute_nodes() if item["id"] == eval_task.get("compute_node_id")),
None,
)
if not node:
failed.append({"eval_task_id": eval_task["id"], "error": "compute node not found"})
continue
try:
client = ComputeNodeClient(node["api_base_url"])
job = await client.get_job(eval_task["compute_job_id"])
result_content = None
# Try to read eval_results.json from the job output directory
if job.get("status") == "completed" and job.get("output_dir"):
try:
full_path = f"{job['output_dir'].rstrip('/')}/eval_results.json"
# Convert absolute path to relative (strip YG_FT_DATA_ROOT prefix)
data_root = "/data/yg-ft/"
if full_path.startswith(data_root):
full_path = full_path[len(data_root):]
rel_path = full_path.lstrip("/")
import httpx
settings_path = f"{node['api_base_url'].rstrip('/')}/modelTF/compute/files/read"
async with httpx.AsyncClient(timeout=30, headers=client.headers()) as http:
read_resp = await http.get(settings_path, params={"path": rel_path})
if read_resp.status_code == 200:
result_content = read_resp.json()
except Exception:
pass
store.apply_eval_job_result(eval_task["id"], job, result_content)
# If job completed, un-mark inference loaded
if job.get("status") in {"completed", "failed", "stopped"}:
store.mark_inference_unloaded(node["id"])
eval_synced += 1
except Exception as exc: # noqa: BLE001
failed.append({"eval_task_id": eval_task["id"], "error": str(exc)})
return {"synced": len(synced) + len(standalone_synced) + eval_synced, "failed": failed,
"items": synced, "standalone": standalone_synced, "eval_synced": eval_synced}