From 64f63ff1bdb0b041f4b0c52447e529e267d7b31a Mon Sep 17 00:00:00 2001 From: KiriAky 107 Date: Sat, 5 Sep 2026 01:06:29 +0800 Subject: [PATCH 01/18] =?UTF-8?q?feat(multimodal):=20=E8=A1=A5=E9=BD=90?= =?UTF-8?q?=E9=98=B6=E6=AE=B5F=E8=BF=90=E8=A1=8C=E7=AE=A1=E7=90=86?= =?UTF-8?q?=E4=B8=8E=E6=94=B6=E5=B0=BE=E9=AA=8C=E6=94=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 1 + backend/app/contracts.py | 1 + backend/app/local_model_routes.py | 9 +- backend/app/local_models/manager.py | 14 +- backend/app/local_models/runtime.py | 110 ++++++--- backend/app/local_models/worker.py | 35 ++- backend/app/media_routes.py | 17 +- backend/app/provider_preview_routes.py | 59 ++++- backend/app/providers/routing.py | 17 ++ backend/app/services/media_notes.py | 41 +++- backend/app/services/model_diagnostics.py | 37 +++ backend/app/services/note_service.py | 10 +- backend/app/services/transcription_service.py | 4 + backend/app/services/usage_service.py | 17 +- backend/scripts/install-model-runtime.ps1 | 9 +- backend/tests/test_multimodal_finalization.py | 221 ++++++++++++++++++ docs/contracts/第二阶段接口契约-开发版.md | 14 ++ .../多模态管线与模型运行开发说明.md | 22 ++ docs/development/阶段F收尾验收记录.md | 55 +++++ .../阶段F-Embedding与知识库问题与解决方案.md | 24 ++ frontend/src/features/media/MediaView.vue | 16 +- .../features/settings/LocalModelSettings.vue | 10 +- .../features/settings/ProviderForm.spec.ts | 18 ++ .../src/features/settings/ProviderForm.vue | 36 ++- .../settings/RequestJsonEditor.spec.ts | 13 ++ .../features/settings/RequestJsonEditor.vue | 43 +++- frontend/src/features/settings/UsageCard.vue | 3 +- frontend/src/services/mediaService.spec.ts | 42 ++++ frontend/src/services/mediaService.ts | 24 +- 29 files changed, 838 insertions(+), 84 deletions(-) create mode 100644 backend/app/services/model_diagnostics.py create mode 100644 backend/tests/test_multimodal_finalization.py create mode 100644 docs/development/阶段F收尾验收记录.md create mode 100644 frontend/src/services/mediaService.spec.ts 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..d3cb20c 100644 --- a/backend/app/local_model_routes.py +++ b/backend/app/local_model_routes.py @@ -1,4 +1,6 @@ +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 @@ -7,9 +9,10 @@ router = APIRouter(prefix="/api/local-models", tags=["Local models"]) @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 +37,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/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..ae73314 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(): @@ -75,28 +89,81 @@ class Runtime: return any(target in paths for paths in self.active_files.values()) async def infer(self, key, operation, payload, *, priority=10): - if read_state(key)["status"] != "installed": - raise ProviderError("LOCAL_MODEL_NOT_INSTALLED", "请先在模型配置中下载本地模型。") - if not interpreter().is_file(): - raise ProviderError("LOCAL_RUNTIME_NOT_INSTALLED", "请先运行本地模型 CPU/CUDA 安装脚本。") - config = configuration() + from app.services import model_diagnostics + config = configuration().model_copy(deep=True) self.counter += 1 ticket = (priority, self.counter) self.waiters.append(ticket) - process = None - attempt = None + queued_at = time.monotonic() + reason = None + from app.services.usage_service import usage_context + from uuid import uuid4 + context = dict(usage_context.get() or {}) + context.setdefault("request_id", uuid4().hex) + usage_token = usage_context.set(context) try: - # One resident model at a time prevents overlapping CPU/GPU allocations. while self.active or ticket != min(self.waiters): await asyncio.sleep(0.05) self.waiters.remove(ticket) self.active[ticket] = key self.active_files[ticket] = {str(Path(payload[name]).resolve()) for name in ("source", "reference") if payload.get(name)} - # Deletion may have occurred while this request was queued. - if read_state(key)["status"] != "installed": - raise ProviderError("LOCAL_MODEL_NOT_INSTALLED", "模型文件已被删除。") - from app.services.usage_service import UsageAttempt - attempt = UsageAttempt("local-models", CATALOG[key].repository, "local", operation, source="local") + queue_seconds = time.monotonic() - queued_at + # Keep the reservation while replacing a failed CUDA process with CPU. + for device in (["cuda", "cpu"] if config.device == "cuda" else ["cpu"]): + started = time.monotonic() + diagnostics = dict(model=CATALOG[key].repository, revision=CATALOG[key].revision, + operation=operation, source="local", requested_device=config.device, + attempted_device=device, queue_seconds=queue_seconds, fallback_reason=reason, request_id=context["request_id"]) + try: + result = await self._execute(key, operation, payload, config.model_copy(update={"device": device}), diagnostics) + diagnostics.update(result.get("diagnostics", {})) + diagnostics.update(requested_device=config.device, status="completed") + if reason: + diagnostics["fallback_reason"] = reason + return result["result"] + except asyncio.CancelledError: + diagnostics.update(status="cancelled", error_code="LOCAL_MODEL_CANCELLED") + raise + except ProviderError as exc: + diagnostics.update(status="failed", error_code=exc.code) + if device == "cuda" and exc.code in {"LOCAL_CUDA_INIT_FAILED", "LOCAL_CUDA_OOM"}: + reason = exc.code + callback = runtime_progress.get() + if callback: + callback({"reset": True, "progress": 0}) + continue + raise + except Exception: + diagnostics.update(status="failed", error_code="LOCAL_MODEL_INVALID_RESPONSE") + raise ProviderError("LOCAL_MODEL_INVALID_RESPONSE", "本地模型返回无效数据。") from None + finally: + diagnostics["requested_device"] = config.device + diagnostics["elapsed_seconds"] = time.monotonic() - started + self.diagnostics.append(model_diagnostics.record(**diagnostics)) + self.diagnostics = self.diagnostics[-100:] + except asyncio.CancelledError: + if ticket not in self.active: + model_diagnostics.record(model=CATALOG[key].repository, operation=operation, + source="local", status="cancelled", error_code="LOCAL_QUEUE_CANCELLED", + requested_device=config.device, queue_seconds=time.monotonic() - queued_at) + raise + finally: + if ticket in self.waiters: + self.waiters.remove(ticket) + self.active.pop(ticket, None) + self.active_files.pop(ticket, None) + usage_context.reset(usage_token) + + async def _execute(self, key, operation, payload, config, diagnostics): + if read_state(key)["status"] != "installed": + raise ProviderError("LOCAL_MODEL_NOT_INSTALLED", "请先下载本地模型。") + if not interpreter().is_file(): + raise ProviderError("LOCAL_RUNTIME_NOT_INSTALLED", "请先安装本地模型运行环境。") + from app.services.usage_service import UsageAttempt + attempt = UsageAttempt("local-models", CATALOG[key].repository, "local", operation, source="local") + diagnostics.update(attempt_id=attempt.attempt_id, request_id=attempt.request_id) + process = None + try: env = {**os.environ, "HF_HUB_OFFLINE": "1", "TRANSFORMERS_OFFLINE": "1", "HF_HUB_DISABLE_TELEMETRY": "1", "OMP_NUM_THREADS": str(config.cpu_threads), "PYTHONIOENCODING": "utf-8"} @@ -118,8 +185,6 @@ class Runtime: process.stdin.close() final = None while line := await process.stdout.readline(): - if len(line) > 16 * 1024 * 1024: - raise ProviderError("LOCAL_MODEL_INVALID_RESPONSE", "本地模型输出超限。") message = json.loads(line) if "progress" in message: callback = runtime_progress.get() @@ -137,26 +202,19 @@ class Runtime: raise ProviderError("LOCAL_MODEL_PROCESS_FAILED", "本地模型进程退出，请检查依赖与资源预算。") if not isinstance(result, dict): raise ProviderError("LOCAL_MODEL_INVALID_RESPONSE", "本地模型进程未返回有效结果。") + diagnostics.update(result.get("diagnostics", {})) if "error_code" in result: raise ProviderError(result["error_code"], result.get("message", "本地推理失败。")) attempt.observe(result) attempt.completed = True - self.diagnostics.append({"model": CATALOG[key].repository, "revision": CATALOG[key].revision, - **result.get("diagnostics", {})}) - self.diagnostics = self.diagnostics[-100:] - return result["result"] + return result finally: - if ticket in self.waiters: - self.waiters.remove(ticket) if process is not None and process.returncode is None: process.kill() await process.wait() if process is not None and hasattr(process, "close"): await process.close() - self.active.pop(ticket, None) - self.active_files.pop(ticket, None) - if attempt: - attempt.persist() + attempt.persist() runtime = Runtime() @@ -188,7 +246,7 @@ class LocalEmbedding: config = (self._config or configuration()).model_copy(deep=True) token = runtime_context.set(config) try: - return await runtime.infer(config.embedding_model, "embedding", {"texts": texts}, priority=0) + return await runtime.infer(config.embedding_model, "embedding", {"texts": texts}, priority=embedding_priority.get()) finally: runtime_context.reset(token) 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/media_routes.py b/backend/app/media_routes.py index 57582e1..aff5917 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,15 @@ 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) + if 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 失败后回退到本地模型。' }}

术语校对

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

- - -
+ +
{{ t('服务器配置', 'Server configuration') }}
+ + + +
- -
+ +
diff --git a/frontend/src/features/media/MediaView.vue b/frontend/src/features/media/MediaView.vue index 11325d7..93c67ed 100644 --- a/frontend/src/features/media/MediaView.vue +++ b/frontend/src/features/media/MediaView.vue @@ -2,6 +2,7 @@ import { computed, onMounted, onUnmounted, ref } from 'vue' import { useRoute } from 'vue-router' import { mediaService, createMediaSubmission, type MediaJob } from '@/services/mediaService' +import { localeTag, t } from '@/i18n' const route = useRoute() const submission = createMediaSubmission() @@ -11,6 +12,7 @@ const selected = ref(null) const file = ref(null) const reference = ref(null) const matchResult = ref('') +const terminologyPlaceholder = computed(() => t('{"错误术语": "正确术语"}', '{"incorrect term": "correct term"}')) const localOnly = ref(false) const diarization = ref(true) const terminology = ref('') @@ -18,17 +20,22 @@ const busy = ref(false) const error = ref('') const notice = ref('') const dirty = ref(false) -const title = ref('课堂转写') +const title = ref(t('课堂转写', 'Class transcript')) const player = ref(null) const position = ref(0) const speed = ref(1) const history = ref([]) let timer: ReturnType | undefined let stopped = false -const labels = {queued: '排队中', running: '转写中', processing: '处理中', completed: '已完成', failed: '失败', cancelled: '已取消'} +const labels = computed(() => ({queued: t('排队中', 'Queued'), running: t('转写中', 'Transcribing'), processing: t('处理中', 'Processing'), completed: t('已完成', 'Completed'), failed: t('失败', 'Failed'), cancelled: t('已取消', 'Cancelled')})) const speakers = computed(() => [...new Set(selected.value?.segments.map(s => s.speaker).filter((s): s is string => !!s) || [])]) const active = (job: MediaJob) => ['queued', 'running', 'processing'].includes(job.status) const stamp = (seconds: number) => `${Math.floor(seconds / 60).toString().padStart(2, '0')}:${Math.floor(seconds % 60).toString().padStart(2, '0')}` +const warningLabel = (warning: string) => ({ + DIARIZATION_UNAVAILABLE: t('当前无法分离说话人', 'Speaker identification is unavailable'), + WORD_TIMESTAMPS_UNAVAILABLE: t('未提供逐字时间戳', 'Word-level timestamps are unavailable'), + DIARIZATION_SEGMENT_LEVEL: t('说话人按音频段估计，同段多人或重叠发言需人工校对', 'Speakers are estimated per segment; multiple or overlapping speakers require manual correction'), +} as Record)[warning] || warning async function refresh() { try { @@ -38,7 +45,7 @@ async function refresh() { if (!stopped) timer = setTimeout(refresh, 2000) } async function choose(job: MediaJob) { - if (dirty.value && !window.confirm('当前校对尚未保存，切换后放弃修改？')) return + if (dirty.value && !window.confirm(t('当前校对尚未保存，切换后放弃修改？', 'The current corrections are unsaved. Discard them and switch?'))) return selected.value = JSON.parse(JSON.stringify(job)); dirty.value = false; history.value = [] } async function action(work: () => Promise) { @@ -52,7 +59,7 @@ async function submit() { let terms = {} if (terminology.value.trim()) { 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 对象，值为替换后的文本。') + if (!terms || typeof terms !== 'object' || Array.isArray(terms) || Object.values(terms).some(v => typeof v !== 'string')) throw new Error(t('术语表需要 JSON 对象，值为替换后的文本。', 'The terminology map must be a JSON object whose values are replacement text.')) } selected.value = await submission.submit(file.value!, {local_only: localOnly.value, diarization: diarization.value, terminology: terms}) @@ -65,10 +72,10 @@ async function purge() { if (!selected.value) return await action(async () => { const impact = await mediaService.impact(selected.value!.attachment_id) - if (!window.confirm(`${impact.message}\n将保留 ${impact.retained_note_ids.length} 篇已保存笔记。确定清理？`)) return + if (!window.confirm(`${impact.message}\n${t('将保留', 'Will retain')} ${impact.retained_note_ids.length} ${t('篇已保存笔记。确定清理？', 'saved notes. Continue cleanup?')}`)) return await mediaService.purge(selected.value!.attachment_id) selected.value = await mediaService.get(selected.value!.job_id) - dirty.value = false; history.value = []; notice.value = '附件与转写内容已清理' + dirty.value = false; history.value = []; notice.value = t('附件与转写内容已清理', 'Attachment and transcript content were removed') }) } async function compareSpeaker() { @@ -79,10 +86,10 @@ async function compareSpeaker() { const sample = await mediaService.upload(file.value!); temporary.push(sample.attachment_id) const known = await mediaService.upload(reference.value!); temporary.push(known.attachment_id) const result = await mediaService.match(sample.attachment_id, known.attachment_id, localOnly.value) - matchResult.value = `相似度 ${result.score.toFixed(3)} · ${result.source === 'local' ? '本地模型' : 'API'}${result.fallback_reason ? ` · 回退：${result.fallback_reason}` : ''}` + matchResult.value = `${t('相似度', 'Similarity')} ${result.score.toFixed(3)} · ${result.source === 'local' ? t('本地模型', 'Local model') : 'API'}${result.fallback_reason ? ` · ${t('回退：', 'Fallback: ')}${result.fallback_reason}` : ''}` } finally { const cleanup = await Promise.allSettled(temporary.map(id => mediaService.purge(id))) - if (cleanup.some(result => result.status === 'rejected')) notice.value = '部分临时参考附件清理失败，请检查后端连接。' + if (cleanup.some(result => result.status === 'rejected')) notice.value = t('部分临时参考附件清理失败，请检查后端连接。', 'Some temporary reference files could not be removed. Check the backend connection.') } }) } @@ -98,52 +105,52 @@ onUnmounted(() => { stopped = true; clearTimeout(timer) })