"""Centralized configuration: one Pydantic ``Settings`` tree. A single ``BaseSettings`` object composed of nested domain groups owns every config value. Modules read config via ``get_settings()`` (a cached singleton) instead of touching the environment directly. The one exception is the provider key-presence probes in ``quber.agents.factory``, which read ``os.getenv`` by design. Design notes: - Each group is a ``BaseSettings`` subclass (not a plain ``BaseModel``) so it loads the *flat* legacy env names — ``ANTHROPIC_API_KEY``, ``POSTGRES_*``, etc. — natively via per-field ``validation_alias``. A plain nested ``BaseModel`` would only populate through an ``env_nested_delimiter`` prefix (``LLM__ANTHROPIC_API_KEY``), which would break every existing ``.env``. - Renamed vars keep their old name as an ``AliasChoices`` fallback so a transition window exists (e.g. ``LOGFIRE_TOKEN`` -> ``PYDANTIC_LOGFIRE_TOKEN``). - ``Settings()`` construction performs no network or credential calls, so it is safe to call at package import (``files.cache.init_s3_cache`` does, via ``quber/__init__.py``). - AWS credentials are intentionally absent: they stay on the boto3 default credential chain (env / ``~/.aws`` / IAM role). Only non-secret AWS knobs (S3 cache dir, TTL, region) live here. """ from __future__ import annotations from functools import lru_cache from pathlib import Path from typing import Annotated, Any, Literal, Optional from pydantic import AliasChoices, BeforeValidator, Field from pydantic_settings import BaseSettings, SettingsConfigDict # Shared config: read `.env` (relative to CWD), ignore unknown env vars, # case-insensitive env matching. Each group reuses this so they all load the # same `.env` and flat environment. BASE_CONFIG = SettingsConfigDict(env_file=".env", extra="ignore", case_sensitive=False) # Truthy token set for `LegacyBool`. Pydantic's native bool parser would raise # on an empty or unrecognized value (e.g. `TRACE_TO_LANGSMITH=`). `LegacyBool` # instead treats these tokens (trimmed, case-insensitive) as True and None or # any other string as False, and never errors. TRUTHY_TOKENS = {"1", "true", "yes", "on"} def coerce_legacy_bool(value: Any) -> Any: if isinstance(value, str): return value.strip().lower() in TRUTHY_TOKENS if value is None: return False return value LegacyBool = Annotated[bool, BeforeValidator(coerce_legacy_bool)] # The model the extraction agents fall back to when neither an explicit model # nor QUBER_LLM_MODEL/ANTHROPIC_MODEL is set. One project-wide default so the # pinned model lives in a single place rather than being repeated in every # agent. A dated id (not a rolling alias) keeps extraction reproducible run to # run, the same reason the structural agents pin temperature 0. DEFAULT_LLM_MODEL = "claude-haiku-4-5-20251001" class LLMSettings(BaseSettings): """LLM model selection, Anthropic/OpenAI credentials, per-agent backends.""" model_config = BASE_CONFIG model: Optional[str] = Field( default=None, validation_alias=AliasChoices("QUBER_LLM_MODEL", "ANTHROPIC_MODEL"), ) anthropic_api_key: Optional[str] = Field(default=None, validation_alias="ANTHROPIC_API_KEY") anthropic_auth_token: Optional[str] = Field(default=None, validation_alias="ANTHROPIC_AUTH_TOKEN") # Shared with the embeddings layer; defined here once (single source of truth). openai_api_key: Optional[str] = Field(default=None, validation_alias="OPENAI_API_KEY") # Per-agent backend selectors. Kept as separate fields (not collapsed to one) # so each agent retains its independent override, matching today's behavior. llm_backend: str = Field(default="api", validation_alias="QUBER_LLM_BACKEND") unifier_backend: str = Field(default="api", validation_alias="QUBER_UNIFIER_BACKEND") classifier_backend: str = Field(default="api", validation_alias="QUBER_CLASSIFIER_BACKEND") detector_backend: str = Field(default="api", validation_alias="QUBER_DETECTOR_BACKEND") grid_locator_backend: str = Field(default="api", validation_alias="QUBER_GRID_LOCATOR_BACKEND") cell_reader_backend: str = Field(default="api", validation_alias="QUBER_CELL_READER_BACKEND") completeness_backend: str = Field(default="text", validation_alias="QUBER_COMPLETENESS_BACKEND") # The capture advisor recommends a Camelot retry adjustment when a # capture drop is detected (values in the region text layer missing from # the grid). `off` disables the repair loop entirely. capture_advisor_backend: str = Field(default="api", validation_alias="QUBER_CAPTURE_ADVISOR_BACKEND") # The status inspector verifies each unboxed cell's proposed condition # against the table image. A condition the image contradicts becomes # `defect`, and one it cannot positively confirm becomes `unverified`. # `off` disables inspection; the proposed statuses are then recorded as-is. status_inspector_backend: str = Field(default="api", validation_alias="QUBER_STATUS_INSPECTOR_BACKEND") # The figure corrector names the text left on a page that belongs to a figure # the scan has already read — axis ticks, gridline labels — so it is removed # rather than indexed beside the reading that supersedes it. `off` removes # nothing and keeps every fragment. figure_correction_backend: str = Field(default="api", validation_alias="QUBER_FIGURE_CORRECTION_BACKEND") # The figure-value readers give every value printed in a figure a second, # independent read against the page image and the scan's prose, so the two # can be tied cell by cell. `off` skips the figure-value reconciliation # pass entirely. figure_values_backend: str = Field(default="api", validation_alias="QUBER_FIGURE_VALUES_BACKEND") # The grid locator reads a labelled grid off the page image — a vision task # the smaller model resolves at too coarse a fidelity (it under-flags the # row-label stub column and wobbles a row at stacked boundaries), so it runs # on a stronger model than the rest of the extraction agents. Overridable; # falls back to the project-wide model when unset. grid_locator_model: Optional[str] = Field( default="claude-sonnet-5", validation_alias="QUBER_GRID_LOCATOR_MODEL" ) # The structure-correction (vet) agent's own model. That agent carries the # most compliance-sensitive output in the pipeline — footnote marker # catalogues, quoted carrying cells, footnote/section judgements — and the # smaller model intermittently omits whole marks lists on real filings. # Overridable per the grid-locator pattern; falls back to the project-wide # model when unset. Validated on a real filing: the smaller model emitted # no marks list at all on some tables; the stronger model emitted them # complete with identical results whether scoped here or run pipeline-wide. vet_model: Optional[str] = Field(default="claude-sonnet-5", validation_alias="QUBER_VET_MODEL") # The figure corrector's own model, pinned for the same reason as the two # above. Asked which text on a page belongs to a figure the scan has read, # the smaller model deleted a four-line annotation about rate sensitivity # while its own stated reason called it an annotation, and missed seven of a # page's forty axis tick labels. The stronger model returned all forty ticks # and left the annotation alone, identically across three runs. figure_correction_model: Optional[str] = Field( default="claude-sonnet-5", validation_alias="QUBER_FIGURE_CORRECTION_MODEL" ) # The local figure-value reader's own model. Reading every number printed # inside a chart off the page image is a vision task, pinned to the # stronger model for the same reason as the figure corrector. Overridable; # falls back to the project-wide model when unset. figure_values_model: Optional[str] = Field( default="claude-sonnet-5", validation_alias="QUBER_FIGURE_VALUES_MODEL" ) class DBSettings(BaseSettings): """Postgres connection parameters for the RAG store.""" model_config = BASE_CONFIG user: str = Field(default="quber", validation_alias="POSTGRES_USER") password: str = Field(default="quber_dev", validation_alias="POSTGRES_PASSWORD") host: str = Field(default="localhost", validation_alias="POSTGRES_HOST") port: int = Field(default=5432, validation_alias="POSTGRES_PORT") db: str = Field(default="quber_rag", validation_alias="POSTGRES_DB") class S3CacheSettings(BaseSettings): """Non-secret S3 ingestion-cache knobs. Credentials stay on the boto3 chain.""" model_config = BASE_CONFIG cache_dir: Path = Field(default=Path(".cache/s3"), validation_alias="QUBER_S3_CACHE_DIR") ttl_seconds: float = Field(default=5 * 86400, validation_alias="QUBER_S3_CACHE_TTL") region: Optional[str] = Field( default=None, validation_alias=AliasChoices("QUBER_S3_REGION", "AWS_REGION"), ) class EmbeddingSettings(BaseSettings): """Embedding provider/device selection. OpenAI key lives in ``LLMSettings``.""" model_config = BASE_CONFIG provider: str = Field(default="local", validation_alias="EMBEDDING_PROVIDER") device: str = Field(default="cuda", validation_alias="EMBEDDING_DEVICE") class LandingSettings(BaseSettings): """Landing.AI ADE parse credentials for the standalone extraction path.""" model_config = BASE_CONFIG # ADE_API_KEY is our .env name; VISION_AGENT_API_KEY is the name the # landingai-ade SDK itself reads, accepted so an SDK-style environment # also works. api_key: Optional[str] = Field( default=None, validation_alias=AliasChoices("ADE_API_KEY", "VISION_AGENT_API_KEY"), ) class TypeSafeSettings(BaseSettings): """TypeSafe scoring API: credential, pinned model, the chunk grade cut, and how hard the playground's ranker drives it.""" model_config = BASE_CONFIG # TYPESAFE_API_KEY is the SDK's own variable name. The key is still passed # explicitly from here into the SDK client, so a missing key is reported by # settings the way a missing Anthropic key is, not by the SDK at the first # request. api_key: Optional[str] = Field(default=None, validation_alias="TYPESAFE_API_KEY") model: str = Field(default="jev-1.13.0", validation_alias="QUBER_TYPESAFE_MODEL") # Chunks whose position on the 0 to 3 rubric is below this do not reach the # answer model. 1.5 is the midpoint between "same topic" and "supports"; # the default sits just under it so a chunk Jev is nearly sure supports the # item is kept. grade_cut: float = Field(default=1.4, validation_alias="QUBER_TYPESAFE_GRADE_CUT") # Calls in flight at once. 32 is the highest value measured, with no # throttling from the API. concurrency: int = Field(default=32, validation_alias="QUBER_TYPESAFE_CONCURRENCY") # The SDK's own default, made explicit: retries after the first attempt on # 429, 5xx, connection and timeout errors. max_retries: int = Field(default=2, validation_alias="QUBER_TYPESAFE_MAX_RETRIES") timeout_seconds: float = Field(default=60.0, validation_alias="QUBER_TYPESAFE_TIMEOUT_SECONDS") class PlaygroundSettings(BaseSettings): """The answering playground: where its local corpus lives, how many ingests may run at once, and which step ranks retrieved chunks.""" model_config = BASE_CONFIG # Staged PDFs, parse artifacts and uploads. Relative paths resolve against # the server's working directory, the same convention as the S3 cache dir. data_dir: Path = Field(default=Path("data/playground"), validation_alias="QUBER_PLAYGROUND_DATA_DIR") # How many uploads run their pipelines at once. On the developer host # every stage is a subprocess, so two concurrent extractions are two # `quber fuse` processes sharing one CUDA device, each also making model # calls. Two overlaps the non-GPU parts of a run without saturating the # device; one is for a machine that cannot spare the contention. When the # hosted playground sends its GPU stages to RunPod, the docling parse runs # on the worker and no local device is shared. upload_concurrency: int = Field(default=2, validation_alias="QUBER_PLAYGROUND_UPLOAD_CONCURRENCY") # The durable home of every document's files when the playground is # hosted: an s3:// prefix holding one / prefix per document, to # which each stage's outputs are copied and from which a served PDF is # fetched when the local copy is gone. Unset, the local data_dir is the # only copy, which is the developer host's mode. artifacts_uri: Optional[str] = Field(default=None, validation_alias="QUBER_PLAYGROUND_ARTIFACTS_URI") # The RunPod serverless endpoint the hosted playground sends its GPU # stages to, the docling parse and the chunk embedding, with the key that # authorizes the submission. The stages go to RunPod only when these two # and artifacts_uri are all set. If any one is unset, both run on this host. runpod_endpoint_id: Optional[str] = Field(default=None, validation_alias="RUNPOD_ENDPOINT_ID") runpod_api_key: Optional[str] = Field(default=None, validation_alias="RUNPOD_API_KEY") # Where the hosted playground reports signed-in activity: a CloudWatch # namespace for one metric, the count of requests that carried a valid # session. The idle alarm watches it, because the public hostname's # load-balancer request count never reaches zero. Unset, nothing is sent. activity_namespace: Optional[str] = Field( default=None, validation_alias="QUBER_PLAYGROUND_ACTIVITY_NAMESPACE" ) # The sign-in the hosted playground asks for: the WorkOS environment's API # key and client ID, the one organization whose members may enter, and the # secret that signs the session cookie. The four are set together; all # unset disables the sign-in, which is the developer host's mode, and a # partial set is refused at startup rather than silently running open. workos_api_key: Optional[str] = Field(default=None, validation_alias="WORKOS_API_KEY") workos_client_id: Optional[str] = Field(default=None, validation_alias="WORKOS_CLIENT_ID") workos_organization_id: Optional[str] = Field(default=None, validation_alias="WORKOS_ORGANIZATION_ID") session_secret: Optional[str] = Field(default=None, validation_alias="PLAYGROUND_SESSION_SECRET") # Which step orders retrieved chunks before the answer model reads them: # the Haiku selection agent over a window of the fused ranking, or Jev # over the whole of it. A backend setting only; it never reaches a client # payload or the UI. ranker: Literal["haiku", "jev"] = Field(default="haiku", validation_alias="QUBER_PLAYGROUND_RANKER") # Questions a batch run answers at once. Each one holds an answer-model # call, and the provider returns 529 under load (observed 2026-07-29 # during a sweep), so the default is deliberate. question_concurrency: int = Field(default=5, validation_alias="QUBER_PLAYGROUND_QUESTION_CONCURRENCY") class ObservabilitySettings(BaseSettings): """Logfire + LangSmith tracing configuration.""" model_config = BASE_CONFIG logfire_token: Optional[str] = Field( default=None, validation_alias=AliasChoices("PYDANTIC_LOGFIRE_TOKEN", "LOGFIRE_TOKEN"), ) enable_logfire: LegacyBool = Field(default=True, validation_alias="QUBER_ENABLE_LOGFIRE") trace_to_langsmith: LegacyBool = Field(default=False, validation_alias="TRACE_TO_LANGSMITH") langsmith_api_key: Optional[str] = Field(default=None, validation_alias="CC_LANGSMITH_API_KEY") langsmith_project: str = Field(default="default", validation_alias="CC_LANGSMITH_PROJECT") langsmith_debug: LegacyBool = Field(default=False, validation_alias="CC_LANGSMITH_DEBUG") class Settings(BaseSettings): """Root settings object. Access via :func:`get_settings`.""" model_config = BASE_CONFIG log_level: str = Field(default="INFO", validation_alias="QUBER_LOG_LEVEL") max_concurrent_tables: int = Field(default=5, validation_alias="QUBER_MAX_CONCURRENT_TABLES") llm: LLMSettings = Field(default_factory=LLMSettings) db: DBSettings = Field(default_factory=DBSettings) s3_cache: S3CacheSettings = Field(default_factory=S3CacheSettings) embeddings: EmbeddingSettings = Field(default_factory=EmbeddingSettings) landing: LandingSettings = Field(default_factory=LandingSettings) typesafe: TypeSafeSettings = Field(default_factory=TypeSafeSettings) playground: PlaygroundSettings = Field(default_factory=PlaygroundSettings) obs: ObservabilitySettings = Field(default_factory=ObservabilitySettings) @lru_cache def get_settings() -> Settings: """Return the process-wide cached :class:`Settings` singleton. Construction reads ``.env`` and the environment only — no network or credential resolution — so this is safe to call at import time. """ return Settings() __all__ = [ "DEFAULT_LLM_MODEL", "DBSettings", "EmbeddingSettings", "LLMSettings", "LandingSettings", "ObservabilitySettings", "PlaygroundSettings", "S3CacheSettings", "Settings", "TypeSafeSettings", "get_settings", ]