RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
honcho/src/config.py

1395 lines
52 KiB
Python
Raw Blame History

import logging
import os
from pathlib import Path
from typing import Annotated, Any, ClassVar, Literal, cast
import tomllib
from dotenv import load_dotenv
from pydantic import AliasChoices, BaseModel, Field, field_validator, model_validator
from pydantic.fields import FieldInfo
from pydantic_settings import (
BaseSettings,
DotEnvSettingsSource,
EnvSettingsSource,
PydanticBaseSettingsSource,
SettingsConfigDict,
)
# Load .env file for local development.
# Make sure this is called before AppSettings is instantiated if you rely on .env for AppSettings construction.
if not os.getenv("PYTHON_DOTENV_DISABLED"):
load_dotenv(override=True)
logger = logging.getLogger(__name__)
ModelTransport = Literal["anthropic", "openai", "gemini"]
EmbeddingTransport = Literal["openai", "gemini"]
EmbeddingDimensionsMode = Literal["auto", "always", "never"]
# OpenAI-compatible models that reject the `dimensions=` request parameter.
_EMBEDDING_KNOWN_REJECTING_MODELS: frozenset[str] = frozenset(
{"text-embedding-ada-002"}
)
def _default_embedding_model_for_transport(transport: EmbeddingTransport) -> str:
if transport == "gemini":
return "gemini-embedding-001"
return "text-embedding-3-small"
def load_toml_config(config_path: str = "config.toml") -> dict[str, Any]:
"""Load configuration from TOML file if it exists."""
if config_path == "config.toml" and os.getenv("HONCHO_CONFIG_TOML_DISABLED"):
return {}
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()
ThinkingEffortLevel = Literal[
"none", "minimal", "low", "medium", "high", "xhigh", "max"
]
class ModelOverrideSettings(BaseModel):
"""Advanced module-level transport overrides."""
api_key: str | None = None
api_key_env: str | None = None
base_url: str | None = None
provider_params: dict[str, Any] = Field(default_factory=dict)
class PromptCachePolicy(BaseModel):
"""Per-call prompt-caching configuration.
Lives in config.py (not src/llm/caching.py) so ModelConfig can reference
it as a field without a circular import. src/llm/caching.py re-exports
this class for existing import paths.
"""
mode: Literal["none", "prefix", "gemini_cached_content"] = "none"
ttl_seconds: int | None = None
key_version: str = "v1"
def _normalize_model_transport(data: Any) -> Any:
"""Normalize 'provider/model' shorthand into separate transport + model fields."""
if not isinstance(data, dict):
return data
raw_data = cast(dict[Any, Any], data)
update: dict[str, Any] = {str(key): value for key, value in raw_data.items()}
model_value = update.get("model")
transport_value = update.get("transport")
if isinstance(model_value, str) and "/" in model_value and transport_value is None:
prefix, bare_model = model_value.split("/", 1)
if prefix in {"anthropic", "openai", "gemini"}:
update["transport"] = prefix
update["model"] = bare_model
return update
def _validate_thinking_constraints(
transport: ModelTransport, thinking_budget_tokens: int | None
) -> None:
"""Enforce transport-specific thinking_budget_tokens rules.
Anthropic requires a minimum of 1024 tokens when thinking is enabled.
Gemini/OpenAI accept any non-negative value (including 0 to disable).
"""
if (
transport == "anthropic"
and thinking_budget_tokens is not None
and 0 < thinking_budget_tokens < 1024
):
raise ValueError("thinking_budget_tokens must be >= 1024 for Anthropic models")
class FallbackModelSettings(BaseModel):
"""Independent fallback model configuration. No inheritance from primary."""
model: str
transport: ModelTransport
temperature: float | None = None
top_p: float | None = None
top_k: int | None = None
frequency_penalty: float | None = None
presence_penalty: float | None = None
seed: int | None = None
thinking_effort: ThinkingEffortLevel | None = Field(
default=None,
validation_alias=AliasChoices("thinking_effort", "reasoning_effort"),
)
thinking_budget_tokens: int | None = None
max_output_tokens: int | None = None
stop_sequences: list[str] | None = None
cache_policy: PromptCachePolicy | None = None
overrides: ModelOverrideSettings = Field(default_factory=ModelOverrideSettings)
@model_validator(mode="before")
@classmethod
def _normalize_legacy_model_format(cls, data: Any) -> Any:
return _normalize_model_transport(data)
@property
def reasoning_effort(self) -> ThinkingEffortLevel | None:
return self.thinking_effort
@model_validator(mode="after")
def _validate_runtime_shape(self) -> "FallbackModelSettings":
_validate_thinking_constraints(self.transport, self.thinking_budget_tokens)
return self
class ConfiguredModelSettings(BaseModel):
"""Operator-configurable persisted model settings."""
model: str
transport: ModelTransport
fallback: FallbackModelSettings | None = None
temperature: float | None = None
top_p: float | None = None
top_k: int | None = None
frequency_penalty: float | None = None
presence_penalty: float | None = None
seed: int | None = None
thinking_effort: ThinkingEffortLevel | None = Field(
default=None,
validation_alias=AliasChoices("thinking_effort", "reasoning_effort"),
)
thinking_budget_tokens: int | None = None
max_output_tokens: int | None = None
stop_sequences: list[str] | None = None
cache_policy: PromptCachePolicy | None = None
overrides: ModelOverrideSettings = Field(default_factory=ModelOverrideSettings)
@model_validator(mode="before")
@classmethod
def _normalize_legacy_model_format(cls, data: Any) -> Any:
return _normalize_model_transport(data)
@property
def reasoning_effort(self) -> ThinkingEffortLevel | None:
"""Backward-compatible alias for the generic thinking effort field."""
return self.thinking_effort
@model_validator(mode="after")
def _validate_runtime_shape(self) -> "ConfiguredModelSettings":
_validate_thinking_constraints(self.transport, self.thinking_budget_tokens)
return self
class ResolvedFallbackConfig(BaseModel):
"""Runtime-resolved fallback config with credentials already resolved."""
model: str
transport: ModelTransport
api_key: str | None = None
base_url: str | None = None
temperature: float | None = None
top_p: float | None = None
top_k: int | None = None
frequency_penalty: float | None = None
presence_penalty: float | None = None
seed: int | None = None
thinking_effort: ThinkingEffortLevel | None = Field(
default=None,
validation_alias=AliasChoices("thinking_effort", "reasoning_effort"),
)
thinking_budget_tokens: int | None = None
provider_params: dict[str, Any] = Field(default_factory=dict)
max_output_tokens: int | None = None
stop_sequences: list[str] | None = None
cache_policy: PromptCachePolicy | None = None
@property
def reasoning_effort(self) -> ThinkingEffortLevel | None:
return self.thinking_effort
class ModelConfig(BaseModel):
"""Reusable model configuration for any non-embedding LLM caller."""
model: str
transport: ModelTransport
fallback: ResolvedFallbackConfig | None = None
api_key: str | None = None
base_url: str | None = None
temperature: float | None = None
top_p: float | None = None
top_k: int | None = None
frequency_penalty: float | None = None
presence_penalty: float | None = None
seed: int | None = None
thinking_effort: ThinkingEffortLevel | None = Field(
default=None,
validation_alias=AliasChoices("thinking_effort", "reasoning_effort"),
)
thinking_budget_tokens: int | None = None
provider_params: dict[str, Any] = Field(default_factory=dict)
max_output_tokens: int | None = None
stop_sequences: list[str] | None = None
cache_policy: PromptCachePolicy | None = None
@model_validator(mode="before")
@classmethod
def _normalize_legacy_model_format(cls, data: Any) -> Any:
return _normalize_model_transport(data)
@property
def reasoning_effort(self) -> ThinkingEffortLevel | None:
"""Backward-compatible alias for the generic thinking effort field."""
return self.thinking_effort
@model_validator(mode="after")
def _validate_thinking_constraints_on_self(self) -> "ModelConfig":
_validate_thinking_constraints(self.transport, self.thinking_budget_tokens)
return self
def for_model(
self,
model_override: str,
*,
transport_override: ModelTransport | None = None,
) -> "ModelConfig":
return self.model_copy(
update={
"model": model_override,
"transport": transport_override or self.transport,
}
)
class ConfiguredEmbeddingModelSettings(BaseModel):
"""Operator-configurable persisted embedding settings."""
model: str = "text-embedding-3-small"
transport: EmbeddingTransport = "openai"
overrides: ModelOverrideSettings = Field(default_factory=ModelOverrideSettings)
dimensions_mode: EmbeddingDimensionsMode = "auto"
@model_validator(mode="before")
@classmethod
def _normalize_legacy_model_format(cls, data: Any) -> Any:
if not isinstance(data, dict):
return data
raw_data = cast(dict[Any, Any], data)
update: dict[str, Any] = {str(key): value for key, value in raw_data.items()}
model_value = update.get("model")
transport_value = update.get("transport")
if (
isinstance(model_value, str)
and "/" in model_value
and transport_value is None
):
prefix, bare_model = model_value.split("/", 1)
if prefix in {"openai", "gemini"}:
update["transport"] = prefix
update["model"] = bare_model
return update
@model_validator(mode="after")
def _default_model_for_transport(self) -> "ConfiguredEmbeddingModelSettings":
if "model" not in self.model_fields_set:
self.model = _default_embedding_model_for_transport(self.transport)
return self
class EmbeddingModelConfig(BaseModel):
"""Runtime embedding configuration with resolved credentials."""
model: str = "text-embedding-3-small"
transport: EmbeddingTransport = "openai"
api_key: str | None = None
base_url: str | None = None
@model_validator(mode="before")
@classmethod
def _normalize_legacy_model_format(cls, data: Any) -> Any:
if not isinstance(data, dict):
return data
raw_data = cast(dict[Any, Any], data)
update: dict[str, Any] = {str(key): value for key, value in raw_data.items()}
model_value = update.get("model")
transport_value = update.get("transport")
if (
isinstance(model_value, str)
and "/" in model_value
and transport_value is None
):
prefix, bare_model = model_value.split("/", 1)
if prefix in {"openai", "gemini"}:
update["transport"] = prefix
update["model"] = bare_model
return update
@model_validator(mode="after")
def _default_model_for_transport(self) -> "EmbeddingModelConfig":
if "model" not in self.model_fields_set:
self.model = _default_embedding_model_for_transport(self.transport)
return self
def _resolve_secret(value: str | None, env_name: str | None) -> str | None:
if value is not None:
return value
if env_name is None:
return None
return os.getenv(env_name)
def _resolve_fallback_config(
fallback: FallbackModelSettings,
) -> ResolvedFallbackConfig:
"""Resolve a FallbackModelSettings into a runtime ResolvedFallbackConfig."""
return ResolvedFallbackConfig(
model=fallback.model,
transport=fallback.transport,
api_key=_resolve_secret(
fallback.overrides.api_key,
fallback.overrides.api_key_env,
),
base_url=fallback.overrides.base_url,
temperature=fallback.temperature,
top_p=fallback.top_p,
top_k=fallback.top_k,
frequency_penalty=fallback.frequency_penalty,
presence_penalty=fallback.presence_penalty,
seed=fallback.seed,
thinking_effort=fallback.thinking_effort,
thinking_budget_tokens=fallback.thinking_budget_tokens,
provider_params=fallback.overrides.provider_params,
max_output_tokens=fallback.max_output_tokens,
stop_sequences=fallback.stop_sequences,
cache_policy=fallback.cache_policy,
)
def resolve_model_config(configured: ConfiguredModelSettings) -> ModelConfig:
"""Resolve persisted model settings into the runtime ModelConfig."""
resolved_fallback = (
_resolve_fallback_config(configured.fallback)
if configured.fallback is not None
else None
)
return ModelConfig(
model=configured.model,
transport=configured.transport,
fallback=resolved_fallback,
api_key=_resolve_secret(
configured.overrides.api_key,
configured.overrides.api_key_env,
),
base_url=configured.overrides.base_url,
temperature=configured.temperature,
top_p=configured.top_p,
top_k=configured.top_k,
frequency_penalty=configured.frequency_penalty,
presence_penalty=configured.presence_penalty,
seed=configured.seed,
thinking_effort=configured.thinking_effort,
thinking_budget_tokens=configured.thinking_budget_tokens,
provider_params=configured.overrides.provider_params,
max_output_tokens=configured.max_output_tokens,
stop_sequences=configured.stop_sequences,
cache_policy=configured.cache_policy,
)
def _default_embedding_api_key(transport: EmbeddingTransport) -> str | None:
"""Fall back to the global LLM API key for the matching transport."""
if transport == "openai":
return settings.LLM.OPENAI_API_KEY
if transport == "gemini":
return settings.LLM.GEMINI_API_KEY
def resolve_embedding_model_config(
configured: ConfiguredEmbeddingModelSettings,
) -> EmbeddingModelConfig:
"""Resolve persisted embedding settings into the runtime config."""
api_key = _resolve_secret(
configured.overrides.api_key,
configured.overrides.api_key_env,
)
if api_key is None:
api_key = _default_embedding_api_key(configured.transport)
return EmbeddingModelConfig(
model=configured.model,
transport=configured.transport,
api_key=api_key,
base_url=configured.overrides.base_url,
)
_TRANSPORT_SPECIFIC_THINKING_KEYS: frozenset[str] = frozenset(
{"thinking_budget_tokens", "thinking_effort"}
)
def _fill_defaults_for_nested_field(
data: dict[str, Any],
field_name: str,
default_factory: Any,
) -> dict[str, Any]:
"""Fill missing keys in a partial nested dict from the field's defaults.
When Pydantic's env_nested_delimiter splits an env var like
``DERIVER_MODEL_CONFIG__THINKING_BUDGET_TOKENS=2048`` it produces
``{"MODEL_CONFIG": {"THINKING_BUDGET_TOKENS": 2048}}``. Without merging
that partial dict would fail validation because required keys like
``model`` and ``transport`` are missing. This helper fills them from
the field's ``default_factory`` so partial overrides work.
If the env override switches ``transport`` to a value that differs from
the default's, transport-specific thinking params
(``thinking_budget_tokens``, ``thinking_effort``) are dropped from the
default before merging. This prevents e.g. a Gemini default's
``thinking_budget_tokens=1024`` from leaking into an OpenAI override,
which would then be rejected by the OpenAI backend (OpenAI uses
``reasoning.effort``, not a token budget). Explicit thinking params in
the env override are preserved.
"""
raw: Any = data.get(field_name) or data.get(field_name.lower())
if not isinstance(raw, dict):
return data
default_obj = default_factory()
if isinstance(default_obj, BaseModel):
default_dict: dict[str, Any] = default_obj.model_dump(by_alias=True)
else:
default_dict = dict(default_obj)
raw_dict = cast(dict[str, Any], raw)
raw_lower = {k.lower(): v for k, v in raw_dict.items()}
default_lower = {k.lower(): v for k, v in default_dict.items()}
override_transport = raw_lower.get("transport")
default_transport = default_lower.get("transport")
if override_transport is not None and override_transport != default_transport:
for k in list(default_dict.keys()):
if k.lower() in _TRANSPORT_SPECIFIC_THINKING_KEYS:
del default_dict[k]
merged: dict[str, Any] = {**default_dict, **raw_dict}
# Preserve the key casing used in data
key = field_name if field_name in data else field_name.lower()
data[key] = merged
return data
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",
"EMBEDDING": "embedding",
"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 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=5, gt=0, le=300)] = (
5 # seconds; kept under CONNECTION_RETRY_MAX_DELAY_SECONDS so a pool
# checkout fails fast enough to allow a retry within the budget
)
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
# Bounded exponential-backoff retry around connection acquisition. Applied
# lazily on the first DB use of any session (HonchoAsyncSession) — both the
# request path and background/tracked_db scopes — without forcing an eager
# checkout. Guards against transient transaction-pooler saturation (e.g.
# Supavisor rejecting with "too many clients") by retrying the checkout
# instead of failing immediately. CONNECTION_RETRY_MAX_DELAY_SECONDS is the
# TOTAL retry
# budget; with a real QueuePool, a single checkout can block up to
# POOL_TIMEOUT, so POOL_TIMEOUT must stay below the budget for a retry to be
# possible (enforced below). With NullPool (the transaction-pooler setup)
# there is no local queue wait, so saturation is a fast OperationalError and
# POOL_TIMEOUT does not apply.
CONNECTION_RETRY_ENABLED: bool = True
CONNECTION_RETRY_MAX_DELAY_SECONDS: Annotated[
float, Field(default=10.0, gt=0.0, le=120.0)
] = 10.0
CONNECTION_RETRY_BACKOFF_INITIAL_SECONDS: Annotated[
float, Field(default=0.1, gt=0.0, le=10.0)
] = 0.1
CONNECTION_RETRY_BACKOFF_MAX_SECONDS: Annotated[
float, Field(default=2.0, gt=0.0, le=30.0)
] = 2.0
@model_validator(mode="after")
def _validate_retry_budget_vs_pool_timeout(self) -> "DBSettings":
# Only meaningful for a real local pool: NullPool has no queue wait, so
# POOL_TIMEOUT is unused there. For a QueuePool, a checkout can block up
# to POOL_TIMEOUT, so it must be < the total retry budget or the first
# attempt consumes the whole budget and no retry ever happens.
if (
self.POOL_CLASS != "null"
and self.CONNECTION_RETRY_ENABLED
and self.POOL_TIMEOUT >= self.CONNECTION_RETRY_MAX_DELAY_SECONDS
):
raise ValueError(
f"DB_POOL_TIMEOUT ({self.POOL_TIMEOUT}s) must be less than "
+ "DB_CONNECTION_RETRY_MAX_DELAY_SECONDS "
+ f"({self.CONNECTION_RETRY_MAX_DELAY_SECONDS}s) so a pooled checkout "
+ "can fail fast enough to be retried within the budget."
)
return self
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
GEMINI_API_KEY: str | None = None
# Base URLs for LLM providers (for OpenAI-compatible proxies like
# OpenRouter, vLLM, Together, Anyscale, self-hosted, etc.)
ANTHROPIC_BASE_URL: str | None = None
OPENAI_BASE_URL: str | None = None
GEMINI_BASE_URL: str | None = None
# 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 EmbeddingSettings(HonchoSettings):
model_config = SettingsConfigDict( # pyright: ignore
env_prefix="EMBEDDING_", env_nested_delimiter="__", extra="ignore"
)
@staticmethod
def _MODEL_CONFIG_DEFAULT() -> ConfiguredEmbeddingModelSettings:
return ConfiguredEmbeddingModelSettings(
transport="openai",
model="text-embedding-3-small",
)
MODEL_CONFIG: ConfiguredEmbeddingModelSettings = Field(
default_factory=_MODEL_CONFIG_DEFAULT
)
VECTOR_DIMENSIONS: Annotated[int, Field(default=1536, gt=0)] = 1536
MAX_INPUT_TOKENS: Annotated[int, Field(default=8192, gt=0)] = 8192
MAX_TOKENS_PER_REQUEST: Annotated[int, Field(default=300_000, gt=0)] = 300_000
@model_validator(mode="before")
@classmethod
def _merge_model_config_defaults(cls, data: Any) -> Any:
if isinstance(data, dict):
_fill_defaults_for_nested_field(
cast(dict[str, Any], data),
"MODEL_CONFIG",
cls._MODEL_CONFIG_DEFAULT,
)
return data # pyright: ignore[reportUnknownVariableType]
def resolve_send_dimensions(self) -> bool:
"""Decide whether OpenAI embedding calls should forward ``dimensions=``.
Lives on the settings instance because ``auto`` mode needs access to
``self.model_fields_set`` to tell whether the operator explicitly set
``VECTOR_DIMENSIONS`` — a standalone resolver over
``ConfiguredEmbeddingModelSettings`` cannot see that.
"""
mode = self.MODEL_CONFIG.dimensions_mode
if mode == "always":
return True
if mode == "never":
return False
if self.MODEL_CONFIG.model in _EMBEDDING_KNOWN_REJECTING_MODELS:
return False
return "VECTOR_DIMENSIONS" in self.model_fields_set
class DeriverSettings(HonchoSettings):
model_config = SettingsConfigDict( # pyright: ignore
env_prefix="DERIVER_", env_nested_delimiter="__", extra="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
# Adaptive polling: when the queue is idle (or the loop is erroring) the
# sleep interval grows from POLLING_SLEEP_INTERVAL_SECONDS toward
# POLLING_SLEEP_MAX_INTERVAL_SECONDS by POLLING_BACKOFF_MULTIPLIER each
# cycle, then snaps back to the base interval as soon as work is found.
# Reduces steady-state query load against the (shared) DB/pooler.
POLLING_BACKOFF_ENABLED: bool = True
POLLING_SLEEP_MAX_INTERVAL_SECONDS: Annotated[
float, Field(default=30.0, gt=0.0, le=300.0)
] = 30.0
POLLING_BACKOFF_MULTIPLIER: Annotated[
float, Field(default=2.0, ge=1.0, le=10.0)
] = 2.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
@staticmethod
def _MODEL_CONFIG_DEFAULT() -> ConfiguredModelSettings:
# Minimal default: transport + model only. Any other knobs would merge
# into operator-supplied env / config.toml overrides via
# _fill_defaults_for_nested_field and clobber intent.
return ConfiguredModelSettings(
transport="openai",
model="gpt-5.4-mini",
)
MODEL_CONFIG: ConfiguredModelSettings = Field(default_factory=_MODEL_CONFIG_DEFAULT)
# Whether to deduplicate documents when creating them
DEDUPLICATE: bool = True
LOG_OBSERVATIONS: bool = False
MAX_INPUT_TOKENS: Annotated[int, Field(default=25000, gt=0, le=25000)] = 25000
MAX_CUSTOM_INSTRUCTIONS_TOKENS: Annotated[
int, Field(default=2000, ge=0, le=2000)
] = 2000
# 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="before")
@classmethod
def _merge_model_config_defaults(cls, data: Any) -> Any:
if isinstance(data, dict):
_fill_defaults_for_nested_field(
cast(dict[str, Any], data),
"MODEL_CONFIG",
cls._MODEL_CONFIG_DEFAULT,
)
return data # pyright: ignore[reportUnknownVariableType]
@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
MODEL_CONFIG: Annotated[
ConfiguredModelSettings,
Field(validation_alias="model_config"),
]
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_anthropic_thinking_budget(self) -> "DialecticLevelSettings":
"""Ensure Anthropic thinking budget is >= 1024 when enabled."""
if (
self.MODEL_CONFIG.transport == "anthropic"
and self.MODEL_CONFIG.thinking_budget_tokens is not None
and self.MODEL_CONFIG.thinking_budget_tokens > 0
and self.MODEL_CONFIG.thinking_budget_tokens < 1024
):
raise ValueError(
"MODEL_CONFIG.thinking_budget_tokens must be >= 1024 for "
+ "Anthropic models when enabled "
+ f"(got {self.MODEL_CONFIG.thinking_budget_tokens})"
)
return self
def _default_dialectic_levels() -> dict[ReasoningLevel, DialecticLevelSettings]:
# Minimal defaults per level: transport + model only. Non-MODEL_CONFIG
# level tuning (MAX_TOOL_ITERATIONS, MAX_OUTPUT_TOKENS, TOOL_CHOICE)
# stays here because it's the per-level behavior, not a model knob —
# operators still override any of it via
# DIALECTIC_LEVELS__<level>__MODEL_CONFIG__* without conflict.
def _default_model_config() -> ConfiguredModelSettings:
return ConfiguredModelSettings(
transport="openai",
model="gpt-5.4-mini",
)
return {
"minimal": DialecticLevelSettings(
MODEL_CONFIG=_default_model_config(),
MAX_TOOL_ITERATIONS=1,
MAX_OUTPUT_TOKENS=250,
TOOL_CHOICE="auto",
),
"low": DialecticLevelSettings(
MODEL_CONFIG=_default_model_config(),
MAX_TOOL_ITERATIONS=5,
TOOL_CHOICE="auto",
),
"medium": DialecticLevelSettings(
MODEL_CONFIG=_default_model_config(),
MAX_TOOL_ITERATIONS=2,
),
"high": DialecticLevelSettings(
MODEL_CONFIG=_default_model_config(),
MAX_TOOL_ITERATIONS=4,
),
"max": DialecticLevelSettings(
MODEL_CONFIG=_default_model_config(),
MAX_TOOL_ITERATIONS=10,
),
}
class DialecticSettings(HonchoSettings):
model_config = SettingsConfigDict( # pyright: ignore
env_prefix="DIALECTIC_", env_nested_delimiter="__", extra="ignore"
)
LEVELS: dict[ReasoningLevel, DialecticLevelSettings] = Field(
default_factory=_default_dialectic_levels
)
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="before")
@classmethod
def _merge_level_defaults(cls, data: Any) -> Any:
"""Merge partial level overrides with built-in defaults."""
if not isinstance(data, dict):
return data
typed_data = cast(dict[str, Any], data)
levels_raw: dict[str, Any] | None = typed_data.get("LEVELS") or typed_data.get(
"levels"
)
if not isinstance(levels_raw, dict):
return data # pyright: ignore[reportUnknownVariableType]
defaults = _default_dialectic_levels()
for level_name_key, level_override_val in levels_raw.items():
level_name = str(level_name_key)
if not isinstance(level_override_val, dict):
continue
level_override = cast(dict[str, Any], level_override_val)
if level_name in defaults:
base: dict[str, Any] = defaults[level_name].model_dump(by_alias=True)
# Recursively merge nested MODEL_CONFIG / model_config too.
# model_dump() always produces the Python field name
# ("MODEL_CONFIG"), but TOML overrides arrive as lowercase
# ("model_config"). Check both casings in the override and
# resolve the base value from whichever casing is present.
for mc_key in ("MODEL_CONFIG", "model_config"):
if mc_key in level_override and isinstance(
level_override[mc_key], dict
):
base_mc: dict[str, Any] = dict(
base.get("MODEL_CONFIG") or base.get("model_config") or {}
)
override_mc = cast(dict[str, Any], level_override[mc_key])
override_lower = {k.lower(): v for k, v in override_mc.items()}
base_lower = {k.lower(): v for k, v in base_mc.items()}
override_transport = override_lower.get("transport")
base_transport = base_lower.get("transport")
if (
override_transport is not None
and override_transport != base_transport
):
for k in list(base_mc.keys()):
if k.lower() in _TRANSPORT_SPECIFIC_THINKING_KEYS:
del base_mc[k]
level_override[mc_key] = {**base_mc, **override_mc}
levels_raw[level_name] = {**base, **level_override}
# Backfill any reasoning levels the operator didn't explicitly set with the default values.
for default_level_name, default_level in defaults.items():
if default_level_name not in levels_raw:
levels_raw[default_level_name] = default_level.model_dump(by_alias=True)
return data # pyright: ignore[reportUnknownVariableType]
@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():
thinking_budget = level_settings.MODEL_CONFIG.thinking_budget_tokens or 0
effective_max = (
level_settings.MAX_OUTPUT_TOKENS
if level_settings.MAX_OUTPUT_TOKENS is not None
else self.MAX_OUTPUT_TOKENS
)
if thinking_budget > 0 and thinking_budget >= effective_max:
raise ValueError(
"MAX_OUTPUT_TOKENS must be greater than MODEL_CONFIG."
+ f"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(HonchoSettings):
model_config = SettingsConfigDict( # pyright: ignore
env_prefix="SUMMARY_", env_nested_delimiter="__", extra="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
@staticmethod
def _MODEL_CONFIG_DEFAULT() -> ConfiguredModelSettings:
# Minimal default; extra knobs would merge into env/TOML overrides.
return ConfiguredModelSettings(
transport="openai",
model="gpt-5.4-mini",
)
MODEL_CONFIG: ConfiguredModelSettings = Field(default_factory=_MODEL_CONFIG_DEFAULT)
@model_validator(mode="before")
@classmethod
def _merge_model_config_defaults(cls, data: Any) -> Any:
if isinstance(data, dict):
_fill_defaults_for_nested_field(
cast(dict[str, Any], data),
"MODEL_CONFIG",
cls._MODEL_CONFIG_DEFAULT,
)
return data # pyright: ignore[reportUnknownVariableType]
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
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
# Sample rate for high-volume events: llm.call.completed, embedding.call.completed,
# agent.iteration, agent.tool.call.completed. Deterministic on run_id so traces
# remain coherent end-to-end. Aggregate envelopes (RepresentationCompleted,
# DialecticCompleted, DreamRun, etc.) are NEVER sampled — they're calibration
# ground truth.
#
# Design trade-off: at rate < 1.0, aggregate events still emit but their
# high-volume children get dropped. Downstream `JOIN ... ON run_id` queries
# will see parents without complete children — this is intentional (the
# aggregates carry totals; detail events are best-effort), but consumers
# MUST NOT rebuild per-call analytics from the sampled children alone or
# they'll undercount. If you tune this below 1.0, audit dashboards/queries
# that join high-volume events to aggregate envelopes first.
HIGH_VOLUME_SAMPLE_RATE: Annotated[float, Field(default=1.0, ge=0.0, le=1.0)] = 1.0
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(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"]
# 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
)
@staticmethod
def _DEDUCTION_MODEL_CONFIG_DEFAULT() -> ConfiguredModelSettings:
# Minimal default; extra knobs would merge into env/TOML overrides.
return ConfiguredModelSettings(
transport="openai",
model="gpt-5.4-mini",
)
DEDUCTION_MODEL_CONFIG: ConfiguredModelSettings = Field(
default_factory=_DEDUCTION_MODEL_CONFIG_DEFAULT
)
@staticmethod
def _INDUCTION_MODEL_CONFIG_DEFAULT() -> ConfiguredModelSettings:
# Minimal default; extra knobs would merge into env/TOML overrides.
return ConfiguredModelSettings(
transport="openai",
model="gpt-5.4-mini",
)
INDUCTION_MODEL_CONFIG: ConfiguredModelSettings = Field(
default_factory=_INDUCTION_MODEL_CONFIG_DEFAULT
)
# Surprisal-based sampling subsystem
SURPRISAL: SurprisalSettings = Field(default_factory=SurprisalSettings)
@model_validator(mode="before")
@classmethod
def _merge_model_config_defaults(cls, data: Any) -> Any:
if isinstance(data, dict):
typed_data = cast(dict[str, Any], data)
_fill_defaults_for_nested_field(
typed_data,
"DEDUCTION_MODEL_CONFIG",
cls._DEDUCTION_MODEL_CONFIG_DEFAULT,
)
_fill_defaults_for_nested_field(
typed_data,
"INDUCTION_MODEL_CONFIG",
cls._INDUCTION_MODEL_CONFIG_DEFAULT,
)
return data # pyright: ignore[reportUnknownVariableType]
@model_validator(mode="after")
def _validate_specialist_token_budgets(self) -> "DreamSettings":
"""Ensure thinking_budget_tokens < max_output_tokens for each specialist."""
for name, cfg in (
("DEDUCTION_MODEL_CONFIG", self.DEDUCTION_MODEL_CONFIG),
("INDUCTION_MODEL_CONFIG", self.INDUCTION_MODEL_CONFIG),
):
if (
cfg.max_output_tokens is not None
and cfg.thinking_budget_tokens is not None
and cfg.max_output_tokens <= cfg.thinking_budget_tokens
):
raise ValueError(
f"dream.{name}.max_output_tokens must be greater than "
+ f"dream.{name}.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
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)
EMBEDDING: EmbeddingSettings = Field(default_factory=EmbeddingSettings)
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 "DIMENSIONS" in self.VECTOR_STORE.model_fields_set:
# VECTOR_STORE_DIMENSIONS is deprecated: EMBEDDING_VECTOR_DIMENSIONS
# is the single source of truth. Log a runtime-visible warning
# so operators see it (DeprecationWarning is filtered by Python's
# default config outside __main__/tests) and also raise the stdlib
# warning so tests can assert on it.
import warnings
message = (
"VECTOR_STORE_DIMENSIONS is deprecated; "
"EMBEDDING_VECTOR_DIMENSIONS is authoritative. "
"Drop VECTOR_STORE_DIMENSIONS from your .env."
)
logger.warning(message)
warnings.warn(message, DeprecationWarning, stacklevel=2)
self.VECTOR_STORE.DIMENSIONS = self.EMBEDDING.VECTOR_DIMENSIONS
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()
Reference in New Issue View Git Blame Copy Permalink