diff --git a/.gitignore b/.gitignore index 5e1932b..f83769f 100644 --- a/.gitignore +++ b/.gitignore @@ -7,7 +7,6 @@ frontend/*.tsbuildinfo # Backend backend/.venv/ backend/.venv-models/ -backend/.venv-models-cuda/ backend/data/models/ backend/data/attachments/ backend/.uv-cache/ diff --git a/backend/app/contracts.py b/backend/app/contracts.py index 7d76dd8..10083ee 100644 --- a/backend/app/contracts.py +++ b/backend/app/contracts.py @@ -1050,7 +1050,6 @@ class TranscriptEditRequest(Contract): class TranscriptNoteRequest(Contract): - update_existing: bool = False title: str = Field(min_length=1, max_length=200) folder: str | None = None include_timestamps: bool = True diff --git a/backend/app/local_model_routes.py b/backend/app/local_model_routes.py index fd79d8c..90f090e 100644 --- a/backend/app/local_model_routes.py +++ b/backend/app/local_model_routes.py @@ -1,30 +1,15 @@ -import asyncio from fastapi import APIRouter -from app.services import model_diagnostics from app.local_models import manager from app.local_models.runtime import RuntimeConfig, configuration, configure, interpreter, runtime router = APIRouter(prefix="/api/local-models", tags=["Local models"]) -@router.get("/runtime-components/cuda") -async def cuda_status(): - from app.local_models import components - return await components.status() - - -@router.post("/runtime-components/cuda", status_code=202) -async def install_cuda(): - from app.local_models import components - return await components.install() - - @router.get("") async def list_models(): - items, diagnostics = await asyncio.gather(asyncio.to_thread(manager.describe), asyncio.to_thread(model_diagnostics.recent)) - return {**items, "runtime_installed": interpreter().is_file(), "config": configuration(), + return {**manager.describe(), "runtime_installed": interpreter().is_file(), "config": configuration(), "active_models": list(runtime.active.values()), "queued_requests": len(runtime.waiters), - "last_inference": diagnostics[-1] if diagnostics else None} + "last_inference": runtime.diagnostics[-1] if runtime.diagnostics else None} @router.put("/config") @@ -49,5 +34,5 @@ async def delete(key: str): @router.get("/diagnostics") async def diagnostics(): - return {"items": await asyncio.to_thread(model_diagnostics.recent), "config": configuration(), "scope": "application_last_200_attempts", + return {"items": runtime.diagnostics, "config": configuration(), "scope": "current_process", "contains": "model_revision_device_timing_resources_only"} diff --git a/backend/app/local_models/components.py b/backend/app/local_models/components.py deleted file mode 100644 index c908620..0000000 --- a/backend/app/local_models/components.py +++ /dev/null @@ -1,111 +0,0 @@ -"""User-triggered installation of the fixed optional CUDA runtime on Windows.""" -import asyncio -import json -import os -import shutil -import subprocess - -from app.config import BACKEND_DIR -from app.errors import ApiError -from app.local_models.process import ThreadedProcess - -ROOT = BACKEND_DIR / '.venv-models-cuda' -state = {'status': 'unchecked', 'stage': '', 'cuda_available': None} -task = None - - -def ready(): - return (ROOT / 'ready.json').is_file() and (ROOT / 'Scripts/python.exe').is_file() - - -async def status(): - global task - if state['status'] == 'unchecked': - state.update(status='checking', stage='检查已有 CUDA 组件') - task = asyncio.create_task(run(False)) - return {**state, 'supported': os.name == 'nt', 'custom_interpreter': bool(os.getenv('APP_MODEL_PYTHON'))} - - -async def install(): - global task - from app.local_models.runtime import runtime - if os.name != 'nt': - raise ApiError(422, 'PLATFORM_UNSUPPORTED', '此安装入口目前支持 Windows。') - if task is not None and not task.done(): - return await status() - if runtime.active or runtime.waiters: - raise ApiError(409, 'MODEL_IN_USE', '请等待本地模型任务结束后再安装组件。') - if state['status'] == 'installed': - return await status() - if not shutil.which('uv'): - raise ApiError(422, 'UV_NOT_INSTALLED', '后端未找到 uv，请先安装 uv 并重启后端。') - state.update(status='installing', stage='准备独立 CUDA 环境', error=None) - task = asyncio.create_task(run(True)) - return await status() - - -async def execute(args, timeout): - process = ThreadedProcess(args, env={**os.environ, 'PYTHONIOENCODING': 'utf-8'}, - limit=8192, creationflags=0x08000000 if os.name == 'nt' else 0) - process.stdin.close() - lines = [] - try: - async with asyncio.timeout(timeout): - while line := await process.stdout.readline(): - value = line.decode('utf-8', errors='replace').strip() - stages = {'COMPONENT:torch': '下载并安装 PyTorch CUDA（约 3 GB）', - 'COMPONENT:dependencies': '安装模型依赖', 'COMPONENT:verify': '验证运行组件'} - if value in stages: - state['stage'] = stages[value] - lines = (lines + [value])[-4:] - await process.wait() - if process.returncode: - raise RuntimeError('component command failed') - return lines - finally: - if process.returncode is None: - if os.name == 'nt': - await asyncio.to_thread(subprocess.run, ['taskkill', '/PID', str(process.process.pid), '/T', '/F'], - stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, - creationflags=0x08000000) - else: - process.kill() - await process.wait() - await process.close() - - -async def run(download): - marker = ROOT / 'ready.json' - try: - if download: - marker.unlink(missing_ok=True) - await execute(['powershell.exe', '-NoProfile', '-NonInteractive', '-File', - str(BACKEND_DIR / 'scripts/install-model-runtime.ps1'), '-Device', 'cuda', - '-RuntimeDirectory', str(ROOT), '-QuietProgress'], 7200) - python = ROOT / 'Scripts/python.exe' - if not python.is_file(): - state.update(status='not_installed', stage='尚未安装') - return - result = await execute([str(python), '-c', - 'import json, torch, torchaudio, sentence_transformers, qwen_asr; ' - 'assert torch.version.cuda; ' - 'print(json.dumps({"torch":torch.__version__,"cuda_available":torch.cuda.is_available()}))'], 180) - info = json.loads(result[-1]) - marker.write_text(json.dumps(info), encoding='utf-8') - state.update(status='installed', stage='组件已安装', error=None, **info) - except asyncio.CancelledError: - marker.unlink(missing_ok=True) - state.update(status='interrupted', stage='安装检查已中断，可重试') - raise - except Exception: - marker.unlink(missing_ok=True) - state.update(status='failed', stage='组件安装或验证失败', - error='请检查网络、磁盘空间和 uv；可以重试。CPU 环境不受影响。') - - -async def shutdown(): - if task is not None and not task.done(): - task.cancel() - await asyncio.gather(task, return_exceptions=True) - if state['status'] in {'checking', 'interrupted'}: - state['status'] = 'unchecked' diff --git a/backend/app/local_models/manager.py b/backend/app/local_models/manager.py index a106a87..215d5f2 100644 --- a/backend/app/local_models/manager.py +++ b/backend/app/local_models/manager.py @@ -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): diff --git a/backend/app/local_models/runtime.py b/backend/app/local_models/runtime.py index e362df0..f31e6ed 100644 --- a/backend/app/local_models/runtime.py +++ b/backend/app/local_models/runtime.py @@ -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(): @@ -70,9 +56,6 @@ def configure(request): def interpreter(): - from app.local_models import components - if not os.getenv("APP_MODEL_PYTHON") and configuration().device == "cuda" and components.ready(): - return components.ROOT / "Scripts/python.exe" return Path(os.getenv("APP_MODEL_PYTHON", str(BACKEND_DIR / ".venv-models" / ("Scripts/python.exe" if os.name == "nt" else "bin/python")))) @@ -92,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"} @@ -188,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() @@ -205,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() @@ -249,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) diff --git a/backend/app/local_models/worker.py b/backend/app/local_models/worker.py index e3b5d50..98a652f 100644 --- a/backend/app/local_models/worker.py +++ b/backend/app/local_models/worker.py @@ -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")) diff --git a/backend/app/main.py b/backend/app/main.py index d307982..f7b97c6 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -26,8 +26,6 @@ async def lifespan(_: FastAPI): yield finally: await transcription_service.shutdown() - from app.local_models import components - await components.shutdown() from app.local_models import manager for _, key in list(manager._downloads): await manager.cancel_download(key) diff --git a/backend/app/media_routes.py b/backend/app/media_routes.py index aff5917..57582e1 100644 --- a/backend/app/media_routes.py +++ b/backend/app/media_routes.py @@ -3,7 +3,6 @@ from __future__ import annotations import asyncio import json -import hashlib from contextlib import closing from pathlib import Path from uuid import uuid4 @@ -23,17 +22,14 @@ MEDIA_SUFFIXES = {".wav", ".mp3", ".flac", ".ogg", ".m4a", ".mp4", ".webm", ".tx @router.post("/attachments", status_code=201) -async def upload_attachment(request: Request, filename: str = Query(min_length=1, max_length=255), - idempotency_key: str | None = Header(None, min_length=16, max_length=100, pattern=r"^[a-zA-Z0-9_-]+$")): +async def upload_attachment(request: Request, filename: str = Query(min_length=1, max_length=255)): suffix = Path(filename).suffix.lower() if suffix not in MEDIA_SUFFIXES: raise ApiError(422, "UNSUPPORTED_MEDIA", "Unsupported attachment extension.") - identity = hashlib.sha256(idempotency_key.encode()).hexdigest() if idempotency_key else uuid4().hex - attachment_id = f"media_{identity}{suffix}" + attachment_id = f"media_{uuid4().hex}{suffix}" destination = attachment_path(attachment_id) destination.parent.mkdir(parents=True, exist_ok=True) - temporary = destination.with_suffix(destination.suffix + f".{uuid4().hex}.upload") - digest = hashlib.sha256() + temporary = destination.with_suffix(destination.suffix + ".upload") size = 0 try: with temporary.open("xb") as stream: @@ -41,15 +37,10 @@ async def upload_attachment(request: Request, filename: str = Query(min_length=1 size += len(chunk) if size > MAX_UPLOAD_BYTES: raise ApiError(413, "ATTACHMENT_TOO_LARGE", "Attachment exceeds 25 MiB.") - digest.update(chunk) stream.write(chunk) if not size: raise ApiError(422, "EMPTY_ATTACHMENT", "Attachment is empty.") - if destination.exists(): - if hashlib.sha256(destination.read_bytes()).digest() != digest.digest(): - raise ApiError(409, "IDEMPOTENCY_CONFLICT", "同一上传标识不能用于不同附件。") - else: - temporary.replace(destination) + temporary.replace(destination) finally: temporary.unlink(missing_ok=True) return {"attachment_id": attachment_id, "filename": Path(filename).name, "size": size} diff --git a/backend/app/provider_preview_routes.py b/backend/app/provider_preview_routes.py index f6f33f5..5b24afa 100644 --- a/backend/app/provider_preview_routes.py +++ b/backend/app/provider_preview_routes.py @@ -1,67 +1,12 @@ from fastapi import APIRouter -from pydantic import BaseModel, Field +from pydantic import BaseModel from app.contracts import ProviderCreateRequest, ProviderConfig, ModelRequest, Message, MessageRole from app.providers.factory import ProviderFactory -from app.request_overrides import RequestOverride, apply_overrides +from app.request_overrides import apply_overrides router = APIRouter(prefix="/api/providers", tags=["Providers"]) -class RulesTransfer(BaseModel): - version: int = Field(default=1, ge=1, le=1) - request_overrides: list[RequestOverride] = Field(max_length=100) - - -@router.post("/request-rules/validate") -async def validate_rules(request: RulesTransfer): - return request - - -class ProbeRequest(BaseModel): - provider: ProviderCreateRequest - stream: bool = True - - -@router.post("/request-probe") -async def probe(request: ProbeRequest): - """Explicit user-triggered inference; no vault context, tools or media uploads.""" - import asyncio - from contextlib import aclosing - from app.container import container - from app.errors import ApiError - from app.providers.base import ProviderError - from app.providers.factory import UnsupportedProviderError - config = ProviderConfig(provider_id="request-probe", **request.provider.model_dump()) - if not config.default_model: - raise ApiError(422, "MODEL_REQUIRED", "请填写要验证的模型 ID。") - try: - adapter = container.provider_factory.build(config) - model_request = ModelRequest(provider_id=config.provider_id, model=config.default_model, - messages=[Message(role=MessageRole.user, content="Reply with OK.")], max_tokens=32) - received = False - async with asyncio.timeout(45): - if request.stream: - async with aclosing(adapter.stream(model_request)) as events: - async for event in events: - if event.event.value in {"TextDelta", "ThinkingDelta"}: - received = received or bool(str(event.data.get("text") or "").strip()) - if event.event.value == "Error": - raise ProviderError("PROVIDER_PROBE_FAILED", "模型返回了错误事件。") - else: - response = await adapter.complete(model_request) - received = bool(response.text and response.text.strip()) - if not received: - raise ApiError(422, "PROVIDER_EMPTY_RESPONSE", "请求未返回有效文本，不能标记验证通过。") - except ProviderError as exc: - raise ApiError(502, exc.code, "推理验证失败，请检查模型、凭据和自定义参数。") from exc - except TimeoutError as exc: - raise ApiError(504, "PROVIDER_TIMEOUT", "推理验证超时。") from exc - except UnsupportedProviderError as exc: - raise ApiError(422, "PROVIDER_TYPE_UNSUPPORTED", "该协议不支持推理验证。") from exc - return {"success": True, "stream": request.stream, "model": config.default_model, - "message": "当前请求配置已通过实际推理验证。"} - - class PreviewRequest(BaseModel): provider: ProviderCreateRequest stream: bool = True diff --git a/backend/app/providers/routing.py b/backend/app/providers/routing.py index d4630fe..dcd009e 100644 --- a/backend/app/providers/routing.py +++ b/backend/app/providers/routing.py @@ -6,8 +6,6 @@ available only for explicitly injected tests and protocol fixtures. from __future__ import annotations import hashlib -import asyncio -import time import json import math from dataclasses import dataclass, field, replace @@ -181,7 +179,6 @@ class ModelRoutingService: payload = apply_overrides(kwargs.get(field, {}), provider.request_overrides, capability) kwargs[field] = payload if field == "json" else {key: json.dumps(value) if isinstance(value, (dict, list, bool)) or value is None else value for key, value in payload.items()} attempt = UsageAttempt(binding.provider_id, binding.model, provider.provider_type.value, capability) - started = time.monotonic() try: async with httpx.AsyncClient(timeout=30, transport=self.transport) as client: async with client.stream("POST", url, headers=headers, **kwargs) as response: @@ -205,11 +202,6 @@ class ModelRoutingService: raise invalid_response() from exc finally: attempt.persist() - from app.services.model_diagnostics import record - task = asyncio.current_task() - status = "completed" if attempt.completed else ("cancelled" if task and task.cancelling() else "failed") - record(model=binding.model, operation=capability, source="api", status=status, - attempt_id=attempt.attempt_id, request_id=attempt.request_id, elapsed_seconds=time.monotonic() - started) if not isinstance(data, dict) or data.get("error"): raise invalid_response() return data, url @@ -263,9 +255,6 @@ class ModelRoutingService: model_id="api-" + hashlib.sha256(identity.encode()).hexdigest()) except ProviderError as exc: reason = exc.code - from app.services.model_diagnostics import record - record(model=binding.model, source="api", status="fallback", error_code=reason, - fallback_reason=reason, operation="model_routing") from app.local_models.runtime import LocalEmbedding local_embedding = self.local_embedding.snapshot() if isinstance(self.local_embedding, LocalEmbedding) else self.local_embedding try: @@ -325,9 +314,6 @@ class ModelRoutingService: return RoutedTranscript(text=text, source="api", segments=segments) except ProviderError as exc: reason = exc.code - from app.services.model_diagnostics import record - record(model=binding.model, source="api", status="fallback", error_code=reason, - fallback_reason=reason, operation="model_routing") try: text = await self.local_speech.transcribe(source, language) if isinstance(text, RoutedTranscript): @@ -360,9 +346,6 @@ class ModelRoutingService: return SpeakerMatchResult(score=score, source="api") except ProviderError as exc: reason = exc.code - from app.services.model_diagnostics import record - record(model=binding.model, source="api", status="fallback", error_code=reason, - fallback_reason=reason, operation="model_routing") try: score = await self.local_speech.match(source, reference) if not finite_number(score) or not 0 <= score <= 1: diff --git a/backend/app/services/media_notes.py b/backend/app/services/media_notes.py index cd3a04d..e2ec13e 100644 --- a/backend/app/services/media_notes.py +++ b/backend/app/services/media_notes.py @@ -19,10 +19,8 @@ async def create_transcript_note(job_id, options): job = require_job(job_id) if job.status != "completed": raise ApiError(409, "TRANSCRIPT_NOT_READY", "Only completed transcripts can become notes.") - options_hash = hashlib.sha256(options.model_copy(update={"update_existing": False}).model_dump_json(exclude={"update_existing"}).encode()).hexdigest() + options_hash = hashlib.sha256(options.model_dump_json().encode()).hexdigest() with closing(connect()) as conn: - conn.execute("CREATE TABLE IF NOT EXISTS media_note_baselines (note_id TEXT PRIMARY KEY, content_hash TEXT NOT NULL)") - previous = conn.execute("SELECT m.note_id,b.content_hash FROM media_notes m LEFT JOIN media_note_baselines b ON b.note_id=m.note_id WHERE m.job_id=? AND m.options_hash=? ORDER BY m.revision DESC LIMIT 1", (job_id, options_hash)).fetchone() row = conn.execute("SELECT note_id FROM media_notes WHERE job_id=? AND revision=? AND options_hash=?", (job_id, job.revision, options_hash)).fetchone() if row: @@ -46,34 +44,15 @@ async def create_transcript_note(job_id, options): if job.local_only: # Persist the indexing policy in the Vault, including later rebuilds. lines = ["---", "embedding_local_only: true", "---", "", *lines] - markdown = "\n".join(lines) - if options.update_existing: - if previous is None or previous[1] is None: - raise ApiError(409, "NOTE_UPDATE_BASELINE_MISSING", "没有可安全更新的导出记录，请先创建新笔记。") - current = await note_service.get_note(previous[0]) - if current is None: - raise ApiError(404, "RESOURCE_NOT_FOUND", "已导出笔记不存在。") - # Recover a successful update if linking failed after the Vault write. - if current.markdown == markdown: - note = current - else: - note = await note_service.update_note(previous[0], markdown=markdown, expected_content_hash=previous[1]) - else: - note = await _create_note(title, markdown, options, marker) + try: + note = await note_service.create_note(title=title, markdown="\n".join(lines), folder=options.folder, tags=["转写"]) + except ApiError as exc: + if exc.code != "RESOURCE_CONFLICT" or "note_id" not in exc.details: + raise + # Recover a crash between successful note creation and linking the job. + note = await note_service.get_note(exc.details["note_id"]) + if note is None or marker not in note.markdown: + raise with closing(connect()) as conn, transaction(conn): conn.execute("INSERT OR IGNORE INTO media_notes VALUES (?,?,?,?)", (job_id, job.revision, options_hash, note.note_id)) - conn.execute("INSERT OR REPLACE INTO media_note_baselines VALUES (?,?)", (note.note_id, hashlib.sha256(markdown.encode()).hexdigest())) return note - - -async def _create_note(title, markdown, options, marker): - try: - note = await note_service.create_note(title=title, markdown=markdown, folder=options.folder, tags=["转写"]) - except ApiError as exc: - if exc.code != "RESOURCE_CONFLICT" or "note_id" not in exc.details: - raise - # Recover a crash between successful note creation and linking the job. - note = await note_service.get_note(exc.details["note_id"]) - if note is None or marker not in note.markdown: - raise - return note diff --git a/backend/app/services/model_diagnostics.py b/backend/app/services/model_diagnostics.py deleted file mode 100644 index 3accbaf..0000000 --- a/backend/app/services/model_diagnostics.py +++ /dev/null @@ -1,37 +0,0 @@ -"""Bounded, durable diagnostics. No payloads, paths, exception text or credentials.""" -import json -import logging -import math -from contextlib import closing -from datetime import datetime, timezone - -from app.database.db import connect, transaction - -TEXT = {"model", "revision", "operation", "source", "requested_device", "actual_device", - "attempted_device", "fallback_reason", "error_code", "status", "request_id", "attempt_id"} -NUMBERS = {"load_seconds", "inference_seconds", "elapsed_seconds", "peak_memory_bytes", "queue_seconds"} - - -def connection(): - conn = connect() - conn.execute("CREATE TABLE IF NOT EXISTS model_diagnostics (id INTEGER PRIMARY KEY AUTOINCREMENT, record_json TEXT NOT NULL)") - return conn - - -def record(**values): - safe = {key: value[:240] for key, value in values.items() if key in TEXT and isinstance(value, str)} - safe.update({key: value for key, value in values.items() - if key in NUMBERS and type(value) in (float, int) and math.isfinite(value) and value >= 0}) - safe["timestamp"] = datetime.now(timezone.utc).isoformat() - try: - with closing(connection()) as conn, transaction(conn): - conn.execute("INSERT INTO model_diagnostics(record_json) VALUES (?)", (json.dumps(safe),)) - conn.execute("DELETE FROM model_diagnostics WHERE id NOT IN (SELECT id FROM model_diagnostics ORDER BY id DESC LIMIT 200)") - except Exception: - logging.getLogger(__name__).warning("Model diagnostic persistence failed") - return safe - - -def recent(): - with closing(connection()) as conn: - return [json.loads(row[0]) for row in conn.execute("SELECT record_json FROM model_diagnostics ORDER BY id")] diff --git a/backend/app/services/note_service.py b/backend/app/services/note_service.py index efd9931..a4f4eeb 100644 --- a/backend/app/services/note_service.py +++ b/backend/app/services/note_service.py @@ -17,7 +17,7 @@ from app.contracts import Note, NoteBlock, NoteSummary from app.database.db import connect, transaction from app.errors import ApiError from app.knowledge.parser import ParsedNote, parse_note -from app.local_models.runtime import LocalEmbedding, background_embeddings +from app.local_models.runtime import LocalEmbedding from app.retrieval import routed_vectors from app.retrieval.vectorstore import SqliteVecStore, VectorRecord from app.services.coordination import serialized_vault_mutation @@ -77,7 +77,6 @@ def _delete_markdown(rel_path: str) -> None: PreparedIndex = tuple[list[list[float]], routed_vectors.RemoteEmbeddings | None] -@background_embeddings async def prepare_note_index(parsed: ParsedNote, *, strict=False) -> PreparedIndex: """Compute vectors before opening a write transaction (including API I/O).""" texts = [block.content for block in parsed.blocks] @@ -181,18 +180,13 @@ async def get_note(note_id: str) -> Note | None: @serialized_vault_mutation async def update_note( - note_id: str, *, title: str | None = None, markdown: str | None = None, tags: list[str] | None = None, expected_content_hash: str | None = None + note_id: str, *, title: str | None = None, markdown: str | None = None, tags: list[str] | None = None ) -> Note: record = repository.get_note_record(note_id) if record is None: raise ApiError(404, "RESOURCE_NOT_FOUND", "note not found", {"note_id": note_id}) old_md = _read_markdown(record.file_path) - if expected_content_hash is not None: - import hashlib - if hashlib.sha256(old_md.encode()).hexdigest() != expected_content_hash: - raise ApiError(409, "NOTE_CONTENT_CONFLICT", "笔记已被编辑，请保留现有内容或导出为新笔记。") - new_md = old_md if markdown is None else markdown # PATCH 语义：tags=None 保持原标签；[] 清空；非空列表替换（区别于 create 的 frontmatter 推导） effective_tags = record.tags if tags is None else tags diff --git a/backend/app/services/transcription_service.py b/backend/app/services/transcription_service.py index aef3d7c..6a816f2 100644 --- a/backend/app/services/transcription_service.py +++ b/backend/app/services/transcription_service.py @@ -134,10 +134,6 @@ async def _execute(job_id, request, routing=None): from app.contracts import TranscriptSegment token = runtime_context.set(RuntimeConfig.model_validate(job.model_snapshot.get("local_runtime", {}))) def progress(message): - if message.get("reset"): - job.segments = []; job.progress = 0 - save(job, "AttemptRestarted") - return job.progress = max(0.0, min(0.99, message["progress"])) job.segments.append(TranscriptSegment.model_validate(message["segment"])) save(job, "SegmentReady") diff --git a/backend/app/services/usage_service.py b/backend/app/services/usage_service.py index fe13413..097e209 100644 --- a/backend/app/services/usage_service.py +++ b/backend/app/services/usage_service.py @@ -3,7 +3,6 @@ from __future__ import annotations import json import logging -import math from contextlib import closing from contextvars import ContextVar from datetime import datetime, timezone @@ -54,7 +53,6 @@ class UsageAttempt: self.capability, self.source = capability, source self.started_at = datetime.now(timezone.utc).isoformat() self.raw = {} - self.audio_seconds = None self.completed = False context = usage_context.get() or {} self.request_id = context.get("request_id") or uuid4().hex @@ -63,9 +61,6 @@ class UsageAttempt: def observe(self, data): if not isinstance(data, dict): return - duration = data.get("audio_seconds", data.get("duration")) - if self.capability in {"transcription", "speaker_matching"} and type(duration) in (int, float) and math.isfinite(duration) and 0 <= duration <= 7200: - self.audio_seconds = max(self.audio_seconds or 0, duration) values = [data.get("usage"), (data.get("message") or {}).get("usage") if isinstance(data.get("message"), dict) else None, (data.get("response") or {}).get("usage") if isinstance(data.get("response"), dict) else None] if self.protocol == "ollama": @@ -92,7 +87,7 @@ class UsageAttempt: miss = inputs - hit if hit is not None and inputs is not None and hit > inputs: hit, miss = None, None - return dict(audio_seconds=self.audio_seconds, input_tokens=inputs, output_tokens=outputs, + return dict(input_tokens=inputs, output_tokens=outputs, total_tokens=inputs + outputs if inputs is not None and outputs is not None else first("total_tokens"), cache_hit_tokens=hit, cache_miss_tokens=miss, cache_write_tokens=write, reasoning_tokens=first("output_tokens_details.reasoning_tokens", "completion_tokens_details.reasoning_tokens")) @@ -108,7 +103,7 @@ class UsageAttempt: def aggregate(start, end, provider_id=None, model=None, source=None): - query = "SELECT counters_json,completed,capability FROM model_usage WHERE started_at>=? AND started_at=? AND started_at import { computed, onMounted, onUnmounted, ref } from 'vue' import { useRoute } from 'vue-router' -import { mediaService, createMediaSubmission, type MediaJob } from '@/services/mediaService' +import { mediaService, type MediaJob } from '@/services/mediaService' const route = useRoute() -const submission = createMediaSubmission() -const updateExisting = ref(false) const jobs = ref([]) const selected = ref(null) const file = ref(null) @@ -42,7 +40,6 @@ async function choose(job: MediaJob) { selected.value = JSON.parse(JSON.stringify(job)); dirty.value = false; history.value = [] } async function action(work: () => Promise) { - if (busy.value) return busy.value = true; error.value = ''; notice.value = '' try { await work() } catch (e) { error.value = (e as Error).message } finally { busy.value = false } } @@ -54,10 +51,11 @@ async function submit() { terms = JSON.parse(terminology.value) if (!terms || typeof terms !== 'object' || Array.isArray(terms) || Object.values(terms).some(v => typeof v !== 'string')) throw new Error('术语表需要 JSON 对象，值为替换后的文本。') } - selected.value = await submission.submit(file.value!, {local_only: localOnly.value, - diarization: diarization.value, terminology: terms}) + const uploaded = await mediaService.upload(file.value!) + selected.value = await mediaService.create({attachment_id: uploaded.attachment_id, local_only: localOnly.value, + diarization: diarization.value, idempotency_key: crypto.randomUUID(), terminology: terms}) dirty.value = false - jobs.value = [selected.value, ...jobs.value.filter(job => job.job_id !== selected.value?.job_id)] + jobs.value.unshift(selected.value) }) } function seek(seconds: number) { if (player.value) { player.value.currentTime = seconds; position.value = seconds } } @@ -106,7 +104,7 @@ onUnmounted(() => { stopped = true; clearTimeout(timer) })

{{ localOnly ? '本次任务不调用远程模型 API，模型需预先下载。' : '若配置了转写 API，将上传所选附件；API 失败后回退到本地模型。' }}

术语校对

在识别完成后替换文本，原始识别结果会保留。