"""运行时可调配置（runtime settings） 与 `app.config.settings`（启动时从环境变量加载、不可变）互补：本模块管理运行时 可通过管理后台动态调整的子集，持久化到 JSON 文件，启动时加载覆盖到内存单例。 设计要点： - RuntimeSettings 只包含「可运行时调整」的字段（模型/解析插件/去重等）， 不包含敏感或启动期固定的字段（端口、数据库连接等）。 - 持久化路径默认 `data/runtime_settings.json`，可由 env `RUNTIME_SETTINGS_PATH` 覆盖。 - 加载失败/文件缺失时回退到默认值，不阻塞启动。 - 写入采用「先临时文件后 rename」原子替换，避免半写损坏。 - 全模块只通过 `get_runtime_settings()` 访问单例，避免直接读 JSON。 """ from __future__ import annotations import json import os import tempfile from pathlib import Path from threading import RLock from typing import Any, Literal import structlog from pydantic import BaseModel, Field from app.config import settings logger = structlog.get_logger() # 默认持久化路径（相对工作目录）；可由 env 覆盖 DEFAULT_RUNTIME_SETTINGS_PATH = "./data/runtime_settings.json" _lock = RLock() # ---------------------------------------------------------------------------- # # 模型配置（文档总结 / 查询 / 分类 各一份独立配置） # ---------------------------------------------------------------------------- # class LlmProviderConfig(BaseModel): """单个用途的 LLM 提供方配置""" provider: Literal["ollama", "openai_compatible"] = Field( default="ollama", description="提供方：ollama 走 Ollama HTTP；openai_compatible 走 OpenAI 兼容 chat/completions" ) base_url: str = Field(default="", description="服务地址；空则 ollama 用 settings.ollama_base_url，openai_compatible 用 settings.openai_base_url") api_key: str = Field(default="", description="API Key（仅 openai_compatible 需要；ollama 忽略）") model: str = Field(default="", description="模型名；空则 ollama 用 settings.ollama_model，openai_compatible 用 settings.embedding_model 同级（如 gpt-4o-mini）") timeout: float = Field(default=120.0, description="请求超时秒数") temperature: float = Field(default=0.3, description="采样温度（0~2）") class ModelSettings(BaseModel): """模型相关运行时配置：三种用途独立配置""" summarize: LlmProviderConfig = Field(default_factory=LlmProviderConfig, description="文档三级总结") query: LlmProviderConfig = Field(default_factory=LlmProviderConfig, description="检索链路 query 解析与结果总结") classify: LlmProviderConfig = Field(default_factory=LlmProviderConfig, description="文档分类判定") # ---------------------------------------------------------------------------- # # 解析插件配置 # ---------------------------------------------------------------------------- # class PluginConfig(BaseModel): """单个解析插件的选择与参数""" plugin: str = Field(default="", description="插件名；空则用默认") params: dict[str, Any] = Field(default_factory=dict, description="插件参数（透传给插件实现）") class ParserSettings(BaseModel): """解析插件运行时配置""" ocr: PluginConfig = Field(default_factory=lambda: PluginConfig(plugin="rapidocr"), description="OCR 插件") pdf: PluginConfig = Field(default_factory=lambda: PluginConfig(plugin="pypdf"), description="PDF 文本层提取插件") docx: PluginConfig = Field(default_factory=lambda: PluginConfig(plugin="python_docx"), description="DOCX 解析插件") # ---------------------------------------------------------------------------- # # 去重策略配置 # ---------------------------------------------------------------------------- # class DedupSettings(BaseModel): """文本去重策略运行时配置""" strategy: Literal["none", "sha256", "simhash"] = Field( default="sha256", description="去重策略：none 关闭；sha256 精确匹配；simhash 近似匹配" ) simhash_threshold: int = Field(default=3, description="simhash 海明距离阈值（仅 strategy=simhash 生效，0~64）") ttl_seconds: int = Field(default=86400, description="去重记录 Redis 保留秒数") # ---------------------------------------------------------------------------- # # 顶层 RuntimeSettings # ---------------------------------------------------------------------------- # class RuntimeSettings(BaseModel): """运行时可调配置顶层模型""" models: ModelSettings = Field(default_factory=ModelSettings) parsers: ParserSettings = Field(default_factory=ParserSettings) dedup: DedupSettings = Field(default_factory=DedupSettings) # ---------------------------------------------------------------------------- # # 单例 + 持久化 # ---------------------------------------------------------------------------- # _runtime_settings: RuntimeSettings | None = None def _resolve_path() -> Path: """解析持久化文件路径：env RUNTIME_SETTINGS_PATH > settings 自定义 > 默认""" path_str = os.environ.get("RUNTIME_SETTINGS_PATH", "") or getattr(settings, "runtime_settings_path", "") or DEFAULT_RUNTIME_SETTINGS_PATH return Path(path_str).expanduser().resolve() def _default_with_env_fallback() -> RuntimeSettings: """构造默认 RuntimeSettings，并把启动 env 中已有的模型相关字段填充进去 这样首次启动（无持久化文件）时，UI 显示的不是空字符串而是 env 当前值。 """ cfg = RuntimeSettings() # 模型默认值沿用 env cfg.models.summarize.base_url = settings.ollama_base_url cfg.models.summarize.model = settings.ollama_model cfg.models.query.base_url = settings.ollama_base_url cfg.models.query.model = settings.ollama_model cfg.models.classify.base_url = settings.ollama_base_url cfg.models.classify.model = settings.ollama_model # 若 env 提供了 OpenAI key/url，预填到 openai_compatible 字段方便切换 if settings.openai_api_key: for usage in ("summarize", "query", "classify"): getattr(cfg.models, usage).api_key = settings.openai_api_key getattr(cfg.models, usage).base_url = settings.openai_base_url if settings.openai_base_url else getattr(cfg.models, usage).base_url return cfg def load_runtime_settings() -> RuntimeSettings: """从磁盘加载 RuntimeSettings；文件缺失或损坏时回退到默认值（带 env 兜底）""" path = _resolve_path() try: if path.exists(): data = json.loads(path.read_text(encoding="utf-8")) return RuntimeSettings.model_validate(data) except Exception: logger.warning("RuntimeSettings 加载失败，回退默认值", path=str(path), exc_info=True) return _default_with_env_fallback() def save_runtime_settings(cfg: RuntimeSettings) -> None: """原子写入 RuntimeSettings 到磁盘（先临时文件后 rename）""" path = _resolve_path() path.parent.mkdir(parents=True, exist_ok=True) data = cfg.model_dump(mode="json") # 写到同目录临时文件再 rename，避免半写损坏 fd, tmp_path = tempfile.mkstemp(prefix=".runtime_settings.", suffix=".tmp", dir=str(path.parent)) try: with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) os.replace(tmp_path, path) except Exception: try: os.unlink(tmp_path) except OSError: pass raise def get_runtime_settings() -> RuntimeSettings: """获取 RuntimeSettings 单例（首次调用时加载）""" global _runtime_settings with _lock: if _runtime_settings is None: _runtime_settings = load_runtime_settings() return _runtime_settings def update_runtime_settings(patch: dict[str, Any]) -> RuntimeSettings: """以 patch 字典更新 RuntimeSettings 并持久化 支持部分更新（顶层键可缺失，缺省保留原值）。例如： update_runtime_settings({"models": {"summarize": {"model": "qwen2.5:3b"}}}) """ with _lock: current = get_runtime_settings() merged = current.model_dump(mode="json") _deep_merge(merged, patch) new_cfg = RuntimeSettings.model_validate(merged) save_runtime_settings(new_cfg) # 替换单例后立即返回，新值对所有后续读取生效 global _runtime_settings _runtime_settings = new_cfg logger.info("RuntimeSettings 已更新并持久化", path=str(_resolve_path())) return new_cfg def _deep_merge(target: dict[str, Any], patch: dict[str, Any]) -> None: """递归把 patch 合并到 target（同 key 字典则递归，否则覆盖）""" for k, v in patch.items(): if k in target and isinstance(target[k], dict) and isinstance(v, dict): _deep_merge(target[k], v) else: target[k] = v def reload_runtime_settings() -> RuntimeSettings: """强制从磁盘重新加载（管理后台触发）""" with _lock: global _runtime_settings _runtime_settings = load_runtime_settings() return _runtime_settings def reset_runtime_settings() -> RuntimeSettings: """重置为默认值并持久化（管理后台触发）""" with _lock: new_cfg = _default_with_env_fallback() save_runtime_settings(new_cfg) global _runtime_settings _runtime_settings = new_cfg return new_cfg