diff --git a/.gitignore b/.gitignore index 84abf25..f83769f 100644 --- a/.gitignore +++ b/.gitignore @@ -6,6 +6,9 @@ frontend/*.tsbuildinfo # Backend backend/.venv/ +backend/.venv-models/ +backend/data/models/ +backend/data/attachments/ backend/.uv-cache/ backend/.pytest_cache/ backend/*.egg-info/ diff --git a/backend/README.md b/backend/README.md index a857f76..9884ac7 100644 --- a/backend/README.md +++ b/backend/README.md @@ -2,7 +2,7 @@ FastAPI + Pydantic 的本地 AI Core / Agent Core。项目使用 uv 管理依赖和虚拟环境。 -当前实现包含 Knowledge/Retrieval、Chat、Agent Runtime、Tool/Permission、Skill/Plugin、stdio MCP Host、Plugin Command/Settings、Provider Adapter、任务、索引和开发阶段凭据加密存储。Provider 支持 Mock、OpenAI Chat/OpenAI-Compatible 与 Ollama；OpenAI Responses、Anthropic Messages、操作系统级 Plugin 沙箱和真实语音模型仍属于后续阶段。 +当前实现包含 Knowledge/Retrieval、Chat、Agent、Tool/Permission、Skill/Plugin、MCP、模型提供商与多模态任务。支持 OpenAI Chat/Compatible、Responses、Anthropic Messages 和 Ollama；真实本地 Embedding、ASR、声纹模型默认 CPU，CUDA 显式选装。操作系统级 Plugin 沙箱仍属于后续阶段。 ```powershell uv sync @@ -23,7 +23,9 @@ uv run uvicorn app.main:app --reload --host 127.0.0.1 --port 8000 uv run pytest ``` -当前基线为 136 项测试通过。Provider API Key 可通过前端设置页写入，也可用 `OPENAI_API_KEY`、`DEEPSEEK_API_KEY` 或 `AINOTE_CREDENTIAL_` 注入；不要把真实密钥写入仓库。`plugin.*` 是 Plugin Settings 的保留凭据命名空间，通用 Provider 凭据接口不能读写。 +阶段 F 后端基线为 472 项测试通过。Provider API Key 可通过前端设置页写入，也可用 `OPENAI_API_KEY`、`DEEPSEEK_API_KEY` 或 `AINOTE_CREDENTIAL_` 注入；不要把真实密钥写入仓库。`plugin.*` 是 Plugin Settings 的保留凭据命名空间，通用 Provider 凭据接口不能读写。 + +本地模型 CPU/CUDA 安装、多模态任务、Token 用量与自定义 JSON 见 [多模态管线与模型运行开发说明](../docs/development/多模态管线与模型运行开发说明.md)。 团队接口清单见 `../docs/contracts/后端接口契约-开发版.md`，机器可读契约以运行时的 `/openapi.json` 为准。 diff --git a/backend/app/agent/runtime.py b/backend/app/agent/runtime.py index e573317..eb7fd03 100644 --- a/backend/app/agent/runtime.py +++ b/backend/app/agent/runtime.py @@ -562,6 +562,7 @@ class AgentRuntime: @staticmethod def _request_metadata(record: RunRecord) -> dict[str, object]: metadata = dict(record.request.metadata) + metadata["run_id"] = record.run.run_id if record.skill_config is not None: metadata["skill_id"] = record.skill_config.skill_id metadata["retrieval"] = record.skill_config.retrieval.model_dump(mode="json") diff --git a/backend/app/benchmarks/service.py b/backend/app/benchmarks/service.py index 2db1de3..f06f3ca 100644 --- a/backend/app/benchmarks/service.py +++ b/backend/app/benchmarks/service.py @@ -116,7 +116,12 @@ async def _validate_index_compatibility(request: RAGRunRequest) -> None: reasons: list[str] = [] if stats["blocks"] == 0: reasons.append("index is empty (no indexed blocks; run /api/index/rebuild first)") - if needs_vector: + from app.local_models.runtime import LocalEmbedding + if needs_vector and isinstance(engine.embedding, LocalEmbedding): + from app.retrieval import routed_vectors + if await routed_vectors.search_remote("索引可用性检查", top_k=1, accept_local=True) is None: + reasons.append("current semantic model space has no complete index") + elif needs_vector: if meta.get("embedding_model") != engine.embedding.model_id: reasons.append( f"embedding model mismatch: index={meta.get('embedding_model')!r}, " diff --git a/backend/app/container.py b/backend/app/container.py index 2f3686d..d5801db 100644 --- a/backend/app/container.py +++ b/backend/app/container.py @@ -88,7 +88,7 @@ def build_container() -> ApplicationContainer: return ApplicationContainer( providers=providers, provider_factory=provider_factory, - model_routing=ModelRoutingService(providers, provider_factory.credentials), + model_routing=_local_model_routing(providers, provider_factory.credentials), credentials=credentials, tools=tools, permissions=permissions, @@ -99,4 +99,9 @@ def build_container() -> ApplicationContainer: ) +def _local_model_routing(providers, credentials): + from app.local_models.runtime import LocalEmbedding, LocalSpeech + return ModelRoutingService(providers, credentials, local_embedding=LocalEmbedding(), local_speech=LocalSpeech()) + + container = build_container() diff --git a/backend/app/contracts.py b/backend/app/contracts.py index 2cebba3..3d68da4 100644 --- a/backend/app/contracts.py +++ b/backend/app/contracts.py @@ -2,7 +2,8 @@ from datetime import datetime from enum import Enum from typing import Annotated, Any, Literal -from pydantic import BaseModel, ConfigDict, Field, SecretStr, field_validator +from pydantic import BaseModel, ConfigDict, Field, SecretStr, field_validator, model_validator +from app.request_overrides import RequestOverride class Contract(BaseModel): @@ -783,6 +784,8 @@ class ProviderConnectionFields(Contract): class ProviderConfig(ProviderConnectionFields): + version: int = Field(default=1, ge=1) + request_overrides: list[RequestOverride] = Field(default_factory=list, max_length=32) provider_id: str provider_type: ProviderType name: str @@ -794,6 +797,7 @@ class ProviderConfig(ProviderConnectionFields): class ProviderCreateRequest(ProviderConnectionFields): + request_overrides: list[RequestOverride] = Field(default_factory=list, max_length=32) provider_type: ProviderType name: str base_url: str | None = None @@ -803,6 +807,8 @@ class ProviderCreateRequest(ProviderConnectionFields): class ProviderUpdateRequest(ProviderConnectionFields): + version: int | None = Field(default=None, ge=1) + request_overrides: list[RequestOverride] | None = Field(default=None, max_length=32) provider_type: ProviderType | None = None name: str | None = None base_url: str | None = None @@ -890,6 +896,7 @@ class EmbeddingResult(Contract): class SpeakerMatchRequest(Contract): attachment_id: str reference_attachment_id: str + local_only: bool = False class SpeakerMatchResult(Contract): @@ -978,18 +985,74 @@ class TranscriptionRequest(Contract): attachment_id: str language: str | None = None diarization: bool = False + local_only: bool = False + word_timestamps: bool = False + idempotency_key: str | None = Field(default=None, min_length=1, max_length=128) + terminology: dict[str, str] = Field(default_factory=dict, max_length=200) + + @field_validator("terminology") + @classmethod + def bound_terminology(cls, value): + if any(not key or len(key) > 200 or len(replacement) > 200 for key, replacement in value.items()): + raise ValueError("术语不能为空，每个术语与替换文本最多 200 字符") + return value + + +class TranscriptSegment(Contract): + segment_id: str + start_time: float = Field(ge=0) + end_time: float = Field(ge=0) + text: str + speaker: str | None = None + language: str | None = None + + @model_validator(mode="after") + def valid_interval(self): + import math + if not math.isfinite(self.start_time) or not math.isfinite(self.end_time) or self.end_time < self.start_time: + raise ValueError("invalid segment time range") + return self class TranscriptionJob(Contract): job_id: str attachment_id: str - status: Literal["queued", "processing", "completed", "failed"] + status: Literal["queued", "processing", "running", "completed", "failed", "cancelled"] text: str | None = None error_code: str | None = None error_message: str | None = None created_at: datetime source: Literal["api", "local", "sidecar"] | None = None fallback_reason: str | None = None + segments: list[TranscriptSegment] = Field(default_factory=list) + original_text: str | None = None + original_segments: list[TranscriptSegment] = Field(default_factory=list) + speaker_names: dict[str, str] = Field(default_factory=dict) + warnings: list[str] = Field(default_factory=list) + progress: float | None = Field(default=None, ge=0, le=1) + revision: int = 1 + started_at: datetime | None = None + updated_at: datetime | None = None + completed_at: datetime | None = None + language: str | None = None + local_only: bool = False + previous_job_id: str | None = None + model_snapshot: dict[str, Any] = Field(default_factory=dict) + corrections: list[dict[str, str]] = Field(default_factory=list) + + +class TranscriptEditRequest(Contract): + revision: int = Field(ge=1) + text: str = Field(max_length=1_000_000) + segments: list[TranscriptSegment] = Field(default_factory=list, max_length=10000) + speaker_names: dict[str, str] = Field(default_factory=dict, max_length=200) + + +class TranscriptNoteRequest(Contract): + title: str = Field(min_length=1, max_length=200) + folder: str | None = None + include_timestamps: bool = True + include_speakers: bool = True class IndexStatus(Contract): diff --git a/backend/app/database/migrations.py b/backend/app/database/migrations.py index 02cdb33..2009ab4 100644 --- a/backend/app/database/migrations.py +++ b/backend/app/database/migrations.py @@ -96,6 +96,30 @@ MIGRATIONS: list[str] = [ CREATE INDEX IF NOT EXISTS idx_agent_events_type ON agent_events(run_id, event, sequence); """, + # v4: durable media jobs, replayable events and revisions. + """ + CREATE TABLE media_jobs ( + job_id TEXT PRIMARY KEY, status TEXT NOT NULL, job_json TEXT NOT NULL, + request_json TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, + idempotency_key TEXT UNIQUE, fingerprint TEXT NOT NULL + ); + CREATE INDEX media_jobs_created ON media_jobs(created_at DESC); + CREATE TABLE media_events ( + job_id TEXT NOT NULL REFERENCES media_jobs(job_id) ON DELETE CASCADE, + sequence INTEGER NOT NULL, event TEXT NOT NULL, data_json TEXT NOT NULL, + timestamp TEXT NOT NULL, PRIMARY KEY(job_id, sequence) + ); + CREATE TABLE media_revisions ( + job_id TEXT NOT NULL REFERENCES media_jobs(job_id) ON DELETE CASCADE, + revision INTEGER NOT NULL, job_json TEXT NOT NULL, + PRIMARY KEY(job_id, revision) + ); + CREATE TABLE media_notes ( + job_id TEXT NOT NULL REFERENCES media_jobs(job_id), revision INTEGER NOT NULL, + options_hash TEXT NOT NULL, note_id TEXT NOT NULL REFERENCES notes(note_id) ON DELETE CASCADE, + PRIMARY KEY(job_id, revision, options_hash) + ); + """, ] diff --git a/backend/app/local_model_routes.py b/backend/app/local_model_routes.py new file mode 100644 index 0000000..90f090e --- /dev/null +++ b/backend/app/local_model_routes.py @@ -0,0 +1,38 @@ +from fastapi import APIRouter +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("") +async def list_models(): + return {**manager.describe(), "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} + + +@router.put("/config") +async def update_config(request: RuntimeConfig): + return configure(request) + + +@router.post("/{key}/download", status_code=202) +async def download(key: str): + return await manager.download(key) + + +@router.post("/{key}/cancel") +async def cancel(key: str): + return await manager.cancel_download(key) + + +@router.delete("/{key}") +async def delete(key: str): + return await manager.delete(key) + + +@router.get("/diagnostics") +async def diagnostics(): + return {"items": runtime.diagnostics, "config": configuration(), "scope": "current_process", + "contains": "model_revision_device_timing_resources_only"} diff --git a/backend/app/local_models/__init__.py b/backend/app/local_models/__init__.py new file mode 100644 index 0000000..81d7b0c --- /dev/null +++ b/backend/app/local_models/__init__.py @@ -0,0 +1 @@ +"""Optional local inference; importing this package does not load model libraries.""" diff --git a/backend/app/local_models/catalog.py b/backend/app/local_models/catalog.py new file mode 100644 index 0000000..e1602c4 --- /dev/null +++ b/backend/app/local_models/catalog.py @@ -0,0 +1,31 @@ +"""Reviewed model identities. Runtime never resolves a moving model revision.""" +from dataclasses import asdict, dataclass + + +@dataclass(frozen=True) +class ModelSpec: + key: str + name: str + capability: str + repository: str + revision: str + license: str + source: str = "huggingface" + dimensions: int | None = None + + def public(self): + return asdict(self) + + +CATALOG = { + spec.key: spec for spec in [ + ModelSpec("bekko", "Bekko Embedding v1 A8M", "embedding", "hotchpotch/bekko-embedding-v1-a8m", + "c721113d59a1d91b447450324f51c4b3332c924a", "MIT", dimensions=384), + ModelSpec("granite", "Granite Embedding 97M Multilingual r2", "embedding", "ibm-granite/granite-embedding-97m-multilingual-r2", + "835ad14087e140460703cf0fae09f97d469d65c2", "Apache-2.0", dimensions=384), + ModelSpec("qwen3-asr", "Qwen3 ASR 0.6B", "transcription", "Qwen/Qwen3-ASR-0.6B", + "5eb144179a02acc5e5ba31e748d22b0cf3e303b0", "Apache-2.0"), + ModelSpec("eres2netv2", "ERes2NetV2 中文声纹", "speaker_matching", "iic/speech_eres2netv2_sv_zh-cn_16k-common", + "3317286545c587ae682dbc166831d9448780eebb", "Apache-2.0", source="modelscope", dimensions=192), + ] +} diff --git a/backend/app/local_models/manager.py b/backend/app/local_models/manager.py new file mode 100644 index 0000000..215d5f2 --- /dev/null +++ b/backend/app/local_models/manager.py @@ -0,0 +1,178 @@ +"""Explicit resumable downloads; inference itself never fetches weights.""" +from __future__ import annotations + +import asyncio +import hashlib +import json +import shutil +from pathlib import Path +from urllib.parse import quote + +import httpx + +from app.config import get_settings +from app.errors import ApiError +from app.local_models.catalog import CATALOG + +_downloads: dict[tuple[str, str], asyncio.Task] = {} + + +def model_path(key: str) -> Path: + if key not in CATALOG: + raise ApiError(404, "MODEL_NOT_FOUND", "Unknown local model.") + return get_settings().data_dir / "models" / key / CATALOG[key].revision + + +def state_path(key): + return model_path(key) / "install-state.json" + + +def read_state(key): + try: + state = json.loads(state_path(key).read_text(encoding="utf-8")) + except (OSError, ValueError): + state = {"status": "not_installed", "downloaded_bytes": 0, "total_bytes": None} + if state["status"] == "downloading" and task_key(key) not in _downloads: + state.update(status="interrupted", error_code="DOWNLOAD_INTERRUPTED") + return state + + +def write_state(key, state): + path = state_path(key) + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(".tmp") + temporary.write_text(json.dumps(state), encoding="utf-8") + temporary.replace(path) + + +def task_key(key): + return str(model_path(key)), key + + +def describe(): + return {"items": [{**spec.public(), **read_state(key)} for key, spec in CATALOG.items()]} + + +async def download(key): + model_path(key) + if task_key(key) not in _downloads and read_state(key)["status"] != "installed": + write_state(key, {"status": "downloading", "downloaded_bytes": 0, "total_bytes": None}) + task = asyncio.create_task(_download(key)) + _downloads[task_key(key)] = task + task.add_done_callback(lambda done: _downloads.pop(task_key(key), None)) + return read_state(key) + + +async def cancel_download(key): + task = _downloads.get(task_key(key)) + if task: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + state = read_state(key) + if state["status"] == "downloading": + state["status"] = "interrupted" + write_state(key, state) + return state + + +async def delete(key): + from app.local_models.runtime import runtime + if runtime.in_use(key): + raise ApiError(409, "MODEL_IN_USE", "Model is serving an active request.") + await cancel_download(key) + path = model_path(key).resolve() + root = (get_settings().data_dir / "models").resolve() + if not path.is_relative_to(root) or path == root: + raise ApiError(400, "INVALID_MODEL_PATH", "Model path escapes storage.") + if path.exists(): + shutil.rmtree(path) + return read_state(key) + + +async def _manifest(client, spec): + if spec.source == "huggingface": + response = await client.get(f"https://huggingface.co/api/models/{spec.repository}/revision/{spec.revision}?blobs=true") + response.raise_for_status() + files = [] + for item in response.json()["siblings"]: + name = item["rfilename"] + if name.startswith(("onnx/", "openvino/", ".")) or not name.endswith((".json", ".txt", ".safetensors", ".md")): + continue + lfs = item.get("lfs") or {} + files.append({"path": name, "size": item["size"], "hash": lfs.get("sha256") or item["blobId"], + "algorithm": "sha256" if lfs else "git-blob", + "url": f"https://huggingface.co/{spec.repository}/resolve/{spec.revision}/{quote(name)}"}) + return files + response = await client.get(f"https://modelscope.cn/api/v1/models/{spec.repository}/repo/files", + params={"Revision": spec.revision, "Recursive": "true"}) + response.raise_for_status() + return [{"path": f["Path"], "size": f["Size"], "hash": f["Sha256"], "algorithm": "sha256", + "url": f"https://modelscope.cn/api/v1/models/{spec.repository}/repo?Revision={spec.revision}&FilePath={quote(f['Path'])}"} + for f in response.json()["Data"]["Files"] + if f["Path"] in {"configuration.json", "pretrained_eres2netv2.ckpt", "README.md"}] + + +def valid_file(path, entry): + if not path.is_file() or path.stat().st_size != entry["size"]: + return False + digest = hashlib.sha256() if entry["algorithm"] == "sha256" else hashlib.sha1() + if entry["algorithm"] == "git-blob": + digest.update(f"blob {entry['size']}\0".encode()) + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() == entry["hash"] + + +async def _download(key): + spec, root = CATALOG[key], model_path(key).resolve() + state = {"status": "downloading", "downloaded_bytes": 0, "total_bytes": None} + try: + async with httpx.AsyncClient(timeout=60, follow_redirects=True) as client: + manifest = await _manifest(client, spec) + if not manifest or not any(f["path"].endswith((".safetensors", ".ckpt")) for f in manifest): + raise ValueError("Missing weights in model manifest") + state["total_bytes"] = sum(f["size"] for f in manifest) + root.mkdir(parents=True, exist_ok=True) + if shutil.disk_usage(root).free < state["total_bytes"] + 100 * 1024 * 1024: + raise ApiError(507, "MODEL_DISK_FULL", "Insufficient free disk space.") + complete = 0 + for entry in manifest: + path = (root / entry["path"]).resolve() + if not path.is_relative_to(root): + raise ValueError("Invalid model manifest path") + path.parent.mkdir(parents=True, exist_ok=True) + if await asyncio.to_thread(valid_file, path, entry): + complete += entry["size"] + continue + partial = path.with_suffix(path.suffix + ".partial") + offset = partial.stat().st_size if partial.exists() else 0 + if offset >= entry["size"]: + partial.unlink() + offset = 0 + async with client.stream("GET", entry["url"], headers={"Range": f"bytes={offset}-"} if offset else {}) as response: + response.raise_for_status() + if offset and response.status_code != 206: + offset = 0 + if response.status_code == 206 and not response.headers.get("content-range", "").startswith(f"bytes {offset}-"): + raise ValueError("Invalid download range") + with partial.open("ab" if offset else "wb") as stream: + async for chunk in response.aiter_bytes(1024 * 1024): + offset += len(chunk) + if offset > entry["size"]: + raise ValueError("Download exceeds manifest size") + stream.write(chunk) + state["downloaded_bytes"] = complete + offset + write_state(key, state) + if not await asyncio.to_thread(valid_file, partial, entry): + partial.unlink(missing_ok=True) + raise ApiError(422, "MODEL_CHECKSUM_FAILED", "Model file checksum did not match.") + partial.replace(path) + complete += entry["size"] + (root / "verified-manifest.json").write_text(json.dumps(manifest), encoding="utf-8") + state.update(status="installed", downloaded_bytes=complete) + except asyncio.CancelledError: + state.update(status="interrupted", error_code="DOWNLOAD_CANCELLED") + except Exception as exc: + state.update(status="failed", error_code=exc.code if isinstance(exc, ApiError) else "MODEL_DOWNLOAD_FAILED") + write_state(key, state) diff --git a/backend/app/local_models/runtime.py b/backend/app/local_models/runtime.py new file mode 100644 index 0000000..80550f0 --- /dev/null +++ b/backend/app/local_models/runtime.py @@ -0,0 +1,198 @@ +"""Bounded, cancellable model subprocesses with CPU as the default device.""" +from __future__ import annotations + +import asyncio +import json +import os +from contextlib import closing +from contextvars import ContextVar +from pathlib import Path +from typing import Literal + +from pydantic import BaseModel, Field + +from app.config import BACKEND_DIR +from app.database.db import connect +from app.errors import ApiError +from app.local_models.catalog import CATALOG +from app.local_models.manager import model_path, read_state +from app.providers.base import ProviderError + + +class RuntimeConfig(BaseModel): + device: Literal["cpu", "cuda"] = "cpu" + cpu_threads: int = Field(default=2, ge=1, le=32) + memory_limit_mb: int = Field(default=8192, ge=1024, le=131072) + gpu_memory_limit_mb: int = Field(default=4096, ge=512, le=65536) + timeout_seconds: int = Field(default=1800, ge=30, le=14400) + embedding_model: Literal["bekko", "granite"] = "bekko" + version: int = Field(default=1, ge=1) + + +runtime_context = ContextVar("runtime_config", default=None) +runtime_progress = ContextVar("runtime_progress", default=None) + + +def configuration(): + if runtime_context.get() is not None: + return runtime_context.get() + with closing(connect()) as conn: + conn.execute("CREATE TABLE IF NOT EXISTS local_runtime_config (id INTEGER PRIMARY KEY CHECK(id=1), config_json TEXT NOT NULL)") + row = conn.execute("SELECT config_json FROM local_runtime_config WHERE id=1").fetchone() + return RuntimeConfig.model_validate_json(row[0]) if row else RuntimeConfig() + + +def configure(request): + from app.database.db import transaction + configuration() + with closing(connect()) as conn, transaction(conn): + row = conn.execute("SELECT config_json FROM local_runtime_config WHERE id=1").fetchone() + previous = RuntimeConfig.model_validate_json(row[0]) if row else RuntimeConfig() + if request.version != previous.version: + raise ApiError(409, "VERSION_CONFLICT", "Local runtime settings changed; reload first.") + request = request.model_copy(update={"version": request.version + 1}) + conn.execute("INSERT OR REPLACE INTO local_runtime_config VALUES (1,?)", (request.model_dump_json(),)) + return request + + +def interpreter(): + return Path(os.getenv("APP_MODEL_PYTHON", str(BACKEND_DIR / ".venv-models" / ("Scripts/python.exe" if os.name == "nt" else "bin/python")))) + + +class Runtime: + def __init__(self): + self.active = {} + self.active_files = {} + self.waiters = [] + self.counter = 0 + self.diagnostics = [] + + def in_use(self, key): + return key in self.active.values() + + def media_in_use(self, path): + target = str(Path(path).resolve()) + 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() + self.counter += 1 + ticket = (priority, self.counter) + self.waiters.append(ticket) + 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)} + # 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"} + process = await asyncio.create_subprocess_exec(str(interpreter()), str(Path(__file__).with_name("worker.py")), + stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL, + env=env, limit=16 * 1024 * 1024, **({"creationflags": 0x08000000} if os.name == "nt" else {})) + request = {"key": key, "operation": operation, "model_path": str(model_path(key).resolve()), + "config": config.model_dump(), "payload": payload} + async def receive(): + process.stdin.write(json.dumps(request).encode()) + await process.stdin.drain() + 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() + if callback: + callback(message) + else: + final = message + await process.wait() + return final + try: + result = await asyncio.wait_for(receive(), config.timeout_seconds) + except TimeoutError as exc: + raise ProviderError("LOCAL_MODEL_TIMEOUT", "本地模型处理超时。") from exc + if process.returncode != 0: + raise ProviderError("LOCAL_MODEL_PROCESS_FAILED", "本地模型进程退出，请检查依赖与资源预算。") + if not isinstance(result, dict): + raise ProviderError("LOCAL_MODEL_INVALID_RESPONSE", "本地模型进程未返回有效结果。") + 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"] + 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() + self.active.pop(ticket, None) + self.active_files.pop(ticket, None) + if attempt: + attempt.persist() + + +runtime = Runtime() + + +class LocalEmbedding: + dim = 384 + + @property + def model_id(self): + spec = CATALOG[configuration().embedding_model] + return f"{spec.repository}@{spec.revision}" + + @property + def version(self): + return CATALOG[configuration().embedding_model].revision + + @property + def available(self): + return read_state(configuration().embedding_model)["status"] == "installed" and interpreter().is_file() + + async def embed_documents(self, texts): + return await runtime.infer(configuration().embedding_model, "embedding", {"texts": texts}, priority=0) + + async def embed_query(self, query): + return (await self.embed_documents([query]))[0] + + +class LocalSpeech: + @property + def available(self): + return self.available_for("transcription") + + def available_for(self, capability): + key = "qwen3-asr" if capability == "transcription" else "eres2netv2" + return read_state(key)["status"] == "installed" and interpreter().is_file() + + async def transcribe(self, source, language): + from app.providers.routing import RoutedTranscript + from app.contracts import TranscriptSegment + result = await runtime.infer("qwen3-asr", "transcription", {"source": str(source.resolve()), "language": language}) + return RoutedTranscript(text=result["text"], source="local", + segments=[TranscriptSegment(**s) for s in result["segments"]]) + + async def match(self, source, reference): + result = await runtime.infer("eres2netv2", "speaker_matching", + {"source": str(source.resolve()), "reference": str(reference.resolve())}, priority=0) + return result["score"] diff --git a/backend/app/local_models/worker.py b/backend/app/local_models/worker.py new file mode 100644 index 0000000..98a652f --- /dev/null +++ b/backend/app/local_models/worker.py @@ -0,0 +1,175 @@ +"""One offline inference process. Heavy libraries stay out of the API process.""" +from __future__ import annotations + +import contextlib +import json +import os +import sys +import threading +import time + + +def decode(path, *, limit_seconds=3600): + import av + import numpy as np + frames = [] + samples = 0 + with av.open(path, options={"protocol_whitelist": "file,pipe"}) as container: + if not container.streams.audio: + raise ValueError("Media has no audio track") + resampler = av.AudioResampler(format="fltp", layout="mono", rate=16000) + for frame in container.decode(audio=0): + for output in resampler.resample(frame): + audio = output.to_ndarray().reshape(-1) + samples += len(audio) + if samples > limit_seconds * 16000: + raise ValueError("Audio exceeds one hour") + frames.append(audio) + for output in resampler.resample(None): + frames.append(output.to_ndarray().reshape(-1)) + if not frames: + raise ValueError("Audio is empty") + audio = np.concatenate(frames).astype(np.float32) + if not np.isfinite(audio).all() or len(audio) < 1600: + raise ValueError("Invalid or too short audio") + return audio + + +def speech_regions(audio): + """Energy-based segmentation, not word alignment; retain original sample offsets.""" + import numpy as np + window = 480 + energies = [float(np.sqrt(np.mean(audio[i:i + window] ** 2))) for i in range(0, len(audio), window)] + threshold = max(0.002, float(np.percentile(energies, 20)) * 2) + active = [i for i, energy in enumerate(energies) if energy >= threshold] + if not active: + return [] + regions, start, previous = [], active[0], active[0] + for index in active[1:]: + if index - previous > 20 or (index - start) * window >= 20 * 16000: + regions.append((max(0, start * window - 2400), min(len(audio), (previous + 1) * window + 2400))) + start = index + previous = index + regions.append((max(0, start * window - 2400), min(len(audio), (previous + 1) * window + 2400))) + return regions + + +def speaker_model(path, device): + import torch + from modelscope.models.audio.sv.ERes2NetV2 import ERes2NetV2 + from pathlib import Path + model = ERes2NetV2(baseWidth=26, scale=2, expansion=2, embed_dim=192) + weights = torch.load(Path(path) / "pretrained_eres2netv2.ckpt", map_location="cpu", weights_only=True) + model.load_state_dict(weights, strict=True) + return model.to(device).eval() + + +def voice_embedding(model, audio, device): + import torch + import torchaudio.compliance.kaldi as kaldi + if len(audio) < 16000: + raise ValueError("Speaker comparison needs at least one second of audio") + features = kaldi.fbank(torch.from_numpy(audio).unsqueeze(0), num_mel_bins=80, sample_frequency=16000) + features -= features.mean(dim=0, keepdim=True) + with torch.inference_mode(): + vector = model(features.unsqueeze(0).to(device)).flatten() + return torch.nn.functional.normalize(vector, dim=0) + + +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)) + process = psutil.Process() + peak = [0] + stop = threading.Event() + + def monitor(): + while not stop.wait(0.2): + used = process.memory_info().rss + peak[0] = max(peak[0], used) + if used > config["memory_limit_mb"] * 1024 ** 2: + os._exit(75) + + threading.Thread(target=monitor, daemon=True).start() + started = time.monotonic() + path, operation = request["model_path"], request["operation"] + try: + usage = {} + if operation == "embedding": + from sentence_transformers import SentenceTransformer + model = SentenceTransformer(path, device=device, local_files_only=True, trust_remote_code=False, + model_kwargs={"attn_implementation": "sdpa"}) + loaded = time.monotonic() + result = model.encode(payload["texts"], batch_size=4, normalize_embeddings=True, show_progress_bar=False).tolist() + # Count the tokenizer's actual encoded input, not characters or words. + usage = {"input_tokens": int(model.tokenize(payload["texts"])["attention_mask"].sum())} + elif operation == "transcription": + from qwen_asr import Qwen3ASRModel + model = Qwen3ASRModel.from_pretrained(path, dtype=torch.float32 if device == "cpu" else torch.float16, + device_map=device, attn_implementation="sdpa", max_inference_batch_size=1, max_new_tokens=512) + loaded = time.monotonic() + audio = decode(payload["source"]) + regions = speech_regions(audio) + language = {"zh": "Chinese", "en": "English", "ja": "Japanese", "yue": "Cantonese"}.get(payload.get("language"), payload.get("language")) + segments = [] + for start, end in regions: + output = model.transcribe(audio=(audio[start:end], 16000), language=language)[0] + if output.text.strip(): + segments.append({"segment_id": f"segment_{len(segments) + 1}", "start_time": start / 16000, + "end_time": end / 16000, "text": output.text, "language": output.language}) + sys.__stdout__.write(json.dumps({"progress": end / len(audio), "segment": segments[-1]}, ensure_ascii=False) + "\n") + sys.__stdout__.flush() + result = {"text": "\n".join(s["text"] for s in segments), "segments": segments} + elif operation == "speaker_matching": + model = speaker_model(path, device) + loaded = time.monotonic() + first = voice_embedding(model, decode(payload["source"]), device) + second = voice_embedding(model, decode(payload["reference"]), device) + # Similarity, not a calibrated identity probability. + result = {"score": max(0.0, min(1.0, float(torch.dot(first, second))))} + elif operation == "diarization": + model = speaker_model(path, device) + loaded = time.monotonic() + audio = decode(payload["source"]) + centroids, speakers = [], [] + for segment in payload["segments"]: + sample = audio[int(segment["start_time"] * 16000):int(segment["end_time"] * 16000)] + if len(sample) < 16000: + speakers.append(None) + continue + vector = voice_embedding(model, sample, device) + similarities = [float(torch.dot(vector, c)) for c in centroids] + best = max(range(len(similarities)), key=similarities.__getitem__) if similarities else None + if best is None or similarities[best] < 0.36: + best = len(centroids) + centroids.append(vector) + speakers.append(f"speaker_{best + 1}") + result = {"speakers": speakers} + else: + raise ValueError("Unknown inference operation") + 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}} + finally: + stop.set() + + +if __name__ == "__main__": + request = json.loads(sys.stdin.buffer.read()) + # Third-party progress/logging must never corrupt the protocol or leak into API errors. + with contextlib.redirect_stdout(sys.stderr): + try: + response = run(request) + except (ImportError, ModuleNotFoundError): + response = {"error_code": "LOCAL_RUNTIME_DEPENDENCY_MISSING", "message": "本地模型运行依赖不完整，请重新运行安装脚本。"} + 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 bfaaf91..f7b97c6 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -9,6 +9,10 @@ from app.config import get_settings from app.container import container from app.errors import ApiError, api_error_handler, http_error_handler, validation_error_handler from app.routes import router as api_router +from app.media_routes import router as media_router +from app.local_model_routes import router as local_model_router +from app.usage_routes import router as usage_router +from app.provider_preview_routes import router as provider_preview_router from app.schemas import HealthResponse, ServiceStatusResponse settings = get_settings() @@ -16,10 +20,17 @@ settings = get_settings() @asynccontextmanager async def lifespan(_: FastAPI): - yield - # 第三方 MCP Server 必须跟随 AI Core 退出，不能遗留孤儿进程。 - container.plugins.shutdown() - container.mcp_servers.shutdown() + from app.services import transcription_service + transcription_service.recover_interrupted() + try: + yield + finally: + await transcription_service.shutdown() + from app.local_models import manager + for _, key in list(manager._downloads): + await manager.cancel_download(key) + container.plugins.shutdown() + container.mcp_servers.shutdown() app = FastAPI( @@ -41,6 +52,10 @@ app.add_exception_handler(ApiError, api_error_handler) app.add_exception_handler(RequestValidationError, validation_error_handler) app.add_exception_handler(StarletteHttpException, http_error_handler) app.include_router(api_router) +app.include_router(media_router) +app.include_router(local_model_router) +app.include_router(usage_router) +app.include_router(provider_preview_router) @app.get("/health", response_model=HealthResponse, tags=["System"]) diff --git a/backend/app/media_routes.py b/backend/app/media_routes.py new file mode 100644 index 0000000..57582e1 --- /dev/null +++ b/backend/app/media_routes.py @@ -0,0 +1,162 @@ +"""Media storage and durable transcription controls.""" +from __future__ import annotations + +import asyncio +import json +from contextlib import closing +from pathlib import Path +from uuid import uuid4 + +from fastapi import APIRouter, Header, Query, Request +from fastapi.responses import FileResponse, StreamingResponse + +from app.contracts import TranscriptEditRequest, TranscriptNoteRequest, TranscriptionJob +from app.database.db import connect, transaction +from app.errors import ApiError +from app.services import transcription_service as jobs +from app.services.attachment_service import attachment_path + +router = APIRouter(prefix="/api/media", tags=["Media"]) +MAX_UPLOAD_BYTES = 25 * 1024 * 1024 +MEDIA_SUFFIXES = {".wav", ".mp3", ".flac", ".ogg", ".m4a", ".mp4", ".webm", ".txt", ".md"} + + +@router.post("/attachments", status_code=201) +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.") + 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 + ".upload") + size = 0 + try: + with temporary.open("xb") as stream: + async for chunk in request.stream(): + size += len(chunk) + if size > MAX_UPLOAD_BYTES: + raise ApiError(413, "ATTACHMENT_TOO_LARGE", "Attachment exceeds 25 MiB.") + stream.write(chunk) + if not size: + raise ApiError(422, "EMPTY_ATTACHMENT", "Attachment is empty.") + temporary.replace(destination) + finally: + temporary.unlink(missing_ok=True) + return {"attachment_id": attachment_id, "filename": Path(filename).name, "size": size} + + +@router.get("/attachments/{attachment_id}") +async def download_attachment(attachment_id: str): + path = attachment_path(attachment_id) + if not path.is_file(): + raise ApiError(404, "ATTACHMENT_NOT_FOUND", "Attachment was not found.") + return FileResponse(path, headers={"X-Content-Type-Options": "nosniff"}) + + +@router.get("/transcriptions") +async def list_jobs(status: str | None = None, limit: int = Query(50, ge=1, le=200), offset: int = Query(0, ge=0)): + if status is not None and status not in jobs.TERMINAL | {"queued", "running", "processing"}: + raise ApiError(422, "INVALID_STATUS", "Unknown transcription status.") + return jobs.list_transcriptions(status, limit, offset) + + +@router.post("/transcriptions/{job_id}/cancel", response_model=TranscriptionJob) +async def cancel_job(job_id: str): + return await jobs.cancel(job_id) + + +@router.post("/transcriptions/{job_id}/retry", response_model=TranscriptionJob, status_code=202) +async def retry_job(job_id: str): + return await jobs.retry(job_id) + + +@router.patch("/transcriptions/{job_id}", response_model=TranscriptionJob) +async def edit_job(job_id: str, request: TranscriptEditRequest): + return jobs.edit(job_id, request) + + +@router.get("/transcriptions/{job_id}/revisions") +async def revisions(job_id: str): + current = jobs.require_job(job_id) + with closing(connect()) as conn: + rows = conn.execute("SELECT job_json FROM media_revisions WHERE job_id=? ORDER BY revision", (job_id,)).fetchall() + return {"items": [TranscriptionJob.model_validate_json(row[0]) for row in rows] + [current]} + + +@router.get("/transcriptions/{job_id}/events") +async def stream_events(job_id: str, request: Request, after: int = Query(-1, ge=-1), + last_event_id: str | None = Header(None)): + jobs.require_job(job_id) + if last_event_id is not None: + try: + after = max(after, int(last_event_id)) + except ValueError as exc: + raise ApiError(422, "INVALID_EVENT_CURSOR", "Last-Event-ID must be an integer.") from exc + + async def stream(): + cursor = after + idle = 0 + while not await request.is_disconnected(): + batch = jobs.events(job_id, cursor) + for event in batch: + cursor = event["sequence"] + yield f"id: {cursor}\nevent: {event['event']}\ndata: {json.dumps(event, ensure_ascii=False)}\n\n" + if len(batch) == 200: + continue + if jobs.require_job(job_id).status in jobs.TERMINAL: + # Re-read once: completion may have been committed after this batch was read. + if jobs.events(job_id, cursor): + continue + return + idle += 1 + if idle % 30 == 0: + yield ": keepalive\n\n" + await asyncio.sleep(0.5) + + return StreamingResponse(stream(), media_type="text/event-stream", + headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}) + + +@router.post("/transcriptions/{job_id}/notes", status_code=201) +async def create_note(job_id: str, request: TranscriptNoteRequest): + from app.services.media_notes import create_transcript_note + return await create_transcript_note(job_id, request) + + +@router.get("/attachments/{attachment_id}/cleanup-impact") +async def cleanup_impact(attachment_id: str): + attachment_path(attachment_id) + with closing(connect()) as conn: + records = conn.execute("SELECT job_json FROM media_jobs").fetchall() + affected = [TranscriptionJob.model_validate_json(row[0]) for row in records] + affected = [job for job in affected if job.attachment_id == attachment_id] + note_ids = [] + for job in affected: + note_ids.extend(row[0] for row in conn.execute("SELECT note_id FROM media_notes WHERE job_id=?", (job.job_id,))) + return {"job_ids": [job.job_id for job in affected], "retained_note_ids": sorted(set(note_ids)), + "message": "清理原附件、转写正文、修订和术语记录；已保存笔记保留，音频链接将失效。"} + + +@router.delete("/attachments/{attachment_id}") +async def cleanup_attachment(attachment_id: str): + from app.local_models.runtime import runtime + impact = await cleanup_impact(attachment_id) + affected = [jobs.require_job(job_id) for job_id in impact["job_ids"]] + if runtime.media_in_use(attachment_path(attachment_id)) or any(job.status not in jobs.TERMINAL for job in affected): + raise ApiError(409, "MEDIA_IN_USE", "Wait for media processing to finish before cleanup.") + for path in (attachment_path(attachment_id), attachment_path(f"{attachment_id}.txt")): + path.unlink(missing_ok=True) + with closing(connect()) as conn, transaction(conn): + for job in affected: + job.text = job.original_text = None + job.segments = []; job.original_segments = []; job.speaker_names = {}; job.corrections = [] + job.model_snapshot = {} + job.status = "cancelled"; job.error_code = "MEDIA_PURGED"; job.error_message = "附件与转写内容已清理。" + job.updated_at = jobs.now() + conn.execute("UPDATE media_jobs SET job_json=?,status=?,request_json='{}' WHERE job_id=?", + (job.model_dump_json(), job.status, job.job_id)) + conn.execute("DELETE FROM media_revisions WHERE job_id=?", (job.job_id,)) + conn.execute("DELETE FROM media_events WHERE job_id=?", (job.job_id,)) + jobs._event(conn, job, "Purged") + return impact diff --git a/backend/app/provider_preview_routes.py b/backend/app/provider_preview_routes.py new file mode 100644 index 0000000..5b24afa --- /dev/null +++ b/backend/app/provider_preview_routes.py @@ -0,0 +1,43 @@ +from fastapi import APIRouter +from pydantic import BaseModel +from app.contracts import ProviderCreateRequest, ProviderConfig, ModelRequest, Message, MessageRole +from app.providers.factory import ProviderFactory +from app.request_overrides import apply_overrides + +router = APIRouter(prefix="/api/providers", tags=["Providers"]) + + +class PreviewRequest(BaseModel): + provider: ProviderCreateRequest + stream: bool = True + capability: str = "chat" + + +@router.post("/request-preview") +async def preview(request: PreviewRequest): + class NoCredentials: + def resolve(self, key): + return None + config = ProviderConfig(provider_id="preview", **request.provider.model_dump()) + if request.capability != "chat": + from app.errors import ApiError + if request.capability not in {"embedding", "transcription", "speaker_matching"}: + raise ApiError(422, "INVALID_CAPABILITY", "Unknown capability.") + payload = {"model": config.default_model or "<模型 ID>"} + payload["input" if request.capability == "embedding" else "file"] = "<运行时输入，不包含正文或文件>" + if request.capability == "speaker_matching": + payload["reference_file"] = "<声纹参考附件>" + else: + from app.providers.factory import UnsupportedProviderError + from app.errors import ApiError + try: + adapter = ProviderFactory(NoCredentials()).build(config) + except UnsupportedProviderError as exc: + raise ApiError(422, "PROVIDER_TYPE_UNSUPPORTED", "该协议不支持请求预览。") from exc + model_request = ModelRequest(provider_id="preview", model=config.default_model or "<模型 ID>", + messages=[Message(role=MessageRole.user, content="<运行时消息，已隐藏>")]) + build = getattr(adapter, "_payload", None) or adapter._chat_payload + payload = build(model_request, stream=request.stream) + return {"body": apply_overrides(payload, config.request_overrides, request.capability, + stream=request.stream if request.capability == "chat" else False), + "contains_credentials": False, "execution": "preview_only"} diff --git a/backend/app/providers/factory.py b/backend/app/providers/factory.py index bab46c8..39b1316 100644 --- a/backend/app/providers/factory.py +++ b/backend/app/providers/factory.py @@ -16,6 +16,30 @@ class ProviderFactory: self.credentials = ProviderCredentialResolver(credentials) def build(self, config: ProviderConfig) -> ModelProvider: + adapter = self._build(config) + adapter.provider_config = config.model_copy(deep=True) + from app.services.usage_service import usage_context + from contextlib import aclosing + from uuid import uuid4 + complete, stream = adapter.complete, adapter.stream + async def complete_with_trace(request): + token = usage_context.set({"request_id": uuid4().hex, "run_id": request.metadata.get("run_id")}) + try: + return await complete(request) + finally: + usage_context.reset(token) + async def stream_with_trace(request): + token = usage_context.set({"request_id": uuid4().hex, "run_id": request.metadata.get("run_id")}) + try: + async with aclosing(stream(request)) as events: + async for event in events: + yield event + finally: + usage_context.reset(token) + adapter.complete, adapter.stream = complete_with_trace, stream_with_trace + return adapter + + def _build(self, config: ProviderConfig) -> ModelProvider: if config.provider_type == ProviderType.openai_responses: from app.providers.openai_responses import OpenAIResponsesProvider return OpenAIResponsesProvider( diff --git a/backend/app/providers/http_base.py b/backend/app/providers/http_base.py index db288ff..6e19037 100644 --- a/backend/app/providers/http_base.py +++ b/backend/app/providers/http_base.py @@ -253,6 +253,18 @@ class HTTPProviderMixin: stream_path = "/chat/completions" stream_format = "sse" + def _custom_payload(self, payload): + from app.request_overrides import apply_overrides + config = getattr(self, "provider_config", None) + return apply_overrides(payload, config.request_overrides, "chat", stream=bool(payload.get("stream"))) if config else payload + + def _usage_attempt(self, payload): + from app.services.usage_service import UsageAttempt + config = getattr(self, "provider_config", None) + protocol = config.provider_type.value if config else "openai_compatible" + return UsageAttempt(config.provider_id if config else "unregistered", str(payload.get("model", "")), protocol, + source="local" if protocol == "ollama" else "api") + def _headers(self) -> dict[str, str]: return {"Content-Type": "application/json"} @@ -268,11 +280,18 @@ class HTTPProviderMixin: async def _request(self, method: str, path: str, **kwargs) -> dict: headers = self._headers() + attempt = None + if isinstance(kwargs.get("json"), dict) and path == self.stream_path: + kwargs["json"] = self._custom_payload(kwargs["json"]) + attempt = self._usage_attempt(kwargs["json"]) try: async with httpx.AsyncClient(timeout=self.timeout_seconds, transport=self.transport) as client: response = await client.request(method, f"{self.base_url}{path}", headers=headers, **kwargs) response.raise_for_status() data = object_value(response.json()) + if attempt: + attempt.observe(data) + attempt.completed = True check_error(data) return data except httpx.TimeoutException as exc: @@ -283,8 +302,13 @@ class HTTPProviderMixin: raise ProviderError("PROVIDER_UNAVAILABLE", "Provider is unavailable.") from exc except (ValueError, TypeError) as exc: raise invalid_response() from exc + finally: + if attempt: + attempt.persist() async def _stream_json(self, payload: dict[str, object]) -> AsyncIterator[dict]: + payload = self._custom_payload(payload) + attempt = self._usage_attempt(payload) headers = self._headers() headers["Accept"] = "text/event-stream" if self.stream_format == "sse" else "application/x-ndjson" try: @@ -295,12 +319,14 @@ class HTTPProviderMixin: if self.stream_format == "sse": async with aclosing(sse_objects(response)) as objects: async for data in objects: + attempt.observe(data) yield data else: async for line in response.aiter_lines(): if line.strip(): data = object_value(json.loads(line)) check_error(data) + attempt.observe(data) yield data except httpx.TimeoutException as exc: raise ProviderError("PROVIDER_TIMEOUT", "Provider request timed out.") from exc @@ -310,3 +336,5 @@ class HTTPProviderMixin: raise ProviderError("PROVIDER_UNAVAILABLE", "Provider is unavailable.") from exc except (ValueError, TypeError) as exc: raise invalid_response() from exc + finally: + attempt.persist() diff --git a/backend/app/providers/routing.py b/backend/app/providers/routing.py index ea7dff3..580da01 100644 --- a/backend/app/providers/routing.py +++ b/backend/app/providers/routing.py @@ -1,14 +1,14 @@ """Capability routing: validated remote results, then an explicit local backend. -Phase E supplies HTTP adapters and injectable local contracts. Hash embeddings are -still a development placeholder; speech models are installed in phase F. +Production injects installed CPU/CUDA backends. Deterministic embeddings remain +available only for explicitly injected tests and protocol fixtures. """ from __future__ import annotations import hashlib import json import math -from dataclasses import dataclass +from dataclasses import dataclass, field, replace from pathlib import Path from typing import Protocol @@ -55,6 +55,7 @@ class RoutedTranscript: text: str source: str fallback_reason: str | None = None + segments: list = field(default_factory=list) def invalid_response() -> ProviderError: @@ -87,6 +88,19 @@ class ModelRoutingService: conn.execute("CREATE TABLE IF NOT EXISTS model_routing (id INTEGER PRIMARY KEY CHECK(id=1), config_json TEXT NOT NULL)") return conn + def snapshot(self): + from copy import copy + from app.providers.registry import RegisteredProvider + frozen = copy(self) + config = self.configuration().model_copy(deep=True) + providers = ProviderRegistry() + for item in self.providers.list_configs(): + original = self.providers.get_any(item.provider_id) + providers._providers[item.provider_id] = RegisteredProvider(item, original.adapter) + frozen.providers = providers + frozen.configuration = lambda: config + return frozen + def configuration(self) -> ModelRoutingConfig: conn = self._connection() try: @@ -98,11 +112,16 @@ class ModelRoutingService: conn.close() def describe(self) -> ModelRoutingResponse: + is_hash = isinstance(self.local_embedding, HashEmbeddingProvider) + embedding_available = getattr(self.local_embedding, "available", True) + def speech_available(capability): + check = getattr(self.local_speech, "available_for", None) + return check(capability) if check else self.local_speech.available return ModelRoutingResponse(config=self.configuration(), local_backends=[ - LocalBackendStatus(capability="embedding", status="placeholder" if isinstance(self.local_embedding, HashEmbeddingProvider) else "ready", - message="当前为 hash-v1 确定性占位向量，真实本地语义模型尚未集成。" if isinstance(self.local_embedding, HashEmbeddingProvider) else "本地 Embedding 模型已就绪。"), - *[LocalBackendStatus(capability=capability, status="ready" if self.local_speech.available else "not_installed", - message="本地模型已就绪。" if self.local_speech.available else "阶段 F 接入本地模型；当前保留回退接口。") + LocalBackendStatus(capability="embedding", status="placeholder" if is_hash else ("ready" if embedding_available else "not_installed"), + message="测试占位向量。" if is_hash else ("本地 Embedding 文件和运行环境已安装。" if embedding_available else "请安装本地模型运行环境并下载 Embedding 权重。")), + *[LocalBackendStatus(capability=capability, status="ready" if speech_available(capability) else "not_installed", + message="本地模型文件和运行环境已安装。" if speech_available(capability) else "请安装运行环境并下载对应本地模型。") for capability in ("transcription", "speaker_matching")], ]) @@ -150,8 +169,16 @@ class ModelRoutingService: url = (provider.base_url or "https://api.openai.com/v1").rstrip("/") + binding.endpoint return url, {"Authorization": f"Bearer {key}"} if key else {} - async def _request(self, binding: ModelBinding, *, remote: tuple[str, dict[str, str]] | None = None, **kwargs) -> tuple[dict, str]: + async def _request(self, binding: ModelBinding, *, remote: tuple[str, dict[str, str]] | None = None, provider_config=None, **kwargs) -> tuple[dict, str]: url, headers = remote or self._remote(binding) + from app.request_overrides import apply_overrides + from app.services.usage_service import UsageAttempt + capability = "embedding" if "json" in kwargs else ("speaker_matching" if "reference_file" in kwargs.get("files", {}) else "transcription") + provider = provider_config or self.providers.get(binding.provider_id).config + field = "json" if capability == "embedding" else "data" + 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) try: async with httpx.AsyncClient(timeout=30, transport=self.transport) as client: async with client.stream("POST", url, headers=headers, **kwargs) as response: @@ -162,6 +189,8 @@ class ModelRoutingService: if len(body) > MAX_RESPONSE_BYTES: raise invalid_response() data = json.loads(body) + attempt.observe(data) + attempt.completed = True except httpx.TimeoutException as exc: raise ProviderError("PROVIDER_TIMEOUT", "Model API timed out.") from exc except httpx.HTTPStatusError as exc: @@ -171,6 +200,8 @@ class ModelRoutingService: raise ProviderError("PROVIDER_UNAVAILABLE", "Model API is unavailable.") from exc except (ValueError, UnicodeError) as exc: raise invalid_response() from exc + finally: + attempt.persist() if not isinstance(data, dict) or data.get("error"): raise invalid_response() return data, url @@ -187,12 +218,13 @@ class ModelRoutingService: dimension = binding.dimensions # Freeze the origin across batches, even if the user edits the provider. remote = self._remote(binding) + provider_config = self.providers.get(binding.provider_id).config.model_copy(deep=True) for start in range(0, len(texts), 32): batch = texts[start:start + 32] payload = {"model": binding.model, "input": batch, "encoding_format": "float"} if binding.dimensions is not None: payload["dimensions"] = binding.dimensions - data, url = await self._request(binding, remote=remote, json=payload) + data, url = await self._request(binding, remote=remote, provider_config=provider_config, json=payload) items = data.get("data") if not isinstance(items, list) or len(items) != len(batch): raise invalid_response() @@ -213,12 +245,20 @@ class ModelRoutingService: raise invalid_response() indexed[index] = [value / norm for value in vector] vectors.extend(indexed[index] for index in range(len(batch))) - identity = json.dumps([url, binding.model, dimension], separators=(",", ":")) + identity_parts = [url, binding.model, dimension] + extensions = [rule.model_dump() for rule in provider_config.request_overrides + if rule.capability == "embedding" and rule.model in (None, binding.model)] + if extensions: + identity_parts.append(extensions) + identity = json.dumps(identity_parts, separators=(",", ":")) return EmbeddingResult(vectors=vectors, source="api", dimensions=dimension, model_id="api-" + hashlib.sha256(identity.encode()).hexdigest()) except ProviderError as exc: reason = exc.code - vectors = await self.local_embedding.embed_documents(texts) + try: + vectors = await self.local_embedding.embed_documents(texts) + except ProviderError as exc: + raise ApiError(503, exc.code, exc.message, {"fallback_reason": reason}) from exc return EmbeddingResult(vectors=vectors, source="local", model_id=self.local_embedding.model_id, dimensions=self.local_embedding.dim, fallback_reason=reason) @@ -234,8 +274,8 @@ class ModelRoutingService: raise ApiError(413, "ATTACHMENT_TOO_LARGE", "Audio attachment must be between 1 byte and 25 MiB.") return handle - async def transcribe(self, source: Path, language: str | None) -> RoutedTranscript: - binding = self.configuration().transcription + async def transcribe(self, source: Path, language: str | None, *, local_only: bool = False) -> RoutedTranscript: + binding = None if local_only else self.configuration().transcription if binding is None: with self._media_file(source): pass @@ -251,19 +291,41 @@ class ModelRoutingService: text = data.get("text") if not isinstance(text, str) or not text.strip(): raise invalid_response() - return RoutedTranscript(text=text, source="api") + segments = [] + raw_segments = data.get("segments", []) + if not isinstance(raw_segments, list) or len(raw_segments) > 10000: + raise invalid_response() + from app.contracts import TranscriptSegment + for index, raw in enumerate(raw_segments): + if not isinstance(raw, dict): + raise invalid_response() + start, end = raw.get("start", raw.get("start_time")), raw.get("end", raw.get("end_time")) + if not finite_number(start) or not finite_number(end) or not isinstance(raw.get("text"), str): + raise invalid_response() + try: + segments.append(TranscriptSegment(segment_id=f"segment_{index + 1}", start_time=start, + end_time=end, text=raw["text"], speaker=raw.get("speaker"))) + except ValueError as exc: + raise invalid_response() from exc + if segments != sorted(segments, key=lambda segment: segment.start_time): + raise invalid_response() + return RoutedTranscript(text=text, source="api", segments=segments) except ProviderError as exc: reason = exc.code try: text = await self.local_speech.transcribe(source, language) + if isinstance(text, RoutedTranscript): + if not text.text.strip(): + raise ProviderError("LOCAL_MODEL_INVALID_RESPONSE", "Local transcription was empty.") + return replace(text, source="local", fallback_reason=reason) if not isinstance(text, str) or not text.strip(): raise ProviderError("LOCAL_MODEL_INVALID_RESPONSE", "Local transcription was empty.") return RoutedTranscript(text=text, source="local", fallback_reason=reason) except ProviderError as exc: raise ApiError(503, exc.code, exc.message, {"fallback_reason": reason}) from exc - async def match_speakers(self, source: Path, reference: Path) -> SpeakerMatchResult: - binding = self.configuration().speaker_matching + async def match_speakers(self, source: Path, reference: Path, *, local_only: bool = False) -> SpeakerMatchResult: + binding = None if local_only else self.configuration().speaker_matching if binding is None: with self._media_file(source), self._media_file(reference): pass diff --git a/backend/app/request_overrides.py b/backend/app/request_overrides.py new file mode 100644 index 0000000..136744b --- /dev/null +++ b/backend/app/request_overrides.py @@ -0,0 +1,68 @@ +"""Declarative request-body extensions with explicit host-owned field conflicts.""" +import copy +import json +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator + +PROTECTED = {"model", "messages", "input", "system", "instructions", "tools", "tool_choice", "parallel_tool_calls", + "functions", "function_call", "file", "audio", "reference_file", "stream", "previous_response_id", + "conversation", "background", "store"} +SECRETS = {"api_key", "apikey", "authorization", "headers", "url", "base_url", "access_token", "secret", "password"} + + +class RequestOverride(BaseModel): + model_config = ConfigDict(extra="forbid") + capability: Literal["chat", "embedding", "transcription", "speaker_matching"] = "chat" + model: str | None = Field(default=None, max_length=200) + stream: bool | None = None + body: dict = Field(default_factory=dict) + + @model_validator(mode="after") + def valid_mode(self): + if self.capability != "chat" and self.stream is True: + raise ValueError("当前 Embedding 与媒体接口不使用流式请求") + return self + + @field_validator("body") + @classmethod + def validate_body(cls, value): + if len(json.dumps(value, allow_nan=False).encode()) > 32768: + raise ValueError("自定义请求 JSON 不得超过 32 KiB") + conflicts = PROTECTED.intersection(value) + if conflicts: + raise ValueError("运行请求管理字段不可覆盖：" + ", ".join(sorted(conflicts))) + def check(item, depth=0): + if depth > 12: + raise ValueError("JSON 嵌套不得超过 12 层") + if isinstance(item, dict): + if any(str(k).lower().replace("-", "_") in SECRETS for k in item): + raise ValueError("密钥、Header 和 URL 请使用独立配置，不得放入请求 JSON") + for child in item.values(): + check(child, depth + 1) + elif isinstance(item, list): + for child in item: + check(child, depth + 1) + check(value) + if "stream_options" in value: + options = value["stream_options"] + if not isinstance(options, dict) or ("include_usage" in options and type(options["include_usage"]) is not bool): + raise ValueError("stream_options 必须是对象，include_usage 必须是布尔值") + return value + + +def deep_merge(base, extension): + result = copy.deepcopy(base) + for key, value in extension.items(): + result[key] = deep_merge(result[key], value) if isinstance(value, dict) and isinstance(result.get(key), dict) else copy.deepcopy(value) + return result + + +def apply_overrides(payload, rules, capability, *, stream=False): + selected = [rule for rule in rules if rule.capability == capability and rule.model in (None, payload.get("model")) + and (rule.stream is None or rule.stream == stream)] + # General defaults precede model overrides; explicit stream conditions are most specific. + selected.sort(key=lambda rule: (rule.model is not None, rule.stream is not None)) + for rule in selected: + payload = deep_merge(payload, rule.body) + return payload diff --git a/backend/app/retrieval/embedding.py b/backend/app/retrieval/embedding.py index 551eb09..a9ce17f 100644 --- a/backend/app/retrieval/embedding.py +++ b/backend/app/retrieval/embedding.py @@ -1,7 +1,6 @@ """Embedding 统一接口与轻量实现。 -真实默认是本地 BGE-M3 类模型，但第一阶段先跑通链路，这里用确定性的特征哈希向量代替。 -后续接入真实模型时实现同样的 EmbeddingProvider 接口替换即可，上层检索逻辑不变。 +生产环境使用 local_models 的真实模型。特征哈希实现仅供测试显式注入。 """ from __future__ import annotations diff --git a/backend/app/retrieval/engine.py b/backend/app/retrieval/engine.py index 1125989..e7a7990 100644 --- a/backend/app/retrieval/engine.py +++ b/backend/app/retrieval/engine.py @@ -20,6 +20,7 @@ from app.contracts import ( ) from app.repository import BlockHit from app.retrieval.embedding import EmbeddingProvider, HashEmbeddingProvider +from app.local_models.runtime import LocalEmbedding from app.retrieval.hybrid import normalize_scores, rrf_fuse from app.retrieval.reranker import LexicalReranker, RankedCandidate, RerankerProvider from app.retrieval import routed_vectors @@ -88,8 +89,13 @@ class RetrievalEngine: and self.embedding is self._routed_defaults[0] and self.vector_store is self._routed_defaults[1] ): - vec_hits = await routed_vectors.search_remote(request.query, top_k=recall) + vec_hits = await routed_vectors.search_remote(request.query, top_k=recall, accept_local=isinstance(self.embedding, LocalEmbedding)) if vec_hits is None: + if isinstance(self.embedding, LocalEmbedding): + if request.mode == SearchMode.hybrid: + return self._search_fts(request) + from app.errors import ApiError + raise ApiError(409, "SEMANTIC_INDEX_UNAVAILABLE", "语义索引未就绪。请配置 Embedding 或下载本地模型后重建索引。") query_vec = await self.embedding.embed_query(request.query) vec_hits = await self.vector_store.search(query_vec, top_k=recall) record_embedding(source="local", model_id=self.embedding.model_id, @@ -287,5 +293,5 @@ def _utc(dt: datetime) -> datetime: # 默认引擎实例：轻量实现跑通链路，后续可替换真实模型实现 engine = RetrievalEngine( - HashEmbeddingProvider(), LexicalReranker(), SqliteVecStore(), route_embeddings=True, + LocalEmbedding(), LexicalReranker(), SqliteVecStore(), route_embeddings=True, ) diff --git a/backend/app/retrieval/routed_vectors.py b/backend/app/retrieval/routed_vectors.py index 7a5be34..7fe0837 100644 --- a/backend/app/retrieval/routed_vectors.py +++ b/backend/app/retrieval/routed_vectors.py @@ -42,6 +42,7 @@ class RemoteEmbeddings: space_id: str dimensions: int vectors: list[list[float]] + source: str = "api" def get_model_routing() -> EmbeddingRuntime | None: @@ -67,7 +68,7 @@ def _unit_vector(vector: list[float], dimensions: int) -> list[float]: return [value / norm for value in scaled] -async def embed_remote(texts: list[str]) -> RemoteEmbeddings | None: +async def embed_remote(texts: list[str], *, accept_local=False) -> RemoteEmbeddings | None: """Return validated API vectors, or None to use the caller's local baseline. Do not use the runtime's local result: the caller may have injected its own @@ -80,7 +81,7 @@ async def embed_remote(texts: list[str]) -> RemoteEmbeddings | None: if runtime is None: return None result = await runtime.embed(texts) - if result.source != "api": + if result.source != "api" and not accept_local: record_embedding(fallback_reason=result.fallback_reason) return None if not isinstance(result.model_id, str) or not result.model_id or result.model_id == "hash-v1": @@ -93,6 +94,7 @@ async def embed_remote(texts: list[str]) -> RemoteEmbeddings | None: space_id=result.model_id, dimensions=result.dimensions, vectors=[_unit_vector(vector, result.dimensions) for vector in result.vectors], + source=result.source, ) except Exception as exc: # Avoid logging provider exceptions containing credentials or note text. @@ -152,13 +154,13 @@ def store_remote( logger.warning("Remote vector storage unavailable (%s); local index retained", type(exc).__name__) -async def search_remote(query: str, *, top_k: int) -> list[VectorHit] | None: +async def search_remote(query: str, *, top_k: int, accept_local=False) -> list[VectorHit] | None: """None means fallback, including any missing/invalid current-block vector. Read coverage and vectors together so concurrent note updates cannot produce an apparently complete subset. Never fill missing remote hits with local hits. """ - batch = await embed_remote([query]) + batch = await embed_remote([query], accept_local=accept_local) if batch is None: return None record_embedding(attempted_space={"model_id": batch.space_id, "dimensions": batch.dimensions}) @@ -190,7 +192,7 @@ async def search_remote(query: str, *, top_k: int) -> list[VectorHit] | None: yield VectorHit(id=row["block_id"], score=max(0.0, min(1.0, score))) result = heapq.nlargest(top_k, hits(), key=lambda hit: hit.score) - record_embedding(source="api", model_id=batch.space_id, + record_embedding(source=batch.source, model_id=batch.space_id, dimensions=batch.dimensions, fallback_reason=None) return result finally: diff --git a/backend/app/routes.py b/backend/app/routes.py index ef9f8d3..18e37b0 100644 --- a/backend/app/routes.py +++ b/backend/app/routes.py @@ -915,6 +915,7 @@ async def create_provider(request: ProviderCreateRequest) -> ProviderConfig: default_model=request.default_model, credential_id=request.credential_id, enabled=request.enabled, + request_overrides=request.request_overrides, capabilities=container.provider_factory.capabilities(request.provider_type), ) try: @@ -943,8 +944,12 @@ async def update_provider( 409, "BUILTIN_PROVIDER_IMMUTABLE", "Mock provider cannot be modified." ) fields = request.model_fields_set + if request.version is not None and request.version != current.version: + raise ApiError(409, "PROVIDER_VERSION_CONFLICT", "提供商配置已变更，请重新加载后保存。") if ("provider_type" in fields and request.provider_type is None) or ("name" in fields and request.name is None) or ( "enabled" in fields and request.enabled is None + ) or ( + "request_overrides" in fields and request.request_overrides is None ): raise ApiError( 422, @@ -952,6 +957,7 @@ async def update_provider( "provider_type, name and enabled cannot be null when explicitly provided.", ) updates = {name: getattr(request, name) for name in fields} + updates["version"] = current.version + 1 if "credential_id" in fields: validate_public_credential_id(request.credential_id) config = ProviderConfig.model_validate( @@ -1100,6 +1106,7 @@ async def create_embeddings(request: EmbeddingRequest) -> EmbeddingResult: async def match_speakers(request: SpeakerMatchRequest) -> SpeakerMatchResult: return await container.model_routing.match_speakers( attachment_path(request.attachment_id), attachment_path(request.reference_attachment_id), + local_only=request.local_only, ) @@ -1111,7 +1118,7 @@ async def match_speakers(request: SpeakerMatchRequest) -> SpeakerMatchResult: ) async def create_transcription(request: TranscriptionRequest) -> TranscriptionJob: return await transcription_service.create_transcription( - request.attachment_id, request.language, diarization=request.diarization + **request.model_dump(), wait=False ) diff --git a/backend/app/services/index_service.py b/backend/app/services/index_service.py index e635577..f6d4aaa 100644 --- a/backend/app/services/index_service.py +++ b/backend/app/services/index_service.py @@ -97,6 +97,7 @@ async def rebuild(request: IndexRebuildRequest) -> IndexJob: task_note_links = dict(conn.execute( "SELECT task_id, note_id FROM tasks WHERE note_id IS NOT NULL" ).fetchall()) + media_links = conn.execute("SELECT job_id,revision,options_hash,note_id FROM media_notes").fetchall() repository.clear_all(conn=conn) await vector_store.clear(conn=conn) for parsed, prepared in prepared_notes: @@ -107,6 +108,9 @@ async def rebuild(request: IndexRebuildRequest) -> IndexJob: "AND EXISTS (SELECT 1 FROM notes WHERE note_id = ?)", (note_id, task_id, note_id), ) + for link in media_links: + conn.execute("INSERT OR IGNORE INTO media_notes SELECT ?,?,?,? WHERE EXISTS (SELECT 1 FROM notes WHERE note_id=?)", + (*link, link["note_id"])) finally: conn.close() except BaseException as exc: diff --git a/backend/app/services/media_notes.py b/backend/app/services/media_notes.py new file mode 100644 index 0000000..cf0f938 --- /dev/null +++ b/backend/app/services/media_notes.py @@ -0,0 +1,55 @@ +"""Idempotent transcript export without overwriting an edited note.""" +import asyncio +import hashlib +from contextlib import closing + +from app.config import get_settings +from app.database.db import connect, transaction +from app.errors import ApiError +from app.services import note_service +from app.services.transcription_service import require_job + +_locks = {} + + +async def create_transcript_note(job_id, options): + identity = (str(get_settings().db_path), job_id) + lock = _locks.setdefault(identity, asyncio.Lock()) + async with lock: + 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() + with closing(connect()) as conn: + 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: + return await note_service.get_note(row[0]) + marker = f"" + title = f"{options.title} · {job_id[-8:]}-r{job.revision}-{options_hash[:6]}" + lines = [marker, f"# {options.title}", "", f"[源音频](/#/media?job={job_id})", ""] + if job.segments: + for segment in job.segments: + prefix = [] + if options.include_timestamps: + seconds = segment.start_time + label = f"{int(seconds // 60):02}:{int(seconds % 60):02}" + prefix.append(f"[{label}](/#/media?job={job_id}&time={seconds})") + if options.include_speakers and segment.speaker: + prefix.append(job.speaker_names.get(segment.speaker, segment.speaker)) + lines.append(" ".join([*prefix, segment.text])) + lines.append("") + else: + lines.append(job.text or "") + 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)) + return note diff --git a/backend/app/services/note_service.py b/backend/app/services/note_service.py index a4a1b27..2380b82 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.retrieval.embedding import HashEmbeddingProvider +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 @@ -28,8 +28,8 @@ from app.services.vault_paths import ( safe_note_filename, ) -# 轻量实现实例（无状态，可直接复用）；接入真实模型后替换为对应 Provider -embedding = HashEmbeddingProvider() +# 真实模型接口不在 API 进程加载权重；测试可显式替换该实例。 +embedding = LocalEmbedding() vector_store = SqliteVecStore() @@ -80,6 +80,10 @@ PreparedIndex = tuple[list[list[float]], routed_vectors.RemoteEmbeddings | None] async def prepare_note_index(parsed: ParsedNote) -> PreparedIndex: """Compute vectors before opening a write transaction (including API I/O).""" texts = [block.content for block in parsed.blocks] + if isinstance(embedding, LocalEmbedding): + # One routed invocation: API first, validated local fallback. No hash vectors. + remote = await routed_vectors.embed_remote(texts, accept_local=True) + return [], remote vectors = await embedding.embed_documents(texts) remote = await routed_vectors.embed_remote(texts) return vectors, remote @@ -127,7 +131,8 @@ async def index_note( await vector_store.upsert(records, conn=conn) routed_vectors.store_remote(conn, [block.block_id for block in parsed.blocks], remote) repository.set_index_meta( - {"embedding_model": embedding.model_id, "embedding_dim": str(embedding.dim)}, + {"embedding_model": remote.space_id if remote and isinstance(embedding, LocalEmbedding) else embedding.model_id, + "embedding_dim": str(remote.dimensions if remote and isinstance(embedding, LocalEmbedding) else embedding.dim)}, conn=conn, ) finally: diff --git a/backend/app/services/transcription_service.py b/backend/app/services/transcription_service.py index 1e1fdda..6a816f2 100644 --- a/backend/app/services/transcription_service.py +++ b/backend/app/services/transcription_service.py @@ -1,65 +1,243 @@ -"""转写作业：API 优先，本地模型回退；保留已有 Host 文本入口。""" - +"""Persistent media jobs and replayable events; HTTP enqueues, tools await.""" from __future__ import annotations - -from collections import OrderedDict +import asyncio +import hashlib +import json +from contextlib import closing from datetime import datetime, timezone from uuid import uuid4 - -from app.contracts import TranscriptionJob +from app.config import get_settings +from app.contracts import TranscriptionJob, TranscriptionRequest, TranscriptEditRequest +from app.database.db import connect, transaction from app.errors import ApiError from app.services.attachment_service import attachment_path -_jobs: OrderedDict[str, TranscriptionJob] = OrderedDict() -MAX_JOBS = 100 +TERMINAL = {"completed", "failed", "cancelled"} +_tasks: dict[tuple[str, str], asyncio.Task] = {} +def now(): + return datetime.now(timezone.utc) -async def create_transcription(attachment_id: str, language: str | None = None, *, diarization: bool = False) -> TranscriptionJob: - from app.container import container - - source = attachment_path(attachment_id) - job = TranscriptionJob( - job_id=f"transcription_{uuid4().hex}", - attachment_id=attachment_id, - status="processing", - created_at=datetime.now(timezone.utc), - ) - try: - if diarization: - # Speaker verification and diarization are different capabilities. - raise ApiError(501, "DIARIZATION_NOT_IMPLEMENTED", "说话人分离将在阶段 F 接入，当前不能忽略 diarization 请求。") - transcript = source if source.suffix.lower() in {".txt", ".md"} else attachment_path(f"{attachment_id}.txt") - # A saved transcript remains an explicit import path, never faked ASR. - if transcript.is_file() and (source == transcript or container.model_routing.configuration().transcription is None): - with transcript.open("rb") as handle: - content = handle.read(1024 * 1024 + 1) - if len(content) > 1024 * 1024: - raise ApiError(413, "TRANSCRIPT_TOO_LARGE", "Transcript exceeds 1 MiB.") - job.text = content.decode("utf-8") - if not job.text.strip(): - raise ApiError(422, "TRANSCRIPT_EMPTY", "Transcript is empty.") - job.source = "sidecar" - else: - result = await container.model_routing.transcribe(source, language) - job.text = result.text - job.source = result.source - job.fallback_reason = result.fallback_reason - job.status = "completed" - except ApiError as exc: - job.status = "failed" - job.error_code = exc.code - job.error_message = exc.message - job.fallback_reason = exc.details.get("fallback_reason") - except (OSError, UnicodeError): - job.status = "failed" - job.error_code = "TRANSCRIPT_UNREADABLE" - job.error_message = "Transcript could not be read." - _jobs[job.job_id] = job - while len(_jobs) > MAX_JOBS: - _jobs.popitem(last=False) - return job.model_copy(deep=True) - +def task_key(job_id): + return str(get_settings().db_path), job_id def get_transcription(job_id: str) -> TranscriptionJob | None: - job = _jobs.get(job_id) - return job.model_copy(deep=True) if job else None + with closing(connect()) as conn: + row = conn.execute("SELECT job_json FROM media_jobs WHERE job_id=?", (job_id,)).fetchone() + return TranscriptionJob.model_validate_json(row[0]) if row else None + +def require_job(job_id): + job = get_transcription(job_id) + if job is None: + raise ApiError(404, "RESOURCE_NOT_FOUND", "Transcription job not found.") + return job + +def _event(conn, job, event, data=None): + sequence = conn.execute("SELECT COALESCE(MAX(sequence),-1)+1 FROM media_events WHERE job_id=?", (job.job_id,)).fetchone()[0] + conn.execute("INSERT INTO media_events VALUES (?,?,?,?,?)", (job.job_id, sequence, event, + json.dumps(data or {"status": job.status, "progress": job.progress}), now().isoformat())) + +def save(job, event): + job.updated_at = now() + with closing(connect()) as conn, transaction(conn): + conn.execute("UPDATE media_jobs SET status=?,job_json=?,updated_at=? WHERE job_id=?", + (job.status, job.model_dump_json(), job.updated_at.isoformat(), job.job_id)) + _event(conn, job, event) + +def list_transcriptions(status=None, limit=50, offset=0): + where, args = (" WHERE status=?", [status]) if status else ("", []) + with closing(connect()) as conn: + total = conn.execute("SELECT COUNT(*) FROM media_jobs" + where, args).fetchone()[0] + rows = conn.execute("SELECT job_json FROM media_jobs" + where + " ORDER BY created_at DESC LIMIT ? OFFSET ?", [*args, limit, offset]).fetchall() + return {"items": [TranscriptionJob.model_validate_json(row[0]) for row in rows], "page": {"total": total, "limit": limit, "offset": offset}} + +def events(job_id, after=-1): + require_job(job_id) + with closing(connect()) as conn: + rows = conn.execute("SELECT * FROM media_events WHERE job_id=? AND sequence>? ORDER BY sequence LIMIT 200", (job_id, after)).fetchall() + return [{"job_id": job_id, "sequence": r["sequence"], "event": r["event"], "data": json.loads(r["data_json"]), "timestamp": r["timestamp"]} for r in rows] + +def recover_interrupted(): + with closing(connect()) as conn: + rows = conn.execute("SELECT job_json FROM media_jobs WHERE status IN ('queued','running','processing')").fetchall() + for row in rows: + job = TranscriptionJob.model_validate_json(row[0]) + if task_key(job.job_id) not in _tasks: + job.status, job.error_code = "failed", "TRANSCRIPTION_INTERRUPTED" + job.error_message = "AI Core stopped before completion. Retry to start a new attempt." + job.completed_at = now() + save(job, "Failed") + +async def shutdown(): + tasks = [t for k, t in list(_tasks.items()) if k[0] == str(get_settings().db_path)] + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + +async def create_transcription(attachment_id, language=None, *, diarization=False, local_only=False, + word_timestamps=False, idempotency_key=None, terminology=None, wait=True, previous_job_id=None): + request = TranscriptionRequest(attachment_id=attachment_id, language=language, diarization=diarization, + local_only=local_only, word_timestamps=word_timestamps, idempotency_key=idempotency_key, terminology=terminology or {}) + source = attachment_path(attachment_id) + actual = source if source.is_file() else attachment_path(f"{attachment_id}.txt") + if not actual.is_file(): + raise ApiError(404, "ATTACHMENT_NOT_FOUND", "Attachment was not found.") + if not 0 < actual.stat().st_size <= 25 * 1024 * 1024: + raise ApiError(413, "ATTACHMENT_TOO_LARGE", "Attachment must be between 1 byte and 25 MiB.") + digest = await asyncio.to_thread(lambda: hashlib.sha256(actual.read_bytes()).hexdigest()) + from app.container import container + from app.local_models.runtime import configuration + from app.local_models.catalog import CATALOG + routing = container.model_routing.snapshot() + route = routing.configuration() + binding = None if local_only else route.transcription + snapshot = {"local_runtime": configuration().model_dump(), "models": {k:v.revision for k,v in CATALOG.items()}, + "transcription": binding.model_dump() if binding else None} + if binding: + provider = routing.providers.get_any(binding.provider_id).config + snapshot["provider"] = provider.model_dump(exclude={"credential_id"}) + fingerprint = hashlib.sha256((digest + request.model_dump_json(exclude={"idempotency_key"}) + json.dumps(snapshot, sort_keys=True)).encode()).hexdigest() + job = TranscriptionJob(job_id=f"transcription_{uuid4().hex}", attachment_id=attachment_id, status="queued", + created_at=now(), updated_at=now(), language=language, local_only=local_only, previous_job_id=previous_job_id, model_snapshot=snapshot) + existing = None + with closing(connect()) as conn, transaction(conn): + if idempotency_key: + existing = conn.execute("SELECT job_json,fingerprint FROM media_jobs WHERE idempotency_key=?", (idempotency_key,)).fetchone() + if existing: + if existing["fingerprint"] != fingerprint: + raise ApiError(409, "IDEMPOTENCY_CONFLICT", "This key was used for different input.") + job = TranscriptionJob.model_validate_json(existing["job_json"]) + else: + conn.execute("INSERT INTO media_jobs VALUES (?,?,?,?,?,?,?,?)", (job.job_id, job.status, + job.model_dump_json(), request.model_dump_json(), job.created_at.isoformat(), job.updated_at.isoformat(), idempotency_key, fingerprint)) + _event(conn, job, "Queued") + key = task_key(job.job_id) + if not existing: + task = asyncio.create_task(_execute(job.job_id, request, routing)) + _tasks[key] = task + task.add_done_callback(lambda finished: _tasks.pop(key, None)) + if wait and key in _tasks: + try: + await _tasks[key] + except asyncio.CancelledError: + await cancel(job.job_id) + raise + return require_job(job.job_id) + return job + +async def _execute(job_id, request, routing=None): + from app.container import container + job = require_job(job_id) + if job.status in TERMINAL: + return + from app.local_models.runtime import runtime_context, runtime_progress, RuntimeConfig + from app.contracts import TranscriptSegment + token = runtime_context.set(RuntimeConfig.model_validate(job.model_snapshot.get("local_runtime", {}))) + def progress(message): + job.progress = max(0.0, min(0.99, message["progress"])) + job.segments.append(TranscriptSegment.model_validate(message["segment"])) + save(job, "SegmentReady") + progress_token = runtime_progress.set(progress) + job.status, job.started_at = "running", now() + save(job, "TranscriptionStarted") + cancelled = False + try: + source = attachment_path(job.attachment_id) + transcript = source if source.suffix.lower() in {".txt", ".md"} else attachment_path(f"{job.attachment_id}.txt") + if transcript.is_file() and (source == transcript or not source.exists()): + def read_transcript(): + with transcript.open("rb") as stream: + return stream.read(1024 * 1024 + 1) + content = await asyncio.to_thread(read_transcript) + if len(content) > 1024 * 1024: + raise ApiError(413, "TRANSCRIPT_TOO_LARGE", "Transcript exceeds 1 MiB.") + job.text, job.source = content.decode("utf-8"), "sidecar" + else: + result = await (routing or container.model_routing).transcribe(source, request.language, local_only=request.local_only) + job.text, job.source, job.fallback_reason = result.text, result.source, result.fallback_reason + job.segments = getattr(result, "segments", []) or [] + if not job.text or not job.text.strip(): + raise ApiError(422, "TRANSCRIPT_EMPTY", "Transcript is empty.") + if request.diarization: + if job.segments: + from app.local_models.runtime import runtime + from app.providers.base import ProviderError + try: + result = await runtime.infer("eres2netv2", "diarization", {"source": str(source.resolve()), + "segments": [s.model_dump() for s in job.segments]}) + for segment, speaker in zip(job.segments, result["speakers"], strict=True): + segment.speaker = speaker + job.warnings.append("DIARIZATION_SEGMENT_LEVEL") + except ProviderError: + job.warnings.append("DIARIZATION_UNAVAILABLE") + else: + job.warnings.append("DIARIZATION_UNAVAILABLE") + if request.word_timestamps: + job.warnings.append("WORD_TIMESTAMPS_UNAVAILABLE") + job.original_text, job.original_segments = job.text, [s.model_copy(deep=True) for s in job.segments] + for original, replacement in request.terminology.items(): + if original and original != replacement and original in job.text: + job.text = job.text.replace(original, replacement) + for segment in job.segments: + segment.text = segment.text.replace(original, replacement) + job.corrections.append({"original": original, "replacement": replacement, "source": "terminology_postprocessing"}) + job.status, job.progress = "completed", 1 + except asyncio.CancelledError: + cancelled = True + job.status, job.error_code = "cancelled", "TRANSCRIPTION_CANCELLED" + except ApiError as exc: + job.status, job.error_code, job.error_message = "failed", exc.code, exc.message + job.fallback_reason = exc.details.get("fallback_reason") + except Exception: + job.status, job.error_code, job.error_message = "failed", "TRANSCRIPTION_FAILED", "Transcription could not be completed." + job.completed_at = now() + save(job, {"completed": "Completed", "cancelled": "Cancelled", "failed": "Failed"}[job.status]) + runtime_context.reset(token) + runtime_progress.reset(progress_token) + if cancelled: + raise asyncio.CancelledError + +async def cancel(job_id): + job = require_job(job_id) + if job.status in TERMINAL: + return job + task = _tasks.get(task_key(job_id)) + if task: + task.cancel() + await asyncio.gather(task, return_exceptions=True) + job = require_job(job_id) + if job.status not in TERMINAL: + job.status, job.error_code, job.completed_at = "cancelled", "TRANSCRIPTION_CANCELLED", now() + save(job, "Cancelled") + return job + +async def retry(job_id): + if require_job(job_id).error_code == "MEDIA_PURGED": + raise ApiError(409, "MEDIA_PURGED", "Purged jobs cannot be retried.") + if require_job(job_id).status not in {"failed", "cancelled"}: + raise ApiError(409, "TRANSCRIPTION_NOT_RETRYABLE", "Only failed or cancelled jobs can be retried.") + with closing(connect()) as conn: + raw = conn.execute("SELECT request_json FROM media_jobs WHERE job_id=?", (job_id,)).fetchone()[0] + request = TranscriptionRequest.model_validate_json(raw) + return await create_transcription(**request.model_dump(exclude={"idempotency_key"}), wait=False, previous_job_id=job_id) + +def edit(job_id, request: TranscriptEditRequest): + with closing(connect()) as conn, transaction(conn): + row = conn.execute("SELECT job_json FROM media_jobs WHERE job_id=?", (job_id,)).fetchone() + if not row: + raise ApiError(404, "RESOURCE_NOT_FOUND", "Transcription job not found.") + job = TranscriptionJob.model_validate_json(row[0]) + if job.status != "completed": + raise ApiError(409, "TRANSCRIPT_NOT_READY", "Only completed transcripts can be edited.") + if job.revision != request.revision: + raise ApiError(409, "VERSION_CONFLICT", "Transcript has changed; reload before saving.") + ids = [s.segment_id for s in request.segments] + if len(ids) != len(set(ids)) or request.segments != sorted(request.segments, key=lambda s: s.start_time): + raise ApiError(422, "INVALID_SEGMENTS", "Segments must have unique IDs and ordered timestamps.") + conn.execute("INSERT INTO media_revisions VALUES (?,?,?)", (job_id, job.revision, job.model_dump_json())) + job.text, job.segments, job.speaker_names = request.text, request.segments, request.speaker_names + job.revision += 1 + job.updated_at = now() + conn.execute("UPDATE media_jobs SET job_json=?,updated_at=? WHERE job_id=?", (job.model_dump_json(), job.updated_at.isoformat(), job_id)) + _event(conn, job, "Revised", {"revision": job.revision}) + return job diff --git a/backend/app/services/usage_service.py b/backend/app/services/usage_service.py new file mode 100644 index 0000000..097e209 --- /dev/null +++ b/backend/app/services/usage_service.py @@ -0,0 +1,132 @@ +"""Application-observed usage per actual HTTP attempt; never an account bill.""" +from __future__ import annotations + +import json +import logging +from contextlib import closing +from contextvars import ContextVar +from datetime import datetime, timezone +from uuid import uuid4 + +from app.database.db import connect + +METRICS = ("input_tokens", "output_tokens", "total_tokens", "cache_hit_tokens", "cache_miss_tokens", "cache_write_tokens", "reasoning_tokens") +logger = logging.getLogger(__name__) +usage_context = ContextVar("usage_context", default=None) + + +def connection(): + conn = connect() + conn.execute("""CREATE TABLE IF NOT EXISTS model_usage ( + attempt_id TEXT PRIMARY KEY, provider_id TEXT NOT NULL, model TEXT NOT NULL, + capability TEXT NOT NULL, source TEXT NOT NULL, started_at TEXT NOT NULL, + completed INTEGER NOT NULL, counters_json TEXT NOT NULL, raw_json TEXT NOT NULL)""") + conn.execute("CREATE INDEX IF NOT EXISTS usage_time_provider ON model_usage(started_at,provider_id,model)") + columns = {row[1] for row in conn.execute("PRAGMA table_info(model_usage)")} + for column in ("request_id", "run_id"): + if column not in columns: + conn.execute(f"ALTER TABLE model_usage ADD COLUMN {column} TEXT") + return conn + + +def numeric_leaves(value, prefix=""): + """Keep known numerical counters only; vendor usage objects may contain arbitrary text.""" + result = {} + if not isinstance(value, dict): + return result + allowed = {"prompt_tokens", "completion_tokens", "input_tokens", "output_tokens", "total_tokens", "cached_tokens", + "cache_read_input_tokens", "cache_creation_input_tokens", "prompt_cache_hit_tokens", "prompt_cache_miss_tokens", + "reasoning_tokens", "prompt_eval_count", "eval_count"} + for key, item in value.items(): + path = f"{prefix}.{key}" if prefix else key + if key in allowed and type(item) is int and 0 <= item <= 2 ** 53: + result[path] = item + elif key in {"prompt_tokens_details", "completion_tokens_details", "input_tokens_details", "output_tokens_details"}: + result.update(numeric_leaves(item, path)) + return result + + +class UsageAttempt: + def __init__(self, provider_id, model, protocol, capability="chat", source="api"): + self.attempt_id = uuid4().hex + self.provider_id, self.model, self.protocol = provider_id, model, protocol + self.capability, self.source = capability, source + self.started_at = datetime.now(timezone.utc).isoformat() + self.raw = {} + self.completed = False + context = usage_context.get() or {} + self.request_id = context.get("request_id") or uuid4().hex + self.run_id = context.get("run_id") + + def observe(self, data): + if not isinstance(data, dict): + return + 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": + values.append(data) + for value in values: + for key, count in numeric_leaves(value).items(): + self.raw[key] = max(self.raw.get(key, 0), count) + if data.get("type") in {"[DONE]", "response.completed", "message_stop"} or data.get("done") is True: + self.completed = True + + def counters(self): + raw = self.raw + def first(*names): + return next((raw[name] for name in names if name in raw), None) + inputs = first("input_tokens", "prompt_tokens", "prompt_eval_count") + outputs = first("output_tokens", "completion_tokens", "eval_count") + hit = first("cache_read_input_tokens", "prompt_cache_hit_tokens", "input_tokens_details.cached_tokens", "prompt_tokens_details.cached_tokens") + write = first("cache_creation_input_tokens") + miss = first("prompt_cache_miss_tokens") + if self.protocol == "anthropic_messages": + miss = inputs + inputs = inputs + hit + write if inputs is not None and hit is not None and write is not None else None + elif miss is None and inputs is not None and hit is not None and 0 <= hit <= inputs: + 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, + 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")) + + def persist(self): + try: + with closing(connection()) as conn: + conn.execute("INSERT OR REPLACE INTO model_usage VALUES (?,?,?,?,?,?,?,?,?,?,?)", ( + self.attempt_id, self.provider_id, self.model, self.capability, self.source, self.started_at, + int(self.completed), json.dumps(self.counters()), json.dumps(self.raw), self.request_id, self.run_id)) + except Exception: + logger.warning("Usage persistence failed; model response remains available") + + +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 文档状态：接口冻结草案 > > 更新日期：2026-09-03 diff --git a/docs/development/多模态管线与模型运行开发说明.md b/docs/development/多模态管线与模型运行开发说明.md new file mode 100644 index 0000000..510c053 --- /dev/null +++ b/docs/development/多模态管线与模型运行开发说明.md @@ -0,0 +1,114 @@ +# 多模态管线与模型运行 + +更新日期：2026-09-04。阶段 F 实现位于 `feat/multimodal-pipeline`，接口以 `/openapi.json` 为准。 + +## 安装 + +API 保留 `backend/.venv`，模型依赖安装到独立的 `backend/.venv-models`。在项目根目录执行： + +```powershell +# 默认 CPU +./backend/scripts/install-model-runtime.ps1 +# CUDA 显式选装，不安装或修改 NVIDIA 驱动 +./backend/scripts/install-model-runtime.ps1 -Device cuda +``` + +脚本固定 torch/torchaudio 2.9.1，分别选择 CPU / cu128 wheel；其他已验证依赖由 `model-requirements.lock` 锁定。不要求 vLLM、FlashAttention。`APP_MODEL_PYTHON` 可指定模型解释器。 + +设置 → 模型提供商 → 本地模型提供下载、续传、删除、设备与预算配置。推理不自动下载；“已下载并校验”不代表设备已通过推理验证，最近实际设备与诊断单独显示。 + +默认 CPU、2 线程、8 GiB 内存预算。独立子进程按需加载，每任务结束释放，取消/超时终止并回收进程。单模型串行执行，排队中的交互向量请求优先于转写，不抢占运行中任务。请求 CUDA 但不可用时回退 CPU，记录原因。任务冻结运行配置。当前要求单 API worker，不支持跨进程调度。 + +## 模型与许可 + +| 能力 | 模型 | 固定 revision | 权重许可 | +| --- | --- | --- | --- | +| 默认 Embedding | hotchpotch/bekko-embedding-v1-a8m | c721113d59a1d91b447450324f51c4b3332c924a | MIT | +| 可选 Embedding | ibm-granite/granite-embedding-97m-multilingual-r2 | 835ad14087e140460703cf0fae09f97d469d65c2 | Apache-2.0 | +| 转写、语言识别 | Qwen/Qwen3-ASR-0.6B | 5eb144179a02acc5e5ba31e748d22b0cf3e303b0 | Apache-2.0 | +| 声纹相似度 | iic/speech_eres2netv2_sv_zh-cn_16k-common | 3317286545c587ae682dbc166831d9448780eebb | Apache-2.0 | + +来源：[Bekko](https://huggingface.co/hotchpotch/bekko-embedding-v1-a8m)、[Granite](https://huggingface.co/ibm-granite/granite-embedding-97m-multilingual-r2)、[Qwen3-ASR](https://huggingface.co/Qwen/Qwen3-ASR-0.6B)、[ERes2NetV2](https://modelscope.cn/models/iic/speech_eres2netv2_sv_zh-cn_16k-common)。下载固定 revision；HF LFS / ModelScope 校验 SHA-256，HF 普通文件校验 Git blob hash。中断保留 .partial，使用 Range 续传；校验失败、磁盘不足和中断分别记录。 + +## 媒体接口 + +| 方法与路径 | 行为 | +| --- | --- | +| POST /api/media/attachments?filename=... | 二进制上传，宿主分配 ID，25 MiB 上限 | +| GET /api/media/attachments/{id} | 受控读取，支持播放器 Range | +| POST /api/media/transcriptions | 202/queued；local_only、diarization、terminology、idempotency_key | +| GET /api/media/transcriptions | 按状态分页查询 | +| GET /api/media/transcriptions/{id} | 状态、分段、原文、修订、进度 | +| GET /api/media/transcriptions/{id}/events | SSE；after / Last-Event-ID 回放 | +| POST /api/media/transcriptions/{id}/cancel | 取消排队或运行任务 | +| POST /api/media/transcriptions/{id}/retry | 新 attempt，保留 previous_job_id | +| PATCH /api/media/transcriptions/{id} | revision 乐观锁校对、重命名 | +| GET /api/media/transcriptions/{id}/revisions | 历史修订 | +| POST /api/media/transcriptions/{id}/notes | Knowledge 写入；任务/修订/选项幂等 | +| POST /api/media/speaker-matches | 两附件声纹比对，支持 local_only | +| GET /api/media/attachments/{id}/cleanup-impact | 清理影响与保留笔记 | +| DELETE /api/media/attachments/{id} | 清理附件、转写正文、修订及术语 | + +任务、模型快照、事件和修订写入 SQLite。重启把未完成任务标为 TRANSCRIPTION_INTERRUPTED，不自动重新上传。幂等摘要包含内容、选项、模型和提供商配置；同键不同输入返回 409。 + +API 优先，无配置或无效结果时本地回退。local_only 禁止远程模型。纯文本附件和既有 sidecar 可导入，但已有真实音频时不使用旁边文本冒充识别。 + +PyAV 提取音轨至 16 kHz 单声道，最长 1 小时，禁止解码器网络协议。能量分段后交给 Qwen3-ASR，返回片段边界，不宣称逐字对齐。ERes2NetV2 提取片段声纹并按相似度聚类；短片段、同段多人、重叠发言需要人工校对。缺失能力返回 DIARIZATION_UNAVAILABLE；未启用逐字对齐返回 WORD_TIMESTAMPS_UNAVAILABLE。 + +术语是识别后的替换规则，保留原始文本和来源。重命名只修改显示名，稳定 ID 不变。笔记包含音频与时间跳转链接；重复导出不覆盖用户编辑。清理保留已导出笔记，音频链接失效，已清理任务不可重试。重建索引保留转写与笔记关联。 + +## 向量空间 + +生产使用真实模型；HashEmbeddingProvider 仅供测试注入。本地/API 向量都写入按模型空间隔离的 routed_block_vectors；不同模型、revision、接口或维度不混用。旧 128 维测试索引不用于真实查询。 + +模型不可用时仍可保存 Markdown/FTS；语义查询返回索引未就绪，混合查询可使用全文检索。切换模型后重建全部索引。Benchmark 验证当前空间完整覆盖。 + +## Token 用量 + +GET /api/usage 使用带时区的 start/end（左闭右开），支持 provider_id、model、source。页面提供今日、7 天、30 天、自定义时段。 + +按实际 attempt 保存 request_id、Agent run_id、模型、来源、时间、原始数值与归一化计数。覆盖 Chat、流式、Agent、Embedding、媒体及本地推理。累计快照取最大值并 upsert；回放不新增请求，真实重试有新 attempt。流中断保留已收到计数。 + +输入缓存按供应商口径归一化，推理不重复加入输出；缓存命中率按完整输入口径加权。缺失为 null，显示每项覆盖数。本地 Embedding 使用真实 tokenizer，其他本地能力不编造 Token。写入失败不影响回复；原始 usage 仅保留数值白名单。本应用观测值不是厂商账单。 + +## 自定义请求 JSON + +request_overrides 每项含 capability、model（空表示全部）、stream（null 表示全部模式）、body。通用规则先于模型规则，同层显式流式规则优先。对象递归合并、数组替换、标量覆盖、null 保持实际值；删除键恢复继承。 + +```json +{ + "capability": "chat", + "model": "special-model", + "stream": true, + "body": { + "stream_options": { "include_usage": true }, + "enable_thinking": false + } +} +``` + +model、消息、系统提示、工具、媒体文件、stream 由宿主管理，冲突拒绝。禁止 body 注入凭据、Header、URL，无模板求值。媒体独立匹配规则，嵌套扩展作为 JSON 文本 multipart 字段，不接收聊天规则。是否支持某扩展由供应商决定。 + +POST /api/providers/request-preview 不联网，隐藏正文/文件且不包含凭据。表单有格式化、校验、删除规则、预览；Provider version 检测保存冲突，适配器冻结配置。原有连接测试只验证模型列表连通性，不等于厂商推理接受扩展字段。 + +## 验证记录 + +2026-09-04，Windows / Python 3.12 / torch 2.9.1+cpu：后端 472 项、前端 93 项测试通过，类型检查和生产构建通过，仍有既有大 bundle 警告。Edge 真实 API 页面、播放器时长/定位、模型与用量卡片无页面异常。 + +真实模型完成音频 → 转写 → 片段声纹 → 笔记 → 语义检索闭环。示例来自固定 ModelScope revision；权重和音频不提交仓库。 + +| 实测 | 结果 | +| --- | --- | +| Bekko 中文小样本 | 384 维；相关相似度 0.495、无关 0.083 | +| Qwen3-ASR 短中文音频 | 加载约 11.1 秒、推理约 6.3 秒、峰值约 5.4 GiB | +| ERes2NetV2 | 同音频 1.000，不同示例说话人 0.090，片段聚类完成 | +| 笔记闭环 | 重复导出同 note_id，语义检索找回同笔记 | + +这是功能冒烟，不是代表性课程语料完整质量评估。CUDA 实机、Granite 对照、逐字强制对齐及重叠语音质量未验证。每任务释放模型有加载成本；长音频准确率、阈值与吞吐需要目标机器专项验收。 + +```powershell +cd backend +.venv/Scripts/python scripts/local-model-smoke.py bekko --download +.venv/Scripts/python scripts/local-model-smoke.py qwen3-asr --download --audio C:/path/to/speech.wav +.venv/Scripts/python scripts/local-model-smoke.py eres2netv2 --download --audio C:/path/to/speech.wav --reference C:/path/to/reference.wav +```