diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..40dc820 --- /dev/null +++ b/app/config.py @@ -0,0 +1,151 @@ +"""配置管理模块""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Dict, Optional + +import yaml + +# 默认配置 +_DEFAULT_CONFIG: Dict[str, Any] = { + "server": {"host": "0.0.0.0", "port": 16051, "debug": False}, + "database": {"url": "sqlite:///./hunzi.db"}, + "llm": { + "providers": { + "default": { + "name": "default", + "base_url": "http://121.40.164.32:18001/v1", + "api_key": "xxxx", + "model": "unsloth/Qwen3.6-27B-Q4_K_M", + "temperature": 0.7, + "max_tokens": 4096, + "supports_vision": True, + "timeout": 120, + } + } + }, + "scheduler": {"enabled": True, "max_concurrent": 5}, + "logging": { + "level": "INFO", + "format": "%(asctime)s [%(levelname)s] %(name)s: %(message)s", + }, +} + + +@dataclass +class LLMProviderConfig: + """单个 LLM 提供商配置""" + name: str + base_url: str + api_key: str + model: str + temperature: float = 0.7 + max_tokens: int = 4096 + supports_vision: bool = True + timeout: int = 120 + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "LLMProviderConfig": + return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__}) + + +@dataclass +class ServerConfig: + host: str = "0.0.0.0" + port: int = 16051 + debug: bool = False + + +@dataclass +class DatabaseConfig: + url: str = "sqlite:///./hunzi.db" + + +@dataclass +class SchedulerConfig: + enabled: bool = True + max_concurrent: int = 5 + + +@dataclass +class AppConfig: + """全局配置""" + server: ServerConfig = field(default_factory=ServerConfig) + database: DatabaseConfig = field(default_factory=DatabaseConfig) + llm_providers: Dict[str, LLMProviderConfig] = field(default_factory=dict) + scheduler: SchedulerConfig = field(default_factory=SchedulerConfig) + log_level: str = "INFO" + log_format: str = "%(asctime)s [%(levelname)s] %(name)s: %(message)s" + + @property + def default_llm(self) -> Optional[LLMProviderConfig]: + if "default" in self.llm_providers: + return self.llm_providers["default"] + return next(iter(self.llm_providers.values()), None) + + +def _deep_merge(base: Dict, override: Dict) -> Dict: + """深度合并字典""" + result = base.copy() + for k, v in override.items(): + if k in result and isinstance(result[k], dict) and isinstance(v, dict): + result[k] = _deep_merge(result[k], v) + else: + result[k] = v + return result + + +def load_config(config_path: Optional[str] = None) -> AppConfig: + """加载配置:文件覆盖默认值""" + config_dict = _DEFAULT_CONFIG.copy() + + if config_path is None: + config_path = os.environ.get( + "HUNZI_CONFIG", + str(Path(__file__).resolve().parent.parent / "config.yaml"), + ) + + path = Path(config_path) + if path.exists(): + with open(path) as f: + file_config = yaml.safe_load(f) or {} + config_dict = _deep_merge(config_dict, file_config) + + # 环境变量覆盖 + if "HUNZI_PORT" in os.environ: + config_dict["server"]["port"] = int(os.environ["HUNZI_PORT"]) + if "HUNZI_DB_URL" in os.environ: + config_dict["database"]["url"] = os.environ["HUNZI_DB_URL"] + + # 构建配置对象 + llm_providers = {} + for name, cfg in config_dict.get("llm", {}).get("providers", {}).items(): + llm_providers[name] = LLMProviderConfig.from_dict(cfg) + + return AppConfig( + server=ServerConfig(**config_dict.get("server", {})), + database=DatabaseConfig(**config_dict.get("database", {})), + llm_providers=llm_providers, + scheduler=SchedulerConfig(**config_dict.get("scheduler", {})), + log_level=config_dict.get("logging", {}).get("level", "INFO"), + log_format=config_dict.get("logging", {}).get("format", ""), + ) + + +# 全局配置实例 +_app_config: Optional[AppConfig] = None + + +def get_config() -> AppConfig: + global _app_config + if _app_config is None: + _app_config = load_config() + return _app_config + + +def set_config(config: AppConfig) -> None: + global _app_config + _app_config = config