diff --git a/backend/app/api/v1/endpoints/platform.py b/backend/app/api/v1/endpoints/platform.py index fa04cca..dcbdac4 100644 --- a/backend/app/api/v1/endpoints/platform.py +++ b/backend/app/api/v1/endpoints/platform.py @@ -33,6 +33,22 @@ def _select_first_online_node(store: Any) -> dict[str, Any] | None: return None +def _select_eval_node(store: Any, preferred_node_id: str | None = None) -> dict[str, Any] | None: + """Select the compute node for an eval job. + + 被评测模型是节点相关的（训练/合并产物只存在于对应算力节点），因此优先使用 + 页面选择的节点或模型所在节点；若该节点不可用则明确失败，绝不派发到其它 + 可能没有模型路径的节点（多算力节点场景下这是评测失败的主因）。 + """ + if preferred_node_id: + node = next((n for n in store.compute_nodes() if n.get("id") == preferred_node_id), None) + if node: + if node.get("enabled") and node.get("scheduler_status") == "online": + return node + return None + return _select_first_online_node(store) + + def _candidate_online_nodes(store: Any, preferred_node_id: str | None = None) -> list[dict[str, Any]]: nodes = [node for node in store.compute_nodes() if node.get("enabled") and node.get("scheduler_status") == "online"] if not preferred_node_id: @@ -1382,13 +1398,16 @@ async def model_eval_start(payload: dict[str, Any] = Body(...)) -> dict[str, Any model_id = str(payload.get("model_id", "")) model_path = "" adapter_path = payload.get("adapter_path", "") + model_node_id = "" try: db_model = store.model(model_id) model_path = db_model.get("path", "") + model_node_id = db_model.get("compute_node_id") or "" 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: + model_node_id = trained.get("compute_node_id") or "" merged_path = trained.get("merged_path", "") base_path = trained.get("base_model_path", "") if trained.get("merged") and merged_path: @@ -1438,16 +1457,22 @@ async def model_eval_start(payload: dict[str, Any] = Body(...)) -> dict[str, Any eval_model_name = dim.get("eval_model", "") api_url = "" api_key = "" + api_model_name = "" 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", "") + if isinstance(eval_model, dict): + api_url = eval_model.get("api_url", "") + api_key = eval_model.get("api_key", "") + # 模型记录里的 model_name 是真实 API 模型名（如 deepseek-chat）， + # 优先传给评测器，避免用平台内部名称调用 LLM API + api_model_name = eval_model.get("model_name") or "" except (KeyError, Exception): pass dimension_cfg = { "type": dim.get("type", ""), "eval_model": eval_model_name, + "api_model": api_model_name or eval_model_name, "eval_method": dim.get("eval_method", ""), "eval_prompt": dim.get("eval_prompt", ""), "api_url": api_url, @@ -1459,11 +1484,13 @@ async def model_eval_start(payload: dict[str, Any] = Body(...)) -> dict[str, Any except KeyError: pass - # 5. Select compute node - node = _select_first_online_node(store) + # 5. Select compute node: 优先页面选择的节点 / 模型所在节点，避免多节点时选错 + preferred_node_id = payload.get("compute_node_id") or payload.get("node_id") or model_node_id + node = _select_eval_node(store, preferred_node_id) 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"}) + message = "no online compute node" if not preferred_node_id else f"model compute node not schedulable: {preferred_node_id}" + 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']}" @@ -1510,8 +1537,8 @@ async def model_eval_start(payload: dict[str, Any] = Body(...)) -> dict[str, Any "compute_node_id": node["id"], "output_dir": output_dir, }) - if job.get("status") in {"queued", "running"}: - store.mark_inference_loaded(node["id"]) + # 评测占用 GPU 由 eval_tasks 派生（gpus()/compute_nodes() 直接统计）， + # 不再复用 mark_inference_loaded 内存标记，避免删除评测后 GPU 状态残留 busy 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)}) diff --git a/backend/app/db/platform_store.py b/backend/app/db/platform_store.py index 02835d4..c08b877 100644 --- a/backend/app/db/platform_store.py +++ b/backend/app/db/platform_store.py @@ -2837,6 +2837,28 @@ class PlatformStore: "SELECT compute_node_id, COUNT(*) AS cnt FROM fine_tune_tasks WHERE status IN ('syncing','queued','running') GROUP BY compute_node_id" ).fetchall() running_map = {r["compute_node_id"]: r["cnt"] for r in running} + # 评测任务同样占用算力节点，纳入运行任务统计 + for row in conn.execute( + "SELECT payload FROM eval_tasks WHERE status IN ('syncing','queued','running')" + ).fetchall(): + node_id = json_loads(row["payload"], {}).get("compute_node_id") + if node_id: + running_map[node_id] = running_map.get(node_id, 0) + 1 + # 推理模型占用算力节点同样计入：优先从 compare_tasks 持久化状态派生 + # （重启后仍准确），并用内存标记兜底（直接 preload 的模型无 compare 记录） + inference_node_ids = set(self._inference_nodes) + for ctr in conn.execute("SELECT payload FROM compare_tasks").fetchall(): + ls = json_loads(ctr["payload"], {}).get("load_status") or {} + if isinstance(ls, str): + try: + ls = json.loads(ls) + except (json.JSONDecodeError, TypeError): + ls = {} + for m in ls.get("loaded_models") or []: + if m.get("status") in {"ready", "running"} and m.get("node_id"): + inference_node_ids.add(m["node_id"]) + for nid in inference_node_ids: + running_map[nid] = running_map.get(nid, 0) + 1 rows = conn.execute("SELECT * FROM compute_nodes ORDER BY scheduler_weight DESC, code").fetchall() return [ { @@ -3054,6 +3076,26 @@ class PlatformStore: "SELECT * FROM fine_tune_tasks WHERE status IN ('syncing','queued','running')" ).fetchall() ] + # 评测任务同样占用节点 GPU + eval_running = [ + json_loads(row["payload"], {}) + for row in conn.execute( + "SELECT payload FROM eval_tasks WHERE status IN ('syncing','queued','running')" + ).fetchall() + ] + # 推理模型占用的节点：优先从 compare_tasks 持久化状态派生（重启后仍准确）， + # 内存标记兜底（直接 preload 的模型无 compare 记录） + inference_node_ids = set(self._inference_nodes) + for ctr in conn.execute("SELECT payload FROM compare_tasks").fetchall(): + ls = json_loads(ctr["payload"], {}).get("load_status") or {} + if isinstance(ls, str): + try: + ls = json.loads(ls) + except (json.JSONDecodeError, TypeError): + ls = {} + for m in ls.get("loaded_models") or []: + if m.get("status") in {"ready", "running"} and m.get("node_id"): + inference_node_ids.add(m["node_id"]) items = [] for row in rows: task = next( @@ -3064,11 +3106,21 @@ class PlatformStore: ), None, ) - busy = task is not None and task.get("status") == "running" - reserved = task is not None and task.get("status") in {"syncing", "queued"} + eval_task = next( + ( + t + for t in eval_running + if t.get("compute_node_id") == row["node_id"] + and row["gpu_index"] == (int(t["gpu_id"]) if t.get("gpu_id") is not None else -1) + ), + None, + ) + busy = (task is not None and task.get("status") == "running") or eval_task is not None + reserved = (task is not None and task.get("status") in {"syncing", "queued"}) or ( + eval_task is not None and eval_task.get("status") in {"syncing", "queued"} + ) # Also mark GPU as busy if an inference model is loaded on this node - inference_busy = self.is_inference_loaded(row["node_id"]) - if inference_busy and not busy: + if row["node_id"] in inference_node_ids and not busy: busy = True reserved = False memory_used = round(row["memory_total_gb"] * (0.72 if busy else 0.18 if reserved else 0.04), 1) @@ -3100,6 +3152,16 @@ class PlatformStore: } ] if task + else [ + { + "pid": int(eval_task.get("process_id") or 0), + "name": "eval_runner", + "memory_used_gb": memory_used, + "task_name": eval_task.get("eval_task_name") or eval_task.get("name") or "评测任务", + "user": "admin", + } + ] + if eval_task else [], } ) diff --git a/backend/app/modules/compute_gateway/sync.py b/backend/app/modules/compute_gateway/sync.py index 4019ba2..1cac998 100644 --- a/backend/app/modules/compute_gateway/sync.py +++ b/backend/app/modules/compute_gateway/sync.py @@ -174,9 +174,7 @@ async def poll_compute_jobs_once() -> dict[str, Any]: 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"]) + # 评测 GPU 占用由 eval_tasks 状态派生，无需维护推理内存标记 eval_synced += 1 except Exception as exc: # noqa: BLE001 failed.append({"eval_task_id": eval_task["id"], "error": str(exc)}) diff --git a/backend/requirements.txt b/backend/requirements.txt index 04e346a..2065541 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -19,3 +19,7 @@ llama-index-core==0.14.23 llama-index-embeddings-huggingface==0.6.1 docling==2.115.0 tiktoken>=0.7.0 + +# 测试与代码检查 +pytest>=8.2.0 +ruff>=0.5.0 diff --git a/backend/tests/test_compare_inference_async.py b/backend/tests/test_compare_inference_async.py index 075e9ac..792d003 100644 --- a/backend/tests/test_compare_inference_async.py +++ b/backend/tests/test_compare_inference_async.py @@ -100,6 +100,28 @@ def _patch_store(monkeypatch, store: FakeInferenceStore) -> None: monkeypatch.setattr(platform, "get_settings", lambda: SimpleNamespace(compute_mode="real")) +def test_select_eval_node_prefers_model_node(monkeypatch) -> None: + from app.api.v1.endpoints.platform import _select_eval_node + + store = FakeInferenceStore(nodes=[_node("n1"), _node("n2")]) + # 指定模型所在节点时优先返回该节点 + assert _select_eval_node(store, "n2")["id"] == "n2" + # 无指定节点时回退到第一个在线节点 + assert _select_eval_node(store, None)["id"] == "n1" + + +def test_select_eval_node_returns_none_when_model_node_offline(monkeypatch) -> None: + from app.api.v1.endpoints.platform import _select_eval_node + + nodes = [_node("n1"), _node("n2")] + nodes[1]["enabled"] = False + store = FakeInferenceStore(nodes=nodes) + # 模型所在节点不可用 → 明确失败，不派发到其它节点 + assert _select_eval_node(store, "n2") is None + # 无指定节点时仍回退第一个在线节点 + assert _select_eval_node(store, None)["id"] == "n1" + + def test_model_compare_load_dispatches_and_returns_starting(monkeypatch) -> None: store = FakeInferenceStore(tasks=[_task("t1", node_id="n1")], nodes=[_node("n1")]) _patch_store(monkeypatch, store) diff --git a/compute/engines/llama_factory/eval_runner.py b/compute/engines/llama_factory/eval_runner.py index 3c974a8..7681817 100644 --- a/compute/engines/llama_factory/eval_runner.py +++ b/compute/engines/llama_factory/eval_runner.py @@ -183,6 +183,9 @@ def _judge_sample( api_url = (config.get("api_url") or "").strip().rstrip("/") api_key = (config.get("api_key") or "").strip() eval_model = (config.get("eval_model") or "").strip() + # 优先使用模型记录里配置的真实 API 模型名（如 deepseek-chat）， + # 否则回退到平台内部模型名 + api_model = (config.get("api_model") or "").strip() or eval_model eval_prompt = (config.get("eval_prompt") or "").strip() score_min = float(config.get("score_min", 0)) score_max = float(config.get("score_max", 5)) @@ -208,7 +211,7 @@ def _judge_sample( import urllib.error body = json.dumps({ - "model": eval_model, + "model": api_model, "messages": [ {"role": "system", "content": system_msg}, {"role": "user", "content": user_msg}, @@ -308,13 +311,15 @@ def run_eval(config: dict[str, Any]) -> dict[str, Any]: print(f"[eval] loading model: {model_path}") from compute.engines.llama_factory.inference import InferenceSession session = InferenceSession() - load_result = session.load( + session.load( model_name_or_path=model_path, adapter_name_or_path=adapter_path, template=template, infer_backend=config.get("infer_backend", "huggingface"), infer_dtype=config.get("infer_dtype", "auto"), ) + # load() 为异步加载（立即返回 loading），必须等待后台线程完成后再进行推理 + load_result = session.wait_until_loaded(timeout=float(config.get("load_timeout", 1800))) if not load_result.get("loaded"): raise RuntimeError(f"model load failed: {load_result.get('error', 'unknown')}") print(f"[eval] model loaded OK") diff --git a/compute/engines/llama_factory/inference.py b/compute/engines/llama_factory/inference.py index dee77cd..72cad82 100644 --- a/compute/engines/llama_factory/inference.py +++ b/compute/engines/llama_factory/inference.py @@ -55,6 +55,31 @@ class InferenceSession: "error": self._error, } + def wait_until_loaded(self, timeout: float | None = None) -> dict[str, Any]: + """Wait for an in-flight async load to finish and return its outcome. + + 供同步消费方（如 eval_runner 子进程）使用：``load()`` 立即返回 loading 后， + 调用本方法等待后台加载线程完成，拿到最终的 loaded/error 结果。 + 若在 timeout 秒内仍未加载完成，返回 ``status == "loading"`` 并附上超时提示。 + """ + with self._state_lock: + thread = self._load_thread + if thread is not None and thread.is_alive(): + thread.join(timeout=timeout) + with self._state_lock: + loaded = self._status == "ready" + status = self._status + error = self._error + if not loaded and status == "loading": + error = error or f"model load timed out after {timeout or 'N/A'}s" + return { + "loaded": loaded, + "status": status, + "model_name": self._model_name, + "adapter_path": self._adapter_path, + "error": error, + } + def load( self, model_name_or_path, diff --git a/compute/tests/test_inference_session.py b/compute/tests/test_inference_session.py index ab55514..0ac38e9 100644 --- a/compute/tests/test_inference_session.py +++ b/compute/tests/test_inference_session.py @@ -125,3 +125,22 @@ def test_chat_stream_while_loading_yields_error(stub_llamafactory) -> None: session.load("/models/qwen") chunks = list(session.chat_stream([{"role": "user", "content": "hi"}])) assert any("still loading" in c for c in chunks) + + +def test_wait_until_loaded_blocks_until_ready(stub_llamafactory) -> None: + session = InferenceSession() + result = session.load("/models/qwen") + assert result["status"] == "loading" + # 同步等待后台加载线程完成 + outcome = session.wait_until_loaded(timeout=3.0) + assert outcome["loaded"] is True + assert outcome["status"] == "ready" + + +def test_wait_until_loaded_reports_load_error(stub_failing_llamafactory) -> None: + session = InferenceSession() + session.load("/models/bad") + outcome = session.wait_until_loaded(timeout=3.0) + assert outcome["loaded"] is False + assert outcome["status"] == "error" + assert "boom" in outcome["error"] diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index 6bac76a..4c93efa 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -283,6 +283,7 @@ export interface StartEvalPayload { eval_type: EvalType model_id: string | number gpu_id: string | number + compute_node_id?: string dataset_id: string | number dimension_id: string | number data_source: 'dataset' | 'inference' diff --git a/frontend/src/views/eval/EvalCreateView.vue b/frontend/src/views/eval/EvalCreateView.vue index 7c0b75e..1e82048 100644 --- a/frontend/src/views/eval/EvalCreateView.vue +++ b/frontend/src/views/eval/EvalCreateView.vue @@ -143,11 +143,15 @@ async function handleSubmit() { submitting.value = true try { const dimensionId = await resolveDimensionId() - await startEval({ + // GPU 选择为「节点:GPU序号」复合值，解析出节点与 GPU 序号， + // 多算力节点时必须把节点信息传给后端，否则会派发到错误的算力节点 + const [gpuNodeId, gpuIndex] = String(taskForm.value.gpu_id).split(':') + const evalResult: any = await startEval({ eval_task_name: taskForm.value.eval_task_name, eval_type: 'custom', model_id: taskForm.value.model_id, - gpu_id: taskForm.value.gpu_id, + gpu_id: Number(gpuIndex) || 0, + compute_node_id: gpuNodeId || '', dataset_id: taskForm.value.data_source === 'dataset' ? taskForm.value.dataset_id : '', dimension_id: dimensionId, data_source: taskForm.value.data_source, @@ -167,6 +171,10 @@ async function handleSubmit() { output_precision: basicMetricForm.value.output_precision, }, }) + if (evalResult?.status === 'failed' || evalResult?.error) { + ElMessage.error(`评测启动失败：${evalResult?.error || '请检查算力节点与模型路径'}`) + return + } ElMessage.success('评测任务已创建并启动') router.push('/model-eval') } catch (error) { diff --git a/frontend/src/views/eval/create/EvalTaskSetupStep.vue b/frontend/src/views/eval/create/EvalTaskSetupStep.vue index 94196d5..8b595c5 100644 --- a/frontend/src/views/eval/create/EvalTaskSetupStep.vue +++ b/frontend/src/views/eval/create/EvalTaskSetupStep.vue @@ -104,9 +104,9 @@ defineExpose({ validate }) diff --git a/frontend/src/views/eval/create/StartEvalStep.vue b/frontend/src/views/eval/create/StartEvalStep.vue index 8c48e6d..ba07946 100644 --- a/frontend/src/views/eval/create/StartEvalStep.vue +++ b/frontend/src/views/eval/create/StartEvalStep.vue @@ -1,4 +1,5 @@