"""可调整的运行设置(SettingsStore)—— 让用户自定义内置小模型 / 大模型 / 管线。 用户可在 Web 界面"模型设置"里调整并持久化到 config/settings.json(gitignore), 重启后保留。调整会触发 v2 管线重建(gateway.api.build_v2_pipeline 重新读取)。 默认值与 config/config.yaml 的 v2 段一致;settings.json 只存用户改动覆盖项。 """ from __future__ import annotations import json from pathlib import Path from typing import Any, Dict, Optional _SETTINGS_PATH = Path(__file__).resolve().parent.parent / "config" / "settings.json" # 可调整项(含默认值);用户改动存这里 DEFAULTS: Dict[str, Any] = { "worker": { "backend": "llama_server", # mock | openai/api | llama_server "model": "models/qwen3.5-4b-q4_k_m.gguf", "base_url": "", # openai 后端填 http://127.0.0.1:11434/v1 等 "port": 8901, # llama_server 端口 "temperature": 0.3, "max_fix_attempts": 2, "code_timeout_s": 10, }, "architect": { "model": "deepseek-chat", "base_url": "https://api.deepseek.com/v1", "api_key_env": "DEEPSEEK_API_KEY", }, "pipeline": { "fast_path": True, "rounds_cap": 6, "api_token_cap": 8000, "breach_policy": "architect_do", }, } class SettingsStore: """用户可调整设置(内存 + settings.json 持久化)。""" def __init__(self, path: Optional[Path] = None): self._path = Path(path) if path else _SETTINGS_PATH self._data: Dict[str, Any] = {} self.load() # ---------- 持久化 ---------- def load(self) -> None: if self._path.exists(): try: self._data = json.loads(self._path.read_text(encoding="utf-8")) except Exception: self._data = {} else: self._data = {} def save(self) -> None: self._path.parent.mkdir(parents=True, exist_ok=True) self._path.write_text( json.dumps(self._data, ensure_ascii=False, indent=2), encoding="utf-8") # ---------- 访问 ---------- def to_dict(self) -> Dict[str, Any]: """返回 默认值 + 用户覆盖 合并后的完整设置。""" merged: Dict[str, Any] = {} for section, defaults in DEFAULTS.items(): ov = self._data.get(section, {}) merged[section] = {**defaults, **(ov if isinstance(ov, dict) else {})} return merged def get(self, section: str, key: str, default: Any = None) -> Any: merged = self.to_dict() return merged.get(section, {}).get(key, default) def update(self, patch: Dict[str, Any]) -> Dict[str, Any]: """应用部分更新(可只传改动的 section/key)。返回合并后的完整设置。""" for section, values in patch.items(): if section not in DEFAULTS or not isinstance(values, dict): continue cur = self._data.setdefault(section, {}) for k, v in values.items(): if k in DEFAULTS[section]: cur[k] = _coerce(v, DEFAULTS[section][k]) self.save() return self.to_dict() def reset(self) -> Dict[str, Any]: """恢复默认。""" self._data = {} self.save() return self.to_dict() def _coerce(value: Any, template: Any) -> Any: """按默认值的类型把输入转成一致类型(数值容错)。""" if isinstance(template, bool): return bool(value) if isinstance(template, int): try: return int(float(value)) except (TypeError, ValueError): return template if isinstance(template, float): try: return float(value) except (TypeError, ValueError): return template return value def load_settings(path: Optional[Path] = None) -> SettingsStore: return SettingsStore(path)