File size: 8,147 Bytes
30e6218 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 | """Application settings.
All configuration is read from environment variables (and an optional ``.env``
file). Nothing here touches the filesystem, opens network connections, or loads
models: settings are plain data so that importing the application is free of
side effects.
"""
from __future__ import annotations
import os
from functools import lru_cache
from pathlib import Path
from typing import List, Literal, Optional
from pydantic import field_validator, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
EmbeddingProviderName = Literal["auto", "none", "openai", "local", "fastembed", "hash"]
GenerationProviderName = Literal["auto", "none", "openai"]
RerankerMode = Literal["none", "heuristic", "cross_encoder", "llm"]
StorageMode = Literal["persistent", "ephemeral"]
class Settings(BaseSettings):
"""Runtime configuration.
Values are validated once at startup. ``resolved_*`` properties expose the
effective provider after ``auto`` resolution so that health endpoints and
logs can report what actually runs.
"""
model_config = SettingsConfigDict(
env_file=os.environ.get("APP_ENV_FILE", ".env"),
env_file_encoding="utf-8",
case_sensitive=False,
extra="ignore",
)
# Application
app_name: str = "Document Intelligence RAG System"
app_version: str = "0.2.0"
app_env: Literal["development", "testing", "production"] = "development"
log_level: str = "INFO"
# Access control. When ``api_key`` is unset the API runs in local mode:
# every endpoint is open, which is only appropriate on a trusted machine.
api_key: Optional[str] = None
cors_origins: str = ""
# Storage
storage_mode: StorageMode = "persistent"
data_dir: str = "./data"
chroma_collection_name: str = "documents"
# Embeddings (dense retrieval). ``auto`` selects ``openai`` when an OpenAI
# key is configured and ``none`` (lexical-only retrieval) otherwise.
embedding_provider: EmbeddingProviderName = "auto"
openai_api_key: Optional[str] = None
openai_base_url: Optional[str] = None
openai_embedding_model: str = "text-embedding-3-small"
openai_timeout_seconds: float = 60.0
local_embedding_model: str = "sentence-transformers/all-MiniLM-L6-v2"
# ``fastembed`` runs the same MiniLM model through ONNX Runtime (no torch).
fastembed_model: str = "sentence-transformers/all-MiniLM-L6-v2"
fastembed_cache_dir: Optional[str] = None
allow_model_download: bool = False
# Generation
generation_provider: GenerationProviderName = "auto"
openai_model: str = "gpt-4o-mini"
generation_max_tokens: int = 600
generation_temperature: float = 0.0
generation_concurrency: int = 4
prompt_version: str = "grounded-v1"
# Reranking. ``cross_encoder`` requires the optional ML dependencies and a
# prepared local model; ``llm`` requires a generation provider.
reranker_mode: RerankerMode = "none"
cross_encoder_model: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"
cross_encoder_device: str = "cpu"
llm_rerank_max_candidates: int = 20
llm_rerank_concurrency: int = 4
# Ingestion limits
max_upload_size: int = 10 * 1024 * 1024
# Cap on documents in the corpus; 0 means unlimited. When the cap is
# reached the oldest non-seeded document is evicted before a new upload.
max_documents: int = 0
# Directory of documents ingested at startup (public demo corpus). Seeded
# documents are never evicted.
demo_seed_dir: Optional[str] = None
max_filename_length: int = 255
max_extracted_chars: int = 2_000_000
max_pdf_pages: int = 500
max_chunks_per_document: int = 2000
max_metadata_keys: int = 32
max_metadata_value_length: int = 512
# Chunking (character based)
chunk_size: int = 1000
chunk_overlap: int = 200
# Retrieval
search_top_k: int = 5
max_top_k: int = 50
candidate_multiplier: int = 4
rrf_k: int = 60
min_similarity: Optional[float] = None
max_query_chars: int = 2000
# Context assembly
max_context_chars: int = 6000
max_excerpt_chars: int = 1200
# Answer cache (exact match; 0 disables)
answer_cache_ttl_seconds: int = 300
answer_cache_max_entries: int = 256
metrics_enabled: bool = True
# Rate limiting on /api routes (requests per sliding 60 s window; 0 disables).
# Per-client identity is the socket address, or the value of
# ``client_ip_header`` when the request carries a valid API key (the
# header is set by a trusted proxy that holds the key).
rate_limit_per_minute: int = 0
rate_limit_global_per_minute: int = 0
client_ip_header: str = "X-Client-IP"
@field_validator("log_level")
@classmethod
def _upper_log_level(cls, value: str) -> str:
value = value.upper()
if value not in {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}:
raise ValueError(f"Unsupported log level: {value}")
return value
@model_validator(mode="after")
def _check_bounds(self) -> "Settings":
if self.chunk_size <= 0:
raise ValueError("chunk_size must be positive")
if self.chunk_overlap < 0 or self.chunk_overlap >= self.chunk_size:
raise ValueError("chunk_overlap must be >= 0 and < chunk_size")
if self.search_top_k <= 0 or self.max_top_k <= 0:
raise ValueError("top_k limits must be positive")
if self.search_top_k > self.max_top_k:
raise ValueError("search_top_k must not exceed max_top_k")
if self.max_upload_size <= 0:
raise ValueError("max_upload_size must be positive")
if self.max_documents < 0:
raise ValueError("max_documents must be >= 0")
if self.rate_limit_per_minute < 0 or self.rate_limit_global_per_minute < 0:
raise ValueError("rate limits must be >= 0")
if self.api_key is not None and len(self.api_key) < 16:
raise ValueError("api_key must be at least 16 characters")
if self.embedding_provider == "openai" and not self.openai_api_key:
raise ValueError("embedding_provider=openai requires OPENAI_API_KEY")
if self.generation_provider == "openai" and not self.openai_api_key:
raise ValueError("generation_provider=openai requires OPENAI_API_KEY")
return self
# Derived paths -------------------------------------------------------
@property
def data_path(self) -> Path:
return Path(self.data_dir)
@property
def uploads_dir(self) -> Path:
return self.data_path / "uploads"
@property
def index_dir(self) -> Path:
return self.data_path / "index"
@property
def manifest_path(self) -> Path:
return self.data_path / "manifest.sqlite3"
# Resolved modes ------------------------------------------------------
@property
def resolved_embedding_provider(self) -> str:
if self.embedding_provider == "auto":
return "openai" if self.openai_api_key else "none"
return self.embedding_provider
@property
def resolved_generation_provider(self) -> str:
if self.generation_provider == "auto":
return "openai" if self.openai_api_key else "none"
return self.generation_provider
@property
def cors_origin_list(self) -> List[str]:
return [o.strip() for o in self.cors_origins.split(",") if o.strip()]
@property
def is_development(self) -> bool:
return self.app_env == "development"
@property
def is_production(self) -> bool:
return self.app_env == "production"
@property
def chunking_config_id(self) -> str:
"""Identity of the chunking configuration used when indexing."""
return f"chars:v2:size={self.chunk_size}:overlap={self.chunk_overlap}"
@lru_cache()
def get_settings() -> Settings:
"""Return the process-wide settings instance."""
return Settings()
def reset_settings_cache() -> None:
"""Clear the cached settings (used by tests that change the environment)."""
get_settings.cache_clear()
|