"""configs/ 目录下 YAML 配置加载器。 目前支持加载 configs/llm_models.yaml 的场景配置(scenes.)。 配置优先级(从高到低): 1. 代码 / 命令行显式参数(如 --provider qwen --model qwen-plus) 2. 本文件 YAML 场景配置(scenes.) 3. 环境变量 / .env(LLM_PROVIDER、DEEPSEEK_MODEL 等,向后兼容) 4. 代码内置默认值 热加载: 缓存以 ``(mtime_ns, size)`` 为准 —— 改完 YAML 保存后,下一次读取即生效, 常驻进程(调度器 / MCP server)无需重启。 说明:API Key 一律放 .env,本文件只保存环境变量名(api_key_env),禁止写密钥。 """ from __future__ import annotations import os import threading from pathlib import Path from loguru import logger #: 指定替代的模型配置文件路径(测试 / 多环境部署用) MODELS_CONFIG_OVERRIDE = "A_SHARE_MODELS_CONFIG" #: 默认模型配置文件(绝对路径,不依赖当前工作目录) DEFAULT_CONFIG_PATH = Path(__file__).resolve().parents[1] / "configs" / "llm_models.yaml" _lock = threading.Lock() _cache: dict[Path, tuple[tuple[int, int] | None, dict]] = {} def config_path() -> Path: """返回当前使用的 ``llm_models.yaml`` 路径。""" override = os.environ.get(MODELS_CONFIG_OVERRIDE) if override: return Path(override).expanduser() return DEFAULT_CONFIG_PATH def _signature(path: Path) -> tuple[int, int] | None: """返回 ``(mtime_ns, size)``;文件不存在时返回 None。""" try: st = path.stat() except OSError: return None return (st.st_mtime_ns, st.st_size) def _load_yaml(path: Path) -> dict: """读取 YAML 为 dict;文件缺失或解析失败返回空 dict(走兜底配置)。 按 ``(mtime_ns, size)`` 失效缓存:文件一旦变化,下次调用即重新解析。 """ sig = _signature(path) with _lock: cached = _cache.get(path) if cached is not None and cached[0] == sig: return cached[1] if not path.is_file(): logger.debug("配置文件不存在,使用内置/环境变量兜底: {}", path) data: dict = {} else: try: import yaml raw = yaml.safe_load(path.read_text(encoding="utf-8")) or {} data = raw if isinstance(raw, dict) else {} except Exception as e: # noqa: BLE001 - YAML 语法错误等 logger.error("解析 {} 失败: {}", path, e) data = {} with _lock: _cache[path] = (sig, data) return data def load_scene_config(scene: str) -> dict: """读取 llm_models.yaml 中 scenes. 的配置 dict。 场景不存在或未配置时返回空 dict(调用方回退 .env / 内置默认)。 scene 为空字符串同样返回空 dict。 """ if not scene: return {} data = _load_yaml(config_path()) scenes = data.get("scenes") or {} cfg = scenes.get(scene) if cfg is None: logger.debug("llm_models.yaml 未配置场景 {!r},使用 .env 兜底", scene) return {} if not isinstance(cfg, dict): logger.warning("llm_models.yaml 场景 {!r} 应为 map,已忽略", scene) return {} return cfg def load_defaults() -> dict: """读取 llm_models.yaml 顶层 defaults(全局默认参数)。""" data = _load_yaml(config_path()) d = data.get("defaults") or {} return d if isinstance(d, dict) else {} def clear_cache() -> None: """清空 YAML 缓存(测试或强制重载时使用;正常热加载无需调用)。""" with _lock: _cache.clear()