Spaces:
Running
Running
| """Provider registry / factory. | |
| Builds the set of available providers from config and detected capabilities. | |
| The router (app/router.py) consumes this registry. | |
| """ | |
| from __future__ import annotations | |
| from ..config import Settings | |
| from .anthropic_provider import AnthropicProvider | |
| from .base import CacheBlock, LLMProvider, LLMRequest, LLMResponse, Usage | |
| from .gemini_provider import GeminiProvider | |
| from .local_provider import LocalProvider | |
| from .mock_provider import MockProvider | |
| __all__ = [ | |
| "CacheBlock", | |
| "LLMProvider", | |
| "LLMRequest", | |
| "LLMResponse", | |
| "Usage", | |
| "ProviderRegistry", | |
| "build_registry", | |
| ] | |
| class ProviderRegistry: | |
| def __init__(self, settings: Settings) -> None: | |
| self.settings = settings | |
| self.mock = MockProvider() | |
| self.anthropic: AnthropicProvider | None = None | |
| self.gemini: GeminiProvider | None = None | |
| self.local: LocalProvider | None = None | |
| if settings.anthropic_api_key: | |
| self.anthropic = AnthropicProvider( | |
| settings.anthropic_api_key, enable_cache=settings.enable_prompt_cache | |
| ) | |
| if settings.gemini_api_key: | |
| self.gemini = GeminiProvider(settings.gemini_api_key, settings.gemini_model) | |
| # Local provider is constructed regardless; .available() decides at call time. | |
| self.local = LocalProvider( | |
| settings.ollama_base_url, settings.local_model, settings.local_backend | |
| ) | |
| # MiniCPM-V as an LLM (text reasoning) — a capable small model. | |
| self.minicpm = None | |
| if settings.minicpm_base_url: | |
| from .minicpm_llm import MiniCPMLLMProvider | |
| self.minicpm = MiniCPMLLMProvider( | |
| settings.minicpm_base_url, settings.minicpm_api_key, settings.minicpm_model) | |
| def capabilities(self) -> dict: | |
| """Report which tiers are live — surfaced at /api/capabilities.""" | |
| anthropic_ok = bool(self.anthropic) and self.anthropic.available() | |
| gemini_ok = bool(self.gemini) and self.gemini.available() | |
| local_ok = bool(self.local) and self.local.available() | |
| minicpm_ok = bool(self.minicpm) and self.minicpm.available() | |
| if anthropic_ok or gemini_ok or minicpm_ok: | |
| active = "hosted" | |
| elif local_ok: | |
| active = "local" | |
| else: | |
| active = "offline" | |
| return { | |
| "active_tier": active, | |
| "providers": { | |
| "mock": {"available": True, "tier": "offline"}, | |
| "anthropic": {"available": anthropic_ok, "tier": "hosted", | |
| "model_smart": self.settings.anthropic_model_smart, | |
| "model_cheap": self.settings.anthropic_model_cheap}, | |
| "gemini": {"available": gemini_ok, "tier": "hosted", | |
| "model": self.settings.gemini_model}, | |
| "local": {"available": local_ok, "tier": "local", | |
| "model": self.settings.local_model, | |
| "backend": self.settings.local_backend}, | |
| "minicpm": {"available": minicpm_ok, "tier": "hosted", | |
| "model": self.settings.minicpm_model}, | |
| }, | |
| "ocr": _ocr_capabilities(), | |
| "browser": {"playwright": _has("playwright")}, | |
| "routing_policy": self.settings.routing_policy, | |
| "prompt_cache": self.settings.enable_prompt_cache, | |
| "semantic_cache": self.settings.enable_semantic_cache, | |
| } | |
| def _has(mod: str) -> bool: | |
| import importlib.util | |
| return importlib.util.find_spec(mod) is not None | |
| def _ocr_capabilities() -> dict: | |
| import shutil | |
| return { | |
| "pymupdf": _has("fitz"), | |
| "pdfplumber": _has("pdfplumber"), | |
| "tesseract": _has("pytesseract") and shutil.which("tesseract") is not None, | |
| "easyocr": _has("easyocr"), | |
| } | |
| def build_registry(settings: Settings) -> ProviderRegistry: | |
| return ProviderRegistry(settings) | |