# ruff: noqa: I001 import logging from pathlib import Path from typing import Annotated, Any, ClassVar, Literal, Protocol import tomllib from dotenv import load_dotenv from pydantic import BaseModel, Field, field_validator, model_validator from pydantic.fields import FieldInfo from pydantic_settings import BaseSettings, DotEnvSettingsSource, EnvSettingsSource, PydanticBaseSettingsSource, SettingsConfigDict from src.utils.types import SupportedProviders # Load .env file for local development. # Make sure this is called before AppSettings is instantiated if you rely on .env for AppSettings construction. load_dotenv(override=True) logger = logging.getLogger(__name__) def load_toml_config(config_path: str = "config.toml") -> dict[str, Any]: """Load configuration from TOML file if it exists.""" config_file = Path(config_path) if config_file.exists(): try: with open(config_file, "rb") as f: return tomllib.load(f) except (tomllib.TOMLDecodeError, OSError) as exc: logger.warning("Failed to load %s: %s", config_path, exc) return {} return {} # Load TOML config once TOML_CONFIG = load_toml_config() class LLMComponentSettings(Protocol): """Protocol for settings classes that use LLM providers with backup support.""" PROVIDER: SupportedProviders MODEL: str BACKUP_PROVIDER: SupportedProviders | None BACKUP_MODEL: str | None class TomlConfigSettingsSource(PydanticBaseSettingsSource): """Custom settings source for loading from TOML file.""" def __init__(self, settings_cls: type[BaseSettings]) -> None: super().__init__(settings_cls) SECTION_MAP: ClassVar[dict[str, str]] = { "DB": "db", "AUTH": "auth", "SENTRY": "sentry", "CACHE": "cache", "LLM": "llm", "AUDIO": "audio", "DERIVER": "deriver", "PEER_CARD": "peer_card", "DIALECTIC": "dialectic", "SUMMARY": "summary", "WEBHOOK": "webhook", "DREAM": "dream", "VECTOR_STORE": "vector_store", "METRICS": "metrics", "TELEMETRY": "telemetry", "": "app", # For AppSettings with no prefix } def get_field_value( self, field: FieldInfo, field_name: str ) -> tuple[Any, str, bool]: # Get the env_prefix from the model config prefix = self.settings_cls.model_config.get("env_prefix", "") if prefix.endswith("_"): prefix = prefix[:-1] # Map prefixes to TOML sections section = self.SECTION_MAP.get(prefix, prefix.lower()) toml_data = TOML_CONFIG.get(section, {}) # Try different case variations field_value = toml_data.get(field_name.lower()) if field_value is None: field_value = toml_data.get(field_name.upper()) if field_value is None: field_value = toml_data.get(field_name) return field_value, field_name, False def __call__(self) -> dict[str, Any]: # Get the env_prefix from the model config prefix = self.settings_cls.model_config.get("env_prefix", "") if prefix.endswith("_"): prefix = prefix[:-1] section = self.SECTION_MAP.get(prefix, prefix.lower()) toml_data = TOML_CONFIG.get(section, {}) # Convert keys to uppercase to match field names return {key.upper(): value for key, value in toml_data.items()} class HonchoSettings(BaseSettings): """Base class for all settings models in Honcho. Defines the source precedence for loading settings. """ @classmethod def settings_customise_sources( # pyright: ignore cls, settings_cls: type[BaseSettings], init_settings: PydanticBaseSettingsSource, env_settings: EnvSettingsSource, dotenv_settings: DotEnvSettingsSource, file_secret_settings: PydanticBaseSettingsSource, ) -> tuple[PydanticBaseSettingsSource, ...]: # Correct precedence: init > env > .env > toml > secrets > defaults return ( init_settings, env_settings, dotenv_settings, TomlConfigSettingsSource(settings_cls), file_secret_settings, ) class BackupLLMSettingsMixin: """Mixin class for settings that support backup LLM provider configuration. Provides backup provider and model fields along with validation to ensure both fields are set together or both are None. """ BACKUP_PROVIDER: SupportedProviders | None = None BACKUP_MODEL: str | None = None @model_validator(mode="after") def _validate_backup_configuration(self): """Ensure both backup fields are set together or both are None.""" if (self.BACKUP_PROVIDER is None) != (self.BACKUP_MODEL is None): raise ValueError( "BACKUP_PROVIDER and BACKUP_MODEL must both be set or both be None" ) return self class DBSettings(HonchoSettings): model_config = SettingsConfigDict(env_prefix="DB_", extra="ignore") # pyright: ignore CONNECTION_URI: str = ( "postgresql+psycopg://postgres:postgres@localhost:5432/postgres" ) SCHEMA: str = "public" POOL_CLASS: str = "default" POOL_PRE_PING: bool = True POOL_SIZE: Annotated[int, Field(default=10, gt=0, le=1000)] = 10 MAX_OVERFLOW: Annotated[int, Field(default=20, ge=0, le=1000)] = 20 POOL_TIMEOUT: Annotated[int, Field(default=30, gt=0, le=300)] = ( 30 # seconds (max 5 minutes) ) POOL_RECYCLE: Annotated[int, Field(default=300, gt=0, le=7200)] = ( 300 # seconds (max 2 hours) ) POOL_USE_LIFO: bool = True SQL_DEBUG: bool = False TRACING: bool = False class AuthSettings(HonchoSettings): model_config = SettingsConfigDict(env_prefix="AUTH_", extra="ignore") # pyright: ignore USE_AUTH: bool = False JWT_SECRET: str | None = None # Must be set if USE_AUTH is true @model_validator(mode="after") # type: ignore def _require_jwt_secret(self) -> "AuthSettings": if self.USE_AUTH and not self.JWT_SECRET: raise ValueError("JWT_SECRET must be set if USE_AUTH is true") return self class SentrySettings(HonchoSettings): model_config = SettingsConfigDict(env_prefix="SENTRY_", extra="ignore") # pyright: ignore ENABLED: bool = False DSN: str | None = None RELEASE: str | None = None # TODO maybe centralize this with release number ENVIRONMENT: str = "development" TRACES_SAMPLE_RATE: Annotated[float, Field(default=0.1, ge=0.0, le=1.0)] = 0.1 PROFILES_SAMPLE_RATE: Annotated[float, Field(default=0.1, ge=0.0, le=1.0)] = 0.1 class LLMSettings(HonchoSettings): model_config = SettingsConfigDict(env_prefix="LLM_", extra="ignore") # pyright: ignore # API Keys for LLM providers ANTHROPIC_API_KEY: str | None = None OPENAI_API_KEY: str | None = None OPENAI_COMPATIBLE_API_KEY: str | None = None GEMINI_API_KEY: str | None = None GROQ_API_KEY: str | None = None OPENAI_COMPATIBLE_BASE_URL: str | None = None # Separate vLLM endpoint (for local models) VLLM_API_KEY: str | None = None VLLM_BASE_URL: str | None = None EMBEDDING_PROVIDER: Literal["openai", "gemini", "openrouter"] = "openai" # General LLM settings DEFAULT_MAX_TOKENS: Annotated[int, Field(default=1000, gt=0, le=100_000)] = 2500 # Maximum characters for tool output to prevent token explosion. # Set to 10,000 chars (~2,500 tokens at 4 chars/token) to stay well under # typical context limits while providing substantial tool output. MAX_TOOL_OUTPUT_CHARS: Annotated[int, Field(default=10000, gt=0, le=100_000)] = ( 10000 ) # Maximum characters for individual message content in tool results. # Keeps each message preview concise while preserving key context. MAX_MESSAGE_CONTENT_CHARS: Annotated[int, Field(default=2000, gt=0, le=10_000)] = ( 2000 ) class AudioSettings(HonchoSettings): model_config = SettingsConfigDict(env_prefix="AUDIO_", extra="ignore") # pyright: ignore PROVIDER: Literal["openai"] = "openai" MODEL: str = "whisper-1" MAX_FILE_SIZE_BYTES: Annotated[int, Field(default=25_000_000, gt=0)] = 25_000_000 class DeriverSettings(BackupLLMSettingsMixin, HonchoSettings): model_config = SettingsConfigDict(env_prefix="DERIVER_", extra="ignore") # pyright: ignore ENABLED: bool = True WORKERS: Annotated[int, Field(default=1, gt=0, le=100)] = 1 POLLING_SLEEP_INTERVAL_SECONDS: Annotated[ float, Field(default=1.0, gt=0.0, le=60.0) ] = 1.0 STALE_SESSION_TIMEOUT_MINUTES: Annotated[int, Field(default=5, gt=0, le=1440)] = 5 # Retention window (seconds) for keeping errored items in the queue QUEUE_ERROR_RETENTION_SECONDS: Annotated[ int, Field(default=30 * 24 * 3600, gt=0) ] = 30 * 24 * 3600 # 30 days default PROVIDER: SupportedProviders = "google" MODEL: str = "gemini-2.5-flash-lite" TEMPERATURE: float | None = None # Whether to deduplicate documents when creating them DEDUPLICATE: bool = True MAX_OUTPUT_TOKENS: Annotated[int, Field(default=4096, gt=0, le=100_000)] = 4096 THINKING_BUDGET_TOKENS: Annotated[int, Field(default=1024, gt=0, le=5000)] = 1024 LOG_OBSERVATIONS: bool = False MAX_INPUT_TOKENS: Annotated[int, Field(default=23000, gt=0, le=23000)] = 23000 # Maximum number of observations to return in working representation # This is applied to both explicit and deductive observations WORKING_REPRESENTATION_MAX_OBSERVATIONS: Annotated[ int, Field(default=100, gt=0, le=1000) ] = 100 REPRESENTATION_BATCH_MAX_TOKENS: Annotated[ int, Field(default=1024, ge=128, le=16_384), ] = 1024 # When enabled, bypasses the batch token threshold and processes work immediately FLUSH_ENABLED: bool = False @model_validator(mode="after") def validate_batch_tokens_vs_context_limit(self): if self.REPRESENTATION_BATCH_MAX_TOKENS > self.MAX_INPUT_TOKENS: raise ValueError( f"REPRESENTATION_BATCH_MAX_TOKENS ({self.REPRESENTATION_BATCH_MAX_TOKENS}) cannot exceed max deriver input tokens ({self.MAX_INPUT_TOKENS})" ) return self class PeerCardSettings(HonchoSettings): model_config = SettingsConfigDict(env_prefix="PEER_CARD_", extra="ignore") # pyright: ignore ENABLED: bool = True # Reasoning levels for dialectic - defined here to avoid circular imports with schemas ReasoningLevel = Literal["minimal", "low", "medium", "high", "max"] REASONING_LEVELS: list[ReasoningLevel] = [ "minimal", "low", "medium", "high", "max", ] class DialecticLevelSettings(BaseModel): """Settings for a specific reasoning level in the dialectic.""" model_config = SettingsConfigDict(populate_by_name=True) # pyright: ignore PROVIDER: Annotated[SupportedProviders, Field(validation_alias="provider")] MODEL: Annotated[str, Field(validation_alias="model")] BACKUP_PROVIDER: Annotated[ SupportedProviders | None, Field(validation_alias="backup_provider") ] = None BACKUP_MODEL: Annotated[str | None, Field(validation_alias="backup_model")] = None THINKING_BUDGET_TOKENS: Annotated[ int, Field(ge=0, le=100_000, validation_alias="thinking_budget_tokens") ] MAX_TOOL_ITERATIONS: Annotated[ int, Field(ge=0, le=50, validation_alias="max_tool_iterations") ] MAX_OUTPUT_TOKENS: Annotated[ int | None, Field(ge=1, le=100_000, validation_alias="max_output_tokens") ] = None # None means use global DIALECTIC.MAX_OUTPUT_TOKENS TOOL_CHOICE: Annotated[str | None, Field(validation_alias="tool_choice")] = ( None # None/auto lets model decide, "any"/"required" forces tool use ) @model_validator(mode="after") def _validate_backup_configuration(self) -> "DialecticLevelSettings": """Ensure both backup fields are set together or both are None.""" if (self.BACKUP_PROVIDER is None) != (self.BACKUP_MODEL is None): raise ValueError( "BACKUP_PROVIDER and BACKUP_MODEL must both be set or both be None" ) return self @model_validator(mode="after") def _validate_anthropic_thinking_budget(self) -> "DialecticLevelSettings": """Ensure Anthropic thinking budget is >= 1024 when enabled.""" if ( self.PROVIDER == "anthropic" and self.THINKING_BUDGET_TOKENS > 0 and self.THINKING_BUDGET_TOKENS < 1024 ): raise ValueError( f"THINKING_BUDGET_TOKENS must be >= 1024 for Anthropic provider when enabled (got {self.THINKING_BUDGET_TOKENS})" ) return self class DialecticSettings(HonchoSettings): model_config = SettingsConfigDict( # pyright: ignore env_prefix="DIALECTIC_", env_nested_delimiter="__", extra="ignore" ) # Per-level settings for provider, model, thinking budget, and tool iterations # TODO: Fill in appropriate values for each reasoning level LEVELS: dict[ReasoningLevel, DialecticLevelSettings] = Field( default_factory=lambda: { "minimal": DialecticLevelSettings( PROVIDER="google", MODEL="gemini-2.5-flash-lite", THINKING_BUDGET_TOKENS=0, MAX_TOOL_ITERATIONS=1, MAX_OUTPUT_TOKENS=250, TOOL_CHOICE="any", ), "low": DialecticLevelSettings( PROVIDER="google", MODEL="gemini-2.5-flash-lite", THINKING_BUDGET_TOKENS=0, MAX_TOOL_ITERATIONS=5, TOOL_CHOICE="any", ), "medium": DialecticLevelSettings( PROVIDER="anthropic", MODEL="claude-haiku-4-5", THINKING_BUDGET_TOKENS=1024, MAX_TOOL_ITERATIONS=2, ), "high": DialecticLevelSettings( PROVIDER="anthropic", MODEL="claude-haiku-4-5", THINKING_BUDGET_TOKENS=1024, MAX_TOOL_ITERATIONS=4, ), "max": DialecticLevelSettings( PROVIDER="anthropic", MODEL="claude-haiku-4-5", THINKING_BUDGET_TOKENS=2048, MAX_TOOL_ITERATIONS=10, ), } ) MAX_OUTPUT_TOKENS: Annotated[int, Field(default=8192, gt=0, le=100_000)] = 8192 MAX_INPUT_TOKENS: Annotated[int, Field(default=100_000, gt=0, le=200_000)] = 100_000 # Token limit for get_recent_history tool within the agent HISTORY_TOKEN_LIMIT: Annotated[int, Field(default=8192, gt=0, le=100_000)] = 8192 # Session history injection: max tokens of recent messages to include when session_id is specified. # Set to 0 to disable automatic session history injection. SESSION_HISTORY_MAX_TOKENS: Annotated[ int, Field(default=4_096, ge=0, le=16_384) ] = 4_096 @model_validator(mode="after") def _validate_token_budgets(self) -> "DialecticSettings": """Ensure the output token limit exceeds all thinking budgets.""" for level, level_settings in self.LEVELS.items(): if self.MAX_OUTPUT_TOKENS <= level_settings.THINKING_BUDGET_TOKENS: raise ValueError( f"MAX_OUTPUT_TOKENS must be greater than THINKING_BUDGET_TOKENS for level '{level}'" ) return self @model_validator(mode="after") def _validate_all_levels_present(self) -> "DialecticSettings": """Ensure all reasoning levels are configured.""" missing = set(REASONING_LEVELS) - set(self.LEVELS.keys()) if missing: raise ValueError(f"Missing configuration for reasoning levels: {missing}") return self class SummarySettings(BackupLLMSettingsMixin, HonchoSettings): model_config = SettingsConfigDict(env_prefix="SUMMARY_", extra="ignore") # pyright: ignore ENABLED: bool = True MESSAGES_PER_SHORT_SUMMARY: Annotated[int, Field(default=20, gt=0, le=100)] = 20 MESSAGES_PER_LONG_SUMMARY: Annotated[int, Field(default=60, gt=0, le=500)] = 60 PROVIDER: SupportedProviders = "google" MODEL: str = "gemini-2.5-flash" MAX_TOKENS_SHORT: Annotated[int, Field(default=1000, gt=0, le=10_000)] = 1000 MAX_TOKENS_LONG: Annotated[int, Field(default=4000, gt=0, le=20_000)] = 4000 THINKING_BUDGET_TOKENS: Annotated[int, Field(default=512, gt=0, le=2000)] = 512 class WebhookSettings(HonchoSettings): model_config = SettingsConfigDict(env_prefix="WEBHOOK_", extra="ignore") # pyright: ignore SECRET: str | None = None # Must be set if configuring webhooks MAX_WORKSPACE_LIMIT: int = 10 class MetricsSettings(HonchoSettings): model_config = SettingsConfigDict(env_prefix="METRICS_", extra="ignore") # pyright: ignore ENABLED: bool = False NAMESPACE: str | None = None class TelemetrySettings(HonchoSettings): """CloudEvents telemetry settings for analytics. These settings configure the CloudEvents emitter for pushing structured events to an analytics backend. """ model_config = SettingsConfigDict(env_prefix="TELEMETRY_", extra="ignore") # pyright: ignore # Master toggle for CloudEvents emission ENABLED: bool = False # CloudEvents HTTP endpoint (e.g., "https://telemetry.honcho.dev/v1/events") ENDPOINT: str | None = None # Optional headers for authentication HEADERS: dict[str, str] | None = None # Batching configuration BATCH_SIZE: Annotated[int, Field(default=100, gt=0, le=1000)] = 100 FLUSH_INTERVAL_SECONDS: Annotated[float, Field(default=1.0, gt=0.0, le=60.0)] = 1.0 FLUSH_THRESHOLD: Annotated[int, Field(default=50, gt=0, le=1000)] = 50 # Retry configuration MAX_RETRIES: Annotated[int, Field(default=3, gt=0, le=10)] = 3 # Buffer configuration MAX_BUFFER_SIZE: Annotated[int, Field(default=10000, gt=0, le=100000)] = 10000 # Namespace for instance identification (propagated from top-level NAMESPACE if not set) NAMESPACE: str | None = None class CacheSettings(HonchoSettings): model_config = SettingsConfigDict(env_prefix="CACHE_", extra="ignore") # pyright: ignore ENABLED: bool = False URL: str = "redis://localhost:6379/0?suppress=true" NAMESPACE: str | None = None DEFAULT_TTL_SECONDS: Annotated[int, Field(default=300, ge=1, le=86_400)] = ( 300 # how long to keep items in cache ) DEFAULT_LOCK_TTL_SECONDS: Annotated[int, Field(default=5, ge=1, le=86_400)] = ( 5 # how long to hold a lock on a resource when fetching DB after cache miss ) class SurprisalSettings(BaseModel): """Settings for tree-based surprisal sampling during dreams.""" ENABLED: bool = False # Tree configuration TREE_TYPE: Literal[ "kdtree", "balltree", "rptree", "covertree", "lsh", "graph", "prototype" ] = "kdtree" TREE_K: Annotated[int, Field(default=5, gt=0, le=20)] = 5 # k for kNN-based trees # Sampling strategy SAMPLING_STRATEGY: Literal["recent", "random", "all"] = "recent" SAMPLE_SIZE: Annotated[int, Field(default=200, gt=0, le=2000)] = 200 # Surprisal filtering (normalized scores: 0.0 = lowest, 1.0 = highest) TOP_PERCENT_SURPRISAL: Annotated[float, Field(default=0.10, gt=0.0, le=1.0)] = ( 0.10 # Top 10% of observations ) # Hybrid mode: min high-surprisal observations to replace standard questions MIN_HIGH_SURPRISAL_FOR_REPLACE: Annotated[int, Field(default=10, gt=0)] = 10 # Observation level filtering INCLUDE_LEVELS: list[str] = ["explicit", "deductive"] class DreamSettings(BackupLLMSettingsMixin, HonchoSettings): model_config = SettingsConfigDict( # pyright: ignore env_prefix="DREAM_", env_nested_delimiter="__", extra="ignore" ) ENABLED: bool = True DOCUMENT_THRESHOLD: Annotated[int, Field(default=50, gt=0, le=1000)] = 50 IDLE_TIMEOUT_MINUTES: Annotated[int, Field(default=60, gt=0, le=1440)] = 60 MIN_HOURS_BETWEEN_DREAMS: Annotated[int, Field(default=8, gt=0, le=72)] = 8 ENABLED_TYPES: list[str] = ["omni"] PROVIDER: SupportedProviders = "anthropic" MODEL: str = "claude-sonnet-4-20250514" MAX_OUTPUT_TOKENS: Annotated[int, Field(default=16_384, gt=0, le=64_000)] = 16_384 THINKING_BUDGET_TOKENS: Annotated[int, Field(default=8192, gt=0, le=32_000)] = 8192 # Agent iteration limit - increased for extended reasoning workflow MAX_TOOL_ITERATIONS: Annotated[int, Field(default=20, gt=0, le=50)] = 20 # Token limit for get_recent_history tool within the agent HISTORY_TOKEN_LIMIT: Annotated[int, Field(default=16_384, gt=0, le=200_000)] = ( 16_384 ) ## NOTE: specialist models use the same provider as the main model # Deduction Specialist: handles logical inference DEDUCTION_MODEL: str = "claude-haiku-4-5" # Induction Specialist: identifies patterns across observations INDUCTION_MODEL: str = "claude-haiku-4-5" # Surprisal-based sampling subsystem SURPRISAL: SurprisalSettings = Field(default_factory=SurprisalSettings) @model_validator(mode="after") def _validate_token_budgets(self) -> "DreamSettings": """Ensure the output token limit exceeds the thinking budget.""" if self.MAX_OUTPUT_TOKENS <= self.THINKING_BUDGET_TOKENS: raise ValueError( "MAX_OUTPUT_TOKENS must be greater than THINKING_BUDGET_TOKENS" ) return self class VectorStoreSettings(HonchoSettings): """Settings for vector store (pgvector, Turbopuffer, or LanceDB).""" model_config = SettingsConfigDict(env_prefix="VECTOR_STORE_", extra="ignore") # pyright: ignore # Vector store type to use TYPE: Literal["pgvector", "turbopuffer", "lancedb"] = "pgvector" MIGRATED: bool = False # Global namespace prefix for all vector namespaces # Namespaces follow the pattern: {NAMESPACE}.{type}.{hash} # where hash is a base64url-encoded SHA-256 of the workspace/peer names # - Documents: {NAMESPACE}.doc.{hash(workspace, observer, observed)} # - Messages: {NAMESPACE}.msg.{hash(workspace)} NAMESPACE: str = "honcho" DIMENSIONS: Annotated[ int, Field( default=1536, gt=0, ), ] = 1536 # Turbopuffer-specific settings TURBOPUFFER_API_KEY: str | None = None TURBOPUFFER_REGION: str | None = None # LanceDB-specific settings (local embedded mode) LANCEDB_PATH: str = "./lancedb_data" RECONCILIATION_INTERVAL_SECONDS: Annotated[int, Field(default=300, gt=0)] = ( 300 # 5 minutes ) @model_validator(mode="after") def _require_api_key_for_turbopuffer(self) -> "VectorStoreSettings": if self.TYPE == "turbopuffer" and not self.TURBOPUFFER_API_KEY: raise ValueError( "VECTOR_STORE_TURBOPUFFER_API_KEY must be set when TYPE is 'turbopuffer'" ) return self class AppSettings(HonchoSettings): # No env_prefix for app-level settings model_config = SettingsConfigDict( # pyright: ignore env_prefix="", env_nested_delimiter="__", extra="ignore" ) # Application-wide settings LOG_LEVEL: str = "INFO" SESSION_OBSERVERS_LIMIT: Annotated[int, Field(default=10, gt=0)] = 10 MAX_FILE_SIZE: Annotated[int, Field(default=5_242_880, gt=0)] = 5_242_880 # 5MB GET_CONTEXT_MAX_TOKENS: Annotated[int, Field(default=100_000, gt=0, le=250_000)] = ( 100_000 ) MAX_MESSAGE_SIZE: Annotated[int, Field(default=25_000, gt=0)] = 25_000 EMBED_MESSAGES: bool = True MAX_EMBEDDING_TOKENS: Annotated[int, Field(default=8192, gt=0)] = 8192 MAX_EMBEDDING_TOKENS_PER_REQUEST: Annotated[int, Field(default=300_000, gt=0)] = ( 300_000 ) LANGFUSE_HOST: str | None = None LANGFUSE_PUBLIC_KEY: str | None = None COLLECT_METRICS_LOCAL: bool = False LOCAL_METRICS_FILE: str = "metrics.jsonl" REASONING_TRACES_FILE: str | None = None # Path to JSONL file for reasoning traces NAMESPACE: str = "honcho" # Top-level namespace for all settings, can be overridden by nested-model settings # Nested settings models DB: DBSettings = Field(default_factory=DBSettings) AUTH: AuthSettings = Field(default_factory=AuthSettings) SENTRY: SentrySettings = Field(default_factory=SentrySettings) LLM: LLMSettings = Field(default_factory=LLMSettings) AUDIO: AudioSettings = Field(default_factory=AudioSettings) DERIVER: DeriverSettings = Field(default_factory=DeriverSettings) DIALECTIC: DialecticSettings = Field(default_factory=DialecticSettings) PEER_CARD: PeerCardSettings = Field(default_factory=PeerCardSettings) SUMMARY: SummarySettings = Field(default_factory=SummarySettings) WEBHOOK: WebhookSettings = Field(default_factory=WebhookSettings) METRICS: MetricsSettings = Field(default_factory=MetricsSettings) TELEMETRY: TelemetrySettings = Field(default_factory=TelemetrySettings) CACHE: CacheSettings = Field(default_factory=CacheSettings) DREAM: DreamSettings = Field(default_factory=DreamSettings) VECTOR_STORE: VectorStoreSettings = Field(default_factory=VectorStoreSettings) @field_validator("LOG_LEVEL") def validate_log_level(cls, v: str) -> str: log_level = v.upper() if log_level not in ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]: raise ValueError(f"Invalid log level: {v}") return log_level @model_validator(mode="after") def propagate_namespace(self) -> "AppSettings": """Propagate top-level NAMESPACE to nested settings if not explicitly set.""" if "NAMESPACE" not in self.CACHE.model_fields_set: self.CACHE.NAMESPACE = self.NAMESPACE if "NAMESPACE" not in self.VECTOR_STORE.model_fields_set: self.VECTOR_STORE.NAMESPACE = self.NAMESPACE if "NAMESPACE" not in self.TELEMETRY.model_fields_set: self.TELEMETRY.NAMESPACE = self.NAMESPACE if "NAMESPACE" not in self.METRICS.model_fields_set: self.METRICS.NAMESPACE = self.NAMESPACE return self # Create a single global instance of the settings settings: AppSettings = AppSettings()