refactor: centralize llm backend configuration
This commit is contained in:
64
utils/ai/llm_registry.py
Normal file
64
utils/ai/llm_registry.py
Normal file
@@ -0,0 +1,64 @@
|
||||
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
|
||||
|
||||
Reference in New Issue
Block a user