from __future__ import annotations import logging from typing import Any from app.core.ai.domain.models import AiCapability, AiModelSpec, AiProviderType logger = logging.getLogger("dscons.ai.registry") class CanonicalModelRegistry: """Single Source of Truth (SSOT) for all enterprise AI models across DSCons.""" _MODELS: dict[str, AiModelSpec] = { # GPT OSS 120B — Primary Vision & Text model (unlimited quota via Antigravity SDK) "gpt-oss-120b": AiModelSpec( canonical_id="gpt-oss-120b", provider_type=AiProviderType.ANTIGRAVITY, provider_model_id="gpt-oss-120b", display_name="GPT OSS 120B (Antigravity SDK — Unlimited Quota)", capabilities=[ AiCapability.TEXT_GENERATION, AiCapability.MULTIMODAL_VISION, AiCapability.STRUCTURED_JSON, AiCapability.CODE_EXECUTION, ], context_window=131072, max_output_tokens=8192, fallback_canonical_id="openrouter/gemini-3.8-flash", ), # Google Antigravity Native / Gemini Models (API key revoked — kept as legacy fallback target) "gemini-3.8-flash": AiModelSpec( canonical_id="gemini-3.8-flash", provider_type=AiProviderType.ANTIGRAVITY, provider_model_id="gemini-3.8-flash", display_name="Google Gemini 3.8 Flash (Native Antigravity SDK — KEY REVOKED)", capabilities=[ AiCapability.TEXT_GENERATION, AiCapability.MULTIMODAL_VISION, AiCapability.STRUCTURED_JSON, AiCapability.CODE_EXECUTION, ], context_window=1048576, max_output_tokens=8192, fallback_canonical_id="openrouter/gemini-3.8-flash", ), # OpenRouter Fast & Lightweight Vision Models (Ideal for Captcha & High-Throughput OCR) "openrouter/gemini-2.5-flash": AiModelSpec( canonical_id="openrouter/gemini-2.5-flash", provider_type=AiProviderType.OPENROUTER, provider_model_id="google/gemini-2.5-flash", display_name="Google Gemini 2.5 Flash (OpenRouter HA Cloud — Fast Captcha Vision)", capabilities=[ AiCapability.TEXT_GENERATION, AiCapability.MULTIMODAL_VISION, AiCapability.STRUCTURED_JSON, ], context_window=1048576, max_output_tokens=8192, fallback_canonical_id="openrouter/gemini-3.8-flash", ), # Người đọc captcha độc lập thứ 2 (dùng cho cơ chế đồng thuận đa model ở gdt_captcha.py). # CỐ Ý không có fallback: nếu model này lỗi mà gateway âm thầm chuyển sang model khác # thì 2 "người đọc" có thể trùng 1 model -> đồng thuận giả. "openrouter/gemini-2.5-flash-lite": AiModelSpec( canonical_id="openrouter/gemini-2.5-flash-lite", provider_type=AiProviderType.OPENROUTER, provider_model_id="google/gemini-2.5-flash-lite", display_name="Google Gemini 2.5 Flash Lite (OpenRouter — Captcha Consensus Reader)", capabilities=[ AiCapability.TEXT_GENERATION, AiCapability.MULTIMODAL_VISION, ], context_window=1048576, max_output_tokens=8192, fallback_canonical_id=None, ), "openrouter/gemini-3.8-flash": AiModelSpec( canonical_id="openrouter/gemini-3.8-flash", provider_type=AiProviderType.OPENROUTER, provider_model_id="google/gemini-3.8-flash", display_name="Google Gemini 3.8 Flash (OpenRouter HA Cloud)", capabilities=[ AiCapability.TEXT_GENERATION, AiCapability.MULTIMODAL_VISION, AiCapability.STRUCTURED_JSON, ], context_window=1048576, max_output_tokens=8192, fallback_canonical_id="openrouter/gemini-3.7-flash", ), "openrouter/gemini-3.7-flash": AiModelSpec( canonical_id="openrouter/gemini-3.7-flash", provider_type=AiProviderType.OPENROUTER, provider_model_id="google/gemini-3.7-flash", display_name="Google Gemini 3.7 Flash (OpenRouter HA Cloud)", capabilities=[ AiCapability.TEXT_GENERATION, AiCapability.MULTIMODAL_VISION, AiCapability.STRUCTURED_JSON, ], context_window=1048576, max_output_tokens=8192, fallback_canonical_id="openrouter/claude-opus-4.8", ), "openrouter/claude-opus-4.8": AiModelSpec( canonical_id="openrouter/claude-opus-4.8", provider_type=AiProviderType.OPENROUTER, provider_model_id="anthropic/claude-opus-4.8", display_name="Claude Opus 4.8 (High Reasoning Fallback)", capabilities=[ AiCapability.TEXT_GENERATION, AiCapability.MULTIMODAL_VISION, AiCapability.STRUCTURED_JSON, ], context_window=200000, max_output_tokens=4096, fallback_canonical_id=None, ), "deepseek-v4-flash": AiModelSpec( canonical_id="deepseek-v4-flash", provider_type=AiProviderType.OPENROUTER, provider_model_id="deepseek/deepseek-v4-flash", display_name="DeepSeek V4 Flash (Low Cost High Throughput)", capabilities=[ AiCapability.TEXT_GENERATION, AiCapability.STRUCTURED_JSON, ], context_window=65536, max_output_tokens=4096, fallback_canonical_id="openrouter/gemini-2.5-flash", ), "deepseek-v4-pro": AiModelSpec( canonical_id="deepseek-v4-pro", provider_type=AiProviderType.OPENROUTER, provider_model_id="deepseek/deepseek-v4-pro", display_name="DeepSeek V4 Pro (Complex Accounting & Math)", capabilities=[ AiCapability.TEXT_GENERATION, AiCapability.STRUCTURED_JSON, ], context_window=65536, max_output_tokens=8192, fallback_canonical_id="openrouter/gemini-2.5-flash", ), "lfm-2.5-2.6b": AiModelSpec( canonical_id="lfm-2.5-2.6b", provider_type=AiProviderType.LMSTUDIO, provider_model_id="lfm-2.5-2.6b", display_name="Local LM Studio (Offline Zero-Latency)", capabilities=[ AiCapability.TEXT_GENERATION, AiCapability.STRUCTURED_JSON, ], context_window=32768, max_output_tokens=2048, fallback_canonical_id="openrouter/gemini-2.5-flash", ), } # Aliases mapping deprecated/legacy IDs to modern canonical specs _ALIASES: dict[str, str] = { "gemini-2.0-flash": "openrouter/gemini-2.5-flash", "gemini-2.0-flash-001": "openrouter/gemini-2.5-flash", "openrouter/gemini-2.0-flash": "openrouter/gemini-2.5-flash", "google/gemini-2.0-flash": "openrouter/gemini-2.5-flash", "gemini-2.5-flash": "openrouter/gemini-2.5-flash", "google/gemini-2.5-flash": "openrouter/gemini-2.5-flash", "gemini-3.1-pro": "gemini-3.8-flash", "gemini-3.1-pro-preview": "gemini-3.8-flash", "gemini-3.6-flash": "gemini-3.8-flash", "google/gemini-2.0-flash-exp:free": "openrouter/gemini-3.8-flash", "google/gemini-3.8-flash": "openrouter/gemini-3.8-flash", "google/gemini-3.7-flash": "openrouter/gemini-3.7-flash", "google/gemini-3.1-pro": "openrouter/gemini-3.8-flash", "stealth/ox-alpha": "gemini-3.8-flash", "claude-opus-4.8": "openrouter/claude-opus-4.8", "anthropic/claude-opus-4.8": "openrouter/claude-opus-4.8", "claude-fable-5": "openrouter/claude-opus-4.8", "anthropic/claude-fable-5": "openrouter/claude-opus-4.8", "deepseek/deepseek-v4-flash": "deepseek-v4-flash", "deepseek/deepseek-v4-pro": "deepseek-v4-pro", "liquid/lfm-2.5-2.6b": "lfm-2.5-2.6b", "lfm2.5-2.6b": "lfm-2.5-2.6b", } @classmethod def resolve_spec(cls, identifier: str | None) -> AiModelSpec: """Resolve any model string, alias, or legacy ID into a strictly validated AiModelSpec.""" if not identifier: return cls._MODELS["openrouter/gemini-2.5-flash"] clean_id = identifier.strip().lower() # Direct canonical hit if clean_id in cls._MODELS: return cls._MODELS[clean_id] # Alias hit if clean_id in cls._ALIASES: target = cls._ALIASES[clean_id] logger.debug("Resolved AI model alias '%s' -> '%s'", clean_id, target) return cls._MODELS[target] # Fuzzy match for Google OpenRouter models if clean_id.startswith("google/"): sub = clean_id.replace("google/", "") if sub in cls._MODELS: return cls._MODELS[sub] if f"openrouter/{sub}" in cls._MODELS: return cls._MODELS[f"openrouter/{sub}"] # Dynamic Plug-and-Play for third-party models provided via OpenRouter (e.g. qwen/..., meta-llama/...) if "/" in clean_id: logger.info("Dynamic Plug-and-Play AI spec created for '%s' via OpenRouter.", clean_id) return AiModelSpec( canonical_id=f"openrouter/{clean_id}", provider_type=AiProviderType.OPENROUTER, provider_model_id=clean_id, display_name=f"Plug-and-Play Model ({clean_id})", capabilities=[AiCapability.TEXT_GENERATION, AiCapability.STRUCTURED_JSON], fallback_canonical_id="openrouter/gemini-2.5-flash", ) # Fallback default logger.warning( "Unregistered AI model '%s'. Defaulting to canonical 'openrouter/gemini-2.5-flash'.", identifier, ) return cls._MODELS["openrouter/gemini-2.5-flash"] @classmethod def get_fallback_spec(cls, spec: AiModelSpec) -> AiModelSpec | None: """Get the next resilient fallback spec in the chain.""" if not spec.fallback_canonical_id: return None return cls.resolve_spec(spec.fallback_canonical_id)