""" Runtime settings pro CodeAgent v5. Nastavení lze měnit ZA BĚHU (bez restartu Space): - přes Gradio tab „Nastavení", - přes API: GET/POST /admin/settings. Změny HF Space Variables/Secrets vždy restartují Space (chování HF), proto runtime konfigurace žije zde: env proměnné slouží jen jako VÝCHOZÍ hodnoty při prvním startu, poté má přednost persistovaný JSON (/data/settings.json, pokud je připojen storage bucket, jinak .agent/settings.json na ephemeral disku). Dvě kategorie polí: - INSTANT — projeví se okamžitě (limity agenta, runner, cache, ...) - ENGINE — vyžadují reload vLLM enginu (model, TP, kvantizace, ...); reload proběhne na pozadí, aplikace běží dál. """ from __future__ import annotations import copy import json import logging import os import tempfile from dataclasses import asdict, dataclass, field, fields from pathlib import Path from threading import RLock logger = logging.getLogger("codeagent.settings") def _env(name: str, default: str = "") -> str: return os.environ.get(name, default) def _env_int(name: str, default: int) -> int: try: return int(os.environ.get(name, default)) except (TypeError, ValueError): return default def _env_float(name: str, default: float) -> float: try: return float(os.environ.get(name, default)) except (TypeError, ValueError): return default def _env_bool(name: str, default: bool) -> bool: raw = os.environ.get(name) if raw is None: return default return raw.lower() in ("1", "true", "yes", "on") # Pole vyžadující reload vLLM enginu. ENGINE_FIELDS = { "model", "model_revision", "download_dir", "tensor_parallel_size", "gpu_memory_utilization", "max_model_len", "quantization", "dtype", "kv_cache_dtype", "enforce_eager", "tool_call_parser", "reasoning_parser", "enable_prefix_caching", "max_num_seqs", "engine_extra_args", } # Pole, která se v GET /admin/settings maskují (secrets). MASKED_FIELDS = {"runner_token"} _MASK = "********" @dataclass class Settings: """Kompletní runtime konfigurace. Env proměnné = výchozí hodnoty.""" # ---- vLLM engine (změna => reload enginu, ne Space) ---- model: str = field(default_factory=lambda: _env( "MODEL_PRIMARY", "Qwen/Qwen3-Coder-Next-FP8")) # Adresář pro stahované váhy modelů (vLLM --download-dir). # Prázdné = auto: /data/models pokud je namountován RW storage bucket # (přežije restart Space), jinak /app/cache/models (ephemeral). download_dir: str = field(default_factory=lambda: _env("MODEL_DOWNLOAD_DIR", "")) # Pin na konkrétní git revizi HF repa (commit hash / tag). Prázdné = # nejnovější. Doporučeno pro produkci u modelů s trust_remote_code. model_revision: str = field(default_factory=lambda: _env("MODEL_REVISION", "")) tensor_parallel_size: int = field(default_factory=lambda: _env_int( "TENSOR_PARALLEL_SIZE", 4)) gpu_memory_utilization: float = field(default_factory=lambda: _env_float( "GPU_MEMORY_UTILIZATION", 0.92)) max_model_len: int = field(default_factory=lambda: _env_int( "MAX_MODEL_LEN", 131072)) # "auto" = nech vLLM detekovat z checkpointu (FP8/AWQ/GPTQ...); # "none" = bez kvantizace; jinak explicitní hodnota pro vLLM. quantization: str = field(default_factory=lambda: _env("QUANTIZATION", "auto")) dtype: str = field(default_factory=lambda: _env("DTYPE", "auto")) kv_cache_dtype: str = field(default_factory=lambda: _env("KV_CACHE_DTYPE", "auto")) enforce_eager: bool = field(default_factory=lambda: _env_bool("ENFORCE_EAGER", False)) # "auto" = odvodit z presetu/rodiny modelu; "" = vypnuto. tool_call_parser: str = field(default_factory=lambda: _env("TOOL_CALL_PARSER", "auto")) reasoning_parser: str = field(default_factory=lambda: _env("REASONING_PARSER", "auto")) enable_prefix_caching: bool = field(default_factory=lambda: _env_bool( "ENABLE_PREFIX_CACHING", True)) max_num_seqs: int = field(default_factory=lambda: _env_int("MAX_NUM_SEQS", 0)) # 0 = default vLLM # Únikový ventil: libovolné další CLI argumenty pro `vllm serve`. engine_extra_args: str = field(default_factory=lambda: _env("ENGINE_EXTRA_ARGS", "")) # ---- agent / inference chování (instant) ---- agent_mode: str = field(default_factory=lambda: _env("AGENT_MODE", "single").lower()) # Routing v hybrid módu: "ai" = lokální model klasifikuje složitost úlohy # (fallback na keywords při chybě), "keywords" = jen statická pravidla. router_mode: str = field(default_factory=lambda: _env("ROUTER_MODE", "ai").lower()) kimi_model: str = field(default_factory=lambda: _env( "KIMI_MODEL", "moonshotai/Kimi-K2.7-Code")) local_context_limit: int = field(default_factory=lambda: _env_int( "LOCAL_CONTEXT_LIMIT", 24000)) temperature: float = field(default_factory=lambda: _env_float("TEMPERATURE", 0.2)) max_output_tokens: int = field(default_factory=lambda: _env_int( "MAX_OUTPUT_TOKENS", 8192)) max_steps: int = field(default_factory=lambda: _env_int("MAX_STEPS", 30)) max_sub_steps: int = field(default_factory=lambda: _env_int("MAX_SUB_STEPS", 15)) # Úroveň výpisů v chatu: full = nástroje + argumenty + výsledky, # compact = jen jména nástrojů, final = pouze finální odpověď. chat_verbosity: str = field(default_factory=lambda: _env("CHAT_VERBOSITY", "full").lower()) # ---- sub-agenti (instant; prompt = "" znamená výchozí) ---- subagent_explorer_enabled: bool = field(default_factory=lambda: _env_bool( "SUBAGENT_EXPLORER_ENABLED", True)) subagent_coder_enabled: bool = field(default_factory=lambda: _env_bool( "SUBAGENT_CODER_ENABLED", True)) subagent_reviewer_enabled: bool = field(default_factory=lambda: _env_bool( "SUBAGENT_REVIEWER_ENABLED", True)) max_explorer_steps: int = field(default_factory=lambda: _env_int("MAX_EXPLORER_STEPS", 20)) max_coder_steps: int = field(default_factory=lambda: _env_int("MAX_CODER_STEPS", 25)) max_reviewer_steps: int = field(default_factory=lambda: _env_int("MAX_REVIEWER_STEPS", 10)) subagent_explorer_prompt: str = field(default_factory=lambda: _env( "SUBAGENT_EXPLORER_PROMPT", "")) subagent_coder_prompt: str = field(default_factory=lambda: _env( "SUBAGENT_CODER_PROMPT", "")) subagent_reviewer_prompt: str = field(default_factory=lambda: _env( "SUBAGENT_REVIEWER_PROMPT", "")) # ---- správa kontextu a tokenů (instant) ---- # trim = deterministická kompakce (stárnutí tool výsledků + vypouštění # nejstarších bloků), off = vypnuto. context_compaction: str = field(default_factory=lambda: _env( "CONTEXT_COMPACTION", "trim").lower()) # 0 = auto: max_model_len - max_output_tokens - 2048 rezerva. context_budget_tokens: int = field(default_factory=lambda: _env_int( "CONTEXT_BUDGET_TOKENS", 0)) # Posledních N bloků konverzace se nikdy nekompaktuje. context_keep_last_steps: int = field(default_factory=lambda: _env_int( "CONTEXT_KEEP_LAST_STEPS", 6)) # Tvrdý strop délky výsledku nástroje při vložení do kontextu. tool_result_max_chars: int = field(default_factory=lambda: _env_int( "TOOL_RESULT_MAX_CHARS", 24000)) # Na kolik znaků se zkrátí staré tool výsledky při kompakci. tool_result_aged_chars: int = field(default_factory=lambda: _env_int( "TOOL_RESULT_AGED_CHARS", 2000)) # ---- runner (instant) ---- runner_url: str = field(default_factory=lambda: _env("LOCAL_RUNNER_URL", "").rstrip("/")) runner_token: str = field(default_factory=lambda: _env("LOCAL_RUNNER_TOKEN", "")) runner_timeout: int = field(default_factory=lambda: _env_int("RUNNER_TIMEOUT", 180)) # ---- ostatní (instant) ---- cache_ttl_seconds: int = field(default_factory=lambda: _env_int("CACHE_TTL_SECONDS", 300)) log_level: str = field(default_factory=lambda: _env("LOG_LEVEL", "INFO").upper()) VALID_AGENT_MODES = {"single", "hybrid"} VALID_ROUTER_MODES = {"ai", "keywords"} VALID_VERBOSITY = {"full", "compact", "final"} VALID_COMPACTION = {"trim", "off"} VALID_DTYPES = {"auto", "half", "float16", "bfloat16", "float32"} VALID_KV_CACHE_DTYPES = {"auto", "fp8", "fp8_e5m2", "fp8_e4m3"} def validate_field(name: str, value): """Zvaliduje a znormalizuje jednu hodnotu. Vrací (ok, normalized|error_msg).""" spec = {f.name: f for f in fields(Settings)}.get(name) if spec is None: return False, f"Neznámé pole '{name}'" typ = spec.type if isinstance(spec.type, type) else None try: if name in ("tensor_parallel_size", "max_model_len", "max_num_seqs", "local_context_limit", "max_output_tokens", "max_steps", "max_sub_steps", "max_explorer_steps", "max_coder_steps", "max_reviewer_steps", "runner_timeout", "cache_ttl_seconds", "context_budget_tokens", "context_keep_last_steps", "tool_result_max_chars", "tool_result_aged_chars"): value = int(value) if value < 0: return False, f"{name} musí být >= 0" if name == "tensor_parallel_size" and not (1 <= value <= 8): return False, "tensor_parallel_size musí být 1–8" if name == "max_model_len" and value < 1024: return False, "max_model_len musí být >= 1024" if name == "context_keep_last_steps" and not (1 <= value <= 50): return False, "context_keep_last_steps musí být 1–50" if name == "tool_result_max_chars" and value < 1000: return False, "tool_result_max_chars musí být >= 1000" if name == "tool_result_aged_chars" and value < 200: return False, "tool_result_aged_chars musí být >= 200" elif name in ("gpu_memory_utilization", "temperature"): value = float(value) if name == "gpu_memory_utilization" and not (0.1 <= value <= 0.99): return False, "gpu_memory_utilization musí být 0.10–0.99" if name == "temperature" and not (0.0 <= value <= 2.0): return False, "temperature musí být 0.0–2.0" elif name in ("enforce_eager", "enable_prefix_caching", "subagent_explorer_enabled", "subagent_coder_enabled", "subagent_reviewer_enabled"): if isinstance(value, str): value = value.lower() in ("1", "true", "yes", "on") else: value = bool(value) elif name == "agent_mode": value = str(value).lower().strip() if value not in VALID_AGENT_MODES: return False, f"agent_mode musí být jedno z {sorted(VALID_AGENT_MODES)}" elif name == "router_mode": value = str(value).lower().strip() if value not in VALID_ROUTER_MODES: return False, f"router_mode musí být jedno z {sorted(VALID_ROUTER_MODES)}" elif name == "chat_verbosity": value = str(value).lower().strip() if value not in VALID_VERBOSITY: return False, f"chat_verbosity musí být jedno z {sorted(VALID_VERBOSITY)}" elif name == "context_compaction": value = str(value).lower().strip() if value not in VALID_COMPACTION: return False, f"context_compaction musí být jedno z {sorted(VALID_COMPACTION)}" elif name == "dtype": value = str(value).lower().strip() if value not in VALID_DTYPES: return False, f"dtype musí být jedno z {sorted(VALID_DTYPES)}" elif name == "kv_cache_dtype": value = str(value).lower().strip() if value not in VALID_KV_CACHE_DTYPES: return False, f"kv_cache_dtype musí být jedno z {sorted(VALID_KV_CACHE_DTYPES)}" elif name == "log_level": value = str(value).upper().strip() if value not in ("DEBUG", "INFO", "WARNING", "ERROR"): return False, "log_level musí být DEBUG/INFO/WARNING/ERROR" elif name == "runner_url": value = str(value).strip().rstrip("/") if value and not value.startswith(("http://", "https://")): return False, "runner_url musí začínat http(s)://" elif name == "model": value = str(value).strip() if not value: return False, "model nesmí být prázdný" elif name == "download_dir": value = str(value).strip().rstrip("/") if value and not value.startswith("/"): return False, ("download_dir musí být absolutní cesta " "(např. /data/models), nebo prázdné = auto") else: value = str(value).strip() if isinstance(value, str) else value _ = typ # typ zde není potřeba, validace je explicitní return True, value except (TypeError, ValueError) as e: return False, f"Neplatná hodnota pro {name}: {e}" def _persist_dir() -> Path: """/data (storage bucket, přežije restart) > .agent (ephemeral).""" data = Path(os.environ.get("SETTINGS_DIR", "/data")) try: if data.is_dir() and os.access(data, os.W_OK): return data except OSError: pass fallback = Path(".agent") fallback.mkdir(parents=True, exist_ok=True) return fallback class SettingsManager: """Thread-safe držák runtime nastavení s JSON persistencí.""" def __init__(self, path: Path | None = None): self._lock = RLock() self.path = path or (_persist_dir() / "settings.json") self._settings = Settings() self.revision = 0 self._load_persisted() # -------------------------------------------------------------- čtení def get(self) -> Settings: """Vrátí kopii aktuálních nastavení (bezpečné ke čtení bez zámku).""" with self._lock: return copy.deepcopy(self._settings) def as_dict(self, mask_secrets: bool = True) -> dict: with self._lock: data = asdict(self._settings) if mask_secrets: for key in MASKED_FIELDS: if data.get(key): data[key] = _MASK return data # -------------------------------------------------------------- zápis def update(self, changes: dict) -> tuple[dict, bool, dict]: """Aplikuje změny. Vrací (applied, engine_reload_needed, errors).""" applied: dict = {} errors: dict = {} engine_reload = False with self._lock: for name, raw in changes.items(): # maskovaná hodnota z UI => beze změny if name in MASKED_FIELDS and raw == _MASK: continue ok, result = validate_field(name, raw) if not ok: errors[name] = result continue current = getattr(self._settings, name) if current == result: continue setattr(self._settings, name, result) applied[name] = result if name in ENGINE_FIELDS: engine_reload = True if applied: self.revision += 1 self._persist() if "log_level" in applied: logging.getLogger("codeagent").setLevel(applied["log_level"]) if applied: safe = {k: (_MASK if k in MASKED_FIELDS else v) for k, v in applied.items()} logger.info("Nastavení změněno (rev %s): %s", self.revision, safe) return applied, engine_reload, errors # -------------------------------------------------------------- persistence def _persist(self): try: self.path.parent.mkdir(parents=True, exist_ok=True) payload = json.dumps(asdict(self._settings), ensure_ascii=False, indent=2) fd, tmp = tempfile.mkstemp(dir=str(self.path.parent), suffix=".tmp") with os.fdopen(fd, "w", encoding="utf-8") as f: f.write(payload) os.replace(tmp, self.path) except OSError as e: logger.warning("Persistence nastavení selhala (%s): %s", self.path, e) def _load_persisted(self): if not self.path.exists(): return try: data = json.loads(self.path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as e: logger.warning("Načtení %s selhalo: %s", self.path, e) return valid_names = {f.name for f in fields(Settings)} for name, value in data.items(): if name not in valid_names: continue ok, result = validate_field(name, value) if ok: setattr(self._settings, name, result) logger.info("Nastavení načteno z %s", self.path)