"""Model Intelligence — discovery, validation, and classification of models.""" from __future__ import annotations import logging import time from dataclasses import dataclass from enum import Enum from typing import Any from cyberbrain26.model_layer.model_registry import ( ModelFreeStatus, ModelRecord, ModelRegistry, ModelStatus, ModelTrust, ModelType, ) from cyberbrain26.model_layer.provider_registry import ( ProviderRecord, ProviderRegistry, ProviderTrust, ) logger = logging.getLogger(__name__) class DiscoveryCategory(str, Enum): """Model category for intelligence classification.""" FREE = "free" NEW = "new" FREE_FOR_TEST = "free_for_test" FREE_FOR_STUDY = "free_for_study" @dataclass class ModelDiscoveryConfig: """Configuration for model discovery/harvesting.""" providers: list[str] | None = None check_interval_hours: float = 24.0 auto_classify: bool = True auto_register: bool = True validate_connectivity: bool = True validate_functional: bool = True categories: list[DiscoveryCategory] | None = None class ModelIntelligence: """Model Intelligence — discovery, validation, and classification. Manages the full model lifecycle from discovery through registry registration. Implements the validation pipeline: DISCOVERY → NORMALIZATION → DEDUPLICATION → METADATA VALIDATION → CONNECTIVITY TEST → FUNCTIONAL TEST → CLASSIFICATION → REGISTRY UPDATE """ def __init__( self, model_registry: ModelRegistry | None = None, provider_registry: ProviderRegistry | None = None, config: ModelDiscoveryConfig | None = None, ) -> None: self.model_registry = model_registry or ModelRegistry() self.provider_registry = provider_registry or ProviderRegistry() self.config = config or ModelDiscoveryConfig() self._discovered: list[str] = [] self._validation_log: list[dict[str, Any]] = [] # ------------------------------------------------------------------ # Discovery # ------------------------------------------------------------------ def discover_from_providers(self) -> list[ModelRecord]: """Discover models from all registered providers.""" discovered: list[ModelRecord] = [] providers = self.provider_registry.list_active() if self.config.providers: providers = [p for p in providers if p.provider in self.config.providers] for provider in providers: models = self._discover_provider_models(provider) discovered.extend(models) logger.info("Discovered %d models from %d providers", len(discovered), len(providers)) return discovered def _discover_provider_models(self, provider: ProviderRecord) -> list[ModelRecord]: """Discover models from a single provider.""" models: list[ModelRecord] = [] for model_id in provider.available_models: record = ModelRecord( model_id=model_id, provider=provider.provider, api_available=provider.api_available, local_available=provider.local_availability, source=provider.source, trust=self._map_provider_trust_to_model_trust(provider.trust), last_checked=time.time(), ) models.append(record) return models @staticmethod def _map_provider_trust_to_model_trust(trust: ProviderTrust) -> ModelTrust: """Map ProviderTrust enum to ModelTrust enum.""" mapping = { ProviderTrust.OFFICIAL: ModelTrust.OFFICIAL, ProviderTrust.VERIFIED: ModelTrust.VERIFIED, ProviderTrust.COMMUNITY: ModelTrust.COMMUNITY, ProviderTrust.UNOFFICIAL: ModelTrust.UNOFFICIAL, ProviderTrust.UNKNOWN: ModelTrust.UNKNOWN, } return mapping.get(trust, ModelTrust.UNKNOWN) def discover_from_config(self, config: dict[str, Any]) -> list[ModelRecord]: """Discover models from configuration dict.""" models: list[ModelRecord] = [] for m in config.get("models", []): record = self._normalize_model_config(m) if record: models.append(record) return models def _normalize_model_config(self, config: dict[str, Any]) -> ModelRecord | None: """Normalize a model config dict to ModelRecord.""" try: model_id = str(config.get("model_id", "")) if not model_id: return None return ModelRecord( model_id=model_id, provider=str(config.get("provider", "unknown")), version=str(config.get("version", "")), model_type=ModelType(config.get("model_type", "chat")), context_window=config.get("context_window"), local_available=config.get("local_available", False), api_available=config.get("api_available", True), free_status=config.get("free_status", "unknown"), ) except Exception as exc: # noqa: BLE001 logger.warning("Failed to normalize model config: %s", exc) return None # ------------------------------------------------------------------ # Normalization # ------------------------------------------------------------------ def normalize(self, model: ModelRecord) -> ModelRecord: """Normalize model metadata to standard format.""" model.model_id = model.model_id.lower().strip() aliases: dict[str, str] = { "google": "gemini", "gpt": "openai", "azure-openai": "azure", "local-ai": "litellm", } if model.provider.lower() in aliases: model.provider = aliases[model.provider.lower()] if model.model_type == ModelType.CHAT and not model.model_id: pass # already CHAT default if model.trust == ModelTrust.UNKNOWN: model.trust = ModelTrust.COMMUNITY if model.status == ModelStatus.UNKNOWN: model.status = ModelStatus.ACTIVE return model # ------------------------------------------------------------------ # Deduplication # ------------------------------------------------------------------ def deduplicate(self, models: list[ModelRecord]) -> list[ModelRecord]: """Deduplicate models by model_id, keeping the most recent.""" seen: dict[str, ModelRecord] = {} for model in models: key = model.model_id if key not in seen or model.last_checked > seen[key].last_checked: seen[key] = model return list(seen.values()) # ------------------------------------------------------------------ # Validation # ------------------------------------------------------------------ def validate_metadata(self, model: ModelRecord) -> dict[str, Any]: """Validate model metadata.""" issues: list[str] = [] if not model.model_id: issues.append("model_id is empty") if not model.provider: issues.append("provider is empty") if model.context_window is not None and model.context_window <= 0: issues.append(f"context_window must be positive, got {model.context_window}") if model.max_tokens is not None and model.max_tokens <= 0: issues.append(f"max_tokens must be positive, got {model.max_tokens}") result = { "model_id": model.model_id, "valid": len(issues) == 0, "issues": issues, "timestamp": time.time(), } self._validation_log.append(result) return result def test_connectivity(self, model: ModelRecord) -> dict[str, Any]: """Test connectivity to model endpoint.""" provider = self.provider_registry.get(model.provider) if provider is None: return { "model_id": model.model_id, "reachable": False, "error": "provider not found", } reachable = provider.api_availability or model.local_available result = { "model_id": model.model_id, "reachable": reachable, "provider_status": provider.status.value, "local": model.local_available, "api_available": provider.api_available, } self._validation_log.append(result) return result def functional_test(self, model: ModelRecord) -> dict[str, Any]: """Run basic functional test on model.""" result: dict[str, Any] = { "model_id": model.model_id, "tested": False, "passed": False, } if not model.api_available and not model.local_available: result["error"] = "model not reachable" self._validation_log.append(result) return result registered = self.model_registry.has(model.model_id) result["tested"] = True result["passed"] = registered and model.status == ModelStatus.ACTIVE result["registered"] = registered self._validation_log.append(result) return result # ------------------------------------------------------------------ # Classification # ------------------------------------------------------------------ def classify(self, model: ModelRecord) -> DiscoveryCategory: """Classify model into a discovery category.""" if model.free_status == ModelFreeStatus.FREE and not model.card_required: return DiscoveryCategory.FREE elif model.free_status == ModelFreeStatus.FREE_FOR_TEST: return DiscoveryCategory.FREE_FOR_TEST elif model.free_status == ModelFreeStatus.FREE_FOR_STUDY: return DiscoveryCategory.FREE_FOR_STUDY else: return DiscoveryCategory.NEW def score_model(self, model: ModelRecord) -> float: """Score model suitability (0.0 to 1.0).""" score = 0.0 if model.free_status == ModelFreeStatus.FREE: score += 25 elif model.free_status == ModelFreeStatus.FREE_FOR_TEST: score += 15 elif model.free_status == ModelFreeStatus.FREE_FOR_STUDY: score += 10 if model.trust == ModelTrust.OFFICIAL: score += 25 elif model.trust == ModelTrust.VERIFIED: score += 20 elif model.trust == ModelTrust.COMMUNITY: score += 10 if model.context_window is not None: if model.context_window >= 128000: score += 20 elif model.context_window >= 32000: score += 15 elif model.context_window >= 8000: score += 10 elif model.context_window >= 2000: score += 5 caps = model.capabilities cap_score = sum([ caps.vision, caps.tool_calling, caps.structured_output, caps.reasoning, caps.coding, caps.multimodal, ]) score += min(cap_score * 3.33, 20) if model.status == ModelStatus.ACTIVE: score += 10 elif model.status == ModelStatus.DEGRADED: score += 5 return min(score, 100.0) / 100.0 # ------------------------------------------------------------------ # Full pipeline # ------------------------------------------------------------------ def run_pipeline(self, models: list[ModelRecord] | None = None) -> list[dict[str, Any]]: """Run the full validation pipeline.""" if models is None: models = self.discover_from_providers() results: list[dict[str, Any]] = [] for model in models: step_result: dict[str, Any] = {"model_id": model.model_id} model = self.normalize(model) step_result["normalized"] = True if self.config.auto_register: self.model_registry.register(model) step_result["registered"] = True validation = self.validate_metadata(model) step_result["metadata_valid"] = validation["valid"] if self.config.validate_connectivity: connectivity = self.test_connectivity(model) step_result["reachable"] = connectivity["reachable"] if self.config.validate_functional: functional = self.functional_test(model) step_result["functional_pass"] = functional.get("passed", False) if self.config.auto_classify: category = self.classify(model) step_result["category"] = category.value step_result["score"] = self.score_model(model) results.append(step_result) logger.info("Pipeline step complete for %s", model.model_id) return results def get_validation_log(self) -> list[dict[str, Any]]: return list(self._validation_log)