LQAgentPlatform/foundation/ai/models/model_config_loader.py

#!/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)