"""Configuration for the template-agnostic v2 backend. All values are overridable via environment variables (prefix-free) or a ``.env`` file at the repository root. See the package README for the deployment model. """ from __future__ import annotations from functools import lru_cache from pathlib import Path from pydantic import Field from pydantic_settings import BaseSettings, SettingsConfigDict # backend/config.py -> repo root is one level up from this package. REPO_ROOT = Path(__file__).resolve().parents[1] class Settings(BaseSettings): """Central v2 settings object.""" model_config = SettingsConfigDict( # Resolve from repo root, not process cwd — uvicorn launched from the wrong # directory (e.g. a path typo) must still pick up DATA_DIR / HF_HOME. env_file=str(REPO_ROOT / ".env"), env_file_encoding="utf-8", extra="ignore", ) # ── LLM (OpenAI) ───────────────────────────────────────────────────────── openai_api_key: str = Field(default="", description="OpenAI API key") mapping_model: str = Field( default="gpt-4o-mini", description="Model used to map surveyor notes onto retrieved master paragraphs.", ) discovery_model: str = Field( default="gpt-4o-mini", description="Model used for master-template schema discovery (JSON mode).", ) grounding_model: str = Field( default="gpt-4o-mini", description="Model used for the PII / grounding audit pass.", ) openai_request_timeout_seconds: float = Field( default=60.0, ge=5.0, le=600.0, description="Hard timeout for OpenAI chat and embedding HTTP calls (seconds).", ) openai_pipeline_timeout_seconds: float = Field( default=20.0, ge=5.0, le=120.0, description=( "Strict per-call timeout (seconds) for chat/embeddings inside the " "report generation pipeline." ), ) max_concurrent_llm_calls: int = Field( default=10, ge=1, le=100, description=( "Maximum concurrent in-flight OpenAI chat completion HTTP requests " "across parallel section workers." ), ) section_concurrency: int = Field( default=4, ge=1, le=54, description=( "Max sections processed in parallel during report generation. Default " "4 keeps jina embedder + reranker + OpenAI within a 4 GB GPU and small " "Windows paging file; 54 fans out every section at once and can hard-" "abort the process (browser shows 'Failed to fetch')." ), ) openai_rate_limit_max_retries: int = Field( default=5, ge=0, le=20, description="Retry attempts after OpenAI HTTP 429 rate-limit responses.", ) openai_rate_limit_backoff_base_seconds: float = Field( default=1.0, ge=0.1, le=60.0, description="Initial exponential backoff base (seconds) for 429 retries.", ) openai_rate_limit_backoff_max_seconds: float = Field( default=60.0, ge=1.0, le=300.0, description="Maximum backoff delay (seconds) between 429 retries.", ) # ── Embeddings (local jina-embeddings-v3 default; OpenAI optional) ──────── embedding_provider: str = Field( default="local", description="Embedding backend: 'local' (sentence-transformers) or 'openai'.", ) local_embedding_model: str = Field( default="jinaai/jina-embeddings-v3", description=( "sentence-transformers model used when embedding_provider='local'. " "jina-embeddings-v3 is a 1024-dim, 8192-token model with a custom " "architecture loaded via trust_remote_code; it uses asymmetric task " "adapters (retrieval.passage for documents, retrieval.query for queries)." ), ) local_embedding_dtype: str = Field( default="bfloat16", description=( "Weight dtype for the local embedder: 'auto' (float16 on GPU, bfloat16 " "on CPU), or an explicit 'float32' / 'float16' / 'bfloat16'. bfloat16 " "halves resident memory for the 0.57B jina model to ~1.14 GB." ), ) local_embedding_trust_remote_code: bool = Field( default=True, description=( "Allow the local embedder to execute the model's remote modeling code. " "Required by jina-embeddings-v3 (custom architecture). Set False to pin " "to models that ship no remote code (e.g. all-MiniLM-L6-v2)." ), ) local_embedding_device: str = Field( default="cpu", description=( "Device for the backend local embedder: 'auto' (CUDA if available, else " "CPU), 'cuda', or 'cpu'. Force 'cpu' on small GPUs (<6 GB) where jina-v3 " "(0.57B) plus a reranker will not fit — the embedder auto-falls back to " "CPU on unrecoverable CUDA OOM regardless, but 'cpu' avoids the churn." ), ) local_embedding_batch_size: int = Field( default=8, description=( "Max texts per embedder forward pass. Small default keeps jina-v3 within " "a 4 GB VRAM budget during reingest. On CUDA OOM the embedder halves this " "(sticky, per-process) down to 1, then falls back to CPU — so this is a " "starting ceiling, not a hard limit. Raise it on larger GPUs for speed." ), ) local_embedding_max_seq_length: int = Field( default=8192, description=( "Token truncation length for the local embedder. jina-v3's native window " "is 8192; keeping it means REFERENCE chunks (reference_paragraph_max_chars " "= 8000 chars ~= 2000 tokens) embed in full with no truncation. Attention " "is O(L^2) under native (non-flash) attention, so on low-VRAM GPUs pair " "this with a small local_embedding_batch_size; the embedder also auto-" "halves the batch and falls back to CPU on OOM. Lower it to force " "truncation, or set 0 for the model default." ), ) hf_offline: bool = Field( default=False, description=( "Run HuggingFace fully offline: load embedder + reranker weights and " "trust_remote_code modules from the local HF cache only, with zero " "network calls. Exported as HF_HUB_OFFLINE=1 + TRANSFORMERS_OFFLINE=1 " "at startup (before any model import). Required in production and on " "slow/blocked networks: for trust_remote_code models (jina-v3, " "jina-reranker-v3) even a fully-cached load otherwise issues an online " "HEAD check for custom_st.py that hangs or fails. Weights must be " "pre-provisioned (baked into the image or a mounted cache). Leave False " "for the first download on a fresh host, then set True." ), ) openai_embedding_model: str = Field( default="text-embedding-3-small", description="OpenAI embedding model used when embedding_provider='openai'.", ) # ── PII scrubbing ──────────────────────────────────────────────────────── spacy_model: str = Field( default="en_core_web_sm", description=( "spaCy NER model for PII scrubbing. en_core_web_sm is bundled via " "requirements.txt; set en_core_web_md or en_core_web_trf in .env for " "higher accuracy when installed." ), ) pii_use_spacy: bool = Field( default=True, description="Enable the spaCy NER pass on top of the always-on regex pass.", ) pii_scrub_audit_dump: bool = Field( default=True, description=( "Write verbose PII scrub audit artifacts at REFERENCE ingest under " "/pii_scrub_audit///: redacted_content.txt " "(the whole document after redaction, section by section) and " "redactions.json (every redacted value with its location — section, " "paragraph, chunk, char offset, context — plus whitelisted survey terms " "and dropped chunks). Also writes a static whitelist_catalog.json. " "pii_mapping.json is always written for scrubbed content regardless of " "this flag. Disable verbose dump via PII_SCRUB_AUDIT_DUMP=false." ), ) pii_scrub_audit_max_surface_chars: int = Field( default=200, ge=0, le=2000, description=( "Max characters of each redacted/whitelisted surface string and context " "snippet in redactions.json (0 = full text)." ), ) # ── Operator bundle (Master Standard report and paragraphs/) ───────────── master_template_dir: str = Field( default="Master Standard report and paragraphs", description="Folder holding the report template PDF and standard-paragraphs Word file.", ) report_template_filename: str = Field( default="SAMPLE LEVEL 3 REPORT NCS.pdf", description=( "Report template (PDF): defines section structure, order and ratings. " "Schema discovery reads this file." ), ) standard_paragraphs_filename: str = Field( default="HB-BS STANDARD PARAS v6 Sept 2015.doc", description=( "Standard paragraphs (Word): firm-approved boilerplate wording per section. " "Ingested into the MASTER RAG tier." ), ) # Backward-compatible alias for the standard-paragraphs file. master_template_filename: str = Field( default="HB-BS STANDARD PARAS v6 Sept 2015.doc", description="Deprecated alias for standard_paragraphs_filename.", ) master_template_auto_ingest: bool = Field( default=False, description=( "Ingest the operator's shared standard-paragraph master into the MASTER " "RAG tier on startup. Default False: in the per-tenant model every tenant " "supplies its own past reports (REFERENCE tier) and generation is sourced " "exclusively from those at all interference levels. Enable only if a firm " "wants a shared boilerplate master seeded for the default tenant. The " "canonical RICS L3 schema is always installed regardless of this flag." ), ) master_template_prebuilt_schema: str = Field( default="", description="Optional path to a prebuilt schema.json; skips discovery when set and present.", ) master_template_prebuilt_faiss: str = Field( default="", description="Optional path to a prebuilt MASTER FAISS dir; skips embedding when set and present.", ) master_template_upload_enabled: bool = Field( default=False, description="Enable the gated admin override routes for replacing the master at runtime.", ) # ── Optional reference uploads (past completed reports, style only) ──────── reference_auto_ingest_enabled: bool = Field( default=False, description=( "When true, scan reference_auto_ingest_dir for extra past reports. " "The report-template PDF and standard-paragraphs Word file are never " "ingested as references." ), ) reference_auto_ingest_dir: str = Field( default="", description="Folder to scan for reference docs; empty means use master_template_dir.", ) reference_auto_ingest_globs: str = Field( default="*.pdf,*.docx,*.doc", description="Comma-separated globs scanned for reference docs (master filename excluded).", ) # ── RAG / retrieval ────────────────────────────────────────────────────── data_dir: str = Field( default_factory=lambda: str(Path.home() / ".rics_v2"), description="Root directory for per-tenant schema + FAISS artifacts.", ) default_tenant_id: str = Field( default="default", description="Tenant that receives the operator master at startup.", ) retrieval_top_k: int = Field(default=5, ge=1, le=50) reference_baseline_top_k: int = Field( default=4, ge=1, le=50, description="Top REFERENCE-tier blocks retrieved as stylistic baseline scaffolding.", ) retrieval_section_boost: float = Field( default=0.25, ge=0.0, le=1.0, description="FAISS score boost when chunk section_id matches the alias-resolved id.", ) retrieval_lexical_boost: float = Field( default=0.04, ge=0.0, le=0.5, description="Per shared content-token boost when reranking paragraphs against notes.", ) hybrid_retrieval_enabled: bool = Field( default=True, description=( "Fuse a sparse BM25 arm with the dense FAISS arm at retrieval time " "(Reciprocal Rank Fusion). When True, lexically-strong chunks that " "embed poorly still enter the candidate pool before reranking. " "Disable to fall back to dense-only retrieval." ), ) hybrid_rrf_k: int = Field( default=60, ge=1, le=1000, description=( "Reciprocal Rank Fusion damping constant. A rank-r (0-based) hit " "contributes 1/(k+r+1); the standard value is 60." ), ) hybrid_bm25_k1: float = Field( default=1.5, ge=0.0, le=5.0, description="BM25 term-frequency saturation (k1). Typical range 1.2-2.0.", ) hybrid_bm25_b: float = Field( default=0.75, ge=0.0, le=1.0, description="BM25 document-length normalisation (b). 0=no normalisation.", ) reference_subchunk_indexing_enabled: bool = Field( default=True, description=( "Small-to-big dense retrieval for the REFERENCE tier. The dense arm " "scores per-parent the BEST overlapping sub-window instead of the " "head-truncated full chunk, mitigating MiniLM's 256-token head-bias on " "long past-report chunks. Subchunk hits collapse to their parent before " "RRF with the UNCHANGED parent-level BM25 arm; scrub/assembly/reranker " "are untouched (search still returns parent chunks). The view is built " "lazily in-memory and rebuilt on meta change (same pattern as the BM25 " "arm), so it rebuilds on first reference search after a restart. " "Enabled by default: the deployed subchunk-hybrid arm cleared the locked " "benchmark gate (late-slice MRR +0.38/+0.37, R@10 +0.27/+0.26 over " "hybrid; early slice unchanged). See backend/tests/golden/" "retrieval_benchmark.py. Set False to revert to dense-on-full-chunk." ), ) reference_subchunk_words: int = Field( default=120, ge=16, le=512, description=( "Sub-window length in whitespace tokens for reference subchunk " "indexing. ~120 words (~160 MiniLM tokens) sits inside the model's " "256-token window so each window embeds without head-truncation." ), ) reference_subchunk_overlap: int = Field( default=30, ge=0, le=256, description=( "Sub-window overlap in whitespace tokens. Overlap keeps a finding that " "straddles a window boundary intact in at least one window." ), ) reference_subchunk_embed_batch: int = Field( default=512, ge=16, le=4096, description=( "Sub-windows embedded per embed_documents() call when building the " "subchunk view. Bounds the transient list-of-lists allocation: a large " "reference tier yields tens of thousands of windows, and embedding them " "all at once spikes host RAM (each MiniLM vector round-trips through a " "Python list before re-packing to float32). Batching writes straight " "into a preallocated matrix so peak memory stays flat regardless of " "corpus size." ), ) dedupe_chunks_on_ingest: bool = Field( default=True, description=( "Drop duplicate chunks at ingest BEFORE scrubbing/embedding so they are " "never processed or stored. A chunk is a duplicate when another chunk " "with the same (section_id, normalised text) was already seen in this " "batch or already exists in the tier. Prevents the index bloat that " "follows repeated uploads / re-ingests (the OOM root cause) and keeps " "BM25 document frequencies + the subchunk view honest." ), ) prompt_literature_few_shot_enabled: bool = Field( default=True, description=( "Inject curated few-shot user/assistant turns from the operator " "literature corpus (My literature April 2026) before the live task." ), ) prompt_chain_of_thought_enabled: bool = Field( default=True, description=( "Append internal chain-of-thought protocols to system prompts. " "Models reason through steps internally; output contracts unchanged." ), ) prompt_dynamic_literature_enabled: bool = Field( default=True, description=( "Retrieve task-relevant exemplars LIVE from the operator literature " "corpus (segmented + hybrid-indexed at runtime) and inject them as " "fenced phrasing references plus any mined draft->edited few-shot " "pairs. Keyed on the live section + notes. Augments the curated " "static few-shot; the foreign-fact reducers remain the safety net." ), ) literature_corpus_filename: str = Field( default="My literature April 2026.docx", description=( "Operator literature corpus file used for dynamic exemplar retrieval. " "Absolute path, or resolved relative to the project root." ), ) literature_exemplar_top_k: int = Field( default=3, ge=0, le=10, description="Max literature reference exhibits injected per LLM call.", ) literature_exemplar_pairs_max: int = Field( default=1, ge=0, le=4, description="Max mined draft->edited few-shot pairs injected per LLM call.", ) literature_exemplar_min_score: float = Field( default=0.2, ge=0.0, le=1.0, description=( "Minimum dense cosine for a literature passage to be injected. Floors " "out topically-irrelevant exhibits so we never pad prompts with noise." ), ) literature_redact_specifics: bool = Field( default=True, description=( "Redact hard specifics (money, dates, measurements, percentages, URLs, " "phone numbers) from literature exhibits before injection so the " "phrasing survives but other-property facts cannot bleed in." ), ) reference_cross_encoder_enabled: bool = Field( default=True, description=( "Apply the jina-reranker-v3 listwise reranker as the final re-scoring " "stage on the reference-mapping shortlist. Degrades gracefully to the " "multi-signal reranker if the model / deps / weights are unavailable." ), ) reference_cross_encoder_model: str = Field( default="jinaai/jina-reranker-v3", description=( "HuggingFace model id of the reranker. jina-reranker-v3 is a listwise " "reranker loaded with trust_remote_code (custom architecture)." ), ) reference_cross_encoder_top_n: int = Field( default=8, ge=2, le=50, description=( "How many top multi-signal candidates to re-score with the " "cross-encoder. Kept small to bound CPU latency." ), ) reference_cross_encoder_doc_chars: int = Field( default=1600, ge=200, le=8000, description="Per-candidate character cap fed to the cross-encoder.", ) reference_cross_encoder_dtype: str = Field( default="auto", description=( "Weight dtype for the reranker: 'auto' (float16 on GPU, bfloat16 on CPU " "to halve RAM), or an explicit 'float32' / 'float16' / 'bfloat16'. Half " "precision keeps the 0.6B model well under ~1.2GB." ), ) reference_cross_encoder_device: str = Field( default="auto", description=( "Device for the jina-reranker-v3 reranker: 'auto' (CUDA when available, " "else CPU), 'cpu' (force CPU), or 'cuda'. On a small GPU (e.g. 4 GB) the " "reranker cannot co-reside with the resident jina embedder — the weight " "load / forward pass can hard-abort the process (native CUDA/driver " "abort Python cannot catch). Set 'cpu' to keep the embedder on GPU while " "the reranker runs in host RAM (bounded latency: top_n candidates only)." ), ) reference_cross_encoder_warmup: bool = Field( default=False, description=( "Load the reranker once at server startup instead of lazily on the " "first reference mapping. Disabled by default: on Windows hosts with a " "small paging file the eager weight load can hard-abort (os error 1455) " "before Python can catch it. Set true on servers with ample RAM to " "avoid a lazy-load spike mid-generation. Guarded by the free-RAM check " "and wrapped non-fatally when enabled." ), ) reference_cross_encoder_min_free_gb: float = Field( default=1.5, ge=0.0, le=64.0, description=( "Minimum free host RAM (GiB) required to load the reranker. The load " "path materialises the full fp16 model in host RAM (~1.2 GB) before " "moving it to the device, and on a memory-tight host attempting it can " "hard-abort the process (OpenBLAS/safetensors OOM) instead of raising. " "Below this floor the reranker is skipped and retrieval degrades to " "multi-signal. Set 0 to disable the check." ), ) retrieval_debug_dump: bool = Field( default=False, description=( "Write the hybrid-retrieval candidates (dense embedder + BM25, RRF-fused) " "to a human-readable file BEFORE jina-reranker-v3 reorders them, so you " "can audit whether the right chunks are retrieved. Off by default; enable " "via RETRIEVAL_DEBUG_DUMP=true. Files land in /retrieval_debug/ " "as retrieval_.log (one appended block per section retrieval). " "Each block includes tiktoken counts (cl100k_base proxy) for the query, " "full chunk, embedder truncation risk, and reranker char-capped feed." ), ) retrieval_debug_max_text_chars: int = Field( default=1500, ge=0, le=20000, description=( "Max characters of each candidate chunk written to the retrieval-debug " "dump (0 = full chunk). Keeps the audit file readable when chunks are " "large (~8000-char reference chunks)." ), ) note_routing_mode: str = Field( default="keyword", description="Note routing: 'keyword' (deterministic regex) or 'rag' (embedding anchors).", ) note_rag_match_min_score: float = Field( default=0.72, ge=0.0, le=1.0, description=( "Minimum cosine similarity for a surveyor note to be mapped onto a " "canonical section anchor. Below this, the note is surfaced as UNASSIGNED." ), ) note_rag_ambiguity_margin: float = Field( default=0.05, ge=0.0, le=0.5, description=( "Minimum cosine gap between the top two section-anchor matches. When " "the margin is narrower, the note is treated as ambiguous and UNASSIGNED." ), ) note_baseline_lexical_min_overlap: float = Field( default=0.12, ge=0.0, le=1.0, description=( "Minimum token-overlap ratio between a note and the section baseline " "to allow in-place mapping without a strong per-note RAG hit." ), ) paragraph_min_chars: int = Field(default=80, ge=1, le=2000) paragraph_max_chars: int = Field(default=1200, ge=100, le=4000) chunk_overlap: int = Field(default=120, ge=0, le=1000) reference_paragraph_max_chars: int = Field( default=8000, ge=500, le=20000, description="Max characters per REFERENCE-tier chunk (long-form past report sections).", ) reference_chunk_overlap: int = Field( default=1500, ge=0, le=5000, description="Character overlap when splitting oversized REFERENCE sections.", ) reference_section_complete_enabled: bool = Field( default=True, description=( "Assemble the WHOLE past-report section as the mapping baseline (every " "chunk for the chosen source+section, in document order) rather than only " "the top-K semantically nearest chunks. Also lets a section that exists in " "the index but was missed by similarity search fall back to a metadata " "fetch before degrading to NOTES_ONLY. Disable to restore top-K-only." ), ) reference_section_complete_max_chars: int = Field( default=12000, ge=1000, le=40000, description=( "Safety cap on an assembled section-complete baseline. Chunks are added " "in document order until this budget is reached, bounding the mapping " "prompt while still covering long (30–50 line) past-report sections." ), ) # ── Generation behaviour ───────────────────────────────────────────────── default_survey_level: int = Field( default=3, ge=1, le=3, description="Default RICS survey tier (Level 3 → full in-place narrative edit).", ) composition_mode: str = Field( default="in_place_edit", description="Report composition strategy: style-informed in-place edit on REFERENCE baseline.", ) max_tokens_mapping: int = Field(default=4096, ge=256, le=16000) max_tokens_grounding: int = Field(default=1024, ge=256, le=8000) max_tokens_repair: int = Field( default=900, ge=256, le=4096, description=( "Output cap for the surgical repair pass. Repair edits a single " "paragraph (subtract-only), so a tight cap speeds completion and " "avoids the timeouts seen when it shared the 4096 mapping budget." ), ) validation_call_timeout_seconds: float = Field( default=60.0, ge=5.0, le=600.0, description=( "Per-call timeout for the auditor and repair LLM passes. These are " "the quality gate and are given more headroom than the strict " "generation-pipeline timeout to avoid wasted circuit-breaker iterations." ), ) max_tokens_discovery: int = Field(default=4000, ge=512, le=16000) notes_expansion_enabled: bool = Field( default=False, description="Run the optional notes-expander pass before mapping.", ) use_llm_paragraph_mapping: bool = Field( default=True, description=( "Style-informed in-place edit on the REFERENCE baseline using the " "mapping editor prompt. When false, deterministic weave only." ), ) grounding_enabled: bool = Field( default=True, description="Run the grounding/PII audit on each mapped section before assembly.", ) grounding_alert_threshold: float = Field( default=0.2, ge=0.0, le=1.0, description=( "Fraction of sections needing review above which the preview flags " "manual_review_required." ), ) ai_transparency_footer_enabled: bool = Field( default=False, description="Append an AI transparency footer to generated DOCX when enabled.", ) template_docx_path: str = Field( default="", description="Optional branded DOCX template path for report export.", ) # ── Report post-processing (final cleanup pass) ─────────────────────────── postprocess_enabled: bool = Field( default=True, description="Master switch for the final report post-processing/normalisation pass.", ) dedup_similarity_threshold: float = Field( default=0.85, ge=0.0, le=1.0, description="Character-trigram Jaccard threshold for near-duplicate paragraph removal (lower = more aggressive).", ) extra_header_patterns: list[str] = Field( default_factory=list, description="Firm-specific header/artifact regex strings to strip during post-processing (keeps the pipeline firm-agnostic).", ) postprocess_placeholder_pattern: str = Field( default=r"\[{1,2}REDACTED_[A-Z]+(?:_\d+)?\]{1,2}", description="Regex matching redaction tokens for post-processing grammar repair.", ) postprocess_debug: bool = Field( default=False, description="When true, the post-processor prints a before/after preview to stdout.", ) # ── Property-type retrieval guard ───────────────────────────────────────── property_guard_enabled: bool = Field( default=True, description="Reject retrieved paragraphs whose terminology is incompatible with the property type (e.g. flat content in a house report).", ) property_blocklist_extra_terms: list[str] = Field( default_factory=list, description="Additional firm-specific terms that mark a retrieved paragraph as property-incompatible.", ) # ── Upload limits (batch + ZIP + notes extract) ─────────────────────────── max_single_upload_bytes: int = Field( default=50 * 1024 * 1024, ge=1024 * 1024, le=200 * 1024 * 1024, description="Max bytes per uploaded reference file (single or inside ZIP).", ) max_zip_members: int = Field( default=40, ge=1, le=200, description="Max non-directory entries allowed inside an uploaded ZIP.", ) max_zip_uncompressed_bytes: int = Field( default=200 * 1024 * 1024, ge=1024 * 1024, le=1024 * 1024 * 1024, description="Max total uncompressed size of all files in a ZIP.", ) max_notes_extract_bytes: int = Field( default=10 * 1024 * 1024, ge=64 * 1024, le=50 * 1024 * 1024, description="Max bytes for /extract-notes uploads.", ) # ── Section photos & vision ──────────────────────────────────────────────── max_section_photo_bytes: int = Field( default=8 * 1024 * 1024, ge=256 * 1024, le=50 * 1024 * 1024, description="Max bytes per uploaded section photo.", ) max_section_photos_per_section: int = Field( default=5, ge=0, le=50, description="Max photos stored per report section.", ) max_section_photos_for_ai: int = Field( default=2, ge=0, le=10, description="Max photos per section the user may select for AI vision analysis.", ) section_photo_vision_enabled: bool = Field( default=True, description="When true and OpenAI is configured, analyze selected section photos at generation.", ) vision_model: str = Field( default="gpt-4o", description="OpenAI vision-capable model for section photo analysis.", ) vision_max_tokens: int = Field( default=1200, ge=256, le=4096, description="Token budget for a section photo vision analysis call.", ) vision_timeout_seconds: float = Field( default=90.0, ge=10.0, le=300.0, description="Hard timeout per vision API call (seconds).", ) vision_max_observations: int = Field( default=12, ge=4, le=40, description="Max observation lines merged from vision per section.", ) # ── Observability (local-only telemetry) ───────────────────────────────── # All observability is local: telemetry is written under the DATA_DIR and, when # tracing is enabled, exported to a localhost OTLP/Phoenix collector. No prompt # or completion text is recorded and nothing leaves the machine. observability_enabled: bool = Field( default=True, description=( "Master switch for local runtime telemetry: per-LLM-call latency / token " "/ cost records and per-report quality metrics, written as JSONL under " "the metrics dir. Passive (no text captured); safe to leave on. Disable " "to make every recorder a no-op." ), ) observability_metrics_dir: str = Field( default="", description=( "Directory for local metrics JSONL (llm_calls.jsonl, reports.jsonl). " "Empty resolves to '/metrics'. Absolute or repo-relative." ), ) observability_console_log_calls: bool = Field( default=False, description=( "Also emit one INFO log line per LLM call (label/model/latency/tokens/" "cost). Off by default to keep generation logs readable; the JSONL sink " "always records regardless." ), ) observability_tracing_enabled: bool = Field( default=False, description=( "Emit OpenTelemetry spans (parent per report; children for retrieval, " "rerank, each section map, each validation iteration) to a local OTLP " "collector such as Arize Phoenix. Off by default and fully no-op when " "the OTel packages or the endpoint are unavailable, so CI/offline runs " "are never affected." ), ) observability_otlp_endpoint: str = Field( default="http://127.0.0.1:6006/v1/traces", description=( "OTLP/HTTP traces endpoint for the local collector (Arize Phoenix " "default). Used only when observability_tracing_enabled is true." ), ) observability_service_name: str = Field( default="rics-rag-backend", description="OpenTelemetry service.name resource attribute for emitted spans.", ) model_pricing: dict[str, list[float]] = Field( default_factory=lambda: { # USD per 1,000,000 tokens, as [input, output]. Embeddings have no # output cost (output element is 0.0). Approximate list-price values — # override via the MODEL_PRICING env var (JSON) for exact accounting. # Matched by exact id first, then by longest key prefix. "gpt-4o-mini": [0.15, 0.60], "gpt-4o": [2.50, 10.00], "gpt-4.1-mini": [0.40, 1.60], "gpt-4.1": [2.00, 8.00], "text-embedding-3-small": [0.02, 0.0], "text-embedding-3-large": [0.13, 0.0], "text-embedding-ada-002": [0.10, 0.0], }, description=( "Token pricing table (USD per 1M tokens, [input, output]) used to " "estimate per-call and per-report cost. Override via env for exact " "rates; unknown models cost 0.0 (recorded but not priced)." ), ) # ── Context construction (flag-gated; eval before enabling) ─────────────── context_reorder_enabled: bool = Field( default=False, description=( "Lost-in-the-middle mitigation for the MULTI-SOURCE reference merge: " "reorder merged sentences edges-in (highest note-relevance at the head " "and tail, lowest in the middle) so salient context is not buried where " "long-context models attend least. Scoped to combine_reference_blocks " "extras only — a single coherent section keeps its document order. Off " "until the golden grounding/coverage + retrieval benchmark confirm no " "regression." ), ) context_compression_enabled: bool = Field( default=False, description=( "Relevance-aware extractive compression when an assembled " "section-complete baseline exceeds reference_section_complete_max_chars: " "keep the sentences with the highest note overlap within budget instead " "of a positional tail-cut. Deterministic/extractive (no LLM, no new " "invention surface). Off until eval-confirmed." ), ) # ── Text normalization (flag-gated; eval before enabling) ───────────────── text_normalize_enabled: bool = Field( default=False, description=( "Apply unified text normalization (NFKC, smart-quote/dash folding, PDF " "de-hyphenation, whitespace collapse) at reference ingest and on the " "retrieval query. Preserves RICS section codes and never alters PII " "scrubbing. Off until the retrieval benchmark confirms no regression." ), ) # ── Auth ───────────────────────────────────────────────────────────────── jwt_secret: str = Field( default="change-me-in-production", description="HMAC secret for signing tenant JWTs.", ) jwt_algorithm: str = Field(default="HS256") jwt_expiry_minutes: int = Field(default=60 * 24, ge=5, le=60 * 24 * 30) admin_token: str = Field( default="", description="Static admin token required (in addition to JWT) for /admin routes.", ) # ── Helpers ────────────────────────────────────────────────────────────── def resolve_path(self, value: str | Path) -> Path: """Resolve a possibly repo-relative path to an absolute path.""" p = Path(value) return p if p.is_absolute() else (REPO_ROOT / p) @property def report_template_path(self) -> Path: """Absolute path to the report template (PDF) — schema authority.""" return ( self.resolve_path(self.master_template_dir) / self.report_template_filename ) @property def standard_paragraphs_path(self) -> Path: """Absolute path to the standard-paragraphs Word file — MASTER RAG authority.""" name = self.standard_paragraphs_filename or self.master_template_filename return self.resolve_path(self.master_template_dir) / name @property def master_template_path(self) -> Path: """Backward-compatible alias for :attr:`standard_paragraphs_path`.""" return self.standard_paragraphs_path @property def reference_dir_path(self) -> Path: """Absolute path to the folder scanned for reference auto-ingest.""" target = self.reference_auto_ingest_dir or self.master_template_dir return self.resolve_path(target) @property def data_dir_path(self) -> Path: return self.resolve_path(self.data_dir) @property def rag_top_k(self) -> int: """Spec alias for :attr:`retrieval_top_k`.""" return self.retrieval_top_k @property def confidence_threshold(self) -> float: """Alias for per-note REFERENCE match gate (CURSOR_REFACTOR parity).""" return self.note_rag_match_min_score @property def branded_template_path(self) -> Path | None: if not self.template_docx_path.strip(): return None p = self.resolve_path(self.template_docx_path) return p if p.is_file() else None @lru_cache def get_settings() -> Settings: """Cached settings singleton.""" return Settings() settings = get_settings()