awoooi/apps/api/src/services/model_registry.py

"""
Model Registry - Phase 12 P1 修復
=================================
集中管理 AI 模型配置，消除 hardcode 模型名稱

功能:
- 從 models.json 讀取配置
- 提供 get_model(provider, purpose) 方法
- Singleton 模式
- 支援依賴注入測試

版本: v1.0
建立: 2026-03-26 23:00 (台北時區)
建立者: Claude Code
最後修改: 2026-03-26 23:00 (台北時區)
修改者: Claude Code
"""

import json
from pathlib import Path
from typing import Protocol

import structlog

logger = structlog.get_logger(__name__)


# =============================================================================
# Interface (支援 DI 測試)
# =============================================================================


class IModelRegistry(Protocol):
    """Model Registry Interface for DI"""

    def get_model(self, provider: str, purpose: str = "default") -> str:
        """取得模型名稱"""
        ...

    def get_fallback_order(self) -> list[str]:
        """取得備援順序"""
        ...

    def get_model_by_complexity(self, complexity: int) -> str:
        """依複雜度取得推薦模型"""
        ...

    def get_provider_config(self, provider: str) -> dict:
        """取得 provider 完整配置"""
        ...


# =============================================================================
# Implementation
# =============================================================================


class ModelRegistry:
    """
    Model Registry 實作

    從 models.json 讀取配置，提供統一的模型查詢介面

    Usage:
        registry = get_model_registry()
        model = registry.get_model("ollama", "rca")  # -> "qwen2.5:7b-instruct"
    """

    def __init__(self, config_path: Path | str | None = None):
        """
        初始化 ModelRegistry

        Args:
            config_path: models.json 路徑，None 使用預設路徑
        """
        if config_path is None:
            # 預設路徑: apps/api/models.json
            config_path = Path(__file__).parent.parent.parent / "models.json"
        elif isinstance(config_path, str):
            config_path = Path(config_path)

        self._config_path = config_path
        self._config: dict = {}
        self._load_config()

        # 複雜度對應模型 (從 config 或使用預設)
        self._complexity_map = self._build_complexity_map()

    def _load_config(self) -> None:
        """載入 models.json"""
        try:
            with open(self._config_path) as f:
                self._config = json.load(f)
            logger.info(
                "model_registry_loaded",
                path=str(self._config_path),
                providers=list(self._config.get("providers", {}).keys()),
            )
        except FileNotFoundError:
            logger.warning(
                "models_json_not_found",
                path=str(self._config_path),
                using="fallback_defaults",
            )
            self._config = self._get_default_config()
        except json.JSONDecodeError as e:
            logger.error(
                "models_json_parse_error",
                path=str(self._config_path),
                error=str(e),
            )
            self._config = self._get_default_config()

    def _get_default_config(self) -> dict:
        """預設配置 (fallback)"""
        return {
            "default_provider": "ollama",
            "fallback_order": ["ollama", "gemini", "claude"],
            "providers": {
                "ollama": {
                    "models": {
                        "default": "qwen2.5:7b-instruct",
                        "rca": "qwen2.5:7b-instruct",
                        "summary": "llama3.2:3b",
                    }
                },
                "gemini": {
                    "models": {
                        "default": "gemini-1.5-flash",
                        "rca": "gemini-1.5-flash",
                        "summary": "gemini-1.5-flash",
                    }
                },
                "claude": {
                    "models": {
                        "default": "claude-3-haiku-20240307",
                        "rca": "claude-3-haiku-20240307",
                        "summary": "claude-3-haiku-20240307",
                    }
                },
            },
        }

    def _build_complexity_map(self) -> dict[int, str]:
        """建立複雜度對應模型映射"""
        # 從 config 或使用預設
        ollama_models = self._config.get("providers", {}).get("ollama", {}).get("models", {})
        default_model = ollama_models.get("default", "qwen2.5:7b-instruct")
        summary_model = ollama_models.get("summary", "llama3.2:3b")

        return {
            1: summary_model,  # 簡單任務，快速回應
            2: default_model,  # 中等任務
            3: default_model,  # 複雜任務
            4: "gemini",  # 需要雲端能力
            5: "claude",  # 極複雜，需要最強模型
        }

    def get_model(self, provider: str, purpose: str = "default") -> str:
        """
        取得模型名稱

        Args:
            provider: 提供者 (ollama, gemini, claude)
            purpose: 用途 (default, rca, summary)

        Returns:
            模型名稱
        """
        providers = self._config.get("providers", {})
        provider_config = providers.get(provider, {})
        models = provider_config.get("models", {})

        # 優先取用途，fallback 到 default
        model = models.get(purpose) or models.get("default")

        if not model:
            # 最終 fallback
            fallback_map = {
                "ollama": "qwen2.5:7b-instruct",
                "gemini": "gemini-1.5-flash",
                "claude": "claude-3-haiku-20240307",
            }
            model = fallback_map.get(provider, provider)
            logger.warning(
                "model_not_found_using_fallback",
                provider=provider,
                purpose=purpose,
                fallback=model,
            )

        return model

    def get_fallback_order(self) -> list[str]:
        """取得備援順序"""
        return self._config.get("fallback_order", ["ollama", "gemini", "claude"])

    def get_model_by_complexity(self, complexity: int) -> str:
        """
        依複雜度取得推薦模型

        Args:
            complexity: 複雜度分數 (1-5)

        Returns:
            推薦模型名稱
        """
        # 確保在範圍內
        complexity = max(1, min(5, complexity))
        return self._complexity_map.get(complexity, self.get_model("ollama", "default"))

    def get_provider_config(self, provider: str) -> dict:
        """取得 provider 完整配置"""
        return self._config.get("providers", {}).get(provider, {})

    def get_default_provider(self) -> str:
        """取得預設 provider"""
        return self._config.get("default_provider", "ollama")

    def get_provider_options(self, provider: str) -> dict:
        """取得 provider 的 options"""
        provider_config = self.get_provider_config(provider)
        return provider_config.get("options", {})

    def get_provider_timeout(self, provider: str) -> int:
        """取得 provider 的 timeout (秒)"""
        provider_config = self.get_provider_config(provider)
        return provider_config.get("timeout_seconds", 30)


# =============================================================================
# Singleton
# =============================================================================

_registry: ModelRegistry | None = None


def get_model_registry() -> ModelRegistry:
    """取得 ModelRegistry 單例"""
    global _registry
    if _registry is None:
        _registry = ModelRegistry()
    return _registry


def reset_model_registry() -> None:
    """重置單例 (用於測試)"""
    global _registry
    _registry = None


# =============================================================================
# Convenience Functions
# =============================================================================


def get_model(provider: str, purpose: str = "default") -> str:
    """便捷函數: 取得模型名稱"""
    return get_model_registry().get_model(provider, purpose)


def get_model_by_complexity(complexity: int) -> str:
    """便捷函數: 依複雜度取得模型"""
    return get_model_registry().get_model_by_complexity(complexity)