Merge branch 'ft_wyt' of http://www.caoxiaozhu.com:13001/YG-Soft/YG_FT into ft_wyt

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)})

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 [],
                     }
                 )

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)})

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

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)

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")

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,

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"]

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'

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) {

frontend/src/views/eval/create/EvalTaskSetupStep.vue

@@ -104,9 +104,9 @@ defineExpose({ validate })
       <el-select v-model="form.gpu_id" placeholder="请选择 GPU" style="width: 100%" :loading="loading">
         <el-option
           v-for="gpu in gpus"
-          :key="gpu.id"
-          :label="`${gpu.name} (GPU ${gpu.id})`"
-          :value="gpu.id ?? 0"
+          :key="`${gpu.node_id || ''}:${gpu.id ?? 0}`"
+          :label="`${gpu.node_name || gpu.node_code || '算力节点'} / ${gpu.name} (GPU ${gpu.id ?? 0})`"
+          :value="`${gpu.node_id || ''}:${gpu.id ?? 0}`"
         />
       </el-select>
     </el-form-item>

frontend/src/views/eval/create/StartEvalStep.vue

@@ -1,4 +1,5 @@
 <script setup lang="ts">
+import { computed } from 'vue'
 import type { DatasetItem, GpuInfo, ModelItem, TrainedModel } from '@/types'
 import type { BasicMetricSetupDraft } from './BasicMetricSetupStep.vue'
 import type { EvalRuleSetupDraft } from './EvalRuleSetupStep.vue'
@@ -17,6 +18,15 @@ const props = defineProps<{
 function nameOf<T extends { id: string | number; name?: string }>(items: T[], id: string | number) {
   return items.find((item) => item.id === id)?.name || String(id || '-')
 }
+
+/** GPU 选择为「节点:GPU序号」复合值，解析并展示为可读标签 */
+const gpuLabel = computed(() => {
+  const key = String(props.task.gpu_id || '')
+  const gpu = props.gpus.find((g) => `${g.node_id || ''}:${g.id ?? 0}` === key)
+  if (gpu) return `${gpu.node_name || gpu.node_code || '算力节点'} / GPU ${gpu.id ?? 0}`
+  const [nodeId, idx] = key.split(':')
+  return nodeId ? `节点 ${nodeId} / GPU ${idx || 0}` : `GPU ${key || 0}`
+})
 </script>
 
 <template>
@@ -24,7 +34,7 @@ function nameOf<T extends { id: string | number; name?: string }>(items: T[], id
     <el-descriptions :column="2" border>
       <el-descriptions-item label="Task">{{ props.task.eval_task_name || '-' }}</el-descriptions-item>
       <el-descriptions-item label="Model">{{ nameOf(props.trainedModels, props.task.model_id) }}</el-descriptions-item>
-      <el-descriptions-item label="GPU">GPU {{ props.task.gpu_id || 0 }}</el-descriptions-item>
+      <el-descriptions-item label="GPU">{{ gpuLabel }}</el-descriptions-item>
       <el-descriptions-item label="Dataset">
         {{ props.task.data_source === 'dataset' ? nameOf(props.evalDatasets, props.task.dataset_id) : 'Inference results' }}
       </el-descriptions-item>