55 lines
2.2 KiB
Python
55 lines
2.2 KiB
Python
|
"""独立 VLLM 部署项目的模型清单与路径解析辅助函数。"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
|
||
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
||
|
|
MANIFEST_PATH = PROJECT_ROOT / "model_manifest.json"
|
||
|
|
|
||
|
|
|
||
|
|
def load_manifest(path: Path = MANIFEST_PATH) -> dict[str, Any]:
|
||
|
|
"""读取本项目自己的模型清单，整个过程不导入原项目代码。"""
|
||
|
|
with path.open("r", encoding="utf-8") as manifest_file:
|
||
|
|
manifest = json.load(manifest_file)
|
||
|
|
if not isinstance(manifest.get("models"), dict) or not manifest["models"]:
|
||
|
|
raise ValueError("model_manifest.json 必须包含非空的 models 对象")
|
||
|
|
return manifest
|
||
|
|
|
||
|
|
|
||
|
|
def resolve_model_id(model: str | None, manifest: dict[str, Any]) -> str:
|
||
|
|
"""将默认值、短别名或完整模型 ID 解析为一个 ASR 模型。"""
|
||
|
|
models = manifest["models"]
|
||
|
|
requested = (model or "default").strip()
|
||
|
|
if requested == "default":
|
||
|
|
requested = str(manifest["default_model"])
|
||
|
|
|
||
|
|
if requested in models:
|
||
|
|
return requested
|
||
|
|
|
||
|
|
for model_id, config in models.items():
|
||
|
|
if requested.lower() == str(config.get("alias", "")).lower():
|
||
|
|
return model_id
|
||
|
|
raise ValueError(f"不支持的 ASR 模型 '{model}'，可选模型：{', '.join(models)}")
|
||
|
|
|
||
|
|
|
||
|
|
def model_directory(model_id: str, manifest: dict[str, Any], models_dir: Path) -> Path:
|
||
|
|
"""根据清单返回 ASR 或辅助模型实际使用的本地目录。"""
|
||
|
|
config = manifest.get("models", {}).get(model_id)
|
||
|
|
if config is None:
|
||
|
|
config = manifest.get("auxiliary_models", {}).get(model_id)
|
||
|
|
if not isinstance(config, dict) or not config.get("directory"):
|
||
|
|
raise ValueError(f"模型 '{model_id}' 在清单中没有配置本地目录")
|
||
|
|
return models_dir / str(config["directory"])
|
||
|
|
|
||
|
|
|
||
|
|
def auxiliary_models(manifest: dict[str, Any]) -> dict[str, dict[str, Any]]:
|
||
|
|
"""返回可独立部署的 VAD、说话人和对齐模型资产。"""
|
||
|
|
models = manifest.get("auxiliary_models", {})
|
||
|
|
if not isinstance(models, dict):
|
||
|
|
raise ValueError("model_manifest.json 的 auxiliary_models 必须是对象")
|
||
|
|
return {str(model_id): config for model_id, config in models.items() if isinstance(config, dict)}
|