diff --git a/.gitignore b/.gitignore index f83769f..5e1932b 100644 --- a/.gitignore +++ b/.gitignore @@ -7,6 +7,7 @@ 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 10083ee..7d76dd8 100644 --- a/backend/app/contracts.py +++ b/backend/app/contracts.py @@ -1050,6 +1050,7 @@ 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 90f090e..fd79d8c 100644 --- a/backend/app/local_model_routes.py +++ b/backend/app/local_model_routes.py @@ -1,15 +1,30 @@ +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(): - return {**manager.describe(), "runtime_installed": interpreter().is_file(), "config": configuration(), + 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(), "active_models": list(runtime.active.values()), "queued_requests": len(runtime.waiters), - "last_inference": runtime.diagnostics[-1] if runtime.diagnostics else None} + "last_inference": diagnostics[-1] if diagnostics else None} @router.put("/config") @@ -34,5 +49,5 @@ async def delete(key: str): @router.get("/diagnostics") async def diagnostics(): - return {"items": runtime.diagnostics, "config": configuration(), "scope": "current_process", + return {"items": await asyncio.to_thread(model_diagnostics.recent), "config": configuration(), "scope": "application_last_200_attempts", "contains": "model_revision_device_timing_resources_only"} diff --git a/backend/app/local_models/components.py b/backend/app/local_models/components.py new file mode 100644 index 0000000..c908620 --- /dev/null +++ b/backend/app/local_models/components.py @@ -0,0 +1,111 @@ +"""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 215d5f2..a106a87 100644 --- a/backend/app/local_models/manager.py +++ b/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): diff --git a/backend/app/local_models/runtime.py b/backend/app/local_models/runtime.py index f31e6ed..ed86af0 100644 --- a/backend/app/local_models/runtime.py +++ b/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(): @@ -55,7 +69,11 @@ def configure(request): return request -def interpreter(): +def interpreter(config=None): + from app.local_models import components + requested_device = (config or configuration()).device + if not os.getenv("APP_MODEL_PYTHON") and requested_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")))) @@ -75,32 +93,86 @@ 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", "请先下载本地模型。") + executable = interpreter(config) + if not executable.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"} - args = (str(interpreter()), str(Path(__file__).with_name("worker.py"))) + args = (str(executable), str(Path(__file__).with_name("worker.py"))) options = {"env": env, "limit": 16 * 1024 * 1024, **({"creationflags": 0x08000000} if os.name == "nt" else {})} try: @@ -118,8 +190,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 +207,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 +251,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) diff --git a/backend/app/local_models/worker.py b/backend/app/local_models/worker.py index 98a652f..e3b5d50 100644 --- a/backend/app/local_models/worker.py +++ b/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")) diff --git a/backend/app/main.py b/backend/app/main.py index f7b97c6..d307982 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -26,6 +26,8 @@ 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 57582e1..221ffb1 100644 --- a/backend/app/media_routes.py +++ b/backend/app/media_routes.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio import json +import hashlib from contextlib import closing from pathlib import Path from uuid import uuid4 @@ -22,14 +23,17 @@ 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)): +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_-]+$")): suffix = Path(filename).suffix.lower() if suffix not in MEDIA_SUFFIXES: raise ApiError(422, "UNSUPPORTED_MEDIA", "Unsupported attachment extension.") - attachment_id = f"media_{uuid4().hex}{suffix}" + identity = hashlib.sha256(idempotency_key.encode()).hexdigest() if idempotency_key else uuid4().hex + attachment_id = f"media_{identity}{suffix}" destination = attachment_path(attachment_id) destination.parent.mkdir(parents=True, exist_ok=True) - temporary = destination.with_suffix(destination.suffix + ".upload") + temporary = destination.with_suffix(destination.suffix + f".{uuid4().hex}.upload") + digest = hashlib.sha256() size = 0 try: with temporary.open("xb") as stream: @@ -37,10 +41,40 @@ 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.") - temporary.replace(destination) + content_hash = digest.hexdigest() + if idempotency_key: + with closing(connect()) as conn: + conn.execute("CREATE TABLE IF NOT EXISTS media_upload_idempotency (idempotency_key TEXT PRIMARY KEY, attachment_id TEXT NOT NULL, filename TEXT NOT NULL, content_hash TEXT NOT NULL)") + conn.execute("BEGIN IMMEDIATE") + try: + row = conn.execute("SELECT attachment_id,filename,content_hash FROM media_upload_idempotency WHERE idempotency_key=?", (idempotency_key,)).fetchone() + if row: + if row["filename"] != Path(filename).name or row["content_hash"] != content_hash: + raise ApiError(409, "IDEMPOTENCY_CONFLICT", "同一上传标识不能用于不同附件。") + existing = attachment_path(row["attachment_id"]) + if not existing.is_file() or hashlib.sha256(existing.read_bytes()).hexdigest() != content_hash: + raise ApiError(409, "IDEMPOTENCY_EXPIRED", "该上传标识对应的附件已不存在，请开始一次新提交。") + attachment_id = row["attachment_id"] + else: + if destination.exists() and hashlib.sha256(destination.read_bytes()).hexdigest() != content_hash: + raise ApiError(409, "IDEMPOTENCY_CONFLICT", "同一上传标识不能用于不同附件。") + if not destination.exists(): + temporary.replace(destination) + conn.execute("INSERT INTO media_upload_idempotency VALUES (?,?,?,?)", + (idempotency_key, attachment_id, Path(filename).name, content_hash)) + conn.execute("COMMIT") + except BaseException: + conn.execute("ROLLBACK") + raise + elif destination.exists(): + if hashlib.sha256(destination.read_bytes()).digest() != digest.digest(): + raise ApiError(409, "IDEMPOTENCY_CONFLICT", "同一上传标识不能用于不同附件。") + else: + 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 5b24afa..f6f33f5 100644 --- a/backend/app/provider_preview_routes.py +++ b/backend/app/provider_preview_routes.py @@ -1,12 +1,67 @@ from fastapi import APIRouter -from pydantic import BaseModel +from pydantic import BaseModel, Field from app.contracts import ProviderCreateRequest, ProviderConfig, ModelRequest, Message, MessageRole from app.providers.factory import ProviderFactory -from app.request_overrides import apply_overrides +from app.request_overrides import RequestOverride, 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 dcd009e..d4630fe 100644 --- a/backend/app/providers/routing.py +++ b/backend/app/providers/routing.py @@ -6,6 +6,8 @@ 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 @@ -179,6 +181,7 @@ 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: @@ -202,6 +205,11 @@ 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 @@ -255,6 +263,9 @@ 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: @@ -314,6 +325,9 @@ 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): @@ -346,6 +360,9 @@ 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 e2ec13e..cd3a04d 100644 --- a/backend/app/services/media_notes.py +++ b/backend/app/services/media_notes.py @@ -19,8 +19,10 @@ 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_dump_json().encode()).hexdigest() + options_hash = hashlib.sha256(options.model_copy(update={"update_existing": False}).model_dump_json(exclude={"update_existing"}).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: @@ -44,15 +46,34 @@ 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] - 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 + 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) 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 new file mode 100644 index 0000000..3accbaf --- /dev/null +++ b/backend/app/services/model_diagnostics.py @@ -0,0 +1,37 @@ +"""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 a4f4eeb..efd9931 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 +from app.local_models.runtime import LocalEmbedding, background_embeddings from app.retrieval import routed_vectors from app.retrieval.vectorstore import SqliteVecStore, VectorRecord from app.services.coordination import serialized_vault_mutation @@ -77,6 +77,7 @@ 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] @@ -180,13 +181,18 @@ 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 + note_id: str, *, title: str | None = None, markdown: str | None = None, tags: list[str] | None = None, expected_content_hash: 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 6a816f2..aef3d7c 100644 --- a/backend/app/services/transcription_service.py +++ b/backend/app/services/transcription_service.py @@ -134,6 +134,10 @@ 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 097e209..fe13413 100644 --- a/backend/app/services/usage_service.py +++ b/backend/app/services/usage_service.py @@ -3,6 +3,7 @@ from __future__ import annotations import json import logging +import math from contextlib import closing from contextvars import ContextVar from datetime import datetime, timezone @@ -53,6 +54,7 @@ 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 @@ -61,6 +63,9 @@ 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": @@ -87,7 +92,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(input_tokens=inputs, output_tokens=outputs, + return dict(audio_seconds=self.audio_seconds, 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")) @@ -103,7 +108,7 @@ class UsageAttempt: def aggregate(start, end, provider_id=None, model=None, source=None): - query = "SELECT counters_json,completed 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, type MediaJob } from '@/services/mediaService' +import { mediaService, createMediaSubmission, 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) @@ -40,6 +42,7 @@ 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 } } @@ -51,11 +54,10 @@ 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 对象，值为替换后的文本。') } - 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}) + selected.value = await submission.submit(file.value!, {local_only: localOnly.value, + diarization: diarization.value, terminology: terms}) dirty.value = false - jobs.value.unshift(selected.value) + jobs.value = [selected.value, ...jobs.value.filter(job => job.job_id !== selected.value?.job_id)] }) } function seek(seconds: number) { if (player.value) { player.value.currentTime = seconds; position.value = seconds } } @@ -104,7 +106,7 @@ onUnmounted(() => { stopped = true; clearTimeout(timer) })

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

术语校对

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