From c7c9ed925bc00d6fc5930adc98711c3ac754d510 Mon Sep 17 00:00:00 2001 From: wuyongtao Date: Tue, 28 Jul 2026 17:29:16 +0800 Subject: [PATCH 1/5] =?UTF-8?q?feat:=20=E6=A8=A1=E5=9E=8B=E6=8E=A8?= =?UTF-8?q?=E7=90=86=E7=AB=AF=E5=88=B0=E7=AB=AF=E9=97=AD=E7=8E=AF=20?= =?UTF-8?q?=E2=80=94=20=E7=9C=9F=E5=AE=9E=E6=B5=81=E5=BC=8F=E6=8E=A8?= =?UTF-8?q?=E7=90=86=20+=20=E9=87=8A=E6=94=BE/=E5=88=A0=E9=99=A4=20+=20GPU?= =?UTF-8?q?=20=E7=8A=B6=E6=80=81=E5=90=8C=E6=AD=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 后端 (platform.py + platform_store.py): - 新增 _build_messages_payload() 转换前端格式为 OpenAI messages - 新增 _stream_chat_proxy() SSE 流式代理到算力节点 - 新增 _unload_from_compute_node() 真正释放算力节点 GPU 显存 - 重写 model_compare_load: 从假 PID/端口改为真正调用算力节点加载模型 - 修复 model_compare_unload: 调用 _unload_from_compute_node 释放 GPU - 修复 model_compare_delete: 先释放 GPU 再删除记录 - 修复 model_compare_stream_chat: 从 mock 改为 StreamingResponse 代理 - 修复 model_chat_local/stream: 消息格式转换 + 路径修正 - PlatformStore 新增 _inference_nodes 追踪，gpus() 同步推理占用状态 - preload/unload 端点标记/清除推理节点占用 算力节点 (compute): - inference.py: 适配新版 LLaMA-Factory API (get_infer_args 4 返回值、ChatModel args dict、stream_chat 新签名) - inference.py: unload() 增加 gc.collect + torch.cuda.empty_cache + synchronize 彻底释放显存 - main.py: inference/load 移除 HTTPException(500)，错误以 200 正常返回 前端: - InferenceChatView: 真实模式下走 SSE 流式推理，mock 模式保留兼容 - InferenceCreateView: 调用 preloadLocalModel + createCompare 真实创建推理任务，失败回退 mock - InferenceListView: 「停止」改为「释放」，删除前先释放算力节点，改进错误提示 - compare.ts: 新增 streamChatReal() fetch SSE，preload 超时提升至 5 分钟 - useStreamChat.ts: send() 支持 useMock 参数，真实模式调用 streamChatReal - GPU 选择过滤: 仅显示在线算力节点上的空闲 GPU Co-Authored-By: Claude --- backend/app/api/v1/endpoints/platform.py | 191 ++++++++++++++---- backend/app/db/platform_store.py | 18 ++ compute/api/main.py | 2 - compute/engines/llama_factory/inference.py | 40 +++- frontend/src/api/modules/compare.ts | 29 ++- frontend/src/composables/useStreamChat.ts | 16 +- frontend/src/types/index.ts | 2 +- .../src/views/inference/InferenceChatView.vue | 38 ++-- .../views/inference/InferenceCreateView.vue | 83 ++++++-- .../src/views/inference/InferenceListView.vue | 25 +-- 10 files changed, 331 insertions(+), 113 deletions(-) diff --git a/backend/app/api/v1/endpoints/platform.py b/backend/app/api/v1/endpoints/platform.py index 4ee8131..ecaec0b 100644 --- a/backend/app/api/v1/endpoints/platform.py +++ b/backend/app/api/v1/endpoints/platform.py @@ -31,6 +31,62 @@ def _select_first_online_node(store: Any) -> dict[str, Any] | None: return None +def _build_messages_payload(payload: dict[str, Any]) -> dict[str, Any]: + """Convert frontend inference payload to compute API messages format. + + Accepts both: + - OpenAI-style: {messages: [{role, content}, ...], temperature, ...} + - Frontend-style: {user_question, system_prompt, temperature, ...} + """ + if payload.get("messages"): + messages = payload["messages"] + # messages already in OpenAI format; pass through with optional system prompt + if payload.get("system_prompt") and not any(m.get("role") == "system" for m in messages): + messages = [{"role": "system", "content": payload["system_prompt"]}] + list(messages) + else: + messages = [] + if payload.get("system_prompt"): + messages.append({"role": "system", "content": payload["system_prompt"]}) + question = payload.get("user_question") or payload.get("question") or "" + if question: + messages.append({"role": "user", "content": question}) + return { + "messages": messages, + "temperature": float(payload.get("temperature", 0.7)), + "top_p": float(payload.get("top_p", 0.95)), + "max_new_tokens": int(payload.get("max_tokens", 2048)), + "do_sample": bool(payload.get("do_sample", True)), + } + + +async def _stream_chat_proxy(payload: dict[str, Any]) -> StreamingResponse: + """Common SSE streaming proxy: convert payload → forward to compute node → stream back.""" + store = get_platform_store() + node = _select_first_online_node(store) + if not node: + return StreamingResponse( + iter(['data: {"error": "no online compute node available for inference"}\n\n']), + media_type="text/event-stream", + ) + client = ComputeNodeClient(node["api_base_url"]) + compute_payload = _build_messages_payload(payload) + + async def stream_proxy(): + async with httpx.AsyncClient(timeout=300) as http: + url = f"{node['api_base_url'].rstrip('/')}{client.route_prefix}/inference/chat/stream" + try: + async with http.stream("POST", url, json=compute_payload, headers=client.headers()) as resp: + if resp.status_code >= 400: + yield f'data: {{"error": "compute node returned {resp.status_code}"}}\n\n'.encode() + return + async for chunk in resp.aiter_bytes(): + yield chunk + except Exception as exc: + yield f'data: {{"error": "stream proxy failed: {exc}"}}\n\n'.encode() + + return StreamingResponse(stream_proxy(), media_type="text/event-stream") + + def fail(status_code: int, message: str) -> HTTPException: return HTTPException(status_code=status_code, detail={"code": status_code, "message": message, "data": None}) @@ -1022,8 +1078,27 @@ async def model_compare_detail(task_id: str) -> dict[str, Any]: raise fail(404, "compare task not found") +async def _unload_from_compute_node() -> dict[str, Any]: + """Best-effort unload the inference model from the first online compute node.""" + store = get_platform_store() + # Clear all inference tracking — only one model can be loaded at a time + for node in store.compute_nodes(): + store.mark_inference_unloaded(node["id"]) + node = _select_first_online_node(store) + if not node: + return {"unloaded": False, "error": "no online compute node"} + try: + client = ComputeNodeClient(node["api_base_url"]) + result = await client._request("POST", "/inference/unload", json_data={}) + return result + except Exception as exc: + return {"unloaded": False, "error": str(exc)} + + @router.delete("/model-compare/{task_id}") async def model_compare_delete(task_id: str) -> dict[str, Any]: + # 删除前先释放算力节点上的模型 + await _unload_from_compute_node() get_platform_store().delete_compare_task(task_id) return ok({"deleted": task_id}) @@ -1053,26 +1128,56 @@ async def model_compare_update_load_status(task_id: str, payload: dict[str, Any] @router.post("/model-compare/{task_id}/load") async def model_compare_load(task_id: str) -> dict[str, Any]: + """真正加载模型到算力节点（不再使用假 PID/端口）。""" try: - task = get_platform_store().compare_task(task_id) + store = get_platform_store() + task = store.compare_task(task_id) models = task.get("models") or [] if isinstance(models, str): try: models = json.loads(models) except json.JSONDecodeError: models = [] - loaded_models = [ - { - "model_id": item.get("model_id"), - "model_name": item.get("model_name"), - "status": "ready", - "pid": 45000 + index, - "port": item.get("port") or 18000 + index, + # 选取在线算力节点 + node = _select_first_online_node(store) + if not node: + return ok({"status": "failed", "error": "no online compute node"}) + client = ComputeNodeClient(node["api_base_url"]) + loaded_models = [] + for item in models: + if not isinstance(item, dict): + continue + model_path = item.get("model_path", "") + if not model_path: + # 尝试从模型库获取路径 + model_id = item.get("model_id", "") + try: + db_model = store.model(model_id) + model_path = db_model.get("path", "") + except KeyError: + pass + if not model_path: + loaded_models.append({**item, "status": "error", "error": "model_path not found"}) + continue + # 真正调用算力节点加载模型 + load_payload = { + "model_name_or_path": model_path, + "template": item.get("template", "qwen"), } - for index, item in enumerate(models) - if isinstance(item, dict) - ] - return ok(get_platform_store().update_compare_task(task_id, {"status": "loaded", "load_status": {"loaded_models": loaded_models}})) + if item.get("adapter_path"): + load_payload["adapter_name_or_path"] = item["adapter_path"] + try: + result = await client._request("POST", "/inference/load", json_data=load_payload) + if result.get("loaded"): + store.mark_inference_loaded(node["id"]) + loaded_models.append({**item, "status": "ready"}) + else: + loaded_models.append({**item, "status": "error", "error": result.get("error", "load failed")}) + except Exception as exc: + loaded_models.append({**item, "status": "error", "error": str(exc)}) + status = "loaded" if any(m.get("status") == "ready" for m in loaded_models) else "failed" + updated = store.update_compare_task(task_id, {"status": status, "load_status": {"loaded_models": loaded_models}}) + return ok(updated) except KeyError: raise fail(404, "compare task not found") @@ -1080,7 +1185,11 @@ async def model_compare_load(task_id: str) -> dict[str, Any]: @router.post("/model-compare/{task_id}/unload") async def model_compare_unload(task_id: str) -> dict[str, Any]: try: - return ok(get_platform_store().update_compare_task(task_id, {"status": "pending", "load_status": {"loaded_models": []}})) + store = get_platform_store() + # 真正释放算力节点上的模型资源 + unload_result = await _unload_from_compute_node() + updated = store.update_compare_task(task_id, {"status": "pending", "load_status": {"loaded_models": []}}) + return ok({"task": updated, "unload": unload_result}) except KeyError: raise fail(404, "compare task not found") @@ -1092,18 +1201,23 @@ async def model_compare_start_model(task_id: str, payload: dict[str, Any] = Body @router.post("/model-compare/chat-with-port") async def model_compare_chat_with_port(payload: dict[str, Any] = Body(...)) -> dict[str, Any]: - question = "" - for message in payload.get("messages") or []: - if message.get("role") == "user": - question = str(message.get("content") or "") - content = f"当前后端已收到推理请求：{question[:120]}" - return ok({"response": content, "content": content}) + """Proxy non-streaming chat to the compute node running the inference model.""" + store = get_platform_store() + node = _select_first_online_node(store) + if not node: + return ok({"response": "no online compute node available for inference", "request": payload}) + try: + client = ComputeNodeClient(node["api_base_url"]) + result = await client._request("POST", "/inference/chat", json_data=_build_messages_payload(payload)) + return ok(result) + except Exception as exc: + return ok({"response": f"inference failed: {exc}", "request": payload}) @router.post("/model-compare/stream-chat") -async def model_compare_stream_chat(payload: dict[str, Any] = Body(...)) -> dict[str, Any]: - question = payload.get("user_question") or payload.get("question") or "" - return ok({"response": f"当前后端已收到流式推理请求：{str(question)[:120]}"}) +async def model_compare_stream_chat(payload: dict[str, Any] = Body(...)) -> StreamingResponse: + """Stream chat from the compute node (SSE proxy).""" + return await _stream_chat_proxy(payload) @router.post("/model-chat/batch") @@ -1120,7 +1234,7 @@ async def model_chat_local(payload: dict[str, Any] = Body(...)) -> dict[str, Any return ok({"response": "no online compute node available for inference", "request": payload}) try: client = ComputeNodeClient(node["api_base_url"]) - result = await client._request("POST", "/inference/chat", json_data=payload) + result = await client._request("POST", "/inference/chat", json_data=_build_messages_payload(payload)) return ok(result) except Exception as exc: return ok({"response": f"inference failed: {exc}", "request": payload}) @@ -1129,28 +1243,15 @@ async def model_chat_local(payload: dict[str, Any] = Body(...)) -> dict[str, Any @router.post("/model-chat/local/chat/stream") async def model_chat_local_stream(payload: dict[str, Any] = Body(...)) -> StreamingResponse: """Stream chat from the compute node.""" - store = get_platform_store() - node = _select_first_online_node(store) - if not node: - return StreamingResponse( - iter(['data: {"error": "no online compute node"}\n\n']), - media_type="text/event-stream", - ) - client = ComputeNodeClient(node["api_base_url"]) - - async def stream_proxy(): - async with httpx.AsyncClient(timeout=300) as http: - url = f"{node['api_base_url'].rstrip('/')}/modelTF/inference/chat/stream" - async with http.stream("POST", url, json=payload, headers=client.headers()) as resp: - async for chunk in resp.aiter_bytes(): - yield chunk - - return StreamingResponse(stream_proxy(), media_type="text/event-stream") + return await _stream_chat_proxy(payload) @router.post("/model-chat/local/preload") async def model_chat_local_preload(payload: dict[str, Any] = Body(...)) -> dict[str, Any]: """Load a model on the compute node for inference.""" + model_path = (payload.get("model_name_or_path") or "").strip() + if not model_path: + return ok({"loaded": False, "error": "model_name_or_path is required"}) store = get_platform_store() node = _select_first_online_node(store) if not node: @@ -1158,6 +1259,8 @@ async def model_chat_local_preload(payload: dict[str, Any] = Body(...)) -> dict[ try: client = ComputeNodeClient(node["api_base_url"]) result = await client._request("POST", "/inference/load", json_data=payload) + if result.get("loaded"): + store.mark_inference_loaded(node["id"]) return ok(result) except Exception as exc: return ok({"loaded": False, "error": str(exc)}) @@ -1167,6 +1270,9 @@ async def model_chat_local_preload(payload: dict[str, Any] = Body(...)) -> dict[ async def model_chat_local_unload() -> dict[str, Any]: """Unload the inference model from the compute node.""" store = get_platform_store() + # Clear all inference tracking + for n in store.compute_nodes(): + store.mark_inference_unloaded(n["id"]) node = _select_first_online_node(store) if not node: return ok({"unloaded": False, "error": "no online compute node"}) @@ -1196,6 +1302,9 @@ async def model_chat_local_status() -> dict[str, Any]: @router.post("/model-chat/trained/preload") async def model_chat_trained_preload(payload: dict[str, Any] = Body(...)) -> dict[str, Any]: """Load a trained model (base + adapter) on the compute node for inference.""" + model_path = (payload.get("model_name_or_path") or "").strip() + if not model_path: + return ok({"loaded": False, "error": "model_name_or_path is required"}) store = get_platform_store() node = _select_first_online_node(store) if not node: @@ -1203,6 +1312,8 @@ async def model_chat_trained_preload(payload: dict[str, Any] = Body(...)) -> dic try: client = ComputeNodeClient(node["api_base_url"]) result = await client._request("POST", "/inference/load", json_data=payload) + if result.get("loaded"): + store.mark_inference_loaded(node["id"]) return ok(result) except Exception as exc: return ok({"loaded": False, "error": str(exc)}) diff --git a/backend/app/db/platform_store.py b/backend/app/db/platform_store.py index ab67a5c..f66789d 100644 --- a/backend/app/db/platform_store.py +++ b/backend/app/db/platform_store.py @@ -293,6 +293,19 @@ class PlatformStore: self.database_url = _psycopg_url(database_url or settings.database_url) self.ensure_schema() self.ensure_seed_data() + # Track which compute nodes have an active inference model loaded + self._inference_nodes: set[str] = set() + + # ── inference node tracking ──────────────────────────────────── + + def mark_inference_loaded(self, node_id: str) -> None: + self._inference_nodes.add(node_id) + + def mark_inference_unloaded(self, node_id: str) -> None: + self._inference_nodes.discard(node_id) + + def is_inference_loaded(self, node_id: str) -> bool: + return node_id in self._inference_nodes @contextmanager def connect(self) -> Iterator["PgConnection"]: @@ -2698,6 +2711,11 @@ class PlatformStore: ) busy = task is not None and task.get("status") == "running" reserved = task is not None and 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: + busy = True + reserved = False memory_used = round(row["memory_total_gb"] * (0.72 if busy else 0.18 if reserved else 0.04), 1) gpu_percent = 86 if busy else 22 if reserved else 3 memory_total = float(row["memory_total_gb"] or 0) diff --git a/compute/api/main.py b/compute/api/main.py index f96dca1..3628525 100644 --- a/compute/api/main.py +++ b/compute/api/main.py @@ -688,8 +688,6 @@ def create_app() -> FastAPI: infer_backend=payload.get("infer_backend", "huggingface"), infer_dtype=payload.get("infer_dtype", "auto"), ) - if not result.get("loaded"): - raise HTTPException(status_code=500, detail=result.get("error", "model load failed")) return result @app.post(f"{route_prefix}/inference/unload") diff --git a/compute/engines/llama_factory/inference.py b/compute/engines/llama_factory/inference.py index 4d1e227..0afb036 100644 --- a/compute/engines/llama_factory/inference.py +++ b/compute/engines/llama_factory/inference.py @@ -59,10 +59,18 @@ class InferenceSession: if adapter_name_or_path: args["adapter_name_or_path"] = adapter_name_or_path args.update(kwargs) - model_args, generating_args = get_infer_args(args) - self._model = ChatModel(model_args) - self._tokenizer = self._model.tokenizer - self._generating_args = generating_args + infer_result = get_infer_args(args) + # ChatModel internally re-parses the args dict via get_infer_args, + # so pass the original args (not the parsed dataclass objects). + self._model = ChatModel(args) + self._tokenizer = getattr(self._model, 'tokenizer', None) or self._model.engine.tokenizer + # Extract generating_args (last element) for later use in chat() + generating_args = infer_result[-1] + if hasattr(generating_args, '__dataclass_fields__'): + self._generating_args = {k: v for k, v in vars(generating_args).items() + if not k.startswith('_')} + else: + self._generating_args = dict(generating_args) self._loaded_at = time.time() self._status = "ready" return {"loaded": True, "status": "ready"} @@ -80,6 +88,16 @@ class InferenceSession: pass self._model = None self._tokenizer = None + # 强制释放 PyTorch CUDA 缓存，真正归还 GPU 显存 + try: + import gc + gc.collect() + import torch + if torch.cuda.is_available(): + torch.cuda.empty_cache() + torch.cuda.synchronize() + except Exception: + pass self._status = "idle" self._model_name = "" self._adapter_path = "" @@ -91,11 +109,12 @@ class InferenceSession: if self._status != "ready" or self._model is None: return {"error": "model not loaded", "response": ""} try: - generate_kwargs = {**self._generating_args, "temperature": temperature, "top_p": top_p, "max_new_tokens": max_new_tokens, "do_sample": do_sample} + generate_kwargs = {"temperature": temperature, "top_p": top_p, "max_new_tokens": max_new_tokens, "do_sample": do_sample} generate_kwargs.update(kwargs) - formatted = self._model.tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) + system = next((m["content"] for m in messages if m["role"] == "system"), None) + user_messages = [m for m in messages if m["role"] != "system"] responses = [] - for response in self._model.stream_chat(formatted, generate_kwargs): + for response in self._model.stream_chat(user_messages, system=system, **generate_kwargs): responses.append(response) full_response = "".join(str(r) for r in responses) return {"response": full_response} @@ -108,9 +127,10 @@ class InferenceSession: yield 'data: {"error": "model not loaded"}\n\n' return try: - generate_kwargs = {**self._generating_args, **kwargs} - formatted = self._model.tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) - for new_text in self._model.stream_chat(formatted, generate_kwargs): + generate_kwargs = {**kwargs} + system = next((m["content"] for m in messages if m["role"] == "system"), None) + user_messages = [m for m in messages if m["role"] != "system"] + for new_text in self._model.stream_chat(user_messages, system=system, **generate_kwargs): yield new_text except Exception as exc: yield 'data: {"error": "' + str(exc) + '"}\n\n' diff --git a/frontend/src/api/modules/compare.ts b/frontend/src/api/modules/compare.ts index 30586cf..35f1ee5 100644 --- a/frontend/src/api/modules/compare.ts +++ b/frontend/src/api/modules/compare.ts @@ -64,6 +64,27 @@ export const streamChat = async (data: any): Promise => { } } +/** 真实流式对话 — 使用 fetch 调用后端 SSE 端点，返回 Response 供 ReadableStream 消费 */ +export const streamChatReal = (data: any): Promise => { + const messages = data.messages || [] + if (!messages.length && data.user_question) { + if (data.system_prompt) { + messages.push({ role: 'system', content: data.system_prompt }) + } + messages.push({ role: 'user', content: data.user_question }) + } + return fetch('/modelTF/model-compare/stream-chat', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + messages, + temperature: data.temperature ?? 0.7, + top_p: data.top_p ?? 0.95, + max_tokens: data.max_tokens ?? 2048, + }), + }) +} + /** 非流式对话（按端口代理） */ export const chatWithPort = (data: any) => post('/model-compare/chat-with-port', data) @@ -73,8 +94,8 @@ export const batchChat = (data: any) => post('/model-chat/batch', data) /** 本地 transformers 模型对话 */ export const localChat = (data: any) => post('/model-chat/local/chat', data) -/** 预加载本地模型 */ -export const preloadLocalModel = (data: any) => post('/model-chat/local/preload', data) +/** 预加载本地模型（模型加载耗时长，超时 5 分钟） */ +export const preloadLocalModel = (data: any) => post('/model-chat/local/preload', data, { timeout: 300000 }) -/** 预加载已训练模型 */ -export const preloadTrainedModel = (data: any) => post('/model-chat/trained/preload', data) +/** 预加载已训练模型（超时 5 分钟） */ +export const preloadTrainedModel = (data: any) => post('/model-chat/trained/preload', data, { timeout: 300000 }) diff --git a/frontend/src/composables/useStreamChat.ts b/frontend/src/composables/useStreamChat.ts index 331cbe9..d322529 100644 --- a/frontend/src/composables/useStreamChat.ts +++ b/frontend/src/composables/useStreamChat.ts @@ -1,5 +1,5 @@ import { ref } from 'vue' -import { streamChat } from '@/api/modules/compare' +import { streamChat, streamChatReal } from '@/api/modules/compare' export interface StreamMessage { /** 用户问题 */ @@ -20,6 +20,11 @@ export interface StreamMessage { error?: string } +export interface SendOptions { + /** 是否使用 mock 模式（默认 true，向后兼容） */ + useMock?: boolean +} + /** * 流式对话 composable * 移植自原 model-chat.html： @@ -65,8 +70,10 @@ export function useStreamChat() { /** * 发起流式对话 * @param payload 后端请求体 { port, model_name, model_path, system_prompt, user_question, ... } + * @param options 可选配置 { useMock?: boolean } */ - async function send(payload: any) { + async function send(payload: any, options?: SendOptions) { + const useMock = options?.useMock ?? true loading.value = true message.value = { question: payload.user_question || '', @@ -82,7 +89,10 @@ export function useStreamChat() { const UPDATE_INTERVAL = 50 // 50ms 节流 try { - const response = await streamChat(payload) + const response = useMock + ? await streamChat(payload) + : await streamChatReal(payload) + if (!response.ok) { throw new Error(`HTTP ${response.status}`) } diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts index 816ce84..991fe29 100644 --- a/frontend/src/types/index.ts +++ b/frontend/src/types/index.ts @@ -361,10 +361,10 @@ export interface GpuInfo { processes?: GpuProcess[] fan_speed?: number clock_mhz?: number - driver_version?: string node_id?: string node_code?: string node_name?: string + driver_version?: string } export interface SystemInfo { diff --git a/frontend/src/views/inference/InferenceChatView.vue b/frontend/src/views/inference/InferenceChatView.vue index 7d6b056..1700abd 100644 --- a/frontend/src/views/inference/InferenceChatView.vue +++ b/frontend/src/views/inference/InferenceChatView.vue @@ -10,8 +10,8 @@ import type { CompareTask, LoadedModel } from '@/types' const route = useRoute() const router = useRouter() const taskId = route.params.id as string -/** 是否为 mock 直通模式（新建推理假数据进入，不走真实任务接口） */ -const isMock = taskId === 'mock' +/** 是否为 mock 模式（新建推理无真实 taskId 或明确为 mock 时进入 mock 模式） */ +const isMock = taskId === 'mock' || !taskId || taskId === 'unknown' /** 当前对话使用的模型名 */ const modelName = ref(route.query.model as string || '') @@ -88,30 +88,20 @@ async function handleSend() { return } - // 真实模式：获取已启动模型的端口/路径 - const models = parseLoadedModels(task.value) - const target = models[0] - if (!target) { - ElMessage.error('未找到已启动的模型') - assistantMsg.content = '未找到已启动的模型，请先返回列表加载模型' - assistantMsg.done = true - assistantMsg.isStreaming = false - return - } - - // 流式状态变化时只同步当前回复，避免固定定时器空转。 + // 真实模式：通过后端 SSE 流式代理到算力节点进行推理 activeAssistant = assistantMsg - await send({ - port: target.port, - model_name: target.model_name, - model_path: '', - system_prompt: systemPrompt.value, - user_question: question, - temperature: temperature.value, - top_p: top_p.value, - max_tokens: maxTokens.value, - }) + await send( + { + model_path: route.query.model_path as string || '', + system_prompt: systemPrompt.value, + user_question: question, + temperature: temperature.value, + top_p: top_p.value, + max_tokens: maxTokens.value, + }, + { useMock: false }, + ) // 完成后同步最终内容 assistantMsg.content = message.value.displayContent || message.value.error || '(无回复)' diff --git a/frontend/src/views/inference/InferenceCreateView.vue b/frontend/src/views/inference/InferenceCreateView.vue index 1285b2f..51f1c7f 100644 --- a/frontend/src/views/inference/InferenceCreateView.vue +++ b/frontend/src/views/inference/InferenceCreateView.vue @@ -5,6 +5,8 @@ import { ElMessage, type FormInstance, type FormRules } from 'element-plus' import PageCard from '@/components/PageCard.vue' import { getModelList, getTrainedModels } from '@/api/modules/model' import { getSystemInfo } from '@/api/modules/system' +import { getComputeNodes, type ComputeNode } from '@/api/modules/compute' +import { createCompare, preloadLocalModel, preloadTrainedModel } from '@/api/modules/compare' import type { ModelItem, TrainedModel, GpuInfo } from '@/types' const router = useRouter() @@ -15,6 +17,7 @@ const startupStatus = ref('') const dbModels = ref([]) const trainedModels = ref([]) const gpus = ref([]) +const computeNodes = ref([]) /** 可选模型（下拉用，区分本地/已训练两类） */ interface SelectableModel { @@ -54,7 +57,17 @@ const trainedOptions = computed(() => })), ) -/** key → 模型映射，便于取选中项 */ +/** 仅显示在线算力节点上的空闲 GPU */ +const onlineNodeIds = computed(() => new Set( + computeNodes.value + .filter((n) => n.enabled && n.scheduler_status === 'online') + .map((n) => n.id), +)) +const idleGpus = computed(() => + gpus.value.filter( + (g) => g.status === 'idle' && (!g.node_id || onlineNodeIds.value.has(g.node_id)), + ), +) const modelMap = computed>(() => { const map: Record = {} for (const m of [...dbOptions.value, ...trainedOptions.value]) map[m.key] = m @@ -90,12 +103,56 @@ async function handleSubmit() { submitting.value = true startupStatus.value = '正在启动模型服务...' try { - // 当前为 mock 环境：不创建任务、不启动后端服务， - // 用假数据直通进入对话界面（模型名通过 query 传递）。 - // 接入真实后端后，可在此恢复 createCompare / startModelsInBackground / monitorStartup 流程。 - await new Promise((resolve) => setTimeout(resolve, 1200)) + // Step 1: 将模型加载到算力节点 + const preloadPayload = { + model_name_or_path: m.model_path, + model_name: m.name, + template: 'qwen', + } + let preloadResult: any + if (m.source === 'trained') { + preloadResult = await preloadTrainedModel(preloadPayload) + } else { + preloadResult = await preloadLocalModel(preloadPayload) + } + + if (preloadResult && (preloadResult as any).error) { + ElMessage.warning(`模型加载失败：${(preloadResult as any).error}`) + submitting.value = false + startupStatus.value = '' + return + } + + // Step 2: 创建推理任务记录 + const taskResult = await createCompare({ + name: form.name || m.name, + description: form.description, + models: [ + { + model_id: String(m.id), + model_name: m.name, + model_path: m.model_path, + source: m.source, + gpu_id: form.gpu_id, + }, + ], + }) + const taskId = taskResult?.id || 'unknown' ElMessage.success('模型已启动') + router.push({ + path: `/model-inference/chat/${taskId}`, + query: { + model: m.name, + source: m.source, + model_path: m.model_path, + }, + }) + } catch (e: any) { + // 真实 API 失败时回退到 mock 模式（方便无算力节点的开发调试） + const m = selectedModel.value! + const reason = e?.message || e?.toString() || '未知错误' + ElMessage.warning(`推理服务启动失败：${reason}，进入 mock 演示模式`) router.push({ path: '/model-inference/chat/mock', query: { model: m.name }, @@ -113,16 +170,18 @@ function handleCancel() { async function loadData() { try { - const [db, trained, sys] = await Promise.all([ + const [db, trained, sys, nodes] = await Promise.all([ getModelList(), getTrainedModels(), getSystemInfo(), + getComputeNodes(), ]) dbModels.value = db || [] trainedModels.value = trained?.models || [] gpus.value = sys?.gpu || [] - // 默认选中第一个 GPU - if (gpus.value.length > 0) form.gpu_id = 0 + computeNodes.value = nodes || [] + // 默认选中第一个空闲 GPU + if (idleGpus.value.length > 0) form.gpu_id = idleGpus.value[0].id ?? 0 } catch { // ignore } @@ -172,10 +231,10 @@ onMounted(loadData) diff --git a/frontend/src/views/inference/InferenceListView.vue b/frontend/src/views/inference/InferenceListView.vue index 2b2cc1f..f4bc7fc 100644 --- a/frontend/src/views/inference/InferenceListView.vue +++ b/frontend/src/views/inference/InferenceListView.vue @@ -7,10 +7,8 @@ import { usePolling } from '@/composables/usePolling' import { getCompareList, deleteCompare, - getCompare, loadCompare, unloadCompare, - stopModelByPid, } from '@/api/modules/compare' import type { CompareTask, LoadedModel } from '@/types' import { statusLabel, statusTagType } from '@/utils/status' @@ -86,26 +84,19 @@ async function handleLoad(row: any) { delayedRefreshTimer = setTimeout(loadData, 1000) } -/** 卸载推理任务 */ +/** 释放推理任务（停止模型服务，释放算力节点 GPU 显存） */ async function handleUnload(row: any) { - await ElMessageBox.confirm('确定要停止模型服务吗？', '确认停止', { type: 'warning' }) + await ElMessageBox.confirm('确定要释放模型服务吗？将停止模型进程并释放 GPU 显存。', '确认释放', { type: 'warning' }) await unloadCompare(row.id) - ElMessage.success('已停止模型服务') + ElMessage.success('已释放模型服务') loadData() } -/** 删除（先停止进程） */ +/** 删除（先释放算力节点再删除记录） */ async function handleDelete(row: any) { - // 先尝试停止已加载的模型进程 - const task = await getCompare(row.id).catch(() => null) - if (task?.load_status) { - const models = parseLoadedModels(task as CompareTask) - for (const m of models) { - if (m.pid) { - await stopModelByPid(m.pid).catch(() => {}) - } - } - } + await ElMessageBox.confirm('确定要删除该推理记录吗？将先释放算力节点再删除。', '确认删除', { type: 'warning' }) + // 先释放算力节点上的模型 + await unloadCompare(row.id).catch(() => {}) await deleteCompare(row.id) dataList.value = dataList.value.filter((item) => item.id !== row.id) await loadData(true) @@ -180,7 +171,7 @@ onUnmounted(() => { 对话 - 停止 + 释放