feat(multimodal): 补齐阶段F运行管理与收尾验收

backend/app/local_models/manager.py

@@ -49,8 +49,20 @@ def task_key(key):
     return str(model_path(key)), key
 
 
+def disk_bytes(key):
+    total = 0
+    try:
+        root = model_path(key).resolve()
+        for path in root.rglob("*"):
+            if not path.is_symlink() and path.is_file() and path.resolve().is_relative_to(root):
+                total += path.stat().st_size
+    except OSError:
+        return None
+    return total
+
+
 def describe():
-    return {"items": [{**spec.public(), **read_state(key)} for key, spec in CATALOG.items()]}
+    return {"items": [{**spec.public(), **read_state(key), "disk_bytes": disk_bytes(key)} for key, spec in CATALOG.items()]}
 
 
 async def download(key):

backend/app/local_models/runtime.py

@@ -4,8 +4,10 @@ from __future__ import annotations
 import asyncio
 import json
 import os
+import time
 from contextlib import closing
 from contextvars import ContextVar
+from functools import wraps
 from pathlib import Path
 from typing import Literal
 
@@ -31,6 +33,18 @@ class RuntimeConfig(BaseModel):
 
 runtime_context = ContextVar("runtime_config", default=None)
 runtime_progress = ContextVar("runtime_progress", default=None)
+embedding_priority = ContextVar("embedding_priority", default=0)
+
+
+def background_embeddings(operation):
+    @wraps(operation)
+    async def wrapped(*args, **kwargs):
+        token = embedding_priority.set(20)
+        try:
+            return await operation(*args, **kwargs)
+        finally:
+            embedding_priority.reset(token)
+    return wrapped
 
 
 def configuration():
@@ -75,28 +89,81 @@ class Runtime:
         return any(target in paths for paths in self.active_files.values())
 
     async def infer(self, key, operation, payload, *, priority=10):
-        if read_state(key)["status"] != "installed":
-            raise ProviderError("LOCAL_MODEL_NOT_INSTALLED", "请先在模型配置中下载本地模型。")
-        if not interpreter().is_file():
-            raise ProviderError("LOCAL_RUNTIME_NOT_INSTALLED", "请先运行本地模型 CPU/CUDA 安装脚本。")
-        config = configuration()
+        from app.services import model_diagnostics
+        config = configuration().model_copy(deep=True)
         self.counter += 1
         ticket = (priority, self.counter)
         self.waiters.append(ticket)
-        process = None
-        attempt = None
+        queued_at = time.monotonic()
+        reason = None
+        from app.services.usage_service import usage_context
+        from uuid import uuid4
+        context = dict(usage_context.get() or {})
+        context.setdefault("request_id", uuid4().hex)
+        usage_token = usage_context.set(context)
         try:
-            # One resident model at a time prevents overlapping CPU/GPU allocations.
             while self.active or ticket != min(self.waiters):
                 await asyncio.sleep(0.05)
             self.waiters.remove(ticket)
             self.active[ticket] = key
             self.active_files[ticket] = {str(Path(payload[name]).resolve()) for name in ("source", "reference") if payload.get(name)}
-            # Deletion may have occurred while this request was queued.
-            if read_state(key)["status"] != "installed":
-                raise ProviderError("LOCAL_MODEL_NOT_INSTALLED", "模型文件已被删除。")
-            from app.services.usage_service import UsageAttempt
-            attempt = UsageAttempt("local-models", CATALOG[key].repository, "local", operation, source="local")
+            queue_seconds = time.monotonic() - queued_at
+            # Keep the reservation while replacing a failed CUDA process with CPU.
+            for device in (["cuda", "cpu"] if config.device == "cuda" else ["cpu"]):
+                started = time.monotonic()
+                diagnostics = dict(model=CATALOG[key].repository, revision=CATALOG[key].revision,
+                    operation=operation, source="local", requested_device=config.device,
+                    attempted_device=device, queue_seconds=queue_seconds, fallback_reason=reason, request_id=context["request_id"])
+                try:
+                    result = await self._execute(key, operation, payload, config.model_copy(update={"device": device}), diagnostics)
+                    diagnostics.update(result.get("diagnostics", {}))
+                    diagnostics.update(requested_device=config.device, status="completed")
+                    if reason:
+                        diagnostics["fallback_reason"] = reason
+                    return result["result"]
+                except asyncio.CancelledError:
+                    diagnostics.update(status="cancelled", error_code="LOCAL_MODEL_CANCELLED")
+                    raise
+                except ProviderError as exc:
+                    diagnostics.update(status="failed", error_code=exc.code)
+                    if device == "cuda" and exc.code in {"LOCAL_CUDA_INIT_FAILED", "LOCAL_CUDA_OOM"}:
+                        reason = exc.code
+                        callback = runtime_progress.get()
+                        if callback:
+                            callback({"reset": True, "progress": 0})
+                        continue
+                    raise
+                except Exception:
+                    diagnostics.update(status="failed", error_code="LOCAL_MODEL_INVALID_RESPONSE")
+                    raise ProviderError("LOCAL_MODEL_INVALID_RESPONSE", "本地模型返回无效数据。") from None
+                finally:
+                    diagnostics["requested_device"] = config.device
+                    diagnostics["elapsed_seconds"] = time.monotonic() - started
+                    self.diagnostics.append(model_diagnostics.record(**diagnostics))
+                    self.diagnostics = self.diagnostics[-100:]
+        except asyncio.CancelledError:
+            if ticket not in self.active:
+                model_diagnostics.record(model=CATALOG[key].repository, operation=operation,
+                    source="local", status="cancelled", error_code="LOCAL_QUEUE_CANCELLED",
+                    requested_device=config.device, queue_seconds=time.monotonic() - queued_at)
+            raise
+        finally:
+            if ticket in self.waiters:
+                self.waiters.remove(ticket)
+            self.active.pop(ticket, None)
+            self.active_files.pop(ticket, None)
+            usage_context.reset(usage_token)
+
+    async def _execute(self, key, operation, payload, config, diagnostics):
+        if read_state(key)["status"] != "installed":
+            raise ProviderError("LOCAL_MODEL_NOT_INSTALLED", "请先下载本地模型。")
+        if not interpreter().is_file():
+            raise ProviderError("LOCAL_RUNTIME_NOT_INSTALLED", "请先安装本地模型运行环境。")
+        from app.services.usage_service import UsageAttempt
+        attempt = UsageAttempt("local-models", CATALOG[key].repository, "local", operation, source="local")
+        diagnostics.update(attempt_id=attempt.attempt_id, request_id=attempt.request_id)
+        process = None
+        try:
             env = {**os.environ, "HF_HUB_OFFLINE": "1", "TRANSFORMERS_OFFLINE": "1",
                    "HF_HUB_DISABLE_TELEMETRY": "1", "OMP_NUM_THREADS": str(config.cpu_threads),
                    "PYTHONIOENCODING": "utf-8"}
@@ -118,8 +185,6 @@ class Runtime:
                 process.stdin.close()
                 final = None
                 while line := await process.stdout.readline():
-                    if len(line) > 16 * 1024 * 1024:
-                        raise ProviderError("LOCAL_MODEL_INVALID_RESPONSE", "本地模型输出超限。")
                     message = json.loads(line)
                     if "progress" in message:
                         callback = runtime_progress.get()
@@ -137,26 +202,19 @@ class Runtime:
                 raise ProviderError("LOCAL_MODEL_PROCESS_FAILED", "本地模型进程退出，请检查依赖与资源预算。")
             if not isinstance(result, dict):
                 raise ProviderError("LOCAL_MODEL_INVALID_RESPONSE", "本地模型进程未返回有效结果。")
+            diagnostics.update(result.get("diagnostics", {}))
             if "error_code" in result:
                 raise ProviderError(result["error_code"], result.get("message", "本地推理失败。"))
             attempt.observe(result)
             attempt.completed = True
-            self.diagnostics.append({"model": CATALOG[key].repository, "revision": CATALOG[key].revision,
-                                     **result.get("diagnostics", {})})
-            self.diagnostics = self.diagnostics[-100:]
-            return result["result"]
+            return result
         finally:
-            if ticket in self.waiters:
-                self.waiters.remove(ticket)
             if process is not None and process.returncode is None:
                 process.kill()
                 await process.wait()
             if process is not None and hasattr(process, "close"):
                 await process.close()
-            self.active.pop(ticket, None)
-            self.active_files.pop(ticket, None)
-            if attempt:
-                attempt.persist()
+            attempt.persist()
 
 
 runtime = Runtime()
@@ -188,7 +246,7 @@ class LocalEmbedding:
         config = (self._config or configuration()).model_copy(deep=True)
         token = runtime_context.set(config)
         try:
-            return await runtime.infer(config.embedding_model, "embedding", {"texts": texts}, priority=0)
+            return await runtime.infer(config.embedding_model, "embedding", {"texts": texts}, priority=embedding_priority.get())
         finally:
             runtime_context.reset(token)
 

backend/app/local_models/worker.py

@@ -76,16 +76,25 @@ def voice_embedding(model, audio, device):
         return torch.nn.functional.normalize(vector, dim=0)
 
 
+class CudaInitializationError(RuntimeError):
+    pass
+
+
 def run(request):
     import torch
     import psutil
     config, payload = request["config"], request["payload"]
     torch.set_num_threads(config["cpu_threads"])
     requested = config["device"]
-    device = "cuda:0" if requested == "cuda" and torch.cuda.is_available() else "cpu"
-    if device != "cpu":
-        total = torch.cuda.get_device_properties(0).total_memory
-        torch.cuda.set_per_process_memory_fraction(min(1.0, config["gpu_memory_limit_mb"] * 1024 ** 2 / total))
+    try:
+        device = "cuda:0" if requested == "cuda" and torch.cuda.is_available() else "cpu"
+        if device != "cpu":
+            torch.cuda.init()
+            total = torch.cuda.get_device_properties(0).total_memory
+            torch.cuda.set_per_process_memory_fraction(min(1.0, config["gpu_memory_limit_mb"] * 1024 ** 2 / total))
+    except Exception as exc:
+        raise CudaInitializationError() from exc
+    request["_actual_device"] = device
     process = psutil.Process()
     peak = [0]
     stop = threading.Event()
@@ -102,6 +111,7 @@ def run(request):
     path, operation = request["model_path"], request["operation"]
     try:
         usage = {}
+        audio_seconds = None
         if operation == "embedding":
             from sentence_transformers import SentenceTransformer
             model = SentenceTransformer(path, device=device, local_files_only=True, trust_remote_code=False,
@@ -116,6 +126,7 @@ def run(request):
                 device_map=device, attn_implementation="sdpa", max_inference_batch_size=1, max_new_tokens=512)
             loaded = time.monotonic()
             audio = decode(payload["source"])
+            audio_seconds = len(audio) / 16000
             regions = speech_regions(audio)
             language = {"zh": "Chinese", "en": "English", "ja": "Japanese", "yue": "Cantonese"}.get(payload.get("language"), payload.get("language"))
             segments = []
@@ -154,7 +165,7 @@ def run(request):
             result = {"speakers": speakers}
         else:
             raise ValueError("Unknown inference operation")
-        return {"result": result, "usage": usage, "diagnostics": {"requested_device": requested, "actual_device": device,
+        return {"result": result, "usage": usage, "audio_seconds": audio_seconds, "diagnostics": {"requested_device": requested, "actual_device": device,
                 "fallback_reason": "CUDA_UNAVAILABLE" if requested == "cuda" and device == "cpu" else None,
                 "load_seconds": loaded - started, "inference_seconds": time.monotonic() - loaded,
                 "peak_memory_bytes": max(peak[0], process.memory_info().rss), "operation": operation}}
@@ -170,6 +181,16 @@ if __name__ == "__main__":
             response = run(request)
         except (ImportError, ModuleNotFoundError):
             response = {"error_code": "LOCAL_RUNTIME_DEPENDENCY_MISSING", "message": "本地模型运行依赖不完整，请重新运行安装脚本。"}
-        except Exception:
-            response = {"error_code": "LOCAL_INFERENCE_FAILED", "message": "本地推理失败，请检查媒体格式、模型和设备配置。"}
+        except Exception as exc:
+            # Only device failures allow the host to retry once in a fresh CPU process.
+            import torch
+            cuda_failure = isinstance(exc, CudaInitializationError)
+            cuda_oom = request.get("_actual_device") == "cuda:0" and isinstance(exc, torch.cuda.OutOfMemoryError)
+            if cuda_failure or cuda_oom:
+                response = {"error_code": "LOCAL_CUDA_OOM" if cuda_oom else "LOCAL_CUDA_INIT_FAILED",
+                            "message": "CUDA 运行失败，将释放进程并重试 CPU。"}
+            else:
+                response = {"error_code": "LOCAL_INFERENCE_FAILED", "message": "本地推理失败，请检查媒体格式、模型和设备配置。"}
+    if "error_code" in response:
+        response["diagnostics"] = {"requested_device": request["config"]["device"], "actual_device": request.get("_actual_device", "unknown")}
     sys.stdout.buffer.write((json.dumps(response, ensure_ascii=False, allow_nan=False) + "\n").encode("utf-8"))