""" Provider Detector for MoE Constitutional Court (ADR-073).
Detects available LLM providers and configures the MoE system to operate in single-provider, dual-provider, or multi-provider mode based on available API keys and customer configuration.
Features:
- Auto-detect available providers from environment variables
- Configure provider mode (single, dual, multi)
- Select appropriate models per persona based on available providers
- Calculate confidence adjustments for reduced diversity
- Support for environment variable overrides
Configuration Priority:
- CODITECT_PROVIDER_MODE environment variable
- Auto-detection from available API keys
- Fallback to single-provider mode if uncertain
Usage: detector = ProviderDetector() mode = detector.detect_mode() model = detector.get_model_for_persona("technical_architect") confidence = detector.adjust_confidence(0.85) """
import json import os from dataclasses import dataclass, field from datetime import datetime, timezone from enum import Enum from pathlib import Path from typing import Dict, List, Optional, Any, Set, Tuple
class ProviderMode(str, Enum): """Provider availability modes.""" SINGLE = "single" DUAL = "dual" MULTI = "multi" AUTO = "auto"
class Provider(str, Enum): """Supported LLM providers.""" ANTHROPIC = "anthropic" OPENAI = "openai" DEEPSEEK = "deepseek" GOOGLE = "google" ALIBABA = "alibaba" META = "meta" MINIMAX = "minimax"
class ModelTier(str, Enum): """Model capability tiers for single-provider diversity.""" FLAGSHIP = "flagship" BALANCED = "balanced" FAST = "fast" REASONING = "reasoning"
@dataclass class ProviderConfig: """Configuration for a single provider.""" provider: Provider api_key_env: str models: List[str] tier: str single_provider_capable: bool hosting_options: List[str] = field(default_factory=list) notes: str = ""
@property
def is_available(self) -> bool:
"""Check if provider is available via environment API key."""
api_key = os.getenv(self.api_key_env)
return api_key is not None and len(api_key) > 0
@dataclass class ModeConfig: """Configuration for a provider mode.""" mode: ProviderMode min_model_families: int diversity_strategy: str confidence_adjustment: float description: str
@dataclass class ProviderDetectionResult: """Result of provider detection.""" mode: ProviderMode available_providers: List[Provider] provider_count: int confidence_adjustment: float diversity_strategy: str detection_timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) override_active: bool = False override_source: Optional[str] = None
def to_dict(self) -> Dict[str, Any]:
"""Convert to dictionary for serialization."""
return {
"mode": self.mode.value,
"available_providers": [p.value for p in self.available_providers],
"provider_count": self.provider_count,
"confidence_adjustment": self.confidence_adjustment,
"diversity_strategy": self.diversity_strategy,
"detection_timestamp": self.detection_timestamp.isoformat(),
"override_active": self.override_active,
"override_source": self.override_source
}
class ProviderDetector: """ Detects available LLM providers and configures MoE operation mode.
Implements ADR-073: MoE Provider Flexibility for single to multi-provider modes.
"""
DEFAULT_ROUTING_PATH = Path(__file__).parent.parent.parent.parent / "config" / "judge-model-routing.json"
# Provider configuration (from judge-model-routing.json v2.0.0)
PROVIDER_CONFIGS: Dict[Provider, ProviderConfig] = {
Provider.ANTHROPIC: ProviderConfig(
provider=Provider.ANTHROPIC,
api_key_env="ANTHROPIC_API_KEY",
models=["claude-opus-4-5", "claude-sonnet-4-5", "claude-sonnet-4", "claude-haiku-4-5"],
tier="flagship",
single_provider_capable=True,
notes="Full model family supports all judge persona requirements"
),
Provider.OPENAI: ProviderConfig(
provider=Provider.OPENAI,
api_key_env="OPENAI_API_KEY",
models=["gpt-4.1", "gpt-4.1-mini", "gpt-4o", "o3", "o4-mini"],
tier="flagship",
single_provider_capable=True,
notes="o3/o4-mini for reasoning, gpt-4.1 for general analysis"
),
Provider.DEEPSEEK: ProviderConfig(
provider=Provider.DEEPSEEK,
api_key_env="DEEPSEEK_API_KEY",
models=["deepseek-v3.2", "deepseek-v3.1", "deepseek-chat", "deepseek-reasoner"],
tier="cost_effective",
single_provider_capable=True,
notes="V3.2 supports thinking mode; most cost-effective option"
),
Provider.GOOGLE: ProviderConfig(
provider=Provider.GOOGLE,
api_key_env="GOOGLE_API_KEY",
models=["gemini-3-pro", "gemini-3-flash", "gemini-2.0-flash"],
tier="flagship",
single_provider_capable=True,
notes="Gemini 3 Pro for reasoning-first agentic workflows"
),
Provider.ALIBABA: ProviderConfig(
provider=Provider.ALIBABA,
api_key_env="DASHSCOPE_API_KEY",
models=["qwen2.5-72b", "qwen2.5-32b", "qwen-max"],
tier="specialized",
single_provider_capable=False,
notes="Specialized for multilingual and domain expertise"
),
Provider.META: ProviderConfig(
provider=Provider.META,
api_key_env="TOGETHER_API_KEY",
models=["llama-4-maverick", "llama-4-scout", "llama-3.3-70b"],
tier="open_source",
single_provider_capable=True,
hosting_options=["together_ai", "groq", "fireworks", "self_hosted"],
notes="Llama 4 Maverick beats GPT-4o; Scout fits on single H100"
),
Provider.MINIMAX: ProviderConfig(
provider=Provider.MINIMAX,
api_key_env="MINIMAX_API_KEY",
models=["MiniMax-M2.5", "MiniMax-M2.5-highspeed", "MiniMax-M2.1", "MiniMax-M2.1-highspeed", "MiniMax-M2"],
tier="cost_effective",
single_provider_capable=True,
notes="Anthropic Messages API-compatible; text-only (no vision); 204K context (ADR-200)"
),
}
# Mode configurations (from ADR-073)
MODE_CONFIGS: Dict[ProviderMode, ModeConfig] = {
ProviderMode.SINGLE: ModeConfig(
mode=ProviderMode.SINGLE,
min_model_families=1,
diversity_strategy="model_tier_diversity",
confidence_adjustment=-0.10,
description="Single provider with model tier diversity"
),
ProviderMode.DUAL: ModeConfig(
mode=ProviderMode.DUAL,
min_model_families=2,
diversity_strategy="provider_alternation",
confidence_adjustment=-0.05,
description="Two providers with alternation strategy"
),
ProviderMode.MULTI: ModeConfig(
mode=ProviderMode.MULTI,
min_model_families=3,
diversity_strategy="full_diversity",
confidence_adjustment=0.0,
description="Full Constitutional Court with 3+ providers"
),
}
# Model tier mappings for single-provider mode
SINGLE_PROVIDER_TIERS: Dict[Provider, Dict[ModelTier, str]] = {
Provider.ANTHROPIC: {
ModelTier.FLAGSHIP: "claude-opus-4-5",
ModelTier.BALANCED: "claude-sonnet-4-5",
ModelTier.FAST: "claude-haiku-4-5",
ModelTier.REASONING: "claude-opus-4-5",
},
Provider.OPENAI: {
ModelTier.FLAGSHIP: "o3",
ModelTier.BALANCED: "gpt-4.1",
ModelTier.FAST: "gpt-4.1-mini",
ModelTier.REASONING: "o3",
},
Provider.DEEPSEEK: {
ModelTier.FLAGSHIP: "deepseek-reasoner",
ModelTier.BALANCED: "deepseek-v3.2",
ModelTier.FAST: "deepseek-chat",
ModelTier.REASONING: "deepseek-reasoner",
},
Provider.GOOGLE: {
ModelTier.FLAGSHIP: "gemini-3-pro",
ModelTier.BALANCED: "gemini-3-flash",
ModelTier.FAST: "gemini-2.0-flash",
ModelTier.REASONING: "gemini-3-pro",
},
Provider.META: {
ModelTier.FLAGSHIP: "llama-4-maverick",
ModelTier.BALANCED: "llama-4-scout",
ModelTier.FAST: "llama-3.3-70b",
ModelTier.REASONING: "llama-4-maverick",
},
Provider.MINIMAX: {
ModelTier.FLAGSHIP: "MiniMax-M2.5",
ModelTier.BALANCED: "MiniMax-M2.1",
ModelTier.FAST: "MiniMax-M2.1-highspeed",
ModelTier.REASONING: "MiniMax-M2.5",
},
}
# Persona to tier mapping (for single-provider mode)
PERSONA_TIER_MAPPING: Dict[str, ModelTier] = {
# High-stakes personas -> flagship
"ai_risk_officer": ModelTier.FLAGSHIP,
"compliance_auditor": ModelTier.FLAGSHIP,
"privacy_dpo_specialist": ModelTier.FLAGSHIP,
# Technical personas -> balanced
"technical_architect": ModelTier.BALANCED,
"security_analyst": ModelTier.BALANCED,
"ai_ethics_reviewer": ModelTier.BALANCED,
"data_governance_specialist": ModelTier.BALANCED,
"domain_expert_healthcare": ModelTier.BALANCED,
"domain_expert_finance": ModelTier.BALANCED,
# Fast screening -> fast
"qa_evaluator": ModelTier.FAST,
}
# Recommended dual-provider pairs (from ADR-073)
RECOMMENDED_PAIRS: List[Tuple[Provider, Provider, str]] = [
(Provider.ANTHROPIC, Provider.OPENAI, "Best general coverage"),
(Provider.ANTHROPIC, Provider.DEEPSEEK, "Cost-optimized"),
(Provider.OPENAI, Provider.GOOGLE, "Enterprise focus"),
(Provider.DEEPSEEK, Provider.META, "Open-source friendly"),
(Provider.GOOGLE, Provider.ANTHROPIC, "Multimodal + reasoning"),
]
def __init__(self, routing_config_path: Optional[Path] = None):
"""
Initialize the provider detector.
Args:
routing_config_path: Optional path to judge-model-routing.json
"""
self.routing_config_path = routing_config_path or self.DEFAULT_ROUTING_PATH
self._routing_config: Optional[Dict] = None
self._detection_result: Optional[ProviderDetectionResult] = None
self._load_routing_config()
def _load_routing_config(self) -> None:
"""Load the model routing configuration."""
if self.routing_config_path.exists():
with open(self.routing_config_path, 'r', encoding='utf-8') as f:
self._routing_config = json.load(f)
else:
self._routing_config = {}
def detect_available_providers(self) -> List[Provider]:
"""
Detect which providers have valid API keys.
Returns:
List of available providers, sorted by tier priority
"""
available = []
for provider, config in self.PROVIDER_CONFIGS.items():
if config.is_available:
available.append(provider)
# Sort by tier priority: flagship > cost_effective > specialized > open_source
tier_order = {"flagship": 0, "cost_effective": 1, "specialized": 2, "open_source": 3}
available.sort(key=lambda p: tier_order.get(self.PROVIDER_CONFIGS[p].tier, 99))
return available
def detect_mode(self, force_mode: Optional[ProviderMode] = None) -> ProviderDetectionResult:
"""
Detect the appropriate provider mode based on available providers.
Args:
force_mode: Optional mode override (ignores auto-detection)
Returns:
ProviderDetectionResult with mode and configuration
"""
# Check for environment override
env_mode = os.getenv("CODITECT_PROVIDER_MODE")
override_active = False
override_source = None
if force_mode:
mode = force_mode
override_active = True
override_source = "parameter"
elif env_mode:
try:
mode = ProviderMode(env_mode.lower())
override_active = True
override_source = "CODITECT_PROVIDER_MODE"
except ValueError:
mode = ProviderMode.AUTO
else:
mode = ProviderMode.AUTO
# Detect available providers
available_providers = self.detect_available_providers()
provider_count = len(available_providers)
# Determine actual mode if AUTO
if mode == ProviderMode.AUTO:
if provider_count >= 3:
mode = ProviderMode.MULTI
elif provider_count == 2:
mode = ProviderMode.DUAL
else:
mode = ProviderMode.SINGLE
# Get mode configuration
mode_config = self.MODE_CONFIGS.get(mode, self.MODE_CONFIGS[ProviderMode.SINGLE])
self._detection_result = ProviderDetectionResult(
mode=mode,
available_providers=available_providers,
provider_count=provider_count,
confidence_adjustment=mode_config.confidence_adjustment,
diversity_strategy=mode_config.diversity_strategy,
override_active=override_active,
override_source=override_source
)
return self._detection_result
def get_model_for_persona(
self,
persona_id: str,
override_model: Optional[str] = None,
override_provider: Optional[Provider] = None
) -> str:
"""
Get the appropriate model for a persona based on provider mode.
Priority:
1. override_model parameter
2. Environment variable: CODITECT_JUDGE_MODEL_<PERSONA_ID>
3. Mode-appropriate selection from routing config
Args:
persona_id: The judge persona ID
override_model: Optional explicit model override
override_provider: Optional provider override for single-provider mode
Returns:
Model identifier string
"""
# Check explicit override
if override_model:
return override_model
# Check environment override
env_key = f"CODITECT_JUDGE_MODEL_{persona_id.upper()}"
env_model = os.getenv(env_key)
if env_model:
return env_model
# Ensure detection has run
if self._detection_result is None:
self.detect_mode()
mode = self._detection_result.mode
available_providers = self._detection_result.available_providers
# Get routing configuration
routing = self._routing_config.get("routing", {}).get(persona_id, {})
if mode == ProviderMode.SINGLE:
return self._get_single_provider_model(persona_id, available_providers, override_provider)
elif mode == ProviderMode.DUAL:
return self._get_dual_provider_model(persona_id, available_providers, routing)
else:
# Multi-provider: use primary model from routing
return routing.get("primary_model", self._get_fallback_model(persona_id, available_providers))
def _get_single_provider_model(
self,
persona_id: str,
available_providers: List[Provider],
override_provider: Optional[Provider] = None
) -> str:
"""
Get model for single-provider mode using tier diversity.
Args:
persona_id: The judge persona ID
available_providers: List of available providers
override_provider: Optional provider override
Returns:
Model identifier for the single available provider
"""
# Determine provider
if override_provider and override_provider in available_providers:
provider = override_provider
elif available_providers:
# Use first available single-provider-capable provider
for p in available_providers:
if self.PROVIDER_CONFIGS[p].single_provider_capable:
provider = p
break
else:
provider = available_providers[0]
else:
# No providers available, return placeholder
return "no-provider-available"
# Check routing config for single_provider_fallback
routing = self._routing_config.get("routing", {}).get(persona_id, {})
fallbacks = routing.get("single_provider_fallback", {})
if provider.value in fallbacks:
return fallbacks[provider.value]
# Use tier mapping
tier = self.PERSONA_TIER_MAPPING.get(persona_id, ModelTier.BALANCED)
provider_tiers = self.SINGLE_PROVIDER_TIERS.get(provider)
if provider_tiers and tier in provider_tiers:
return provider_tiers[tier]
# Ultimate fallback: first model in provider's list
return self.PROVIDER_CONFIGS[provider].models[0]
def _get_dual_provider_model(
self,
persona_id: str,
available_providers: List[Provider],
routing: Dict
) -> str:
"""
Get model for dual-provider mode using provider alternation.
Args:
persona_id: The judge persona ID
available_providers: List of available providers (should be 2)
routing: Routing configuration for the persona
Returns:
Model identifier
"""
if len(available_providers) < 2:
return self._get_single_provider_model(persona_id, available_providers, None)
# Determine which provider to use based on persona's model_family
target_family = routing.get("model_family", "anthropic")
# Map model families to providers
family_to_provider = {
"anthropic": Provider.ANTHROPIC,
"openai": Provider.OPENAI,
"deepseek": Provider.DEEPSEEK,
"google": Provider.GOOGLE,
"alibaba": Provider.ALIBABA,
"meta": Provider.META,
"minimax": Provider.MINIMAX,
}
target_provider = family_to_provider.get(target_family)
# If target provider is available, use primary model
if target_provider and target_provider in available_providers:
return routing.get("primary_model", self._get_fallback_model(persona_id, available_providers))
# Otherwise, use fallback for one of the available providers
for provider in available_providers:
fallbacks = routing.get("single_provider_fallback", {})
if provider.value in fallbacks:
return fallbacks[provider.value]
# Ultimate fallback
return self._get_fallback_model(persona_id, available_providers)
def _get_fallback_model(self, persona_id: str, available_providers: List[Provider]) -> str:
"""Get fallback model when routing config doesn't match."""
if not available_providers:
return "claude-sonnet-4-5" # Default fallback
provider = available_providers[0]
tier = self.PERSONA_TIER_MAPPING.get(persona_id, ModelTier.BALANCED)
provider_tiers = self.SINGLE_PROVIDER_TIERS.get(provider)
if provider_tiers and tier in provider_tiers:
return provider_tiers[tier]
return self.PROVIDER_CONFIGS[provider].models[0]
def adjust_confidence(self, confidence: float) -> float:
"""
Apply mode-based confidence adjustment.
Args:
confidence: Original confidence value (0.0 - 1.0)
Returns:
Adjusted confidence value
"""
if self._detection_result is None:
self.detect_mode()
adjustment = self._detection_result.confidence_adjustment
adjusted = confidence + adjustment
# Clamp to valid range
return max(0.0, min(1.0, adjusted))
def get_panel_composition(self, panel_type: str = "standard") -> Dict[str, str]:
"""
Get recommended panel composition based on mode and panel type.
Args:
panel_type: Type of panel (code_review, ai_governance_review, high_risk_ai_review)
Returns:
Dictionary mapping persona_id to recommended model
"""
if self._detection_result is None:
self.detect_mode()
# Get panel configuration from routing config
panel_assembly = self._routing_config.get("panel_assembly", {})
mode_key = f"{self._detection_result.mode.value}_provider_panels"
panels = panel_assembly.get(mode_key, {})
panel_config = panels.get(panel_type, {})
composition = {}
judges = panel_config.get("judges", [])
for persona_id in judges:
composition[persona_id] = self.get_model_for_persona(persona_id)
return composition
def get_diversity_report(self) -> Dict[str, Any]:
"""
Generate a report on provider diversity status.
Returns:
Dictionary with diversity analysis
"""
if self._detection_result is None:
self.detect_mode()
available = self._detection_result.available_providers
mode = self._detection_result.mode
# Calculate diversity metrics
tiers_represented = set()
for provider in available:
tiers_represented.add(self.PROVIDER_CONFIGS[provider].tier)
# Check recommended pairs
matching_pairs = []
for p1, p2, description in self.RECOMMENDED_PAIRS:
if p1 in available and p2 in available:
matching_pairs.append({
"providers": [p1.value, p2.value],
"description": description
})
return {
"mode": mode.value,
"provider_count": len(available),
"providers": [p.value for p in available],
"tiers_represented": list(tiers_represented),
"confidence_adjustment": self._detection_result.confidence_adjustment,
"diversity_strategy": self._detection_result.diversity_strategy,
"single_provider_capable": [
p.value for p in available
if self.PROVIDER_CONFIGS[p].single_provider_capable
],
"recommended_pairs_available": matching_pairs,
"full_diversity_achieved": mode == ProviderMode.MULTI,
"recommendations": self._get_diversity_recommendations(available, mode)
}
def _get_diversity_recommendations(
self,
available: List[Provider],
mode: ProviderMode
) -> List[str]:
"""Generate recommendations for improving diversity."""
recommendations = []
if mode == ProviderMode.SINGLE:
recommendations.append(
"Consider adding a second provider for cross-provider verification. "
f"Recommended pairs with {available[0].value if available else 'your provider'}: "
f"{self._get_recommended_additions(available)}"
)
recommendations.append(
"Single-provider mode has -10% confidence adjustment. "
"Multi-provider mode eliminates this penalty."
)
elif mode == ProviderMode.DUAL:
recommendations.append(
"Add a third provider to achieve full Constitutional Court diversity. "
f"Consider: {self._get_recommended_additions(available)}"
)
recommendations.append(
"Dual-provider mode has -5% confidence adjustment."
)
else:
recommendations.append(
"Full diversity achieved. No confidence penalty applied."
)
return recommendations
def _get_recommended_additions(self, available: List[Provider]) -> str:
"""Get recommended providers to add."""
available_set = set(available)
missing = []
# Prioritize flagship providers
priority_order = [
Provider.ANTHROPIC, Provider.OPENAI, Provider.GOOGLE,
Provider.DEEPSEEK, Provider.META, Provider.MINIMAX
]
for provider in priority_order:
if provider not in available_set:
missing.append(provider.value)
if len(missing) >= 2:
break
return ", ".join(missing) if missing else "none"
def validate_provider_health(self) -> Dict[str, Any]:
"""
Validate health of configured providers.
Returns:
Dictionary with provider health status
"""
health = {}
for provider, config in self.PROVIDER_CONFIGS.items():
api_key = os.getenv(config.api_key_env)
health[provider.value] = {
"configured": api_key is not None,
"api_key_env": config.api_key_env,
"api_key_present": bool(api_key),
"api_key_length": len(api_key) if api_key else 0,
"tier": config.tier,
"single_provider_capable": config.single_provider_capable,
"available_models": config.models,
}
return health
Convenience functions
def detect_provider_mode() -> ProviderDetectionResult: """ Convenience function to detect provider mode.
Returns:
ProviderDetectionResult with detected mode and configuration
"""
detector = ProviderDetector()
return detector.detect_mode()
def get_model_for_persona(persona_id: str) -> str: """ Convenience function to get model for a persona.
Args:
persona_id: The judge persona ID
Returns:
Model identifier string
"""
detector = ProviderDetector()
return detector.get_model_for_persona(persona_id)
def adjust_confidence(confidence: float) -> float: """ Convenience function to adjust confidence based on provider mode.
Args:
confidence: Original confidence value
Returns:
Adjusted confidence value
"""
detector = ProviderDetector()
return detector.adjust_confidence(confidence)
def get_diversity_report() -> Dict[str, Any]: """ Convenience function to get diversity report.
Returns:
Dictionary with diversity analysis
"""
detector = ProviderDetector()
return detector.get_diversity_report()
def check_provider_health() -> Dict[str, Any]: """ Convenience function to check provider health.
Returns:
Dictionary with provider health status
"""
detector = ProviderDetector()
return detector.validate_provider_health()
Module-level singleton for efficiency
_default_detector: Optional[ProviderDetector] = None
def get_default_detector() -> ProviderDetector: """ Get or create the default ProviderDetector singleton.
Returns:
Shared ProviderDetector instance
"""
global _default_detector
if _default_detector is None:
_default_detector = ProviderDetector()
return _default_detector
def reset_default_detector() -> None: """Reset the default detector (useful for testing).""" global _default_detector _default_detector = None