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

This reverts commit 64f63ff1bd.
2026-09-05 02:02:26 +08:00
parent f697364aaf
commit 510936431a
29 changed files with 84 additions and 838 deletions
@@ -49,20 +49,8 @@ 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), "disk_bytes": disk_bytes(key)} for key, spec in CATALOG.items()]}
+    return {"items": [{**spec.public(), **read_state(key)} for key, spec in CATALOG.items()]}


 async def download(key):
@@ -4,10 +4,8 @@ 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

@@ -33,18 +31,6 @@ 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():
@@ -89,81 +75,28 @@ class Runtime:
        return any(target in paths for paths in self.active_files.values())

    async def infer(self, key, operation, payload, *, priority=10):
-        from app.services import model_diagnostics
-        config = configuration().model_copy(deep=True)
+        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()
        self.counter += 1
        ticket = (priority, self.counter)
        self.waiters.append(ticket)
-        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)
+        process = None
+        attempt = None
        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)}
-            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:
+            # 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")
            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"}
@@ -185,6 +118,8 @@ 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()
@@ -202,19 +137,26 @@ 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
-            return result
+            self.diagnostics.append({"model": CATALOG[key].repository, "revision": CATALOG[key].revision,
+                                     **result.get("diagnostics", {})})
+            self.diagnostics = self.diagnostics[-100:]
+            return result["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()
-            attempt.persist()
+            self.active.pop(ticket, None)
+            self.active_files.pop(ticket, None)
+            if attempt:
+                attempt.persist()


 runtime = Runtime()
@@ -246,7 +188,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=embedding_priority.get())
+            return await runtime.infer(config.embedding_model, "embedding", {"texts": texts}, priority=0)
        finally:
            runtime_context.reset(token)

@@ -76,25 +76,16 @@ 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"]
-    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
+    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))
    process = psutil.Process()
    peak = [0]
    stop = threading.Event()
@@ -111,7 +102,6 @@ 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,
@@ -126,7 +116,6 @@ 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 = []
@@ -165,7 +154,7 @@ def run(request):
            result = {"speakers": speakers}
        else:
            raise ValueError("Unknown inference operation")
-        return {"result": result, "usage": usage, "audio_seconds": audio_seconds, "diagnostics": {"requested_device": requested, "actual_device": device,
+        return {"result": result, "usage": usage, "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}}
@@ -181,16 +170,6 @@ if __name__ == "__main__":
            response = run(request)
        except (ImportError, ModuleNotFoundError):
            response = {"error_code": "LOCAL_RUNTIME_DEPENDENCY_MISSING", "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")}
+        except Exception:
+            response = {"error_code": "LOCAL_INFERENCE_FAILED", "message": "本地推理失败，请检查媒体格式、模型和设备配置。"}
    sys.stdout.buffer.write((json.dumps(response, ensure_ascii=False, allow_nan=False) + "\n").encode("utf-8"))