from __future__ import annotations from pathlib import Path from typing import Any, Dict, Optional import yaml class LLMRegistry: """从项目根 config.yaml 读取集中式 LLM 后端配置。""" _cache: Dict[str, Any] = {"mtime": None, "data": {}} @classmethod def get_root_config_path(cls) -> Path: return Path(__file__).resolve().parents[2] / "config.yaml" @classmethod def load_root_config(cls) -> Dict[str, Any]: path = cls.get_root_config_path() if not path.exists(): return {} stat = path.stat() if cls._cache["mtime"] == stat.st_mtime and cls._cache["data"]: return cls._cache["data"] with open(path, "r", encoding="utf-8") as fp: data = yaml.safe_load(fp) or {} cls._cache = {"mtime": stat.st_mtime, "data": data} return data @classmethod def get_llm_config(cls) -> Dict[str, Any]: config = cls.load_root_config() llm_config = config.get("llm", {}) or {} return llm_config if isinstance(llm_config, dict) else {} @classmethod def get_backend(cls, backend_name: str) -> Dict[str, Any]: if not backend_name: return {} llm_config = cls.get_llm_config() backends = llm_config.get("backends", {}) or {} backend = backends.get(backend_name, {}) or {} return dict(backend) if isinstance(backend, dict) else {} @classmethod def resolve(cls, local_config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: local = dict(local_config or {}) backend_name = ( local.get("backend") or local.get("backend_name") or local.get("backend_ref") or "" ) if not backend_name: return local merged = cls.get_backend(str(backend_name).strip()) merged.update(local) merged["backend"] = backend_name return merged