"""Configuration validation utilities.""" from typing import Dict, List, Optional, Tuple # Valid routing policies VALID_ROUTING_POLICIES = [ "auto", ":fastest", ":cheapest", "groq", "together", "replicate", "cerebras", "fireworks", "deepinfra", "meta-llama/Llama-3.3-70B-Instruct", ] # Valid model tiers VALID_MODEL_TIERS = ["fast", "capable", "vision"] # Valid LLM providers VALID_LLM_PROVIDERS = ["openai", "anthropic", "huggingface", "qwen"] def validate_routing_policy(policy: str) -> Tuple[bool, Optional[str]]: """ Validate a routing policy. Args: policy: Routing policy string Returns: Tuple of (is_valid, error_message) """ if not policy: return False, "Routing policy cannot be empty" # Check if it's a known policy if policy in VALID_ROUTING_POLICIES: return True, None # Check if it's a model path (provider/model format) if "/" in policy: parts = policy.split("/") if len(parts) >= 2: return True, None # Assume it's a valid model path # Unknown policy return ( False, f"Invalid routing policy: {policy}. Must be one of {VALID_ROUTING_POLICIES} or a provider/model path", ) def validate_model_tier(tier: str) -> Tuple[bool, Optional[str]]: """ Validate a model tier. Args: tier: Model tier string Returns: Tuple of (is_valid, error_message) """ if tier not in VALID_MODEL_TIERS: return False, f"Invalid model tier: {tier}. Must be one of {VALID_MODEL_TIERS}" return True, None def validate_agent_routing_config( config: Dict[str, Dict[str, str]], ) -> Tuple[bool, Optional[str]]: """ Validate agent routing configuration. Args: config: Dictionary mapping agent names to routing config Returns: Tuple of (is_valid, error_message) """ if not isinstance(config, dict): return False, "Configuration must be a dictionary" for agent_name, agent_config in config.items(): if not isinstance(agent_config, dict): return False, f"Configuration for {agent_name} must be a dictionary" # Validate routing policy if "routing_policy" not in agent_config: return False, f"Missing routing_policy for {agent_name}" is_valid, error = validate_routing_policy(agent_config["routing_policy"]) if not is_valid: return False, f"{agent_name}: {error}" # Validate model tier (optional) if "model_tier" in agent_config: is_valid, error = validate_model_tier(agent_config["model_tier"]) if not is_valid: return False, f"{agent_name}: {error}" return True, None def validate_llm_provider(provider: str) -> Tuple[bool, Optional[str]]: """ Validate LLM provider. Args: provider: Provider name Returns: Tuple of (is_valid, error_message) """ if provider not in VALID_LLM_PROVIDERS: return ( False, f"Invalid LLM provider: {provider}. Must be one of {VALID_LLM_PROVIDERS}", ) return True, None def validate_configuration(config: Dict) -> Tuple[bool, Optional[str]]: """ Validate complete configuration. Args: config: Full configuration dictionary Returns: Tuple of (is_valid, error_message) """ # Validate LLM provider if "llm_provider" in config: is_valid, error = validate_llm_provider(config["llm_provider"]) if not is_valid: return False, error # Validate routing policy (global) if "routing_policy" in config: is_valid, error = validate_routing_policy(config["routing_policy"]) if not is_valid: return False, error # Validate per-agent routing config if "agent_routing_config" in config: is_valid, error = validate_agent_routing_config(config["agent_routing_config"]) if not is_valid: return False, error # Validate indicator parameters if "indicator_parameters" in config: params = config["indicator_parameters"] if not isinstance(params, dict): return False, "indicator_parameters must be a dictionary" # Check RSI period if "rsi_period" in params: rsi = params["rsi_period"] if not isinstance(rsi, int) or rsi < 2 or rsi > 100: return False, "RSI period must be between 2 and 100" # Check MACD parameters if "macd_fast" in params: fast = params["macd_fast"] if not isinstance(fast, int) or fast < 2 or fast > 50: return False, "MACD fast period must be between 2 and 50" if "macd_slow" in params: slow = params["macd_slow"] if not isinstance(slow, int) or slow < 10 or slow > 100: return False, "MACD slow period must be between 10 and 100" if "macd_signal" in params: signal = params["macd_signal"] if not isinstance(signal, int) or signal < 2 or signal > 50: return False, "MACD signal period must be between 2 and 50" return True, None