feat: 模型推理异步加载与对话链路修复,同步基线
模型推理全异步化改造: - 计算节点 InferenceSession 改为后台线程异步加载模型,load 立即返回, 加载期间事件循环保持响应(/inference/status 与 /health 不阻塞) - 后端模型加载改为异步派发 + 轮询对账器(reconcile_inference_loads), 任务状态由 starting 自动推进到 ready/error,解决多节点启动超时 (timeout of 120000ms exceeded) - 推理删除/卸载改为任务感知 + 短超时,删除先删记录再 best-effort 卸载, 不再被不可达节点阻塞;同节点新模型替换旧任务标记失效 - 流式对话透传 task_id/node_id 路由到真正加载模型的算力节点, useStreamChat 解析 SSE 错误帧以干净文案展示 - 对话历史按任务 id 本地持久化,退出重进可恢复;移除页脚提示文本 - 新增后端推理异步加载与计算节点异步状态机单元测试 Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -1,15 +1,101 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
from app.db.platform_store import get_platform_store
|
||||
from app.modules.compute_gateway.client import ComputeNodeClient
|
||||
|
||||
# starting 状态允许的最大轮询次数(约 40 * 3s ≈ 2 分钟),超过即判定节点不可达
|
||||
MAX_STARTING_ATTEMPTS = 40
|
||||
|
||||
|
||||
def _node_for_task(task: dict[str, Any]) -> dict[str, Any] | None:
|
||||
return next((node for node in get_platform_store().compute_nodes() if node["id"] == task.get("compute_node_id")), None)
|
||||
|
||||
|
||||
def _parse_inference_load_status(task: dict[str, Any]) -> tuple[list[dict[str, Any]], dict[str, Any]]:
|
||||
load_status = task.get("load_status") or {}
|
||||
if isinstance(load_status, str):
|
||||
try:
|
||||
load_status = json.loads(load_status)
|
||||
except (json.JSONDecodeError, TypeError):
|
||||
load_status = {}
|
||||
return load_status.get("loaded_models") or [], load_status
|
||||
|
||||
|
||||
async def reconcile_inference_loads(store: Any) -> list[dict[str, Any]]:
|
||||
"""推进处于 starting 状态的推理加载。
|
||||
|
||||
模型加载已改为异步派发:/model-compare/{id}/load 立即返回,这里在每次
|
||||
轮询时查询对应计算节点的 /inference/status,把任务从 starting 推进到
|
||||
ready/error。使用短超时,单节点不可达不会阻塞整轮轮询。
|
||||
"""
|
||||
reconciled: list[dict[str, Any]] = []
|
||||
now = time.time()
|
||||
for task in store.compare_tasks():
|
||||
items, _ = _parse_inference_load_status(task)
|
||||
if not any(item.get("status") == "starting" for item in items):
|
||||
continue
|
||||
# dirty 只要处理过任一 starting 项就置位:load_attempts / last_polled_at
|
||||
# 必须落库,否则节点不可达时计数不会累积,封顶逻辑永远触发不了
|
||||
dirty = False
|
||||
for item in items:
|
||||
if item.get("status") != "starting":
|
||||
continue
|
||||
# 节流:同一 item 每 3s 只查询一次
|
||||
if now - float(item.get("last_polled_at") or 0) < 3:
|
||||
continue
|
||||
item["last_polled_at"] = now
|
||||
item["load_attempts"] = int(item.get("load_attempts") or 0) + 1
|
||||
dirty = True
|
||||
node = next((n for n in store.compute_nodes() if n["id"] == item.get("node_id")), None)
|
||||
if not node:
|
||||
item["status"] = "error"
|
||||
item["error"] = "compute node deleted"
|
||||
store.mark_inference_unloaded(item.get("node_id") or "")
|
||||
continue
|
||||
if not node.get("enabled") or node.get("scheduler_status") != "online":
|
||||
item["status"] = "error"
|
||||
item["error"] = "compute node offline"
|
||||
store.mark_inference_unloaded(node["id"])
|
||||
continue
|
||||
try:
|
||||
status = await ComputeNodeClient(node["api_base_url"]).inference_status()
|
||||
except Exception as exc: # noqa: BLE001 - node unreachable; keep retrying until cap
|
||||
if int(item.get("load_attempts") or 0) >= MAX_STARTING_ATTEMPTS:
|
||||
item["status"] = "error"
|
||||
item["error"] = f"compute node unreachable: {exc}"
|
||||
store.mark_inference_unloaded(node["id"])
|
||||
continue
|
||||
node_status = status.get("status")
|
||||
if node_status == "ready":
|
||||
item["status"] = "ready"
|
||||
item.pop("error", None)
|
||||
store.mark_inference_loaded(node["id"])
|
||||
elif node_status == "error":
|
||||
item["status"] = "error"
|
||||
item["error"] = status.get("error") or "model load failed on compute node"
|
||||
store.mark_inference_unloaded(node["id"])
|
||||
elif node_status == "idle":
|
||||
# 节点重启导致已加载模型丢失
|
||||
item["status"] = "error"
|
||||
item["error"] = "model disappeared from compute node (node may have restarted)"
|
||||
store.mark_inference_unloaded(node["id"])
|
||||
# node_status == "loading" -> 保持 starting,下轮再查
|
||||
if dirty:
|
||||
if any(i.get("status") in {"ready", "running"} for i in items):
|
||||
new_status = "loaded"
|
||||
elif any(i.get("status") == "starting" for i in items):
|
||||
new_status = "starting" # 仍在加载中,保持 starting
|
||||
else:
|
||||
new_status = "failed"
|
||||
store.update_compare_task(task["id"], {"status": new_status, "load_status": {"loaded_models": items}})
|
||||
reconciled.append({"task_id": task["id"], "status": new_status})
|
||||
return reconciled
|
||||
|
||||
|
||||
async def fetch_eval_result_content(client: ComputeNodeClient, node: dict[str, Any], job: dict[str, Any]) -> dict[str, Any] | None:
|
||||
output_dir = job.get("output_dir")
|
||||
if not output_dir:
|
||||
@@ -95,5 +181,13 @@ async def poll_compute_jobs_once() -> dict[str, Any]:
|
||||
except Exception as exc: # noqa: BLE001
|
||||
failed.append({"eval_task_id": eval_task["id"], "error": str(exc)})
|
||||
|
||||
# ── Inference load reconciliation ─────────────────────────────────────
|
||||
try:
|
||||
inference_reconciled = await reconcile_inference_loads(store)
|
||||
except Exception as exc: # noqa: BLE001 - keep polling alive
|
||||
failed.append({"inference_reconcile": str(exc)})
|
||||
inference_reconciled = []
|
||||
|
||||
return {"synced": len(synced) + len(standalone_synced) + eval_synced, "failed": failed,
|
||||
"items": synced, "standalone": standalone_synced, "eval_synced": eval_synced}
|
||||
"items": synced, "standalone": standalone_synced, "eval_synced": eval_synced,
|
||||
"inference_reconciled": inference_reconciled}
|
||||
|
||||
Reference in New Issue
Block a user