"""Central configuration for Lexora. Every tunable in the retrieval and generation pipeline is declared here so that a reviewer can see the entire behavioural surface of the system in one file, and so the eval harness can vary knobs (chunk size, rerank on/off) without touching pipeline code. Values are read from the environment, then `.env` at the repo root. See `.env.example`. """ from __future__ import annotations import functools from enum import StrEnum from pathlib import Path from typing import Final, Literal from pydantic import field_validator from pydantic_settings import BaseSettings, SettingsConfigDict # apps/api/app/core/settings.py -> apps/api/app/core -> app -> apps/api -> apps -> repo REPO_ROOT: Final[Path] = Path(__file__).resolve().parents[4] class LLMMode(StrEnum): """How Lexora obtains its language-model behaviour. ``auto`` use the Anthropic API when ``ANTHROPIC_API_KEY`` is set, else ``offline``. ``anthropic`` require the Anthropic API; fail loudly at startup if the key is absent. ``offline`` never call a network model. The query gate falls back to a deterministic rewriter/screener and generation to a deterministic extractive composer that quotes retrieved articles verbatim. The retrieval stack, the citation verifier and the refusal gate are fully exercised; only the natural-language phrasing of the final answer differs. Every response produced this way is labelled ``offline-extractive`` end to end, so a number measured in this mode can never be mistaken for a Claude number. """ AUTO = "auto" ANTHROPIC = "anthropic" OFFLINE = "offline" class Settings(BaseSettings): """Runtime configuration. Immutable once constructed.""" model_config = SettingsConfigDict( env_file=(REPO_ROOT / ".env"), env_file_encoding="utf-8", env_prefix="LEXORA_", extra="ignore", frozen=True, ) # ── paths ──────────────────────────────────────────────────────────────── repo_root: Path = REPO_ROOT corpus_dir: Path = REPO_ROOT / "corpus" var_dir: Path = REPO_ROOT / "var" # ── embedding (local, ONNX, zero cost) ─────────────────────────────────── embedding_model: str = "BAAI/bge-small-en-v1.5" embedding_dim: int = 384 # bge-small-en-v1.5 truncates its input at 512 tokens. Chunking must respect this # or the tail of every long chunk is silently dropped from the vector. embedding_max_tokens: int = 512 # bge asks for this prefix on the QUERY side only; documents are embedded bare. embedding_query_prefix: str = "Represent this sentence for searching relevant passages: " # ── reranking (local cross-encoder, ONNX) ──────────────────────────────── # Default chosen by measurement, not by reputation. Benchmarked on this corpus # against BAAI/bge-reranker-base (the larger, better-known checkpoint): # # model hit@1 worst trap score ms/query # ms-marco-MiniLM-L-6-v2 6/7 -4.55 2314 # bge-reranker-base 5/7 -1.95 12452 # # The smaller cross-encoder wins on ranking accuracy, is 5x faster, and — the part # that matters most here — separates out-of-corpus questions far more sharply, which # is what gives the refusal gate a usable margin. bge-reranker-base stays fully # supported: set LEXORA_RERANKER_MODEL to switch. See AUDIT.md "Reranker selection". reranker_model: str = "Xenova/ms-marco-MiniLM-L-6-v2" reranker_backend: Literal["fastembed", "sentence-transformers"] = "fastembed" # Cross-encoder cost is linear in padded sequence length. Capping the passage side # bounds worst-case latency; 384 leaves the p90 chunk (414 tokens) almost untouched. rerank_max_tokens: int = 384 # Candidates are sorted by length and scored in small batches so a single long # passage cannot pad the whole batch up to its length. Verified to produce # bit-identical scores to one large batch, for a ~39% latency reduction. rerank_batch_size: int = 4 # ── chunking ───────────────────────────────────────────────────────────── chunk_target_tokens: int = 600 chunk_overlap_tokens: int = 80 chunk_min_tokens: int = 24 # ── retrieval ──────────────────────────────────────────────────────────── dense_top_k: int = 20 sparse_top_k: int = 20 rrf_k: int = 60 fused_top_k: int = 20 rerank_top_k: int = 5 # ── refusal gate ───────────────────────────────────────────────────────── # Raw cross-encoder logit the best chunk must clear for the corpus to be considered # to cover the question. Model-specific — the scales of two rerankers are unrelated — # and calibrated against the labelled eval set rather than guessed. Regenerate with # `make calibrate` after changing `reranker_model`; see AUDIT.md "Refusal calibration". refusal_score_floor: float = -3.6 # Domain floor on the best DENSE similarity in the candidate set — an orthogonal # signal to the cross-encoder. Measured on the labelled set, the bi-encoder separates # in-corpus from out-of-corpus questions far better than the cross-encoder does # (0.951 vs 0.918 best achievable accuracy), because it captures whether a question # belongs to the corpus's subject domain at all rather than how topically relevant # the best passage is. See AUDIT.md "Refusal calibration". # Set to 0.0 = disabled. A dense floor was fitted (0.706) and REJECTED: it separated # the eval set well (0.951 accuracy) but was overfitted to that set's phrasing. Short # colloquial queries score low cosine against everything regardless of domain — # "How much end-of-service gratuity after 6 years?" scores 0.641 while retrieving # Article 51 at rank 1 — so the floor refused correct answers. Kept as a knob, off by # default, because the measurement is worth preserving; see AUDIT.md. refusal_dense_floor: float = 0.0 # How many near-miss chunks the amber refusal card shows, to prove it searched. refusal_near_miss_count: int = 3 # ── vector store ───────────────────────────────────────────────────────── # Unset -> embedded on-disk Qdrant under var/qdrant (zero credentials, zero cost). # Set -> Qdrant Cloud free tier. Identical client API either way. qdrant_url: str | None = None qdrant_api_key: str | None = None qdrant_collection: str = "lexora_chunks" # ── models ─────────────────────────────────────────────────────────────── llm_mode: LLMMode = LLMMode.AUTO anthropic_api_key: str | None = None answer_model: str = "claude-sonnet-4-6" guard_model: str = "claude-haiku-4-5-20251001" answer_max_tokens: int = 1200 guard_max_tokens: int = 400 # Which body of text this pipeline is reading: "corpus" or "workspace". It selects # the SourceProfile in rag/generate.py, which names the source in both the answer # system prompt and the refusal copy. Overridden per-request in resolve_pipeline # alongside refusal_score_floor -- a workspace holds the user's own upload, and # telling them it answers "only from the UAE Federal Labour Law" is a false claim # about their document. source_profile: str = "corpus" # Factual QA over statutes: never sample. temperature: float = 0.0 llm_timeout_s: float = 60.0 # ── observability ──────────────────────────────────────────────────────── langfuse_public_key: str | None = None langfuse_secret_key: str | None = None langfuse_host: str = "https://cloud.langfuse.com" # ── service ────────────────────────────────────────────────────────────── cors_allow_origins: str = "http://localhost:3020" # 10/minute throttled legitimate interactive use: the UI offers four one-click # questions, and someone working through them plus a few follow-ups hits the wall # inside a single minute -- as the demo pre-flight did, on its eighth request. # 30/minute still throttles a scraper hard while leaving room for a person. rate_limit: str = "30/minute" max_question_chars: int = 1000 max_history_turns: int = 6 request_timeout_s: float = 30.0 # ── workspace (bring your own document) ────────────────────────────────── # Uploaded documents are session-scoped and never written to disk, so every limit # here is also a memory limit: the ceiling is roughly # `workspace_max_sessions * workspace_max_docs * workspace_max_bytes` held at once, # and the container has already been measured against a 1 GB budget (AUDIT.md §6.4). workspace_max_bytes: int = 20_000_000 workspace_max_pages: int = 200 workspace_max_docs: int = 8 workspace_max_sessions: int = 200 # Idle sessions are purged on the next request rather than by a background task: a # sweeper thread would need its own lifecycle, and traffic is the only thing that # creates sessions in the first place. workspace_session_ttl_s: float = 3600.0 workspace_fetch_timeout_s: float = 20.0 workspace_max_redirects: int = 5 # An uploaded document has no calibrated floor. The law corpus's -3.6 was fitted # against 61 labelled questions about *that* corpus and does not transfer, so the # workspace uses a deliberately permissive floor and the UI says the number is not # calibrated. Refusing confidently on an un-evaluated corpus would be the same # overclaim the project exists to avoid. workspace_refusal_score_floor: float = -8.0 @field_validator("source_profile") @classmethod def _source_profile_must_be_known(cls, v: str) -> str: # Fail loudly rather than fall back. A typo here is invisible at runtime -- the # pipeline would quietly describe a user's own upload as "the UAE Federal Labour # Law", which is the exact false claim this field exists to prevent. from app.rag.generate import SOURCE_PROFILE_NAMES if v not in SOURCE_PROFILE_NAMES: raise ValueError( f"unknown source_profile {v!r}; expected one of " f"{', '.join(sorted(SOURCE_PROFILE_NAMES))}" ) return v @field_validator("temperature") @classmethod def _temperature_must_be_zero_for_factual_qa(cls, v: float) -> float: if v != 0.0: raise ValueError( "Lexora is a grounded QA system over statutes; temperature must be 0. " "Sampling trades faithfulness for fluency, which is the exact failure " "mode this project exists to eliminate." ) return v def model_post_init(self, context: object, /) -> None: del context if self.chunk_overlap_tokens < 0: raise ValueError("chunk_overlap_tokens must be >= 0") if self.chunk_overlap_tokens >= self.chunk_target_tokens: raise ValueError( f"chunk_overlap_tokens ({self.chunk_overlap_tokens}) must be smaller than " f"chunk_target_tokens ({self.chunk_target_tokens}); otherwise the splitter " "cannot make forward progress." ) # ── derived paths ──────────────────────────────────────────────────────── @property def pdf_dir(self) -> Path: return self.corpus_dir / "pdf" @property def manifest_path(self) -> Path: return self.corpus_dir / "manifest.json" @property def index_dir(self) -> Path: return self.var_dir / "index" @property def qdrant_path(self) -> Path: return self.var_dir / "qdrant" @property def models_cache_dir(self) -> Path: return self.var_dir / "models" @property def chunks_path(self) -> Path: """Canonical chunk store: every chunk with full text and metadata.""" return self.index_dir / "chunks.jsonl" @property def bm25_path(self) -> Path: """Persisted sparse index (tokenised corpus + doc-id order).""" return self.index_dir / "bm25.json" @property def vectors_path(self) -> Path: """Dense vectors for exactly the chunks in ``chunks_path``, in the same order. The third portable artefact. Committed for the same reason as the other two, and for one more: embedding is not batch-invariant. Padding length varies with batch composition, so re-embedding the corpus elsewhere reproduces the vectors only to ~4e-4 per component — enough to move a borderline score across the calibrated refusal floor. Shipping the vectors means the deployed index is the index the evaluation scored, bit for bit, rather than a near-copy of it. JSON rather than ``.npy`` deliberately. The Hugging Face Hub refuses binary files outside Xet/LFS, and routing this through LFS would mean every consumer needs ``lfs: true`` on checkout — including the CI job that builds the image, where a pointer file would produce a container that silently re-embedded instead of failing. Text costs ~1.4 MB instead of 272 KB and removes that whole failure mode. float32 values round-trip through JSON exactly (test_index_vectors.py). """ return self.index_dir / "vectors.json" @property def index_meta_path(self) -> Path: return self.index_dir / "index_meta.json" @property def eval_dir(self) -> Path: return self.repo_root / "eval" @property def eval_results_dir(self) -> Path: return self.repo_root / "eval" / "results" # ── derived behaviour ──────────────────────────────────────────────────── @property def effective_chunk_max_tokens(self) -> int: """Hard ceiling on a chunk, in embedding-model tokens. A chunk longer than the encoder's context window is truncated *inside* the encoder, so its tail contributes nothing to the vector while still being shown to the user as if it had been retrieved on its merits. Capping here keeps the indexed text and the embedded text identical. """ return min(self.chunk_target_tokens, self.embedding_max_tokens) @property def use_anthropic(self) -> bool: if self.llm_mode is LLMMode.OFFLINE: return False if self.llm_mode is LLMMode.ANTHROPIC: return True return bool(self.anthropic_api_key) @property def cors_origins(self) -> list[str]: return [o.strip() for o in self.cors_allow_origins.split(",") if o.strip()] @property def langfuse_enabled(self) -> bool: return bool(self.langfuse_public_key and self.langfuse_secret_key) @functools.lru_cache(maxsize=1) def get_settings() -> Settings: """Process-wide settings singleton.""" return Settings()