"""설정 로딩 — config.yaml + credentials.env""" import os from pathlib import Path import yaml from pydantic import BaseModel class UploadConfig(BaseModel): max_bytes: int = 100_000_000 content_length_slack_ratio: float = 1.05 stream_chunk_bytes: int = 1_048_576 # orphan cleanup (`*.uploading` — 크래시/abort 후 잔존물) orphan_max_age_sec: int = 3600 cleanup_warn_threshold: int = 10 class AIModelConfig(BaseModel): endpoint: str model: str max_tokens: int = 4096 timeout: int = 60 daily_budget_usd: float | None = None require_explicit_trigger: bool = False # B-0: 4B/26B 에 부여한 실사용 컨텍스트 상한 (char). triage=120k, primary=260k. # classify_worker 가 에스컬레이션 판정 시 참고. 0/None 이면 상한 무시. context_char_limit: int | None = None class DeepSummaryBacklogConfig(BaseModel): """B-1 R2 — deep_summary enqueue 폭발 억제 임계치.""" ratio_threshold: float = 0.3 # 지난 window 의 deep_n/classify_n pending_threshold: int = 5 # deep_summary pending+processing window_minutes: int = 30 class AIConfig(BaseModel): gateway_endpoint: str # B-0: 3-tier routing. triage(4B) 상시, primary(26B) escalation-only, fallback(4B) 최후. triage: AIModelConfig primary: AIModelConfig fallback: AIModelConfig premium: AIModelConfig embedding: AIModelConfig rerank: AIModelConfig # Phase 3.5a: exaone classifier (optional — 없으면 score-only gate) classifier: AIModelConfig | None = None # Phase 3.5b: exaone verifier (optional — 없으면 grounding-only) verifier: AIModelConfig | None = None # Legacy: vision 슬롯 (현재 사용처 0 — Document Server 는 OCR/STT 별도 서비스). # 제거 진행 중이므로 optional 로 관대한 로딩 유지. vision: AIModelConfig | None = None # B-1 R2: backlog guard 임계치 deep_summary_backlog: DeepSummaryBacklogConfig = DeepSummaryBacklogConfig() class Settings(BaseModel): # DB database_url: str = "" # AI ai: AIConfig | None = None # NAS nas_mount_path: str = "/documents" nas_pkm_root: str = "/documents/PKM" # 인증 jwt_secret: str = "" totp_secret: str = "" # Phase 3.5: eval runner shared secret — X-Source=eval / X-Eval-Case-Id 헤더 신뢰 검증. # 비어있으면 모든 eval 헤더 거부 (부재 = 비활성). eval_runner_token: str = "" # kordoc kordoc_endpoint: str = "http://kordoc-service:3100" # OCR (Surya) ocr_endpoint: str = "http://ocr-service:3200" # STT (faster-whisper, §3) stt_endpoint: str = "http://stt-service:3300" # §3 file_watcher: Roon 음원 경로 (prefix match 로 skip). # 빈 문자열이면 skip 없음. 예: "/documents/PKM/../Music/roon-library" 또는 # NFS 경유 별도 마운트된 Roon 라이브러리. roon_library_path: str = "" # 분류 체계 taxonomy: dict = {} document_types: list[str] = [] # 업로드 한도 (authoritative policy) upload: UploadConfig = UploadConfig() def load_settings() -> Settings: """config.yaml + 환경변수에서 설정 로딩""" # 환경변수 (docker-compose에서 주입) database_url = os.getenv("DATABASE_URL", "") jwt_secret = os.getenv("JWT_SECRET", "") totp_secret = os.getenv("TOTP_SECRET", "") eval_runner_token = os.getenv("EVAL_RUNNER_TOKEN", "") kordoc_endpoint = os.getenv("KORDOC_ENDPOINT", "http://kordoc-service:3100") ocr_endpoint = os.getenv("OCR_ENDPOINT", "http://ocr-service:3200") stt_endpoint = os.getenv("STT_ENDPOINT", "http://stt-service:3300") roon_library_path = os.getenv("ROON_LIBRARY_PATH", "") # config.yaml — Docker 컨테이너 내부(/app/config.yaml) 또는 프로젝트 루트 config_path = Path("/app/config.yaml") if not config_path.exists(): config_path = Path(__file__).parent.parent.parent / "config.yaml" ai_config = None nas_mount = "/documents" nas_pkm = "/documents/PKM" if config_path.exists(): with open(config_path) as f: raw = yaml.safe_load(f) if "ai" in raw: ai_raw = raw["ai"] models = ai_raw.get("models", {}) # B-0: triage 는 config.yaml 에 없을 수도 있는 신규 슬롯. 구버전 호환을 위해 # 없으면 fallback 를 triage 로 대체 (동일 모델 재사용). triage_raw = models.get("triage") or models.get("fallback") if triage_raw is None: raise ValueError("config.yaml: ai.models.triage (or fallback) required") ai_config = AIConfig( gateway_endpoint=ai_raw.get("gateway", {}).get("endpoint", ""), triage=AIModelConfig(**triage_raw), primary=AIModelConfig(**models["primary"]), fallback=AIModelConfig(**models["fallback"]), premium=AIModelConfig(**models["premium"]), embedding=AIModelConfig(**models["embedding"]), rerank=AIModelConfig(**models["rerank"]), vision=(AIModelConfig(**models["vision"]) if "vision" in models else None), classifier=( AIModelConfig(**models["classifier"]) if "classifier" in models else None ), verifier=( AIModelConfig(**models["verifier"]) if "verifier" in models else None ), deep_summary_backlog=DeepSummaryBacklogConfig( **ai_raw.get("deep_summary_backlog", {}) ), ) if "nas" in raw: nas_mount = raw["nas"].get("mount_path", nas_mount) nas_pkm = raw["nas"].get("pkm_root", nas_pkm) taxonomy = raw.get("taxonomy", {}) if config_path.exists() and raw else {} document_types = raw.get("document_types", []) if config_path.exists() and raw else [] upload_cfg = ( UploadConfig(**raw["upload"]) if config_path.exists() and raw and "upload" in raw else UploadConfig() ) return Settings( database_url=database_url, ai=ai_config, nas_mount_path=nas_mount, nas_pkm_root=nas_pkm, jwt_secret=jwt_secret, totp_secret=totp_secret, eval_runner_token=eval_runner_token, kordoc_endpoint=kordoc_endpoint, ocr_endpoint=ocr_endpoint, stt_endpoint=stt_endpoint, roon_library_path=roon_library_path, taxonomy=taxonomy, document_types=document_types, upload=upload_cfg, ) settings = load_settings()