119 lines
4.4 KiB
Python
119 lines
4.4 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,
|
||
},
|
||
"execution": {
|
||
"mode": "rule", # rule（L0 零参数）| hybrid
|
||
"planner": "rule", # rule | hf
|
||
"expert_backend": "rule", # rule（规则执行器）| hf | api
|
||
"model_level": "L0", # L0 | L1 | L2
|
||
"max_plan_depth": 3,
|
||
},
|
||
"classifier": {"type": "rule", "model": "Qwen/Qwen3-0.6B", "confidence_floor": 0.55},
|
||
"domains": ["code", "math", "legal", "medical", "finance", "life", "education", "general"],
|
||
# 两级路由：大领域分组（用户接口指定 group → 组内路由模型 → 组内专业小模型）
|
||
# 组内路由模型只识别本组领域，体积约为统一路由模型的 1/4
|
||
"domain_groups": {
|
||
"tech": ["code", "math"],
|
||
"professional": ["legal", "medical", "finance"],
|
||
"lifestyle": ["life", "education"],
|
||
"general": ["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"},
|
||
"finance": {"type": "mock", "model": "Qwen/Qwen3-4B-Instruct"},
|
||
"life": {"type": "mock", "model": "Qwen/Qwen3-1.7B-Instruct"},
|
||
"education": {"type": "mock", "model": "Qwen/Qwen3-1.7B-Instruct"},
|
||
"general": {"type": "mock", "model": "Qwen/Qwen3-1.7B-Instruct"},
|
||
},
|
||
"fallback": {
|
||
"type": "mock",
|
||
"model": "deepseek-v4-flash",
|
||
"base_url": "https://api.deepseek.com",
|
||
"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
|