101 lines
3.5 KiB
Python
101 lines
3.5 KiB
Python
"""配置加载：优先 YAML（若安装了 pyyaml），否则回退 JSON。
|
|
|
|
设计原则：router_system 核心零依赖，因此 pyyaml 是"可选"的。
|
|
默认 config/config.yaml 存在；若 pyyaml 不可用，可提供同名 .json。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Optional
|
|
|
|
DEFAULT_CONFIG_PATH = Path(__file__).resolve().parent.parent / "config" / "config.yaml"
|
|
|
|
_DEFAULTS: Dict[str, Any] = {
|
|
"system": {"name": "multi-expert-router", "version": "0.1.0"},
|
|
"router": {
|
|
"low_confidence_threshold": 0.60, # 分类置信度低于此值 -> 直接走大模型
|
|
"judge_fallback_threshold": 0.70, # Judge 质量分低于此值 -> 升级大模型
|
|
"default_temperature": 0.2,
|
|
},
|
|
"classifier": {"type": "rule", "model": "Qwen/Qwen3-0.6B", "confidence_floor": 0.55},
|
|
"domains": ["code", "math", "legal", "medical", "general"],
|
|
"experts": {
|
|
"code": {"type": "mock", "model": "Qwen/Qwen2.5-Coder-7B-Instruct"},
|
|
"math": {"type": "mock", "model": "Qwen/Qwen3-4B-Instruct"},
|
|
"legal": {"type": "mock", "model": "Qwen/Qwen3-4B-Instruct"},
|
|
"medical": {"type": "mock", "model": "Qwen/Qwen3-4B-Instruct"},
|
|
"general": {"type": "mock", "model": "Qwen/Qwen3-1.7B-Instruct"},
|
|
},
|
|
"fallback": {
|
|
"type": "mock",
|
|
"model": "deepseek-chat",
|
|
"base_url": "https://api.deepseek.com/v1",
|
|
"api_key_env": "DEEPSEEK_API_KEY",
|
|
},
|
|
"judge": {"type": "rule", "model": "Qwen/Qwen3-1.7B-Instruct"},
|
|
"cache": {
|
|
"enabled": True,
|
|
"semantic_enabled": True,
|
|
"similarity_threshold": 0.88,
|
|
"promote_frequency": 5,
|
|
},
|
|
}
|
|
|
|
|
|
def load_defaults() -> Dict[str, Any]:
|
|
return _DEFAULTS
|
|
|
|
|
|
def _try_load_yaml(path: Path) -> Optional[Dict[str, Any]]:
|
|
try:
|
|
import yaml # type: ignore
|
|
except ImportError:
|
|
return None
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
data = yaml.safe_load(f)
|
|
return data if isinstance(data, dict) else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _try_load_json(path: Path) -> Optional[Dict[str, Any]]:
|
|
json_path = path.with_suffix(".json")
|
|
if not json_path.exists():
|
|
return None
|
|
try:
|
|
with open(json_path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
return data if isinstance(data, dict) else None
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _merge_defaults(data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""将用户配置与内置默认配置做一层合并（用户优先）。"""
|
|
merged = dict(_DEFAULTS)
|
|
for k, v in data.items():
|
|
if isinstance(v, dict) and isinstance(merged.get(k), dict):
|
|
merged[k] = {**merged[k], **v}
|
|
else:
|
|
merged[k] = v
|
|
return merged
|
|
|
|
|
|
def load_config(path: Optional[Path | str] = None) -> Dict[str, Any]:
|
|
"""加载配置，返回 dict。文件不存在或解析失败时返回内置默认配置。"""
|
|
cfg_path = Path(path) if path else DEFAULT_CONFIG_PATH
|
|
if cfg_path.exists():
|
|
data = _try_load_yaml(cfg_path) or _try_load_json(cfg_path)
|
|
if data is not None:
|
|
return _merge_defaults(data)
|
|
return dict(_DEFAULTS)
|
|
|
|
|
|
def get_api_key(cfg: Dict[str, Any]) -> Optional[str]:
|
|
"""从环境变量读取 API Key（用于 api 类型后端）。"""
|
|
env_name = cfg.get("api_key_env") or "API_KEY"
|
|
return os.environ.get(env_name) or None
|
|
|