From 0292bf5138558dc136808f5840ef8ad92baa3ec7 Mon Sep 17 00:00:00 2001 From: wuyongtao Date: Tue, 4 Aug 2026 18:21:16 +0800 Subject: [PATCH 1/2] =?UTF-8?q?fix:=20=E6=A8=A1=E5=9E=8B=E8=AF=84=E6=B5=8B?= =?UTF-8?q?=E5=BC=82=E6=AD=A5=E5=8A=A0=E8=BD=BD=E7=AD=89=E5=BE=85=E4=B8=8E?= =?UTF-8?q?=E5=A4=9A=E8=8A=82=E7=82=B9=E8=B7=AF=E7=94=B1=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - eval_runner 等待异步模型加载完成（InferenceSession.wait_until_loaded）， 修复 "model load failed: unknown" - 评测算力节点选择：优先页面选择的节点 / 模型所在节点（_select_eval_node）， 多节点时不再派发到不可达节点导致连接超时 - 前端评测 GPU 选择改为节点感知（节点:GPU 复合值），透传 compute_node_id， 并检查 startEval 结果展示真实错误 - 大模型评价（judge）使用模型记录的真实 API 模型名（api_model）， 避免用平台内部名调用 LLM API 导致 HTTP 400 - 新增后端节点选择与 compute wait_until_loaded 单元测试 Co-Authored-By: Claude --- backend/app/api/v1/endpoints/platform.py | 39 ++++++++++++++++--- backend/tests/test_compare_inference_async.py | 22 +++++++++++ compute/engines/llama_factory/eval_runner.py | 9 ++++- compute/engines/llama_factory/inference.py | 25 ++++++++++++ compute/tests/test_inference_session.py | 19 +++++++++ frontend/src/types/index.ts | 1 + frontend/src/views/eval/EvalCreateView.vue | 12 +++++- .../views/eval/create/EvalTaskSetupStep.vue | 6 +-- .../src/views/eval/create/StartEvalStep.vue | 12 +++++- 9 files changed, 131 insertions(+), 14 deletions(-) diff --git a/backend/app/api/v1/endpoints/platform.py b/backend/app/api/v1/endpoints/platform.py index fa04cca..69a4f3a 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']}" 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 @@