"""应用配置:全部从 .env 读取,代码不硬编码任何连接/模型值。 .env 路径由本文件位置推导(绝对路径),与启动工作目录无关;每个子配置各自声明 env_file,保证嵌套配置也读文件(否则默认实例会吞掉环境变量);任一必需键缺失, 启动即报错(fail-fast),避免带着错误配置静默运行。 """ from pathlib import Path from typing import Optional from pydantic_settings import BaseSettings, SettingsConfigDict from sqlalchemy.engine import URL # 项目根 = settings.py 的上两级;无论从哪个 cwd 启动都能定位 .env _ENV_FILE: Path = Path(__file__).resolve().parent.parent / ".env" class MysqlCfg(BaseSettings): host: str port: int user: str password: str database: str pool_size: int max_overflow: int pool_recycle: int connect_timeout: int pool_timeout: int model_config = SettingsConfigDict(env_prefix="MYSQL_", env_file=_ENV_FILE, extra="ignore") @property def url(self) -> URL: return URL.create( "mysql+aiomysql", username=self.user, password=self.password, host=self.host, port=self.port, database=self.database, query={"charset": "utf8mb4"}, ) class RedisCfg(BaseSettings): host: str port: int db: int password: Optional[str] = None # 可缺省(空串/未设置表示无密码) socket_connect_timeout: int socket_timeout: int health_check_interval: int max_connections: int model_config = SettingsConfigDict(env_prefix="REDIS_", env_file=_ENV_FILE, extra="ignore") class Neo4jCfg(BaseSettings): uri: str user: str password: str max_connection_pool_size: int connection_acquisition_timeout: int connection_timeout: int max_transaction_retry_time: int model_config = SettingsConfigDict(env_prefix="NEO4J_", env_file=_ENV_FILE, extra="ignore") class MilvusCfg(BaseSettings): uri: str token: Optional[str] = None # 可缺省(无需鉴权时留空) user: Optional[str] = None password: Optional[str] = None db_name: Optional[str] = None connect_timeout: int # 建连/通道就绪超时(构造是急切连接,必须短) timeout: int # 数据操作超时,调用处可覆盖 model_config = SettingsConfigDict(env_prefix="MILVUS_", env_file=_ENV_FILE, extra="ignore") class LLMCfg(BaseSettings): """大模型配置:本地 Ollama / OpenAI 兼容 API 双模式见 tool/llm.py。""" mode: str # —— 本地 Ollama —— ollama_base: str ollama_chat_model: str ollama_embed_model: str # —— OpenAI 兼容 API(base_url 以 /v1 结尾)—— api_base: str api_key: str api_chat_model: str api_embed_model: str # —— 通用超参 —— temperature: float max_tokens: int timeout: int max_retries: int retry_backoff_sec: float # 指数退避基数 fallback_chat_model: str # 备用模型:主模型失败后自动切换(空则不启用) model_config = SettingsConfigDict(env_prefix="LLM_", env_file=_ENV_FILE, extra="ignore") class JwtCfg(BaseSettings): """JWT 鉴权配置(service/auth.py)。""" secret: str # 签名密钥,生产必须换强随机值 algorithm: str # 签名算法,如 HS256 expire_seconds: int # Token 有效期(秒) model_config = SettingsConfigDict(env_prefix="JWT_", env_file=_ENV_FILE, extra="ignore") class DBMaintenanceCfg(BaseSettings): """数据库启动策略:连接重试与严格模式(config/database/__init__.py)。""" strict_startup: bool = False # true=任一核心库启动失败即阻止应用启动(生产建议开启) conn_retries: int = 3 # 各库 init 最大尝试次数 retry_backoff_sec: float = 1.0 # 指数退避基数:1s/2s/4s model_config = SettingsConfigDict(env_prefix="DB_", env_file=_ENV_FILE, extra="ignore") class Settings(BaseSettings): app_env: str db: DBMaintenanceCfg = DBMaintenanceCfg() jwt: JwtCfg = JwtCfg() mysql: MysqlCfg = MysqlCfg() redis: RedisCfg = RedisCfg() neo4j: Neo4jCfg = Neo4jCfg() milvus: MilvusCfg = MilvusCfg() llm: LLMCfg = LLMCfg() model_config = SettingsConfigDict(env_file=_ENV_FILE, extra="ignore") settings = Settings()