152 lines
4.2 KiB
Python
152 lines
4.2 KiB
Python
"""配置管理模块"""
|
|
|
|
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
|