feat: 推理与评测支持多卡 GPU 选择,训练任务实时 GPU 监控

- 新增 GPU 选择归一化 helper,统一前端各形态的选择(gpu_indices/gpus/gpu_id)
- 评测与推理支持同一节点内多卡选择,校验所选 GPU 空闲后再派发
- 新增 /fine-tune/{id}/gpu-status 接口,训练日志页展示实时 GPU 指标
- 算力节点推理加载支持 CUDA_VISIBLE_DEVICES 多卡可见,nvidia-smi 进程级监控
- GPU 占用跟踪细化为按卡记录,覆盖评测任务、推理模型与对比任务

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
wuyongtao
2026-08-19 10:44:56 +08:00
parent c3e96ae61b
commit 9379f93633
13 changed files with 360 additions and 48 deletions

View File

@@ -57,6 +57,33 @@ def _select_first_online_node(store: Any) -> dict[str, Any] | None:
return None return None
def _normalize_gpu_indices(payload: dict[str, Any], *, allow_primary: bool = True) -> list[int]:
"""Normalize all frontend GPU selection shapes to sorted integer indexes."""
raw = payload.get("gpu_indices")
if raw is None:
raw = payload.get("gpus")
if raw is None and allow_primary and payload.get("gpu_id") is not None:
raw = [payload.get("gpu_id")]
if raw is None or raw == "":
return []
if isinstance(raw, str):
raw = [item.strip() for item in raw.split(",") if item.strip()]
if not isinstance(raw, (list, tuple, set)):
raw = [raw]
result: set[int] = set()
for item in raw:
if isinstance(item, str) and ":" in item:
item = item.rsplit(":", 1)[-1]
try:
index = int(item)
except (TypeError, ValueError) as exc:
raise ValueError(f"invalid GPU index: {item}") from exc
if index < 0:
raise ValueError("GPU index must be non-negative")
result.add(index)
return sorted(result)
async def _wait_for_object_storage() -> None: async def _wait_for_object_storage() -> None:
"""Wait for MinIO before starting a resource task.""" """Wait for MinIO before starting a resource task."""
settings = get_settings() settings = get_settings()
@@ -1527,10 +1554,10 @@ async def start_fine_tune(
if node_id and gpu_indices: if node_id and gpu_indices:
if not store.check_gpu_access(current_user["id"], node_id, gpu_indices): if not store.check_gpu_access(current_user["id"], node_id, gpu_indices):
raise fail(403, "无权使用所选 GPU请联系管理员分配") raise fail(403, "无权使用所选 GPU请联系管理员分配")
# 记录创建者
if node_id and not gpu_indices: if node_id and not gpu_indices:
payload["allowed_gpu_indices"] = store.assigned_gpu_indexes(current_user["id"], node_id) payload["allowed_gpu_indices"] = store.assigned_gpu_indexes(current_user["id"], node_id)
payload["strict_node_selection"] = bool(node_id) # 页面明确选择节点时,调度器必须保持节点约束;否则可能落到其它节点。
payload["strict_node_selection"] = bool(payload.get("compute_node_id") or payload.get("node_id"))
payload.setdefault("created_by", current_user.get("id")) payload.setdefault("created_by", current_user.get("id"))
try: try:
return ok(await _submit_fine_tune_task(store, payload)) return ok(await _submit_fine_tune_task(store, payload))
@@ -1661,6 +1688,42 @@ async def fine_tune_diagnostics(task_id: str) -> dict[str, Any]:
) )
@router.get("/fine-tune/{task_id}/gpu-status")
async def fine_tune_gpu_status(task_id: str, current_user: dict[str, Any] = Depends(get_current_user)) -> dict[str, Any]:
"""Return live GPU metrics for the task's selected node and cards."""
store = get_platform_store()
try:
task = store.task(task_id)
except KeyError:
raise fail(404, "fine tune task not found")
if not has_resource_access("fine-tune", task_id, current_user, "read"):
raise fail(403, "no permission to access this task")
node = _node_for_task(task)
selected = set(_normalize_gpu_indices({"gpus": task.get("gpus") or []}, allow_primary=False))
if not node:
return ok({"source": "unavailable", "items": [], "selected_gpus": sorted(selected)})
try:
if get_settings().compute_mode == "simulator":
live_items = store.gpus()
else:
live_items = await ComputeNodeClient(node["api_base_url"]).gpu_resources()
items = []
for item in live_items:
index = int(item.get("gpu_index", item.get("id", -1)))
if selected and index not in selected:
continue
items.append({
**item,
"id": index,
"node_id": node["id"],
"node_code": node.get("code"),
"node_name": node.get("name"),
})
return ok({"source": "compute", "items": items, "selected_gpus": sorted(selected)})
except Exception as exc: # noqa: BLE001 - let UI retain last good snapshot
return ok({"source": "unavailable", "items": [], "selected_gpus": sorted(selected), "error": str(exc)})
@router.put("/fine-tune/{task_id}") @router.put("/fine-tune/{task_id}")
async def update_fine_tune(task_id: str, payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]: async def update_fine_tune(task_id: str, payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
if not has_resource_access("fine-tune", task_id, current_user, "write"): if not has_resource_access("fine-tune", task_id, current_user, "write"):
@@ -1808,6 +1871,12 @@ async def model_eval_detail(task_id: str, current_user: dict = Depends(get_curre
async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]: async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
"""Start an evaluation task: submit eval job to compute node.""" """Start an evaluation task: submit eval job to compute node."""
store = get_platform_store() store = get_platform_store()
try:
gpu_indices = _normalize_gpu_indices(payload)
except ValueError as exc:
raise fail(400, str(exc))
if not gpu_indices:
raise fail(400, "请选择至少一张 GPU")
# 1. Create eval task record # 1. Create eval task record
payload.setdefault("created_by", current_user.get("id")) payload.setdefault("created_by", current_user.get("id"))
task = store.create_eval_task({**payload, "status": "pending"}) task = store.create_eval_task({**payload, "status": "pending"})
@@ -1910,6 +1979,17 @@ async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: di
store.update_eval_task(task["id"], {"status": "failed", "error": message}) store.update_eval_task(task["id"], {"status": "failed", "error": message})
return ok({"task_id": task["id"], "status": "failed", "error": message}) return ok({"task_id": task["id"], "status": "failed", "error": message})
node_gpus = {
int(item.get("id", item.get("gpu_index", -1))): item
for item in store.gpus()
if item.get("node_id") == node["id"]
}
unavailable = [index for index in gpu_indices if node_gpus.get(index, {}).get("status") != "idle"]
if unavailable:
message = f"selected GPU is not idle on compute node {node.get('code')}: {unavailable}"
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 # 6. Build eval job payload
output_dir = f"/data/yg-ft/outputs/{task['id']}" output_dir = f"/data/yg-ft/outputs/{task['id']}"
job_payload = { job_payload = {
@@ -1923,7 +2003,9 @@ async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: di
"output_dir": output_dir, "output_dir": output_dir,
"basic_metrics": payload.get("basic_metrics", {}), "basic_metrics": payload.get("basic_metrics", {}),
"dimension": dimension_cfg, "dimension": dimension_cfg,
"gpus": [int(payload.get("gpu_id", 0))], "gpu_id": gpu_indices[0],
"gpu_indices": gpu_indices,
"gpus": gpu_indices,
"temperature": payload.get("temperature", 0.1), "temperature": payload.get("temperature", 0.1),
"max_new_tokens": payload.get("max_new_tokens", 512), "max_new_tokens": payload.get("max_new_tokens", 512),
"compute_node_id": node["id"], "compute_node_id": node["id"],
@@ -2222,6 +2304,15 @@ async def model_compare_load(task_id: str, current_user: dict = Depends(get_curr
"model_name_or_path": model_path, "model_name_or_path": model_path,
"template": item.get("template", "qwen"), "template": item.get("template", "qwen"),
} }
try:
item_gpu_indices = _normalize_gpu_indices(item)
except ValueError as exc:
loaded_models.append({**item, "status": "error", "error": str(exc)})
continue
if not item_gpu_indices:
loaded_models.append({**item, "status": "error", "error": "no GPU selected"})
continue
load_payload["gpu_indices"] = item_gpu_indices
if item.get("adapter_path"): if item.get("adapter_path"):
load_payload["adapter_name_or_path"] = item["adapter_path"] load_payload["adapter_name_or_path"] = item["adapter_path"]
if get_settings().compute_mode == "simulator": if get_settings().compute_mode == "simulator":
@@ -2230,14 +2321,28 @@ async def model_compare_load(task_id: str, current_user: dict = Depends(get_curr
# 只派发HTTP 响应成功即视为已接受节点会异步加载loaded 字段忽略 # 只派发HTTP 响应成功即视为已接受节点会异步加载loaded 字段忽略
item_dispatched = False item_dispatched = False
errors = [] errors = []
for node in _candidate_online_nodes(store, preferred_node_id): candidate_nodes = _candidate_online_nodes(store, preferred_node_id)
if preferred_node_id:
candidate_nodes = candidate_nodes[:1]
for node in candidate_nodes:
try: try:
node_gpu_map = {
int(gpu.get("id", gpu.get("gpu_index", -1))): gpu
for gpu in store.gpus()
if gpu.get("node_id") == node["id"]
}
unavailable = [
index for index in item_gpu_indices
if node_gpu_map.get(index, {}).get("status") != "idle"
]
if unavailable:
raise RuntimeError(f"selected GPU is not idle on compute node {node.get('code')}: {unavailable}")
if get_settings().minio_enabled: if get_settings().minio_enabled:
await _wait_for_object_storage() await _wait_for_object_storage()
client = ComputeNodeClient(node["api_base_url"]) client = ComputeNodeClient(node["api_base_url"])
await client.inference_load(load_payload) await client.inference_load(load_payload)
store.mark_inference_loaded(node["id"]) store.mark_inference_loaded(node["id"], item_gpu_indices)
loaded_models.append({**item, "status": "starting", "node_id": node["id"], "node_name": node.get("name")}) loaded_models.append({**item, "gpu_indices": item_gpu_indices, "gpus": item_gpu_indices, "status": "starting", "node_id": node["id"], "node_name": node.get("name")})
item_dispatched = True item_dispatched = True
break break
except Exception as exc: # noqa: BLE001 - try next candidate node except Exception as exc: # noqa: BLE001 - try next candidate node
@@ -2340,7 +2445,7 @@ async def model_chat_local_preload(payload: dict[str, Any] = Body(...)) -> dict[
# 计算节点现在异步加载HTTP 接受loading/ready即视为派发成功 # 计算节点现在异步加载HTTP 接受loading/ready即视为派发成功
result = await client.inference_load(payload) result = await client.inference_load(payload)
if result.get("loaded") or result.get("status") in {"loading", "ready"}: if result.get("loaded") or result.get("status") in {"loading", "ready"}:
store.mark_inference_loaded(node["id"]) store.mark_inference_loaded(node["id"], _normalize_gpu_indices(payload))
return ok(result) return ok(result)
except Exception as exc: except Exception as exc:
return ok({"loaded": False, "error": str(exc)}) return ok({"loaded": False, "error": str(exc)})
@@ -2420,7 +2525,7 @@ async def model_chat_trained_preload(payload: dict[str, Any] = Body(...), curren
# 计算节点现在异步加载HTTP 接受loading/ready即视为派发成功 # 计算节点现在异步加载HTTP 接受loading/ready即视为派发成功
result = await client.inference_load({**payload, "compute_node_id": node["id"]}) result = await client.inference_load({**payload, "compute_node_id": node["id"]})
if result.get("loaded") or result.get("status") in {"loading", "ready"}: if result.get("loaded") or result.get("status") in {"loading", "ready"}:
store.mark_inference_loaded(node["id"]) store.mark_inference_loaded(node["id"], _normalize_gpu_indices(payload))
return ok(result) return ok(result)
except Exception as exc: except Exception as exc:
return ok({"loaded": False, "error": str(exc)}) return ok({"loaded": False, "error": str(exc)})

View File

@@ -454,15 +454,21 @@ class PlatformStore:
self.ensure_seed_data() self.ensure_seed_data()
# Track which compute nodes have an active inference model loaded # Track which compute nodes have an active inference model loaded
self._inference_nodes: set[str] = set() self._inference_nodes: set[str] = set()
self._inference_gpu_indexes: dict[str, set[int]] = {}
self._last_runtime_refresh = 0.0 self._last_runtime_refresh = 0.0
# ── inference node tracking ──────────────────────────────────── # ── inference node tracking ────────────────────────────────────
def mark_inference_loaded(self, node_id: str) -> None: def mark_inference_loaded(self, node_id: str, gpu_indexes: list[int] | None = None) -> None:
self._inference_nodes.add(node_id) self._inference_nodes.add(node_id)
if gpu_indexes is not None:
self._inference_gpu_indexes[node_id] = {int(item) for item in gpu_indexes}
else:
self._inference_gpu_indexes.pop(node_id, None)
def mark_inference_unloaded(self, node_id: str) -> None: def mark_inference_unloaded(self, node_id: str) -> None:
self._inference_nodes.discard(node_id) self._inference_nodes.discard(node_id)
self._inference_gpu_indexes.pop(node_id, None)
def is_inference_loaded(self, node_id: str) -> bool: def is_inference_loaded(self, node_id: str) -> bool:
return node_id in self._inference_nodes return node_id in self._inference_nodes
@@ -2766,7 +2772,35 @@ class PlatformStore:
"SELECT gpu_index FROM gpu_allocations WHERE node_id=? AND status IN ('allocated','running')", "SELECT gpu_index FROM gpu_allocations WHERE node_id=? AND status IN ('allocated','running')",
(node_id,), (node_id,),
).fetchall() ).fetchall()
return {int(row["gpu_index"]) for row in rows} active = {int(row["gpu_index"]) for row in rows}
# Evaluation jobs use the same Compute ProcessManager GPU lock but do
# not have fine-tune allocation rows; derive their selected cards here
# so a training task cannot race onto an evaluation GPU.
for row in conn.execute(
"SELECT payload FROM eval_tasks WHERE status IN ('syncing','queued','running')"
).fetchall():
payload = json_loads(row["payload"], {})
if payload.get("compute_node_id") != node_id:
continue
selected = payload.get("gpu_indices") or payload.get("gpus")
if selected is None and payload.get("gpu_id") is not None:
selected = [payload.get("gpu_id")]
active.update(int(item) for item in selected or [])
# Loaded inference models also reserve only their selected cards.
active.update(self._inference_gpu_indexes.get(node_id, set()))
for row in conn.execute("SELECT payload FROM compare_tasks").fetchall():
payload = json_loads(row["payload"], {})
load_status = payload.get("load_status") or {}
if isinstance(load_status, str):
load_status = json_loads(load_status, {})
for item in load_status.get("loaded_models") or []:
if item.get("node_id") != node_id or item.get("status") not in {"starting", "ready", "running"}:
continue
selected = item.get("gpu_indices") or item.get("gpus")
if selected is None and item.get("gpu_id") is not None:
selected = [item.get("gpu_id")]
active.update(int(gpu) for gpu in selected or [])
return active
def _node_gpu_indexes(self, conn: PgConnection, node: dict[str, Any]) -> set[int]: def _node_gpu_indexes(self, conn: PgConnection, node: dict[str, Any]) -> set[int]:
rows = conn.execute("SELECT gpu_index FROM gpus WHERE node_id=?", (node["id"],)).fetchall() rows = conn.execute("SELECT gpu_index FROM gpus WHERE node_id=?", (node["id"],)).fetchall()
@@ -3195,6 +3229,9 @@ class PlatformStore:
# 推理模型占用算力节点同样计入:优先从 compare_tasks 持久化状态派生 # 推理模型占用算力节点同样计入:优先从 compare_tasks 持久化状态派生
# (重启后仍准确),并用内存标记兜底(直接 preload 的模型无 compare 记录) # (重启后仍准确),并用内存标记兜底(直接 preload 的模型无 compare 记录)
inference_node_ids = set(self._inference_nodes) inference_node_ids = set(self._inference_nodes)
inference_gpu_indexes: dict[str, set[int]] = {
node_id: set(indexes) for node_id, indexes in self._inference_gpu_indexes.items()
}
for ctr in conn.execute("SELECT payload FROM compare_tasks").fetchall(): for ctr in conn.execute("SELECT payload FROM compare_tasks").fetchall():
ls = json_loads(ctr["payload"], {}).get("load_status") or {} ls = json_loads(ctr["payload"], {}).get("load_status") or {}
if isinstance(ls, str): if isinstance(ls, str):
@@ -3204,8 +3241,13 @@ class PlatformStore:
ls = {} ls = {}
for m in ls.get("loaded_models") or []: for m in ls.get("loaded_models") or []:
if m.get("status") in {"ready", "running"} and m.get("node_id"): if m.get("status") in {"ready", "running"} and m.get("node_id"):
inference_node_ids.add(m["node_id"]) node_id = m["node_id"]
for nid in inference_node_ids: selected = m.get("gpu_indices") or m.get("gpus")
if selected:
inference_gpu_indexes.setdefault(node_id, set()).update(int(item) for item in selected)
else:
inference_node_ids.add(node_id)
for nid in set(inference_node_ids) | set(inference_gpu_indexes):
running_map[nid] = running_map.get(nid, 0) + 1 running_map[nid] = running_map.get(nid, 0) + 1
rows = conn.execute("SELECT * FROM compute_nodes ORDER BY scheduler_weight DESC, code").fetchall() rows = conn.execute("SELECT * FROM compute_nodes ORDER BY scheduler_weight DESC, code").fetchall()
return [ return [
@@ -3434,6 +3476,9 @@ class PlatformStore:
# 推理模型占用的节点:优先从 compare_tasks 持久化状态派生(重启后仍准确), # 推理模型占用的节点:优先从 compare_tasks 持久化状态派生(重启后仍准确),
# 内存标记兜底(直接 preload 的模型无 compare 记录) # 内存标记兜底(直接 preload 的模型无 compare 记录)
inference_node_ids = set(self._inference_nodes) inference_node_ids = set(self._inference_nodes)
inference_gpu_indexes: dict[str, set[int]] = {
node_id: set(indexes) for node_id, indexes in self._inference_gpu_indexes.items()
}
for ctr in conn.execute("SELECT payload FROM compare_tasks").fetchall(): for ctr in conn.execute("SELECT payload FROM compare_tasks").fetchall():
ls = json_loads(ctr["payload"], {}).get("load_status") or {} ls = json_loads(ctr["payload"], {}).get("load_status") or {}
if isinstance(ls, str): if isinstance(ls, str):
@@ -3443,7 +3488,12 @@ class PlatformStore:
ls = {} ls = {}
for m in ls.get("loaded_models") or []: for m in ls.get("loaded_models") or []:
if m.get("status") in {"ready", "running"} and m.get("node_id"): if m.get("status") in {"ready", "running"} and m.get("node_id"):
inference_node_ids.add(m["node_id"]) node_id = m["node_id"]
selected = m.get("gpu_indices") or m.get("gpus")
if selected:
inference_gpu_indexes.setdefault(node_id, set()).update(int(item) for item in selected)
else:
inference_node_ids.add(node_id)
items = [] items = []
for row in rows: for row in rows:
task = next( task = next(
@@ -3459,7 +3509,10 @@ class PlatformStore:
t t
for t in eval_running for t in eval_running
if t.get("compute_node_id") == row["node_id"] 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) and row["gpu_index"] in {
int(item)
for item in (t.get("gpu_indices") or t.get("gpus") or ([t["gpu_id"]] if t.get("gpu_id") is not None else []))
}
), ),
None, None,
) )
@@ -3468,7 +3521,8 @@ class PlatformStore:
eval_task is not None and eval_task.get("status") in {"syncing", "queued"} 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 # Also mark GPU as busy if an inference model is loaded on this node
if row["node_id"] in inference_node_ids and not busy: inference_on_gpu = row["node_id"] in inference_node_ids or row["gpu_index"] in inference_gpu_indexes.get(row["node_id"], set())
if inference_on_gpu and not busy:
busy = True busy = True
reserved = False reserved = False
memory_used = round(row["memory_total_gb"] * (0.72 if busy else 0.18 if reserved else 0.04), 1) memory_used = round(row["memory_total_gb"] * (0.72 if busy else 0.18 if reserved else 0.04), 1)

View File

@@ -240,6 +240,10 @@ class ComputeNodeClient:
async def inference_status(self) -> dict[str, Any]: async def inference_status(self) -> dict[str, Any]:
return await self._request("GET", "/inference/status", timeout=INFERENCE_STATUS_TIMEOUT) return await self._request("GET", "/inference/status", timeout=INFERENCE_STATUS_TIMEOUT)
async def gpu_resources(self) -> list[dict[str, Any]]:
"""Read live per-GPU metrics from this compute node."""
return await self.gpus()
async def inference_unload(self) -> dict[str, Any]: async def inference_unload(self) -> dict[str, Any]:
return await self._request("POST", "/inference/unload", json_data={}, timeout=INFERENCE_UNLOAD_TIMEOUT) return await self._request("POST", "/inference/unload", json_data={}, timeout=INFERENCE_UNLOAD_TIMEOUT)

View File

@@ -73,7 +73,8 @@ async def reconcile_inference_loads(store: Any) -> list[dict[str, Any]]:
if node_status == "ready": if node_status == "ready":
item["status"] = "ready" item["status"] = "ready"
item.pop("error", None) item.pop("error", None)
store.mark_inference_loaded(node["id"]) selected_gpus = item.get("gpu_indices") or item.get("gpus")
store.mark_inference_loaded(node["id"], selected_gpus)
elif node_status == "error": elif node_status == "error":
item["status"] = "error" item["status"] = "error"
item["error"] = status.get("error") or "model load failed on compute node" item["error"] = status.get("error") or "model load failed on compute node"

View File

@@ -221,6 +221,35 @@ def create_app() -> FastAPI:
return fallback_gpu_resources() return fallback_gpu_resources()
items: list[dict[str, Any]] = [] items: list[dict[str, Any]] = []
processes_by_uuid: dict[str, list[dict[str, Any]]] = {}
try:
process_result = subprocess.run(
[
"nvidia-smi",
"--query-compute-apps=gpu_uuid,pid,process_name,used_memory",
"--format=csv,noheader,nounits",
],
check=True,
capture_output=True,
text=True,
timeout=5,
)
for process_line in process_result.stdout.splitlines():
process_parts = [part.strip() for part in process_line.split(",")]
if len(process_parts) < 4:
continue
process_uuid, pid, process_name, used_memory = process_parts[:4]
processes_by_uuid.setdefault(process_uuid, []).append(
{
"pid": int(_safe_float(pid)),
"name": process_name,
"memory_used_gb": round(_safe_float(used_memory) / 1024, 2),
}
)
except Exception:
# Some driver/runtime combinations do not expose compute-apps;
# utilization and memory metrics remain useful without processes.
pass
for line in result.stdout.splitlines(): for line in result.stdout.splitlines():
parts = [part.strip() for part in line.split(",")] parts = [part.strip() for part in line.split(",")]
if len(parts) < 9: if len(parts) < 9:
@@ -244,7 +273,7 @@ def create_app() -> FastAPI:
"temperature": int(_safe_float(temp)), "temperature": int(_safe_float(temp)),
"power_w": round(_safe_float(power), 1), "power_w": round(_safe_float(power), 1),
"power_limit_w": round(_safe_float(power_limit), 1), "power_limit_w": round(_safe_float(power_limit), 1),
"processes": [], "processes": processes_by_uuid.get(uuid, []),
} }
) )
return items return items
@@ -747,6 +776,23 @@ def create_app() -> FastAPI:
infer_backend: str (default: "huggingface") infer_backend: str (default: "huggingface")
infer_dtype: str (default: "auto") infer_dtype: str (default: "auto")
""" """
requested_gpus = payload.get("gpu_indices")
if requested_gpus is None:
requested_gpus = payload.get("gpus") or []
try:
requested_gpus = sorted({int(item) for item in requested_gpus})
except (TypeError, ValueError) as exc:
raise HTTPException(status_code=400, detail=f"invalid GPU selection: {exc}") from exc
if any(item < 0 for item in requested_gpus):
raise HTTPException(status_code=400, detail="GPU index must be non-negative")
if requested_gpus:
known_gpus = {int(item.get("gpu_index", item.get("id", -1))) for item in gpu_resources()}
missing = sorted(set(requested_gpus) - known_gpus)
if missing:
raise HTTPException(status_code=409, detail=f"requested GPU not found: {missing}")
conflict = sorted(set(requested_gpus).intersection(process_manager.locked_gpus()))
if conflict:
raise HTTPException(status_code=409, detail=f"GPU already used by another compute job: {conflict}")
session = get_inference_session() session = get_inference_session()
result = session.load( result = session.load(
model_name_or_path=payload.get("model_name_or_path", ""), model_name_or_path=payload.get("model_name_or_path", ""),
@@ -754,6 +800,7 @@ def create_app() -> FastAPI:
template=payload.get("template", "qwen"), template=payload.get("template", "qwen"),
infer_backend=payload.get("infer_backend", "huggingface"), infer_backend=payload.get("infer_backend", "huggingface"),
infer_dtype=payload.get("infer_dtype", "auto"), infer_dtype=payload.get("infer_dtype", "auto"),
gpu_indices=requested_gpus,
) )
return result return result

View File

@@ -1,5 +1,6 @@
from __future__ import annotations from __future__ import annotations
import os
import threading import threading
import time import time
import uuid import uuid
@@ -36,6 +37,7 @@ class InferenceSession:
self._generating_args: dict[str, Any] = {} self._generating_args: dict[str, Any] = {}
self._model_name: str = "" self._model_name: str = ""
self._adapter_path: str = "" self._adapter_path: str = ""
self._gpu_indices: list[int] = []
self._loaded_at: float = 0.0 self._loaded_at: float = 0.0
@property @property
@@ -53,6 +55,7 @@ class InferenceSession:
"loaded_at": self._loaded_at, "loaded_at": self._loaded_at,
"request_id": self._request_id, "request_id": self._request_id,
"error": self._error, "error": self._error,
"gpu_indices": list(self._gpu_indices),
} }
def wait_until_loaded(self, timeout: float | None = None) -> dict[str, Any]: def wait_until_loaded(self, timeout: float | None = None) -> dict[str, Any]:
@@ -87,8 +90,12 @@ class InferenceSession:
template="qwen", template="qwen",
infer_backend="huggingface", infer_backend="huggingface",
infer_dtype="auto", infer_dtype="auto",
gpu_indices=None,
**kwargs, **kwargs,
) -> dict[str, Any]: ) -> dict[str, Any]:
requested_gpus = sorted({int(item) for item in (gpu_indices or [])})
if any(item < 0 for item in requested_gpus):
return {"loaded": False, "status": "error", "error": "GPU index must be non-negative"}
with self._state_lock: with self._state_lock:
if self._status == "loading": if self._status == "loading":
# A model is already loading — dedupe, reuse the same request id. # A model is already loading — dedupe, reuse the same request id.
@@ -97,6 +104,7 @@ class InferenceSession:
self._status = "loading" self._status = "loading"
self._error = "" self._error = ""
self._request_id = uuid.uuid4().hex[:12] self._request_id = uuid.uuid4().hex[:12]
self._gpu_indices = requested_gpus
self._cancel_requested = False self._cancel_requested = False
self._load_args = { self._load_args = {
"model_name_or_path": model_name_or_path, "model_name_or_path": model_name_or_path,
@@ -115,13 +123,21 @@ class InferenceSession:
def _load_worker(self) -> None: def _load_worker(self) -> None:
"""Build the ChatModel off the state lock so info() never blocks.""" """Build the ChatModel off the state lock so info() never blocks."""
with self._state_lock:
requested_gpus = list(self._gpu_indices)
model = None model = None
tokenizer = None tokenizer = None
generating_args: dict[str, Any] = {} generating_args: dict[str, Any] = {}
error = "" error = ""
previous_visible_devices = os.environ.get("CUDA_VISIBLE_DEVICES")
try: try:
# Set visibility before LLaMA-Factory/PyTorch initializes CUDA.
if requested_gpus:
os.environ["CUDA_VISIBLE_DEVICES"] = ",".join(str(item) for item in requested_gpus)
if self._teardown_old: if self._teardown_old:
self._release_model() self._release_model()
with self._state_lock:
self._gpu_indices = requested_gpus
from llamafactory.chat import ChatModel from llamafactory.chat import ChatModel
from llamafactory.hparams import get_infer_args from llamafactory.hparams import get_infer_args
@@ -138,6 +154,12 @@ class InferenceSession:
generating_args = dict(generating_args) generating_args = dict(generating_args)
except Exception as exc: # noqa: BLE001 - surface load failure via status except Exception as exc: # noqa: BLE001 - surface load failure via status
error = str(exc) error = str(exc)
finally:
if requested_gpus:
if previous_visible_devices is None:
os.environ.pop("CUDA_VISIBLE_DEVICES", None)
else:
os.environ["CUDA_VISIBLE_DEVICES"] = previous_visible_devices
with self._state_lock: with self._state_lock:
if error: if error:
self._model = None self._model = None
@@ -152,6 +174,7 @@ class InferenceSession:
self._model = None self._model = None
self._tokenizer = None self._tokenizer = None
self._status = "idle" self._status = "idle"
self._gpu_indices = []
return return
self._model = model self._model = model
self._tokenizer = tokenizer self._tokenizer = tokenizer
@@ -189,6 +212,7 @@ class InferenceSession:
self._adapter_path = "" self._adapter_path = ""
self._loaded_at = 0.0 self._loaded_at = 0.0
self._error = "" self._error = ""
self._gpu_indices = []
def unload(self) -> dict[str, Any]: def unload(self) -> dict[str, Any]:
with self._state_lock: with self._state_lock:
@@ -208,6 +232,7 @@ class InferenceSession:
self._adapter_path = "" self._adapter_path = ""
self._loaded_at = 0.0 self._loaded_at = 0.0
self._error = "" self._error = ""
self._gpu_indices = []
return {"unloaded": True, "status": "idle"} return {"unloaded": True, "status": "idle"}
def chat(self, messages, temperature=0.95, top_p=0.7, max_new_tokens=1024, do_sample=True, **kwargs) -> dict[str, Any]: def chat(self, messages, temperature=0.95, top_p=0.7, max_new_tokens=1024, do_sample=True, **kwargs) -> dict[str, Any]:

View File

@@ -42,12 +42,23 @@ export interface FineTunePreflightResult {
sync_results?: Array<Record<string, unknown>> sync_results?: Array<Record<string, unknown>>
} }
export interface FineTuneGpuStatus {
source: string
items: Array<Record<string, unknown>>
selected_gpus: number[]
error?: string
}
/** 训练任务列表 */ /** 训练任务列表 */
export const getFineTuneList = () => get<FineTuneTask[]>('/fine-tune') export const getFineTuneList = () => get<FineTuneTask[]>('/fine-tune')
/** 训练任务详情 */ /** 训练任务详情 */
export const getFineTune = (id: string | number) => get<FineTuneTask>(`/fine-tune/${id}`) export const getFineTune = (id: string | number) => get<FineTuneTask>(`/fine-tune/${id}`)
/** 获取任务所在 Compute 节点的实时 GPU 指标 */
export const getFineTuneGpuStatus = (id: string | number) =>
get<FineTuneGpuStatus>(`/fine-tune/${id}/gpu-status`)
/** 任务名查重 */ /** 任务名查重 */
export const checkFineTuneName = (name: string) => export const checkFineTuneName = (name: string) =>
get<{ exists: boolean }>('/fine-tune/check-name', { name }) get<{ exists: boolean }>('/fine-tune/check-name', { name })

View File

@@ -218,6 +218,8 @@ export interface LoadedModel {
port?: number port?: number
node_id?: string node_id?: string
node_name?: string node_name?: string
gpu_indices?: number[]
gpus?: number[]
error?: string error?: string
} }
@@ -239,6 +241,8 @@ export interface CompareModelRef {
gpu_id: number gpu_id: number
node_id?: string node_id?: string
node_name?: string node_name?: string
gpu_indices?: number[]
gpus?: number[]
source?: string source?: string
port?: number port?: number
} }
@@ -282,7 +286,9 @@ export interface StartEvalPayload {
eval_task_name: string eval_task_name: string
eval_type: EvalType eval_type: EvalType
model_id: string | number model_id: string | number
gpu_id: string | number gpu_id: string | number | string[]
gpu_indices?: number[]
gpus?: number[]
compute_node_id?: string compute_node_id?: string
dataset_id: string | number dataset_id: string | number
dimension_id: string | number dimension_id: string | number

View File

@@ -39,7 +39,7 @@ const createdDimensionId = ref<string | number>('')
const taskForm = ref<EvalTaskSetupDraft>({ const taskForm = ref<EvalTaskSetupDraft>({
eval_task_name: '', eval_task_name: '',
model_id: '', model_id: '',
gpu_id: '', gpu_id: [],
data_source: 'dataset', data_source: 'dataset',
dataset_id: '', dataset_id: '',
leaderboard: false, leaderboard: false,
@@ -145,12 +145,24 @@ async function handleSubmit() {
const dimensionId = await resolveDimensionId() const dimensionId = await resolveDimensionId()
// GPU 选择为「节点:GPU序号」复合值解析出节点与 GPU 序号, // GPU 选择为「节点:GPU序号」复合值解析出节点与 GPU 序号,
// 多算力节点时必须把节点信息传给后端,否则会派发到错误的算力节点 // 多算力节点时必须把节点信息传给后端,否则会派发到错误的算力节点
const [gpuNodeId, gpuIndex] = String(taskForm.value.gpu_id).split(':') const selectedGpuKeys = Array.isArray(taskForm.value.gpu_id)
? taskForm.value.gpu_id
: [String(taskForm.value.gpu_id)]
const gpuSelections = selectedGpuKeys
.map((key) => {
const [nodeId, gpuIndex] = String(key).split(':')
return { nodeId, gpuIndex: Number(gpuIndex) }
})
.filter((item) => item.nodeId && Number.isInteger(item.gpuIndex) && item.gpuIndex >= 0)
const gpuNodeId = gpuSelections[0]?.nodeId || ''
const gpuIndices = gpuSelections.map((item) => item.gpuIndex)
const evalResult: any = await startEval({ const evalResult: any = await startEval({
eval_task_name: taskForm.value.eval_task_name, eval_task_name: taskForm.value.eval_task_name,
eval_type: 'custom', eval_type: 'custom',
model_id: taskForm.value.model_id, model_id: taskForm.value.model_id,
gpu_id: Number(gpuIndex) || 0, gpu_id: gpuIndices[0] ?? 0,
gpu_indices: gpuIndices,
gpus: gpuIndices,
compute_node_id: gpuNodeId || '', compute_node_id: gpuNodeId || '',
dataset_id: taskForm.value.data_source === 'dataset' ? taskForm.value.dataset_id : '', dataset_id: taskForm.value.data_source === 'dataset' ? taskForm.value.dataset_id : '',
dimension_id: dimensionId, dimension_id: dimensionId,

View File

@@ -6,7 +6,7 @@ import type { DatasetItem, GpuInfo, TrainedModel } from '@/types'
export interface EvalTaskSetupDraft { export interface EvalTaskSetupDraft {
eval_task_name: string eval_task_name: string
model_id: string | number model_id: string | number
gpu_id: string | number gpu_id: string | number | string[]
data_source: 'dataset' | 'inference' data_source: 'dataset' | 'inference'
dataset_id: string | number dataset_id: string | number
leaderboard: boolean leaderboard: boolean
@@ -23,6 +23,13 @@ defineProps<{
const form = defineModel<EvalTaskSetupDraft>({ required: true }) const form = defineModel<EvalTaskSetupDraft>({ required: true })
const formRef = ref<FormInstance>() const formRef = ref<FormInstance>()
function handleGpuChange(value: string | number | string[]) {
const keys = Array.isArray(value) ? value.map(String) : [String(value || '')]
const nodeId = keys[0]?.split(':', 1)[0]
if (!nodeId || !Array.isArray(form.value.gpu_id)) return
form.value.gpu_id = keys.filter((key) => key.split(':', 1)[0] === nodeId)
}
const rules: FormRules<EvalTaskSetupDraft> = { const rules: FormRules<EvalTaskSetupDraft> = {
eval_task_name: [ eval_task_name: [
{ required: true, message: '请输入任务名称', trigger: 'blur' }, { required: true, message: '请输入任务名称', trigger: 'blur' },
@@ -101,7 +108,16 @@ defineExpose({ validate })
</el-form-item> </el-form-item>
<el-form-item label="选择 GPU" prop="gpu_id"> <el-form-item label="选择 GPU" prop="gpu_id">
<el-select v-model="form.gpu_id" placeholder="请选择 GPU" style="width: 100%" :loading="loading"> <el-select
v-model="form.gpu_id"
multiple
collapse-tags
collapse-tags-tooltip
placeholder="请选择同一算力节点内的一张或多张 GPU"
style="width: 100%"
:loading="loading"
@change="handleGpuChange"
>
<el-option <el-option
v-for="gpu in gpus" v-for="gpu in gpus"
:key="`${gpu.node_id || ''}:${gpu.id ?? 0}`" :key="`${gpu.node_id || ''}:${gpu.id ?? 0}`"

View File

@@ -21,7 +21,7 @@ function nameOf<T extends { id: string | number; name?: string }>(items: T[], id
/** GPU 选择为「节点:GPU序号」复合值解析并展示为可读标签 */ /** GPU 选择为「节点:GPU序号」复合值解析并展示为可读标签 */
const gpuLabel = computed(() => { const gpuLabel = computed(() => {
const key = String(props.task.gpu_id || '') const key = Array.isArray(props.task.gpu_id) ? String(props.task.gpu_id[0] || '') : String(props.task.gpu_id || '')
const gpu = props.gpus.find((g) => `${g.node_id || ''}:${g.id ?? 0}` === key) 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}` if (gpu) return `${gpu.node_name || gpu.node_code || '算力节点'} / GPU ${gpu.id ?? 0}`
const [nodeId, idx] = key.split(':') const [nodeId, idx] = key.split(':')

View File

@@ -4,8 +4,7 @@ import { useRouter } from 'vue-router'
import { ElMessage, type FormInstance, type FormRules } from 'element-plus' import { ElMessage, type FormInstance, type FormRules } from 'element-plus'
import PageCard from '@/components/PageCard.vue' import PageCard from '@/components/PageCard.vue'
import { getModelList, getTrainedModels } from '@/api/modules/model' import { getModelList, getTrainedModels } from '@/api/modules/model'
import { getSystemInfo } from '@/api/modules/system' import { getComputeGpus, getComputeNodes, type ComputeNode } from '@/api/modules/compute'
import { getComputeNodes, type ComputeNode } from '@/api/modules/compute'
import { createCompare, loadCompare } from '@/api/modules/compare' import { createCompare, loadCompare } from '@/api/modules/compare'
import type { ModelItem, TrainedModel, GpuInfo } from '@/types' import type { ModelItem, TrainedModel, GpuInfo } from '@/types'
@@ -83,8 +82,8 @@ const form = reactive({
description: '', description: '',
/** 选中的模型 key单选 */ /** 选中的模型 key单选 */
model_key: '', model_key: '',
/** 使用的 GPU */ /** 使用的 GPU(同一节点内可多选) */
gpu_key: '', gpu_keys: [] as string[],
}) })
const rules: FormRules = { const rules: FormRules = {
@@ -94,12 +93,26 @@ const rules: FormRules = {
/** 当前选中的模型对象 */ /** 当前选中的模型对象 */
const selectedModel = computed(() => modelMap.value[form.model_key]) const selectedModel = computed(() => modelMap.value[form.model_key])
const selectedGpu = computed(() => idleGpus.value.find((g) => `${g.node_id || ''}:${g.id ?? 0}` === form.gpu_key)) const selectedGpus = computed(() => idleGpus.value.filter((gpu) => form.gpu_keys.includes(gpuKey(gpu))))
function gpuKey(gpu: GpuInfo) {
return `${gpu.node_id || ''}:${gpu.id ?? 0}`
}
function handleGpuChange(keys: string[]) {
const nodeId = keys[0]?.split(':', 1)[0]
if (!nodeId) return
const filtered = keys.filter((key) => key.split(':', 1)[0] === nodeId)
if (filtered.length !== keys.length) {
ElMessage.info('一次推理只能使用同一算力节点内的 GPU已忽略其它节点的选择')
}
form.gpu_keys = filtered
}
watch(selectedModel, (model) => { watch(selectedModel, (model) => {
if (!model?.compute_node_id) return if (!model?.compute_node_id) return
const gpu = idleGpus.value.find((item) => item.node_id === model.compute_node_id) const gpu = idleGpus.value.find((item) => item.node_id === model.compute_node_id)
if (gpu) form.gpu_key = `${gpu.node_id || ''}:${gpu.id ?? 0}` if (gpu) form.gpu_keys = [gpuKey(gpu)]
}) })
async function handleSubmit() { async function handleSubmit() {
@@ -111,6 +124,10 @@ async function handleSubmit() {
ElMessage.warning('请选择模型') ElMessage.warning('请选择模型')
return return
} }
if (!selectedGpus.value.length) {
ElMessage.warning('请至少选择一张空闲 GPU')
return
}
submitting.value = true submitting.value = true
startupStatus.value = '正在创建推理任务...' startupStatus.value = '正在创建推理任务...'
try { try {
@@ -130,9 +147,11 @@ async function handleSubmit() {
model_name: m.name, model_name: m.name,
model_path: m.model_path, model_path: m.model_path,
source: m.source, source: m.source,
gpu_id: selectedGpu.value?.id ?? 0, gpu_id: selectedGpus.value[0]?.id ?? 0,
node_id: selectedGpu.value?.node_id || m.compute_node_id, gpu_indices: selectedGpus.value.map((gpu) => Number(gpu.id ?? 0)),
node_name: selectedGpu.value?.node_name || m.compute_node_name, gpus: selectedGpus.value.map((gpu) => Number(gpu.id ?? 0)),
node_id: selectedGpus.value[0]?.node_id || m.compute_node_id,
node_name: selectedGpus.value[0]?.node_name || m.compute_node_name,
}, },
], ],
}) })
@@ -169,17 +188,17 @@ async function loadData() {
const [db, trained, sys, nodes] = await Promise.all([ const [db, trained, sys, nodes] = await Promise.all([
getModelList(), getModelList(),
getTrainedModels(), getTrainedModels(),
getSystemInfo(), getComputeGpus(),
getComputeNodes(), getComputeNodes(),
]) ])
dbModels.value = db || [] dbModels.value = db || []
trainedModels.value = trained?.models || [] trainedModels.value = trained?.models || []
gpus.value = sys?.gpu || [] gpus.value = (sys || []) as unknown as GpuInfo[]
computeNodes.value = nodes || [] computeNodes.value = nodes || []
// 默认选中第一个空闲 GPU // 默认选中第一个空闲 GPU
if (idleGpus.value.length > 0) { if (idleGpus.value.length > 0) {
const firstGpu = idleGpus.value[0] const firstGpu = idleGpus.value[0]
form.gpu_key = `${firstGpu.node_id || ''}:${firstGpu.id ?? 0}` form.gpu_keys = [gpuKey(firstGpu)]
} }
} catch { } catch {
// ignore // ignore
@@ -228,12 +247,20 @@ onMounted(loadData)
</el-form-item> </el-form-item>
<el-form-item label="GPU"> <el-form-item label="GPU">
<el-select v-model="form.gpu_key" style="width: 400px"> <el-select
v-model="form.gpu_keys"
multiple
collapse-tags
collapse-tags-tooltip
style="width: 400px"
placeholder="请选择同一算力节点内的一张或多张 GPU"
@change="handleGpuChange"
>
<el-option <el-option
v-for="g in idleGpus" v-for="g in idleGpus"
:key="`${g.node_id || ''}:${g.id ?? 0}`" :key="gpuKey(g)"
:label="`${g.node_name || g.node_code || '算力节点'} / ${g.name} (GPU${g.id ?? 0}) [空闲]`" :label="`${g.node_name || g.node_code || '算力节点'} / ${g.name} (GPU${g.id ?? 0}) [空闲]`"
:value="`${g.node_id || ''}:${g.id ?? 0}`" :value="gpuKey(g)"
/> />
</el-select> </el-select>
</el-form-item> </el-form-item>

View File

@@ -8,10 +8,9 @@ import TrainingTaskOverview from './training-log/TrainingTaskOverview.vue'
import { usePolling } from '@/composables/usePolling' import { usePolling } from '@/composables/usePolling'
import '@/plugins/echarts-training-log' import '@/plugins/echarts-training-log'
import { useModelsStore } from '@/stores/models' import { useModelsStore } from '@/stores/models'
import { getFineTune, getFineTuneDiagnostics, getFineTuneLogs, getFineTuneMetrics, type TrainingDiagnostic } from '@/api/modules/fineTune' import { getFineTune, getFineTuneDiagnostics, getFineTuneGpuStatus, getFineTuneLogs, getFineTuneMetrics, type TrainingDiagnostic } from '@/api/modules/fineTune'
import { getTrainingLogFiles, getTrainingLogContent } from '@/api/modules/log' import { getTrainingLogFiles, getTrainingLogContent } from '@/api/modules/log'
import { getDataset } from '@/api/modules/dataset' import { getDataset } from '@/api/modules/dataset'
import { getSystemInfo } from '@/api/modules/system'
import { TRAIN_TYPE_MAP, TRAIN_METHOD_MAP } from '@/constants' import { TRAIN_TYPE_MAP, TRAIN_METHOD_MAP } from '@/constants'
import { import {
buildMetricChartOption, buildMetricChartOption,
@@ -205,15 +204,20 @@ async function loadDataset(datasetId: string | number) {
} }
} }
async function loadGpuStatus() { async function loadGpuStatus(currentTask: FineTuneTask) {
try { try {
const systemInfo = await getSystemInfo() const live = await getFineTuneGpuStatus(currentTask.id)
gpuPool.value = systemInfo.gpu ?? [] if (live.source === 'compute' && live.items.length) {
gpuPool.value = live.items as unknown as GpuInfo[]
gpuUpdatedAt.value = new Date() gpuUpdatedAt.value = new Date()
gpuLoadError.value = '' gpuLoadError.value = ''
return
}
gpuPool.value = []
gpuLoadError.value = live.error || 'Compute 节点暂未返回实时 GPU 指标'
} catch { } catch {
gpuLoadError.value = 'GPU 监控数据暂时不可用' gpuLoadError.value = 'GPU 监控数据暂时不可用'
if (!gpuUpdatedAt.value) gpuPool.value = [] gpuPool.value = []
} }
} }
@@ -306,7 +310,7 @@ async function refreshAll() {
const datasetPromise = currentTask.train_dataset_id const datasetPromise = currentTask.train_dataset_id
? loadDataset(currentTask.train_dataset_id) ? loadDataset(currentTask.train_dataset_id)
: Promise.resolve() : Promise.resolve()
await Promise.all([datasetPromise, loadLog(currentTask), loadGpuStatus(), loadDiagnostics(currentTask)]) await Promise.all([datasetPromise, loadLog(currentTask), loadGpuStatus(currentTask), loadDiagnostics(currentTask)])
await loadMetrics(currentTask) await loadMetrics(currentTask)
} finally { } finally {
loading.value = false loading.value = false