#!/usr/bin/env python # -*- coding: utf-8 -*- """ 模型功能配置加载器 从 model_setting.yaml 加载模型功能配置 提供按功能获取模型配置的接口 使用方式: from foundation.ai.models.model_config_loader import get_model_for_function, get_thinking_mode_for_function model = get_model_for_function("doc_classification_secondary") thinking = get_thinking_mode_for_function("doc_classification_secondary") """ from pathlib import Path from typing import Dict, Any, Optional from dataclasses import dataclass import yaml # 延迟导入 logger,避免循环依赖 _logger = None def _get_logger(): global _logger if _logger is None: try: from foundation.observability.logger.loggering import review_logger as logger _logger = logger except ImportError: import logging _logger = logging.getLogger(__name__) return _logger @dataclass class ModelFunctionConfig: """模型功能配置""" model: str enable_thinking: Optional[bool] = None description: str = "" class ModelConfigLoader: """模型配置加载器(单例)""" _instance: Optional["ModelConfigLoader"] = None _config: Optional[Dict[str, Any]] = None def __new__(cls) -> "ModelConfigLoader": if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._initialized = False return cls._instance def __init__(self): if self._initialized: return self._initialized = True self._load_config() def _get_config_path(self) -> Path: """获取配置文件路径""" # 配置文件位于项目根目录 config/ 下 # 本文件位于 foundation/ai/models/,需要向上3层到项目根目录 return Path(__file__).parent.parent.parent.parent / "config" / "model_setting.yaml" def _load_config(self): """加载 YAML 配置文件""" config_path = self._get_config_path() if not config_path.exists(): _get_logger().warning(f"[ModelConfig] 配置文件不存在: {config_path},使用默认配置") self._config = self._get_default_config() return try: with open(config_path, 'r', encoding='utf-8') as f: self._config = yaml.safe_load(f) _get_logger().info(f"[ModelConfig] 已加载模型配置: {config_path}") except Exception as e: _get_logger().error(f"[ModelConfig] 加载配置文件失败: {e}") self._config = self._get_default_config() def _get_default_config(self) -> Dict[str, Any]: """获取默认配置""" return { "default": { "model": "qwen3_5_35b_a3b", "enable_thinking": False }, "model_settings": {} } def get_model_config(self, function_name: str) -> ModelFunctionConfig: """ 获取指定功能的模型配置 Args: function_name: 功能名称(如 doc_classification_secondary) Returns: ModelFunctionConfig: 模型配置 """ settings = self._config.get("model_settings", {}) default = self._config.get("default", {}) # 获取功能配置,如果不存在则使用默认 func_config = settings.get(function_name, default) # 合并默认值 model = func_config.get("model", default.get("model", "qwen3_5_35b_a3b")) enable_thinking = func_config.get("enable_thinking", default.get("enable_thinking", False)) description = func_config.get("description", "") return ModelFunctionConfig( model=model, enable_thinking=enable_thinking, description=description ) def get_model_name(self, function_name: str) -> str: """获取指定功能的模型名称""" return self.get_model_config(function_name).model def get_enable_thinking(self, function_name: str) -> Optional[bool]: """获取指定功能是否启用思考模式""" return self.get_model_config(function_name).enable_thinking def get_available_models(self) -> list: """获取可用模型列表""" return self._config.get("available_models", []) def list_functions(self) -> Dict[str, str]: """列出所有已配置的功能及其描述""" settings = self._config.get("model_settings", {}) return { name: config.get("description", "无描述") for name, config in settings.items() } # 全局单例 model_config_loader = ModelConfigLoader() # 便捷函数 def get_model_for_function(function_name: str) -> str: """获取指定功能使用的模型名称""" return model_config_loader.get_model_name(function_name) def get_thinking_mode_for_function(function_name: str) -> Optional[bool]: """获取指定功能的思考模式配置""" return model_config_loader.get_enable_thinking(function_name) def get_full_config_for_function(function_name: str) -> ModelFunctionConfig: """获取指定功能的完整配置""" return model_config_loader.get_model_config(function_name)