feat: 模型推理端到端闭环 — 真实流式推理 + 释放/删除 + GPU 状态同步

后端 (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 <noreply@anthropic.com>
This commit is contained in:
wuyongtao
2026-07-28 17:29:16 +08:00
parent f917a025e1
commit c7c9ed925b
10 changed files with 331 additions and 113 deletions

View File

@@ -31,6 +31,62 @@ def _select_first_online_node(store: Any) -> dict[str, Any] | None:
return 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: def fail(status_code: int, message: str) -> HTTPException:
return HTTPException(status_code=status_code, detail={"code": status_code, "message": message, "data": None}) 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") 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}") @router.delete("/model-compare/{task_id}")
async def model_compare_delete(task_id: str) -> dict[str, Any]: async def model_compare_delete(task_id: str) -> dict[str, Any]:
# 删除前先释放算力节点上的模型
await _unload_from_compute_node()
get_platform_store().delete_compare_task(task_id) get_platform_store().delete_compare_task(task_id)
return ok({"deleted": 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") @router.post("/model-compare/{task_id}/load")
async def model_compare_load(task_id: str) -> dict[str, Any]: async def model_compare_load(task_id: str) -> dict[str, Any]:
"""真正加载模型到算力节点(不再使用假 PID/端口)。"""
try: try:
task = get_platform_store().compare_task(task_id) store = get_platform_store()
task = store.compare_task(task_id)
models = task.get("models") or [] models = task.get("models") or []
if isinstance(models, str): if isinstance(models, str):
try: try:
models = json.loads(models) models = json.loads(models)
except json.JSONDecodeError: except json.JSONDecodeError:
models = [] models = []
loaded_models = [ # 选取在线算力节点
{ node = _select_first_online_node(store)
"model_id": item.get("model_id"), if not node:
"model_name": item.get("model_name"), return ok({"status": "failed", "error": "no online compute node"})
"status": "ready", client = ComputeNodeClient(node["api_base_url"])
"pid": 45000 + index, loaded_models = []
"port": item.get("port") or 18000 + index, 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 item.get("adapter_path"):
if isinstance(item, dict) load_payload["adapter_name_or_path"] = item["adapter_path"]
] try:
return ok(get_platform_store().update_compare_task(task_id, {"status": "loaded", "load_status": {"loaded_models": loaded_models}})) 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: except KeyError:
raise fail(404, "compare task not found") 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") @router.post("/model-compare/{task_id}/unload")
async def model_compare_unload(task_id: str) -> dict[str, Any]: async def model_compare_unload(task_id: str) -> dict[str, Any]:
try: 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: except KeyError:
raise fail(404, "compare task not found") 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") @router.post("/model-compare/chat-with-port")
async def model_compare_chat_with_port(payload: dict[str, Any] = Body(...)) -> dict[str, Any]: async def model_compare_chat_with_port(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
question = "" """Proxy non-streaming chat to the compute node running the inference model."""
for message in payload.get("messages") or []: store = get_platform_store()
if message.get("role") == "user": node = _select_first_online_node(store)
question = str(message.get("content") or "") if not node:
content = f"当前后端已收到推理请求:{question[:120]}" return ok({"response": "no online compute node available for inference", "request": payload})
return ok({"response": content, "content": content}) 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") @router.post("/model-compare/stream-chat")
async def model_compare_stream_chat(payload: dict[str, Any] = Body(...)) -> dict[str, Any]: async def model_compare_stream_chat(payload: dict[str, Any] = Body(...)) -> StreamingResponse:
question = payload.get("user_question") or payload.get("question") or "" """Stream chat from the compute node (SSE proxy)."""
return ok({"response": f"当前后端已收到流式推理请求:{str(question)[:120]}"}) return await _stream_chat_proxy(payload)
@router.post("/model-chat/batch") @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}) return ok({"response": "no online compute node available for inference", "request": payload})
try: try:
client = ComputeNodeClient(node["api_base_url"]) 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) return ok(result)
except Exception as exc: except Exception as exc:
return ok({"response": f"inference failed: {exc}", "request": payload}) 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") @router.post("/model-chat/local/chat/stream")
async def model_chat_local_stream(payload: dict[str, Any] = Body(...)) -> StreamingResponse: async def model_chat_local_stream(payload: dict[str, Any] = Body(...)) -> StreamingResponse:
"""Stream chat from the compute node.""" """Stream chat from the compute node."""
store = get_platform_store() return await _stream_chat_proxy(payload)
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")
@router.post("/model-chat/local/preload") @router.post("/model-chat/local/preload")
async def model_chat_local_preload(payload: dict[str, Any] = Body(...)) -> dict[str, Any]: async def model_chat_local_preload(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
"""Load a model on the compute node for inference.""" """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() store = get_platform_store()
node = _select_first_online_node(store) node = _select_first_online_node(store)
if not node: if not node:
@@ -1158,6 +1259,8 @@ async def model_chat_local_preload(payload: dict[str, Any] = Body(...)) -> dict[
try: try:
client = ComputeNodeClient(node["api_base_url"]) client = ComputeNodeClient(node["api_base_url"])
result = await client._request("POST", "/inference/load", json_data=payload) result = await client._request("POST", "/inference/load", json_data=payload)
if result.get("loaded"):
store.mark_inference_loaded(node["id"])
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)})
@@ -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]: async def model_chat_local_unload() -> dict[str, Any]:
"""Unload the inference model from the compute node.""" """Unload the inference model from the compute node."""
store = get_platform_store() 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) node = _select_first_online_node(store)
if not node: if not node:
return ok({"unloaded": False, "error": "no online compute 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") @router.post("/model-chat/trained/preload")
async def model_chat_trained_preload(payload: dict[str, Any] = Body(...)) -> dict[str, Any]: 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.""" """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() store = get_platform_store()
node = _select_first_online_node(store) node = _select_first_online_node(store)
if not node: if not node:
@@ -1203,6 +1312,8 @@ async def model_chat_trained_preload(payload: dict[str, Any] = Body(...)) -> dic
try: try:
client = ComputeNodeClient(node["api_base_url"]) client = ComputeNodeClient(node["api_base_url"])
result = await client._request("POST", "/inference/load", json_data=payload) result = await client._request("POST", "/inference/load", json_data=payload)
if result.get("loaded"):
store.mark_inference_loaded(node["id"])
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

@@ -293,6 +293,19 @@ class PlatformStore:
self.database_url = _psycopg_url(database_url or settings.database_url) self.database_url = _psycopg_url(database_url or settings.database_url)
self.ensure_schema() self.ensure_schema()
self.ensure_seed_data() 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 @contextmanager
def connect(self) -> Iterator["PgConnection"]: def connect(self) -> Iterator["PgConnection"]:
@@ -2698,6 +2711,11 @@ class PlatformStore:
) )
busy = task is not None and task.get("status") == "running" busy = task is not None and task.get("status") == "running"
reserved = task is not None and task.get("status") in {"syncing", "queued"} 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) 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 gpu_percent = 86 if busy else 22 if reserved else 3
memory_total = float(row["memory_total_gb"] or 0) memory_total = float(row["memory_total_gb"] or 0)

View File

@@ -688,8 +688,6 @@ def create_app() -> FastAPI:
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"),
) )
if not result.get("loaded"):
raise HTTPException(status_code=500, detail=result.get("error", "model load failed"))
return result return result
@app.post(f"{route_prefix}/inference/unload") @app.post(f"{route_prefix}/inference/unload")

View File

@@ -59,10 +59,18 @@ class InferenceSession:
if adapter_name_or_path: if adapter_name_or_path:
args["adapter_name_or_path"] = adapter_name_or_path args["adapter_name_or_path"] = adapter_name_or_path
args.update(kwargs) args.update(kwargs)
model_args, generating_args = get_infer_args(args) infer_result = get_infer_args(args)
self._model = ChatModel(model_args) # ChatModel internally re-parses the args dict via get_infer_args,
self._tokenizer = self._model.tokenizer # so pass the original args (not the parsed dataclass objects).
self._generating_args = generating_args 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._loaded_at = time.time()
self._status = "ready" self._status = "ready"
return {"loaded": True, "status": "ready"} return {"loaded": True, "status": "ready"}
@@ -80,6 +88,16 @@ class InferenceSession:
pass pass
self._model = None self._model = None
self._tokenizer = 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._status = "idle"
self._model_name = "" self._model_name = ""
self._adapter_path = "" self._adapter_path = ""
@@ -91,11 +109,12 @@ class InferenceSession:
if self._status != "ready" or self._model is None: if self._status != "ready" or self._model is None:
return {"error": "model not loaded", "response": ""} return {"error": "model not loaded", "response": ""}
try: 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) 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 = [] 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) responses.append(response)
full_response = "".join(str(r) for r in responses) full_response = "".join(str(r) for r in responses)
return {"response": full_response} return {"response": full_response}
@@ -108,9 +127,10 @@ class InferenceSession:
yield 'data: {"error": "model not loaded"}\n\n' yield 'data: {"error": "model not loaded"}\n\n'
return return
try: try:
generate_kwargs = {**self._generating_args, **kwargs} generate_kwargs = {**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)
for new_text in self._model.stream_chat(formatted, generate_kwargs): 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 yield new_text
except Exception as exc: except Exception as exc:
yield 'data: {"error": "' + str(exc) + '"}\n\n' yield 'data: {"error": "' + str(exc) + '"}\n\n'

View File

@@ -64,6 +64,27 @@ export const streamChat = async (data: any): Promise<any> => {
} }
} }
/** 真实流式对话 — 使用 fetch 调用后端 SSE 端点,返回 Response 供 ReadableStream 消费 */
export const streamChatReal = (data: any): Promise<Response> => {
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) 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 模型对话 */ /** 本地 transformers 模型对话 */
export const localChat = (data: any) => post('/model-chat/local/chat', data) export const localChat = (data: any) => post('/model-chat/local/chat', data)
/** 预加载本地模型 */ /** 预加载本地模型(模型加载耗时长,超时 5 分钟) */
export const preloadLocalModel = (data: any) => post('/model-chat/local/preload', data) export const preloadLocalModel = (data: any) => post('/model-chat/local/preload', data, { timeout: 300000 })
/** 预加载已训练模型 */ /** 预加载已训练模型(超时 5 分钟) */
export const preloadTrainedModel = (data: any) => post('/model-chat/trained/preload', data) export const preloadTrainedModel = (data: any) => post('/model-chat/trained/preload', data, { timeout: 300000 })

View File

@@ -1,5 +1,5 @@
import { ref } from 'vue' import { ref } from 'vue'
import { streamChat } from '@/api/modules/compare' import { streamChat, streamChatReal } from '@/api/modules/compare'
export interface StreamMessage { export interface StreamMessage {
/** 用户问题 */ /** 用户问题 */
@@ -20,6 +20,11 @@ export interface StreamMessage {
error?: string error?: string
} }
export interface SendOptions {
/** 是否使用 mock 模式(默认 true向后兼容 */
useMock?: boolean
}
/** /**
* 流式对话 composable * 流式对话 composable
* 移植自原 model-chat.html * 移植自原 model-chat.html
@@ -65,8 +70,10 @@ export function useStreamChat() {
/** /**
* 发起流式对话 * 发起流式对话
* @param payload 后端请求体 { port, model_name, model_path, system_prompt, user_question, ... } * @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 loading.value = true
message.value = { message.value = {
question: payload.user_question || '', question: payload.user_question || '',
@@ -82,7 +89,10 @@ export function useStreamChat() {
const UPDATE_INTERVAL = 50 // 50ms 节流 const UPDATE_INTERVAL = 50 // 50ms 节流
try { try {
const response = await streamChat(payload) const response = useMock
? await streamChat(payload)
: await streamChatReal(payload)
if (!response.ok) { if (!response.ok) {
throw new Error(`HTTP ${response.status}`) throw new Error(`HTTP ${response.status}`)
} }

View File

@@ -361,10 +361,10 @@ export interface GpuInfo {
processes?: GpuProcess[] processes?: GpuProcess[]
fan_speed?: number fan_speed?: number
clock_mhz?: number clock_mhz?: number
driver_version?: string
node_id?: string node_id?: string
node_code?: string node_code?: string
node_name?: string node_name?: string
driver_version?: string
} }
export interface SystemInfo { export interface SystemInfo {

View File

@@ -10,8 +10,8 @@ import type { CompareTask, LoadedModel } from '@/types'
const route = useRoute() const route = useRoute()
const router = useRouter() const router = useRouter()
const taskId = route.params.id as string const taskId = route.params.id as string
/** 是否为 mock 直通模式(新建推理假数据进入,不走真实任务接口 */ /** 是否为 mock 模式(新建推理无真实 taskId 或明确为 mock 时进入 mock 模式 */
const isMock = taskId === 'mock' const isMock = taskId === 'mock' || !taskId || taskId === 'unknown'
/** 当前对话使用的模型名 */ /** 当前对话使用的模型名 */
const modelName = ref(route.query.model as string || '') const modelName = ref(route.query.model as string || '')
@@ -88,30 +88,20 @@ async function handleSend() {
return return
} }
// 真实模式:获取已启动模型的端口/路径 // 真实模式:通过后端 SSE 流式代理到算力节点进行推理
const models = parseLoadedModels(task.value)
const target = models[0]
if (!target) {
ElMessage.error('未找到已启动的模型')
assistantMsg.content = '未找到已启动的模型,请先返回列表加载模型'
assistantMsg.done = true
assistantMsg.isStreaming = false
return
}
// 流式状态变化时只同步当前回复,避免固定定时器空转。
activeAssistant = assistantMsg activeAssistant = assistantMsg
await send({ await send(
port: target.port, {
model_name: target.model_name, model_path: route.query.model_path as string || '',
model_path: '', system_prompt: systemPrompt.value,
system_prompt: systemPrompt.value, user_question: question,
user_question: question, temperature: temperature.value,
temperature: temperature.value, top_p: top_p.value,
top_p: top_p.value, max_tokens: maxTokens.value,
max_tokens: maxTokens.value, },
}) { useMock: false },
)
// 完成后同步最终内容 // 完成后同步最终内容
assistantMsg.content = message.value.displayContent || message.value.error || '(无回复)' assistantMsg.content = message.value.displayContent || message.value.error || '(无回复)'

View File

@@ -5,6 +5,8 @@ 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 { 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' import type { ModelItem, TrainedModel, GpuInfo } from '@/types'
const router = useRouter() const router = useRouter()
@@ -15,6 +17,7 @@ const startupStatus = ref('')
const dbModels = ref<ModelItem[]>([]) const dbModels = ref<ModelItem[]>([])
const trainedModels = ref<TrainedModel[]>([]) const trainedModels = ref<TrainedModel[]>([])
const gpus = ref<GpuInfo[]>([]) const gpus = ref<GpuInfo[]>([])
const computeNodes = ref<ComputeNode[]>([])
/** 可选模型(下拉用,区分本地/已训练两类) */ /** 可选模型(下拉用,区分本地/已训练两类) */
interface SelectableModel { interface SelectableModel {
@@ -54,7 +57,17 @@ const trainedOptions = computed<SelectableModel[]>(() =>
})), })),
) )
/** 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<Record<string, SelectableModel>>(() => { const modelMap = computed<Record<string, SelectableModel>>(() => {
const map: Record<string, SelectableModel> = {} const map: Record<string, SelectableModel> = {}
for (const m of [...dbOptions.value, ...trainedOptions.value]) map[m.key] = m for (const m of [...dbOptions.value, ...trainedOptions.value]) map[m.key] = m
@@ -90,12 +103,56 @@ async function handleSubmit() {
submitting.value = true submitting.value = true
startupStatus.value = '正在启动模型服务...' startupStatus.value = '正在启动模型服务...'
try { try {
// 当前为 mock 环境:不创建任务、不启动后端服务, // Step 1: 将模型加载到算力节点
// 用假数据直通进入对话界面(模型名通过 query 传递)。 const preloadPayload = {
// 接入真实后端后,可在此恢复 createCompare / startModelsInBackground / monitorStartup 流程。 model_name_or_path: m.model_path,
await new Promise((resolve) => setTimeout(resolve, 1200)) 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('模型已启动') 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({ router.push({
path: '/model-inference/chat/mock', path: '/model-inference/chat/mock',
query: { model: m.name }, query: { model: m.name },
@@ -113,16 +170,18 @@ function handleCancel() {
async function loadData() { async function loadData() {
try { try {
const [db, trained, sys] = await Promise.all([ const [db, trained, sys, nodes] = await Promise.all([
getModelList(), getModelList(),
getTrainedModels(), getTrainedModels(),
getSystemInfo(), getSystemInfo(),
getComputeNodes(),
]) ])
dbModels.value = db || [] dbModels.value = db || []
trainedModels.value = trained?.models || [] trainedModels.value = trained?.models || []
gpus.value = sys?.gpu || [] gpus.value = sys?.gpu || []
// 默认选中第一个 GPU computeNodes.value = nodes || []
if (gpus.value.length > 0) form.gpu_id = 0 // 默认选中第一个空闲 GPU
if (idleGpus.value.length > 0) form.gpu_id = idleGpus.value[0].id ?? 0
} catch { } catch {
// ignore // ignore
} }
@@ -172,10 +231,10 @@ onMounted(loadData)
<el-form-item label="GPU"> <el-form-item label="GPU">
<el-select v-model="form.gpu_id" style="width: 400px"> <el-select v-model="form.gpu_id" style="width: 400px">
<el-option <el-option
v-for="(g, idx) in gpus" v-for="g in idleGpus"
:key="idx" :key="g.id ?? 0"
:label="`${g.name} (GPU${idx})`" :label="`${g.name} (GPU${g.id ?? 0}) [空闲]`"
:value="idx" :value="g.id ?? 0"
/> />
</el-select> </el-select>
</el-form-item> </el-form-item>

View File

@@ -7,10 +7,8 @@ import { usePolling } from '@/composables/usePolling'
import { import {
getCompareList, getCompareList,
deleteCompare, deleteCompare,
getCompare,
loadCompare, loadCompare,
unloadCompare, unloadCompare,
stopModelByPid,
} from '@/api/modules/compare' } from '@/api/modules/compare'
import type { CompareTask, LoadedModel } from '@/types' import type { CompareTask, LoadedModel } from '@/types'
import { statusLabel, statusTagType } from '@/utils/status' import { statusLabel, statusTagType } from '@/utils/status'
@@ -86,26 +84,19 @@ async function handleLoad(row: any) {
delayedRefreshTimer = setTimeout(loadData, 1000) delayedRefreshTimer = setTimeout(loadData, 1000)
} }
/** 卸载推理任务 */ /** 释放推理任务(停止模型服务,释放算力节点 GPU 显存) */
async function handleUnload(row: any) { async function handleUnload(row: any) {
await ElMessageBox.confirm('确定要停止模型服务吗?', '确认停止', { type: 'warning' }) await ElMessageBox.confirm('确定要释放模型服务吗?将停止模型进程并释放 GPU 显存。', '确认释放', { type: 'warning' })
await unloadCompare(row.id) await unloadCompare(row.id)
ElMessage.success('已停止模型服务') ElMessage.success('已释放模型服务')
loadData() loadData()
} }
/** 删除(先停止进程 */ /** 删除(先释放算力节点再删除记录 */
async function handleDelete(row: any) { async function handleDelete(row: any) {
// 先尝试停止已加载的模型进程 await ElMessageBox.confirm('确定要删除该推理记录吗?将先释放算力节点再删除。', '确认删除', { type: 'warning' })
const task = await getCompare(row.id).catch(() => null) // 先释放算力节点上的模型
if (task?.load_status) { await unloadCompare(row.id).catch(() => {})
const models = parseLoadedModels(task as CompareTask)
for (const m of models) {
if (m.pid) {
await stopModelByPid(m.pid).catch(() => {})
}
}
}
await deleteCompare(row.id) await deleteCompare(row.id)
dataList.value = dataList.value.filter((item) => item.id !== row.id) dataList.value = dataList.value.filter((item) => item.id !== row.id)
await loadData(true) await loadData(true)
@@ -180,7 +171,7 @@ onUnmounted(() => {
<i class="fa fa-comments-o" style="margin-right: 4px" />对话 <i class="fa fa-comments-o" style="margin-right: 4px" />对话
</el-button> </el-button>
<el-button type="warning" link size="small" @click="handleUnload(row)"> <el-button type="warning" link size="small" @click="handleUnload(row)">
<i class="fa fa-stop-circle-o" style="margin-right: 4px" />停止 <i class="fa fa-stop-circle-o" style="margin-right: 4px" />释放
</el-button> </el-button>
</template> </template>
<template v-else> <template v-else>