File size: 6,070 Bytes
72fef8d 5849dd3 72fef8d caea6f0 72fef8d 8f51101 caea6f0 72fef8d caea6f0 72fef8d caea6f0 72fef8d caea6f0 72fef8d 8f51101 caea6f0 8f51101 caea6f0 8f51101 caea6f0 72fef8d 5849dd3 8f51101 caea6f0 72fef8d | 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 | """Environment-backed configuration for LibBee."""
from functools import lru_cache
from pathlib import Path
from typing import Literal
from pydantic import field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
LIBBEE_VERSION = "4.1.2"
class Settings(BaseSettings):
model_config = SettingsConfigDict(case_sensitive=False, env_file=".env", env_file_encoding="utf-8", extra="ignore")
openai_api_key: str = ""
anthropic_api_key: str = ""
primo_api_key: str = ""
# Optional HF variable. "auto" preserves the existing EU/NA/AP discovery;
# selecting the institution's gateway avoids that discovery on each search.
primo_api_region: Literal["auto", "eu", "na", "ap", "ca", "aps", "cn"] = "auto"
openalex_api_key: str = ""
contact_email: str = "libse@ku.ac.ae"
# The Worker sends this secret in X-LibBee-Origin-Token. Production
# protected routes fail closed when the value is missing.
origin_token: str = ""
environment: Literal["development", "production", "test"] = "production"
allowed_cors_origins: str = "http://localhost:8080,http://127.0.0.1:5500"
admin_password: str = ""
session_secret: str = "change-me"
maintenance_mode: bool = False
maintenance_message: str = "LibBee is receiving an update. Please try again shortly."
max_results: int = 5
max_question_chars: int = 2000
max_history_messages: int = 12
max_message_chars: int = 4000
max_concurrent_agent_requests: int = 20
# Includes the entire JSON envelope and UTF-8 history. The original 32 KiB
# ceiling rejected otherwise valid conversations under the model limits.
max_request_body_bytes: int = 524_288
# /data survives HF Space restarts when persistent storage is attached.
runtime_dir: str = "/data/libbee_runtime"
feedback_store_path: str = ""
metrics_store_path: str = ""
config_store_path: str = ""
rag_index_dir: str = ""
knowledge_dir: str = ""
# The authenticated admin page receives only a short-lived signed token;
# the permanent shared secret is never serialized into browser HTML.
cloudflare_worker_url: str = ""
cloudflare_admin_token: str = ""
# HF talks to LibCal directly because some hosted environments cannot open
# outbound connections to shared workers.dev hostnames.
libcal_client_id: str = ""
libcal_client_secret: str = ""
libcal_events_client_id: str = ""
libcal_events_client_secret: str = ""
@field_validator("max_results")
@classmethod
def reasonable_max_results(cls, value: int) -> int:
if not 1 <= value <= 20:
raise ValueError("max_results must be between 1 and 20")
return value
@field_validator("max_question_chars", "max_message_chars")
@classmethod
def reasonable_text_limit(cls, value: int, info) -> int:
ceiling = 2000 if info.field_name == "max_question_chars" else 4000
if not 256 <= value <= ceiling:
raise ValueError(f"{info.field_name} must be between 256 and {ceiling}")
return value
@field_validator("max_history_messages")
@classmethod
def reasonable_history_limit(cls, value: int) -> int:
if not 0 <= value <= 12:
raise ValueError("max_history_messages must be between 0 and 12")
return value
@field_validator("max_concurrent_agent_requests")
@classmethod
def reasonable_concurrency(cls, value: int) -> int:
if not 1 <= value <= 100:
raise ValueError("max_concurrent_agent_requests must be between 1 and 100")
return value
@field_validator("max_request_body_bytes")
@classmethod
def reasonable_body_limit(cls, value: int) -> int:
if not 16_384 <= value <= 2_097_152:
raise ValueError("max_request_body_bytes must be between 16 KiB and 2 MiB")
return value
@property
def runtime_defaults(self) -> dict:
"""One source of defaults for admin, public config, and agent replies."""
return {
"max_results": self.max_results,
"maintenance_mode": self.maintenance_mode,
"welcome_message": "Hi! I'm LibBee, the Khalifa University Library AI Assistant.",
"custom_instructions": "",
"announcement": "",
"maintenance_message": self.maintenance_message,
}
@property
def cors_origins(self) -> list[str]:
return [value.strip().rstrip("/") for value in self.allowed_cors_origins.split(",") if value.strip()]
@property
def runtime_path(self) -> Path:
requested = Path(self.runtime_dir)
# Honour explicit paths, including nested directories not created yet.
# Only the absent default HF mount falls back to ephemeral local disk.
if self.runtime_dir == "/data/libbee_runtime" and not Path("/data").is_dir():
return Path("/tmp/libbee_runtime")
return requested
@property
def feedback_path(self) -> Path:
return Path(self.feedback_store_path) if self.feedback_store_path else self.runtime_path / "feedback.jsonl"
@property
def metrics_path(self) -> Path:
return Path(self.metrics_store_path) if self.metrics_store_path else self.runtime_path / "metrics.json"
@property
def config_path(self) -> Path:
return Path(self.config_store_path) if self.config_store_path else self.runtime_path / "runtime_config.json"
@property
def rag_cache_dir(self) -> Path:
return Path(self.rag_index_dir) if self.rag_index_dir else self.runtime_path / "rag_cache"
@property
def kb_dir(self) -> Path:
if self.knowledge_dir:
return Path(self.knowledge_dir)
if Path("/app/data/knowledge").exists():
return Path("/app/data/knowledge")
return Path("data/knowledge")
@lru_cache
def get_settings() -> Settings:
settings = Settings()
settings.runtime_path.mkdir(parents=True, exist_ok=True)
settings.rag_cache_dir.mkdir(parents=True, exist_ok=True)
return settings
|