app.ai_providers.factory

  1"""AI provider factory for creating provider instances from configuration.
  2
  3This module contains the ``create_provider`` function that reads the
  4application ``config.yaml`` and constructs the appropriate provider
  5class based on the configured provider name.
  6
  7"""
  8
  9from __future__ import annotations
 10
 11from typing import Any
 12
 13from .base import (
 14    AIProviderError,
 15    DEFAULT_CLAUDE_MODEL,
 16    DEFAULT_CLOUD_REQUEST_TIMEOUT_SECONDS,
 17    DEFAULT_KIMI_BASE_URL,
 18    DEFAULT_KIMI_MODEL,
 19    DEFAULT_LOCAL_BASE_URL,
 20    DEFAULT_LOCAL_MODEL,
 21    DEFAULT_LOCAL_REQUEST_TIMEOUT_SECONDS,
 22    DEFAULT_OPENAI_MODEL,
 23    AIProvider,
 24    _normalize_api_key_value,
 25    _resolve_api_key,
 26    _resolve_api_key_candidates,
 27    _resolve_timeout_seconds,
 28)
 29
 30def create_provider(config: dict[str, Any]) -> AIProvider:
 31    """Create and return an AI provider instance based on application config.
 32
 33    Reads the ``ai.provider`` key from the configuration dictionary and
 34    constructs the corresponding provider class with settings from the
 35    provider-specific sub-section.
 36
 37    Args:
 38        config: The application configuration dictionary, expected to
 39            contain an ``"ai"`` section with a ``"provider"`` key.
 40
 41    Returns:
 42        A configured ``AIProvider`` instance ready for use.
 43
 44    Raises:
 45        ValueError: If the ``ai`` section is missing or the provider
 46            name is not supported.
 47        AIProviderError: If the selected provider cannot be initialized.
 48    """
 49    ai_config = config.get("ai", {})
 50    if not isinstance(ai_config, dict):
 51        raise ValueError("Invalid configuration: `ai` section must be a dictionary.")
 52
 53    provider_name = str(ai_config.get("provider", "claude")).strip().lower()
 54
 55    if provider_name == "claude":
 56        return _create_claude_provider(ai_config)
 57
 58    if provider_name == "openai":
 59        return _create_openai_provider(ai_config)
 60
 61    if provider_name == "local":
 62        return _create_local_provider(ai_config)
 63
 64    if provider_name == "kimi":
 65        return _create_kimi_provider(ai_config)
 66
 67    raise ValueError(
 68        f"Unsupported AI provider '{provider_name}'. Expected one of: claude, openai, kimi, local."
 69    )
 70
 71
 72def _create_claude_provider(ai_config: dict[str, Any]) -> AIProvider:
 73    """Create a ClaudeProvider from the ``ai.claude`` config section.
 74
 75    Args:
 76        ai_config: The ``ai`` section of the application configuration.
 77
 78    Returns:
 79        A configured ``ClaudeProvider`` instance.
 80    """
 81    from .claude_provider import ClaudeProvider
 82
 83    claude_config = ai_config.get("claude", {})
 84    if not isinstance(claude_config, dict):
 85        raise ValueError("Invalid configuration: `ai.claude` must be a dictionary.")
 86    api_key = _resolve_api_key(
 87        claude_config.get("api_key", ""),
 88        "ANTHROPIC_API_KEY",
 89    )
 90    return ClaudeProvider(
 91        api_key=api_key,
 92        model=str(claude_config.get("model", DEFAULT_CLAUDE_MODEL)),
 93        attach_csv_as_file=bool(claude_config.get("attach_csv_as_file", True)),
 94        request_timeout_seconds=_resolve_timeout_seconds(
 95            claude_config.get("request_timeout_seconds", DEFAULT_CLOUD_REQUEST_TIMEOUT_SECONDS),
 96            DEFAULT_CLOUD_REQUEST_TIMEOUT_SECONDS,
 97        ),
 98    )
 99
100
101def _create_openai_provider(ai_config: dict[str, Any]) -> AIProvider:
102    """Create an OpenAIProvider from the ``ai.openai`` config section.
103
104    Args:
105        ai_config: The ``ai`` section of the application configuration.
106
107    Returns:
108        A configured ``OpenAIProvider`` instance.
109    """
110    from .openai_provider import OpenAIProvider
111
112    openai_config = ai_config.get("openai", {})
113    if not isinstance(openai_config, dict):
114        raise ValueError("Invalid configuration: `ai.openai` must be a dictionary.")
115    api_key = _resolve_api_key(
116        openai_config.get("api_key", ""),
117        "OPENAI_API_KEY",
118    )
119    return OpenAIProvider(
120        api_key=api_key,
121        model=str(openai_config.get("model", DEFAULT_OPENAI_MODEL)),
122        attach_csv_as_file=bool(openai_config.get("attach_csv_as_file", True)),
123        request_timeout_seconds=_resolve_timeout_seconds(
124            openai_config.get("request_timeout_seconds", DEFAULT_CLOUD_REQUEST_TIMEOUT_SECONDS),
125            DEFAULT_CLOUD_REQUEST_TIMEOUT_SECONDS,
126        ),
127    )
128
129
130def _create_local_provider(ai_config: dict[str, Any]) -> AIProvider:
131    """Create a LocalProvider from the ``ai.local`` config section.
132
133    Args:
134        ai_config: The ``ai`` section of the application configuration.
135
136    Returns:
137        A configured ``LocalProvider`` instance.
138    """
139    from .local_provider import LocalProvider
140
141    local_config = ai_config.get("local", {})
142    if not isinstance(local_config, dict):
143        raise ValueError("Invalid configuration: `ai.local` must be a dictionary.")
144    return LocalProvider(
145        base_url=str(local_config.get("base_url", DEFAULT_LOCAL_BASE_URL)),
146        model=str(local_config.get("model", DEFAULT_LOCAL_MODEL)),
147        api_key=_normalize_api_key_value(local_config.get("api_key", "not-needed")) or "not-needed",
148        attach_csv_as_file=bool(local_config.get("attach_csv_as_file", True)),
149        request_timeout_seconds=_resolve_timeout_seconds(
150            local_config.get("request_timeout_seconds", DEFAULT_LOCAL_REQUEST_TIMEOUT_SECONDS),
151            DEFAULT_LOCAL_REQUEST_TIMEOUT_SECONDS,
152        ),
153    )
154
155
156def _create_kimi_provider(ai_config: dict[str, Any]) -> AIProvider:
157    """Create a KimiProvider from the ``ai.kimi`` config section.
158
159    Args:
160        ai_config: The ``ai`` section of the application configuration.
161
162    Returns:
163        A configured ``KimiProvider`` instance.
164    """
165    from .kimi_provider import KimiProvider
166
167    kimi_config = ai_config.get("kimi", {})
168    if not isinstance(kimi_config, dict):
169        raise ValueError("Invalid configuration: `ai.kimi` must be a dictionary.")
170    api_key = _resolve_api_key_candidates(
171        kimi_config.get("api_key", ""),
172        ("MOONSHOT_API_KEY", "KIMI_API_KEY"),
173    )
174    return KimiProvider(
175        api_key=api_key,
176        model=str(kimi_config.get("model", DEFAULT_KIMI_MODEL)),
177        base_url=str(kimi_config.get("base_url", DEFAULT_KIMI_BASE_URL)),
178        attach_csv_as_file=bool(kimi_config.get("attach_csv_as_file", True)),
179        request_timeout_seconds=_resolve_timeout_seconds(
180            kimi_config.get("request_timeout_seconds", DEFAULT_CLOUD_REQUEST_TIMEOUT_SECONDS),
181            DEFAULT_CLOUD_REQUEST_TIMEOUT_SECONDS,
182        ),
183    )