wehub-resource-sync
1197 lines
49 KiB
Python
1197 lines
49 KiB
Python
"""Per-model settings management for oMLX.
|
||
|
||
This module provides dataclasses and a manager for storing and retrieving
|
||
per-model configuration settings, including sampling parameters, pinned/default
|
||
flags, and metadata.
|
||
"""
|
||
|
||
import copy
|
||
import json
|
||
import logging
|
||
import threading
|
||
from dataclasses import dataclass, field, fields
|
||
from pathlib import Path
|
||
from typing import Any, Callable, Dict, Optional
|
||
|
||
from .model_profiles import (
|
||
MODEL_SPECIFIC_PROFILE_FIELDS,
|
||
filter_profile_fields,
|
||
filter_universal_fields,
|
||
slugify_profile_api_name,
|
||
validate_profile_name,
|
||
utcnow,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Current settings file format version
|
||
SETTINGS_VERSION = 1
|
||
PROFILES_VERSION = 1
|
||
TEMPLATES_VERSION = 1
|
||
|
||
|
||
@dataclass
|
||
class ModelSettings:
|
||
"""Per-model configuration settings.
|
||
|
||
Attributes:
|
||
max_context_window: Maximum prompt token count before rejection (None = use global default).
|
||
max_tokens: Maximum number of tokens to generate (None = use global default).
|
||
temperature: Sampling temperature (None = use global default).
|
||
top_p: Nucleus sampling probability (None = use global default).
|
||
top_k: Top-k sampling parameter (None = use global default).
|
||
min_p: Minimum probability threshold (None = use global default).
|
||
repetition_penalty: Repetition penalty (None = use default 1.0, i.e. disabled).
|
||
presence_penalty: Presence penalty (None = use global default).
|
||
force_sampling: Force sampling even with temperature=0.
|
||
max_tool_result_tokens: Maximum tokens in tool result (None = use global default).
|
||
chat_template_kwargs: Extra chat template keyword arguments.
|
||
forced_ct_kwargs: Keys in chat_template_kwargs that cannot be overridden.
|
||
ttl_seconds: Auto-unload after idle seconds (None = no TTL).
|
||
model_type_override: "llm", "vlm", "embedding", "reranker", or None (auto-detect).
|
||
model_alias: API-visible alternative to the directory name.
|
||
index_cache_freq: IndexCache: every Nth layer keeps indexer (DeepSeek DSA
|
||
only; GLM-5.2 uses its native checkpoint schedule).
|
||
enable_thinking: Explicit toggle for thinking/reasoning mode (None = auto).
|
||
thinking_budget_enabled: Whether a thinking token budget is active.
|
||
thinking_budget_tokens: Max tokens for thinking/reasoning.
|
||
reasoning_parser: xgrammar builtin name: "qwen", "harmony", "llama", etc.
|
||
guided_grammar_enabled: Whether a default guided grammar is active.
|
||
guided_grammar: Default EBNF grammar for constrained decoding.
|
||
turboquant_kv_enabled: Enable TurboQuant KV cache compression.
|
||
turboquant_kv_bits: TurboQuant bit depth (2/2.5/3/3.5/4/6/8).
|
||
turboquant_skip_last: Skip last KVCache layer to prevent corruption.
|
||
specprefill_enabled: Enable SpecPrefill (experimental sparse prefill for MoE).
|
||
specprefill_draft_model: Path to draft model for SpecPrefill.
|
||
specprefill_keep_pct: Keep rate for SpecPrefill (0.1–0.5).
|
||
specprefill_threshold: Min tokens to trigger SpecPrefill.
|
||
dflash_enabled: Enable DFlash speculative decoding.
|
||
dflash_draft_model: Path/repo for DFlash draft checkpoint.
|
||
dflash_draft_quant_enabled: Enable draft model quantization.
|
||
dflash_draft_quant_weight_bits: Quantization weight bits (2, 4, 8).
|
||
dflash_draft_quant_activation_bits: Quantization activation bits (16, 32).
|
||
dflash_draft_quant_group_size: Quantization group size (32, 64, 128).
|
||
dflash_max_ctx: Token threshold to fall back to BatchedEngine (None = unlimited).
|
||
dflash_in_memory_cache: Enable DFlash L1 (RAM) prefix cache.
|
||
dflash_in_memory_cache_max_entries: L1 cache max entries (default 4, matches dflash balanced profile).
|
||
dflash_in_memory_cache_max_bytes: L1 cache byte budget.
|
||
dflash_ssd_cache: Enable DFlash L2 (SSD) prefix cache spill (uses omlx SSD cache dir).
|
||
dflash_ssd_cache_max_bytes: L2 (SSD) disk budget; dflash evicts oldest entries when exceeded.
|
||
dflash_draft_window_size: Draft model sliding-attention window (None = dflash default 1024).
|
||
Helps stabilise acceptance rate on long-context prompts.
|
||
dflash_draft_sink_size: Attention-sink tokens always kept regardless of window
|
||
(None = dflash default 64).
|
||
dflash_verify_mode: Verifier algorithm — "dflash", "adaptive", "ddtree", or "off"
|
||
(None = dflash default "adaptive"). "adaptive" can shrink block size when
|
||
acceptance drops.
|
||
mtp_enabled: Enable native multi-token prediction (mlx-lm PR 990 / PR 15 monkey-patch).
|
||
When True, BatchGenerator uses MTP draft+verify for singleton decode and
|
||
for multi-row decode batches whose cache positions are aligned. Unaligned
|
||
continuous batches fall back to standard decoding automatically. Compatible
|
||
model_types: qwen3_5*, qwen3_6*, deepseek_v4*. Mutually exclusive with
|
||
dflash_enabled.
|
||
vlm_mtp_enabled: Enable VLM MTP speculative decoding via an external assistant
|
||
drafter (mlx-vlm 191d7c8+). Target = Gemma4 VLM body, drafter must be a
|
||
"gemma4_assistant" model.
|
||
vlm_mtp_draft_model: Path/repo of the assistant drafter (e.g. "gemma-4-26B-A4B-it-assistant").
|
||
vlm_mtp_draft_block_size: Tokens drafted per round (None = mlx-vlm default).
|
||
is_pinned: Keep model loaded in memory.
|
||
is_default: Use this model when no model is specified.
|
||
display_name: Human-readable name for UI display.
|
||
description: Optional description of the model.
|
||
active_profile_name: Name of the currently-applied profile (None = no profile).
|
||
"""
|
||
|
||
# Sampling parameters (None means use global default)
|
||
max_context_window: Optional[int] = None
|
||
max_tokens: Optional[int] = None
|
||
temperature: Optional[float] = None
|
||
top_p: Optional[float] = None
|
||
top_k: Optional[int] = None
|
||
repetition_penalty: Optional[float] = None
|
||
min_p: Optional[float] = None
|
||
presence_penalty: Optional[float] = None
|
||
force_sampling: bool = False
|
||
max_tool_result_tokens: Optional[int] = None
|
||
chat_template_kwargs: Optional[Dict[str, Any]] = None
|
||
forced_ct_kwargs: Optional[list[str]] = (
|
||
None # Keys that cannot be overridden by API requests
|
||
)
|
||
ttl_seconds: Optional[int] = None # Auto-unload after idle seconds (None = no TTL)
|
||
model_type_override: Optional[str] = (
|
||
None # "llm", "vlm", "embedding", "reranker", or None (auto-detect)
|
||
)
|
||
model_alias: Optional[str] = (
|
||
None # API-visible name (alternative to directory name)
|
||
)
|
||
index_cache_freq: Optional[int] = (
|
||
None # IndexCache: every Nth layer keeps indexer (DeepSeek DSA only)
|
||
)
|
||
enable_thinking: Optional[bool] = (
|
||
None # Explicit toggle for thinking/reasoning mode (None = auto)
|
||
)
|
||
preserve_thinking: Optional[bool] = (
|
||
None # Keep <think> blocks in historical turns (None = auto, True when template supports it)
|
||
)
|
||
thinking_budget_enabled: bool = False
|
||
thinking_budget_tokens: Optional[int] = None
|
||
reasoning_parser: Optional[str] = (
|
||
None # xgrammar builtin name: "qwen", "harmony", "llama", etc.
|
||
)
|
||
guided_grammar_enabled: bool = False
|
||
guided_grammar: Optional[str] = None
|
||
|
||
# TurboQuant KV cache (mlx-vlm backend)
|
||
turboquant_kv_enabled: bool = False
|
||
turboquant_kv_bits: float = 4 # 2, 2.5, 3, 3.5, 4, 6, 8
|
||
turboquant_skip_last: bool = (
|
||
True # Skip last KVCache layer (prevents corruption on sensitive models)
|
||
)
|
||
|
||
# SpecPrefill (experimental: attention-based sparse prefill for MoE models)
|
||
specprefill_enabled: bool = False
|
||
specprefill_draft_model: Optional[str] = (
|
||
None # Path to draft model (must share tokenizer)
|
||
)
|
||
specprefill_keep_pct: Optional[float] = None # Keep rate (0.1-0.5, default 0.2)
|
||
specprefill_threshold: Optional[int] = None # Min tokens to trigger (default 8192)
|
||
|
||
# DFlash (block diffusion speculative decoding)
|
||
dflash_enabled: bool = False
|
||
dflash_draft_model: Optional[str] = None # Path/repo for DFlash draft checkpoint
|
||
dflash_draft_quant_enabled: Optional[bool] = None
|
||
dflash_draft_quant_weight_bits: Optional[int] = None # 2, 4, 8
|
||
dflash_draft_quant_activation_bits: Optional[int] = None # 16, 32
|
||
dflash_draft_quant_group_size: Optional[int] = None # 32, 64, 128
|
||
dflash_max_ctx: Optional[int] = (
|
||
None # None = unlimited; trigger BatchedEngine fallback when prompt_len >= this
|
||
)
|
||
# DFlash prefix cache (private to dflash; separate from omlx tiered cache because
|
||
# snapshots include draft model GDN state and target hidden chunks omlx never tracks)
|
||
dflash_in_memory_cache: bool = True
|
||
dflash_in_memory_cache_max_entries: int = (
|
||
4 # Matches dflash balanced profile default
|
||
)
|
||
dflash_in_memory_cache_max_bytes: int = (
|
||
8 * 1024 * 1024 * 1024
|
||
) # 8 GiB (balanced profile default)
|
||
dflash_ssd_cache: bool = (
|
||
False # Requires in-memory cache and an omlx paged SSD cache dir
|
||
)
|
||
dflash_ssd_cache_max_bytes: int = 20 * 1024 * 1024 * 1024 # 20 GiB L2 disk budget
|
||
# DFlash runtime tuning knobs. None = let dflash-mlx pick its own DEFAULT_RUNTIME_CONFIG
|
||
# value (currently window=1024, sink=64, verify_mode="adaptive"). Surfaced for long-context
|
||
# agentic workloads where acceptance drops on the default sliding window.
|
||
dflash_draft_window_size: Optional[int] = None
|
||
dflash_draft_sink_size: Optional[int] = None
|
||
dflash_verify_mode: Optional[str] = None # "dflash" | "adaptive" | "ddtree" | "off"
|
||
|
||
# Native MTP (mlx-lm PR 990 / PR 15 monkey-patch). When enabled, BatchGenerator
|
||
# uses MTP draft+verify for singleton decode and aligned multi-row decode batches.
|
||
# Compatible model_types: qwen3_5*, qwen3_6*, deepseek_v4*. Mutually exclusive
|
||
# with dflash.
|
||
mtp_enabled: bool = False
|
||
# Maximum chained MTP draft tokens per verify cycle (speculative depth).
|
||
# None = default (3). Effective for Qwen3.5/3.6 native MTP only;
|
||
# DeepSeek-V4 native MTP always runs depth 1. An adaptive controller
|
||
# picks 1..max per sequence from rolling acceptance/latency estimates;
|
||
# set to 1 for a fixed depth-1 cycle.
|
||
mtp_num_draft_tokens: Optional[int] = None
|
||
|
||
# VLM MTP speculative decoding via external MTP drafter (mlx-vlm f96138e+).
|
||
# Supported drafter types: gemma4_assistant (for Gemma 4 VLMs), qwen3_5_mtp
|
||
# (for Qwen 3.5/3.6). Both resolve to draft_kind="mtp" in mlx-vlm.
|
||
# Mutually exclusive with all other speculative paths because the wrapper
|
||
# bypasses mlx-lm BatchGenerator at decode time.
|
||
vlm_mtp_enabled: bool = False
|
||
vlm_mtp_draft_model: Optional[str] = (
|
||
None # Path / model id of the assistant drafter
|
||
)
|
||
vlm_mtp_draft_block_size: Optional[int] = (
|
||
None # Tokens per draft round (None = mlx-vlm default)
|
||
)
|
||
|
||
# Model management flags
|
||
is_pinned: bool = False
|
||
is_default: bool = False # Only one model can be default
|
||
is_hidden: bool = False # Hidden from /v1/models (still shown, badged, in admin)
|
||
is_favorite: bool = False # Listed first in /v1/models and admin lists
|
||
|
||
# Security: opt-in per model. When True, mlx-lm/mlx-vlm/mlx-embeddings/reranker
|
||
# loaders are allowed to execute custom Python from the model repository
|
||
# (modeling_*.py, tokenization_*.py). Off by default — see issue #926.
|
||
trust_remote_code: bool = False
|
||
|
||
# Metadata
|
||
display_name: Optional[str] = None
|
||
description: Optional[str] = None
|
||
active_profile_name: Optional[str] = None # Name of the currently-applied profile
|
||
|
||
def __post_init__(self) -> None:
|
||
# Native MTP is mutually exclusive with DFlash (also speculative).
|
||
# Reject the combo at construction time so the conflict surfaces in
|
||
# the admin UI / API rather than at model load. TurboQuant KV is
|
||
# compatible: its attention patch routes MTP's decode-shaped
|
||
# multi-row verify through the quantized decode kernels.
|
||
if self.mtp_enabled and self.dflash_enabled:
|
||
raise ValueError(
|
||
"mtp_enabled and dflash_enabled cannot both be True; choose one "
|
||
"speculative-decoding path per model"
|
||
)
|
||
# vlm_mtp wraps mlx-vlm's MTP loop and bypasses mlx-lm BatchGenerator
|
||
# at decode time, so it cannot coexist with any other speculative path
|
||
# or with TurboQuant (which mutates the same cache objects).
|
||
if self.vlm_mtp_enabled:
|
||
conflicts = [
|
||
("dflash_enabled", self.dflash_enabled),
|
||
("specprefill_enabled", self.specprefill_enabled),
|
||
("mtp_enabled", self.mtp_enabled),
|
||
("turboquant_kv_enabled", self.turboquant_kv_enabled),
|
||
]
|
||
for name, value in conflicts:
|
||
if value:
|
||
raise ValueError(
|
||
f"vlm_mtp_enabled and {name} cannot both be True; "
|
||
"choose one speculative path per model"
|
||
)
|
||
|
||
def to_dict(self) -> dict:
|
||
"""Convert to dictionary, excluding None values.
|
||
|
||
Returns:
|
||
Dictionary representation with None values filtered out.
|
||
"""
|
||
result = {}
|
||
for f in fields(self):
|
||
value = getattr(self, f.name)
|
||
if value is not None:
|
||
result[f.name] = value
|
||
return result
|
||
|
||
@classmethod
|
||
def from_dict(cls, data: dict) -> "ModelSettings":
|
||
"""Create ModelSettings from a dictionary.
|
||
|
||
Args:
|
||
data: Dictionary containing settings values.
|
||
|
||
Returns:
|
||
New ModelSettings instance with values from dict.
|
||
"""
|
||
# Get valid field names
|
||
valid_fields = {f.name for f in fields(cls)}
|
||
|
||
# Filter to only valid keys
|
||
filtered_data = {k: v for k, v in data.items() if k in valid_fields}
|
||
|
||
return cls(**filtered_data)
|
||
|
||
|
||
class ModelSettingsManager:
|
||
"""Manager for per-model settings with file persistence.
|
||
|
||
Handles loading, saving, and accessing model settings from a JSON file.
|
||
Thread-safe for concurrent access.
|
||
|
||
Attributes:
|
||
base_path: Base directory for settings storage.
|
||
settings_file: Path to the settings JSON file.
|
||
"""
|
||
|
||
def __init__(self, base_path: Path):
|
||
"""Initialize the settings manager.
|
||
|
||
Args:
|
||
base_path: Base directory for settings storage.
|
||
"""
|
||
self.base_path = Path(base_path)
|
||
self.settings_file = self.base_path / "model_settings.json"
|
||
self.profiles_file = self.base_path / "model_profiles.json"
|
||
self.templates_file = self.base_path / "global_templates.json"
|
||
self._lock = threading.Lock()
|
||
self._settings: Dict[str, ModelSettings] = {}
|
||
self._profiles: Dict[str, Dict[str, Dict[str, Any]]] = {}
|
||
self._templates: Dict[str, Dict[str, Any]] = {}
|
||
|
||
# Ensure base directory exists
|
||
self.base_path.mkdir(parents=True, exist_ok=True)
|
||
|
||
# Load existing settings
|
||
self._load()
|
||
self._load_profiles()
|
||
self._load_templates()
|
||
|
||
def _load(self) -> None:
|
||
"""Load settings from the JSON file.
|
||
|
||
If the file doesn't exist or is invalid, starts with empty settings.
|
||
"""
|
||
if not self.settings_file.exists():
|
||
logger.debug(f"Settings file not found: {self.settings_file}")
|
||
self._settings = {}
|
||
return
|
||
|
||
try:
|
||
with open(self.settings_file, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
|
||
# Check version
|
||
version = data.get("version", 1)
|
||
if version != SETTINGS_VERSION:
|
||
logger.warning(
|
||
f"Settings file version {version} differs from current {SETTINGS_VERSION}"
|
||
)
|
||
|
||
# Load model settings
|
||
models_data = data.get("models", {})
|
||
self._settings = {}
|
||
|
||
for model_id, model_data in models_data.items():
|
||
try:
|
||
self._settings[model_id] = ModelSettings.from_dict(model_data)
|
||
except Exception as e:
|
||
logger.warning(
|
||
f"Failed to load settings for model '{model_id}': {e}"
|
||
)
|
||
|
||
logger.info(f"Loaded settings for {len(self._settings)} models")
|
||
|
||
except json.JSONDecodeError as e:
|
||
logger.error(f"Invalid JSON in settings file: {e}")
|
||
self._settings = {}
|
||
except Exception as e:
|
||
logger.error(f"Failed to load settings file: {e}")
|
||
self._settings = {}
|
||
|
||
def _save(self) -> None:
|
||
"""Save settings to the JSON file.
|
||
|
||
Must be called while holding the lock.
|
||
"""
|
||
data = {
|
||
"version": SETTINGS_VERSION,
|
||
"models": {
|
||
model_id: settings.to_dict()
|
||
for model_id, settings in self._settings.items()
|
||
},
|
||
}
|
||
|
||
try:
|
||
# Write to temp file first, then rename for atomicity
|
||
temp_file = self.settings_file.with_suffix(".tmp")
|
||
with open(temp_file, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, indent=2, ensure_ascii=False)
|
||
|
||
temp_file.replace(self.settings_file)
|
||
logger.debug(f"Saved settings for {len(self._settings)} models")
|
||
|
||
except Exception as e:
|
||
logger.error(f"Failed to save settings file: {e}")
|
||
raise
|
||
|
||
def get_settings(self, model_id: str) -> ModelSettings:
|
||
"""Get settings for a specific model.
|
||
|
||
Args:
|
||
model_id: The model identifier.
|
||
|
||
Returns:
|
||
ModelSettings for the model, or default settings if not found.
|
||
"""
|
||
with self._lock:
|
||
if model_id in self._settings:
|
||
# Return a copy to prevent external modification
|
||
settings = self._settings[model_id]
|
||
return ModelSettings.from_dict(settings.to_dict())
|
||
|
||
return ModelSettings()
|
||
|
||
def get_settings_for_request(
|
||
self,
|
||
model_id: str,
|
||
resolved_model_id: Optional[str] = None,
|
||
) -> ModelSettings:
|
||
"""Get settings for an API-requested model name.
|
||
|
||
Exposed profile model IDs return the base model's settings merged
|
||
with the profile's universal (request-time) overrides. Engine-
|
||
construction fields are handled separately by
|
||
get_exposed_profile_runtime_settings_for_request(), which may trigger
|
||
a transient engine variant reload without mutating persisted settings.
|
||
Any other name falls back to the settings of the already-resolved
|
||
physical model.
|
||
"""
|
||
with self._lock:
|
||
candidates = [model_id]
|
||
if "/" in model_id:
|
||
candidates.append(model_id.split("/", 1)[1])
|
||
for candidate in candidates:
|
||
profile_match = self._find_exposed_profile_locked(candidate)
|
||
if profile_match is not None:
|
||
base_model_id, profile = profile_match
|
||
return self._settings_with_profile_locked(base_model_id, profile)
|
||
|
||
return self.get_settings(resolved_model_id or model_id)
|
||
|
||
def set_settings(self, model_id: str, settings: ModelSettings) -> None:
|
||
"""Set settings for a specific model.
|
||
|
||
If the new settings have is_default=True, clears is_default from all
|
||
other models to maintain the exclusive default constraint.
|
||
|
||
Args:
|
||
model_id: The model identifier.
|
||
settings: The settings to apply.
|
||
"""
|
||
with self._lock:
|
||
# Handle exclusive default constraint
|
||
if settings.is_default:
|
||
for mid, s in self._settings.items():
|
||
if mid != model_id and s.is_default:
|
||
s.is_default = False
|
||
logger.info(
|
||
f"Cleared is_default from model '{mid}' "
|
||
f"(new default: '{model_id}')"
|
||
)
|
||
|
||
# Store a copy of the settings
|
||
self._settings[model_id] = ModelSettings.from_dict(settings.to_dict())
|
||
logger.info(f"Updated settings for model '{model_id}'")
|
||
|
||
self._save()
|
||
|
||
def delete_settings(self, model_id: str) -> bool:
|
||
"""Remove all persisted state for a model (settings + profiles).
|
||
|
||
Called when a model is deleted so its alias and other settings are
|
||
released and can be reused by another model.
|
||
|
||
Args:
|
||
model_id: The model identifier.
|
||
|
||
Returns:
|
||
True if any state was removed, False if nothing was stored.
|
||
"""
|
||
with self._lock:
|
||
removed = False
|
||
if model_id in self._settings:
|
||
del self._settings[model_id]
|
||
self._save()
|
||
removed = True
|
||
if model_id in self._profiles:
|
||
del self._profiles[model_id]
|
||
self._save_profiles()
|
||
removed = True
|
||
if removed:
|
||
logger.info(f"Deleted settings for model '{model_id}'")
|
||
return removed
|
||
|
||
def get_default_model_id(self) -> Optional[str]:
|
||
"""Get the ID of the default model.
|
||
|
||
Returns:
|
||
The model ID marked as default, or None if no default is set.
|
||
"""
|
||
with self._lock:
|
||
for model_id, settings in self._settings.items():
|
||
if settings.is_default:
|
||
return model_id
|
||
return None
|
||
|
||
def get_pinned_model_ids(self) -> list[str]:
|
||
"""Get list of all pinned model IDs.
|
||
|
||
Returns:
|
||
List of model IDs that are marked as pinned.
|
||
"""
|
||
with self._lock:
|
||
return [
|
||
model_id
|
||
for model_id, settings in self._settings.items()
|
||
if settings.is_pinned
|
||
]
|
||
|
||
def get_all_settings(self) -> Dict[str, ModelSettings]:
|
||
"""Get a copy of all model settings.
|
||
|
||
Returns:
|
||
Dictionary mapping model IDs to their settings (deep copy).
|
||
"""
|
||
with self._lock:
|
||
return {
|
||
model_id: ModelSettings.from_dict(settings.to_dict())
|
||
for model_id, settings in self._settings.items()
|
||
}
|
||
|
||
# ==================== Profiles ====================
|
||
|
||
def _load_profiles(self) -> None:
|
||
if not self.profiles_file.exists():
|
||
self._profiles = {}
|
||
return
|
||
try:
|
||
with open(self.profiles_file, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
version = data.get("version", 1)
|
||
if version != PROFILES_VERSION:
|
||
logger.warning(
|
||
f"Profiles file version {version} differs from current {PROFILES_VERSION}"
|
||
)
|
||
self._profiles = data.get("profiles", {}) or {}
|
||
# Migration: strip ttl_seconds from existing profile settings and
|
||
# add api_name for exposed-model IDs without changing internal keys.
|
||
changed = False
|
||
for model_id, profiles in self._profiles.items():
|
||
model_changed = False
|
||
used_api_names: set[str] = set()
|
||
for name, profile in profiles.items():
|
||
settings = profile.get("settings")
|
||
if settings and "ttl_seconds" in settings:
|
||
del settings["ttl_seconds"]
|
||
changed = True
|
||
model_changed = True
|
||
current_api_name = profile.get("api_name")
|
||
if current_api_name:
|
||
try:
|
||
validate_profile_name(current_api_name)
|
||
base_api_name = current_api_name
|
||
except Exception:
|
||
base_api_name = slugify_profile_api_name(
|
||
profile.get("display_name") or name,
|
||
fallback="profile",
|
||
)
|
||
else:
|
||
base_api_name = slugify_profile_api_name(
|
||
profile.get("display_name") or name,
|
||
fallback="profile",
|
||
)
|
||
api_name = self._dedupe_profile_api_name(
|
||
base_api_name,
|
||
used_api_names,
|
||
)
|
||
if current_api_name != api_name:
|
||
profile["api_name"] = api_name
|
||
changed = True
|
||
model_changed = True
|
||
if model_changed:
|
||
logger.info(f"Migrated profile api names for model '{model_id}'")
|
||
if changed:
|
||
self._save_profiles()
|
||
except Exception as e:
|
||
logger.error(f"Failed to load profiles file: {e}")
|
||
self._profiles = {}
|
||
|
||
def _save_profiles(self) -> None:
|
||
"""Write profiles to disk atomically (temp file + rename)."""
|
||
data = {"version": PROFILES_VERSION, "profiles": self._profiles}
|
||
temp_file = self.profiles_file.with_suffix(".tmp")
|
||
try:
|
||
with open(temp_file, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, indent=2, ensure_ascii=False, default=str)
|
||
temp_file.replace(self.profiles_file)
|
||
except Exception as e:
|
||
logger.error(f"Failed to save profiles file: {e}")
|
||
if temp_file.exists():
|
||
temp_file.unlink(missing_ok=True)
|
||
raise
|
||
|
||
@staticmethod
|
||
def _dedupe_profile_api_name(base: str, used: set[str]) -> str:
|
||
validate_profile_name(base)
|
||
candidate = base
|
||
index = 2
|
||
while candidate in used:
|
||
suffix = f"-{index}"
|
||
root = base[: 32 - len(suffix)].rstrip("-_") or "profile"
|
||
candidate = f"{root}{suffix}"
|
||
index += 1
|
||
used.add(candidate)
|
||
return candidate
|
||
|
||
@staticmethod
|
||
def _profile_api_name(profile: Dict[str, Any]) -> str:
|
||
api_name = profile.get("api_name") or profile["name"]
|
||
validate_profile_name(api_name)
|
||
return api_name
|
||
|
||
def _allocate_profile_api_name_locked(
|
||
self,
|
||
profiles: Dict[str, Dict[str, Any]],
|
||
value: str | None,
|
||
*,
|
||
display_name: str | None,
|
||
internal_name: str,
|
||
exclude_name: str | None = None,
|
||
) -> str:
|
||
if value:
|
||
validate_profile_name(value)
|
||
base = value
|
||
else:
|
||
base = slugify_profile_api_name(
|
||
display_name or internal_name,
|
||
fallback="profile",
|
||
)
|
||
used = {
|
||
self._profile_api_name(profile)
|
||
for name, profile in profiles.items()
|
||
if name != exclude_name
|
||
}
|
||
return self._dedupe_profile_api_name(base, used)
|
||
|
||
def _profile_model_id(self, model_id: str, api_name: str) -> str:
|
||
return f"{model_id}:{api_name}"
|
||
|
||
def _display_profile_model_id_locked(
|
||
self, model_id: str, profile: Dict[str, Any]
|
||
) -> str:
|
||
"""Advertised form of an exposed profile's model ID.
|
||
|
||
Uses the base model's alias when set, mirroring how /v1/models
|
||
lists the base model itself. The directory-name form remains
|
||
accepted for requests, exactly like the base model's directory
|
||
name.
|
||
"""
|
||
base = self._settings.get(model_id)
|
||
display_base = base.model_alias if base and base.model_alias else model_id
|
||
return self._profile_model_id(display_base, self._profile_api_name(profile))
|
||
|
||
def _find_exposed_profile_locked(
|
||
self, model_id: str
|
||
) -> Optional[tuple[str, Dict[str, Any]]]:
|
||
for base_model_id, profiles in self._profiles.items():
|
||
base = self._settings.get(base_model_id)
|
||
alias = base.model_alias if base else None
|
||
for profile in profiles.values():
|
||
if not profile.get("expose_as_model"):
|
||
continue
|
||
api_name = self._profile_api_name(profile)
|
||
if model_id == self._profile_model_id(base_model_id, api_name):
|
||
return base_model_id, profile
|
||
if alias and model_id == self._profile_model_id(alias, api_name):
|
||
return base_model_id, profile
|
||
return None
|
||
|
||
def _settings_with_profile_locked(
|
||
self, model_id: str, profile: Dict[str, Any]
|
||
) -> ModelSettings:
|
||
base = self._settings.get(model_id)
|
||
merged = base.to_dict() if base is not None else {}
|
||
# Request-time settings still use only universal fields. Engine-
|
||
# construction fields are handled separately by
|
||
# get_exposed_profile_runtime_settings_for_request(), which can
|
||
# trigger an engine variant reload without persisting base settings.
|
||
merged.update(filter_universal_fields(profile.get("settings", {}) or {}))
|
||
return ModelSettings.from_dict(merged)
|
||
|
||
def _runtime_settings_with_profile_locked(
|
||
self, model_id: str, profile: Dict[str, Any]
|
||
) -> ModelSettings:
|
||
base = self._settings.get(model_id)
|
||
merged = base.to_dict() if base is not None else {}
|
||
merged.update(filter_profile_fields(profile.get("settings", {}) or {}))
|
||
return ModelSettings.from_dict(merged)
|
||
|
||
def get_exposed_profile_source_model_id(self, model_id: str) -> Optional[str]:
|
||
"""Return the base model for an exposed profile model ID, if any."""
|
||
with self._lock:
|
||
candidates = [model_id]
|
||
if "/" in model_id:
|
||
candidates.append(model_id.split("/", 1)[1])
|
||
for candidate in candidates:
|
||
match = self._find_exposed_profile_locked(candidate)
|
||
if match is not None:
|
||
return match[0]
|
||
return None
|
||
|
||
def get_exposed_profile_runtime_settings_for_request(
|
||
self,
|
||
model_id: str,
|
||
) -> Optional[tuple[str, ModelSettings]]:
|
||
"""Return full runtime settings for an exposed profile request.
|
||
|
||
Unlike ``get_settings_for_request()``, this includes model-specific
|
||
engine-construction fields. It is used only for transient engine
|
||
loading and never mutates the base model's persisted settings.
|
||
"""
|
||
with self._lock:
|
||
candidates = [model_id]
|
||
if "/" in model_id:
|
||
candidates.append(model_id.split("/", 1)[1])
|
||
for candidate in candidates:
|
||
match = self._find_exposed_profile_locked(candidate)
|
||
if match is not None:
|
||
base_model_id, profile = match
|
||
return (
|
||
base_model_id,
|
||
self._runtime_settings_with_profile_locked(
|
||
base_model_id, profile
|
||
),
|
||
)
|
||
return None
|
||
|
||
def get_exposed_profile_model_ids(
|
||
self,
|
||
*,
|
||
exclude_model_id: str | None = None,
|
||
exclude_profile_name: str | None = None,
|
||
) -> set[str]:
|
||
"""Return every request ID accepted by exposed profiles."""
|
||
with self._lock:
|
||
model_ids: set[str] = set()
|
||
for base_model_id, profiles in self._profiles.items():
|
||
base = self._settings.get(base_model_id)
|
||
alias = base.model_alias if base else None
|
||
for profile in profiles.values():
|
||
if not profile.get("expose_as_model"):
|
||
continue
|
||
if (
|
||
exclude_model_id == base_model_id
|
||
and exclude_profile_name == profile["name"]
|
||
):
|
||
continue
|
||
api_name = self._profile_api_name(profile)
|
||
model_ids.add(self._profile_model_id(base_model_id, api_name))
|
||
if alias:
|
||
model_ids.add(self._profile_model_id(alias, api_name))
|
||
return model_ids
|
||
|
||
def _profile_request_ids_locked(
|
||
self,
|
||
model_id: str,
|
||
profile: Dict[str, Any],
|
||
) -> set[str]:
|
||
base = self._settings.get(model_id)
|
||
base_ids = {model_id}
|
||
if base and base.model_alias:
|
||
base_ids.add(base.model_alias)
|
||
api_name = self._profile_api_name(profile)
|
||
return {self._profile_model_id(base_id, api_name) for base_id in base_ids}
|
||
|
||
def _validate_exposed_profile_ids_available_locked(
|
||
self,
|
||
model_id: str,
|
||
profile: Dict[str, Any],
|
||
*,
|
||
exclude_profile_name: str | None = None,
|
||
reserved_model_ids: set[str] | None = None,
|
||
) -> None:
|
||
if not profile.get("expose_as_model"):
|
||
return
|
||
candidate_ids = self._profile_request_ids_locked(model_id, profile)
|
||
|
||
if reserved_model_ids:
|
||
for candidate_id in candidate_ids:
|
||
if candidate_id != model_id and candidate_id in reserved_model_ids:
|
||
raise ValueError(
|
||
f"Exposed profile model ID '{candidate_id}' conflicts "
|
||
"with a model directory name"
|
||
)
|
||
|
||
for mid, settings in self._settings.items():
|
||
if settings.model_alias and settings.model_alias in candidate_ids:
|
||
raise ValueError(
|
||
f"Exposed profile model ID '{settings.model_alias}' "
|
||
f"conflicts with model alias for '{mid}'"
|
||
)
|
||
|
||
existing_ids: set[str] = set()
|
||
for base_model_id, profiles in self._profiles.items():
|
||
for other in profiles.values():
|
||
if not other.get("expose_as_model"):
|
||
continue
|
||
if base_model_id == model_id and other["name"] == exclude_profile_name:
|
||
continue
|
||
existing_ids.update(
|
||
self._profile_request_ids_locked(base_model_id, other)
|
||
)
|
||
conflict = candidate_ids & existing_ids
|
||
if conflict:
|
||
conflict_id = sorted(conflict)[0]
|
||
raise ValueError(f"Exposed profile model ID '{conflict_id}' already exists")
|
||
|
||
def list_exposed_profile_models(self) -> list[dict]:
|
||
"""Return profile records promoted to independently visible model IDs."""
|
||
with self._lock:
|
||
exposed = []
|
||
for base_model_id, profiles in self._profiles.items():
|
||
for profile in profiles.values():
|
||
if not profile.get("expose_as_model"):
|
||
continue
|
||
item = dict(profile)
|
||
item["model_id"] = self._display_profile_model_id_locked(
|
||
base_model_id, profile
|
||
)
|
||
item["source_model_id"] = base_model_id
|
||
item["settings"] = self._settings_with_profile_locked(
|
||
base_model_id, item
|
||
).to_dict()
|
||
exposed.append(item)
|
||
return exposed
|
||
|
||
@staticmethod
|
||
def _has_engine_fields(profile: Dict[str, Any]) -> bool:
|
||
"""True when the profile overrides any engine-construction field.
|
||
|
||
Those fields are ignored by the request-time sampling overlay but are
|
||
used for transient engine variant reloads on exposed-profile requests.
|
||
UIs use this flag to indicate that a profile changes load-time state.
|
||
"""
|
||
settings = profile.get("settings", {}) or {}
|
||
return any(
|
||
k in MODEL_SPECIFIC_PROFILE_FIELDS and v is not None
|
||
for k, v in settings.items()
|
||
)
|
||
|
||
def list_profiles(self, model_id: str) -> list[dict]:
|
||
"""Return all profiles for ``model_id`` as serializable dicts."""
|
||
with self._lock:
|
||
per_model = self._profiles.get(model_id, {})
|
||
return [
|
||
{
|
||
**p,
|
||
"model_id": self._display_profile_model_id_locked(model_id, p),
|
||
"has_engine_fields": self._has_engine_fields(p),
|
||
}
|
||
for p in per_model.values()
|
||
]
|
||
|
||
def get_profile(self, model_id: str, name: str) -> Optional[dict]:
|
||
with self._lock:
|
||
return dict(self._profiles.get(model_id, {}).get(name, {})) or None
|
||
|
||
def save_profile(
|
||
self,
|
||
model_id: str,
|
||
name: str,
|
||
display_name: str,
|
||
description: Optional[str],
|
||
settings: Dict[str, Any],
|
||
source_template: Optional[str] = None,
|
||
expose_as_model: bool = False,
|
||
api_name: Optional[str] = None,
|
||
reserved_model_ids: Optional[set[str]] = None,
|
||
) -> dict:
|
||
"""Create a new profile. Raises if name is invalid or already exists."""
|
||
validate_profile_name(name)
|
||
filtered = filter_profile_fields(settings or {})
|
||
with self._lock:
|
||
per_model = self._profiles.setdefault(model_id, {})
|
||
if name in per_model:
|
||
raise ValueError(
|
||
f"Profile '{name}' already exists for model '{model_id}'"
|
||
)
|
||
now = utcnow().isoformat()
|
||
profile_api_name = self._allocate_profile_api_name_locked(
|
||
per_model,
|
||
api_name,
|
||
display_name=display_name,
|
||
internal_name=name,
|
||
)
|
||
profile_record = {
|
||
"name": name,
|
||
"display_name": display_name or name,
|
||
"api_name": profile_api_name,
|
||
"description": description,
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
"settings": filtered,
|
||
"source_template": source_template,
|
||
"expose_as_model": bool(expose_as_model),
|
||
}
|
||
self._validate_exposed_profile_ids_available_locked(
|
||
model_id,
|
||
profile_record,
|
||
reserved_model_ids=reserved_model_ids,
|
||
)
|
||
per_model[name] = profile_record
|
||
self._save_profiles()
|
||
return dict(per_model[name])
|
||
|
||
def update_profile(
|
||
self,
|
||
model_id: str,
|
||
name: str,
|
||
*,
|
||
new_name: Optional[str] = None,
|
||
display_name: Optional[str] = None,
|
||
description: Optional[str] = None,
|
||
settings: Optional[Dict[str, Any]] = None,
|
||
source_template: Optional[str] = None,
|
||
expose_as_model: Optional[bool] = None,
|
||
api_name: Optional[str] = None,
|
||
reserved_model_ids: Optional[set[str]] = None,
|
||
) -> Optional[dict]:
|
||
"""Update a profile's metadata/settings. Returns updated dict or None if not found."""
|
||
with self._lock:
|
||
per_model = self._profiles.get(model_id, {})
|
||
if name not in per_model:
|
||
return None
|
||
profile = dict(per_model[name])
|
||
target_name = name
|
||
rename_mode = False
|
||
if new_name is not None and new_name != name:
|
||
validate_profile_name(new_name)
|
||
if new_name in per_model:
|
||
raise ValueError(
|
||
f"Profile '{new_name}' already exists for model '{model_id}'"
|
||
)
|
||
target_name = new_name
|
||
profile["name"] = new_name
|
||
rename_mode = True
|
||
if display_name is not None:
|
||
profile["display_name"] = display_name
|
||
if api_name is not None:
|
||
profile["api_name"] = self._allocate_profile_api_name_locked(
|
||
per_model,
|
||
api_name,
|
||
display_name=profile.get("display_name"),
|
||
internal_name=target_name,
|
||
exclude_name=name,
|
||
)
|
||
if description is not None:
|
||
profile["description"] = description
|
||
if settings is not None:
|
||
profile["settings"] = filter_profile_fields(settings)
|
||
if source_template is not None:
|
||
profile["source_template"] = source_template or None
|
||
if expose_as_model is not None:
|
||
profile["expose_as_model"] = bool(expose_as_model)
|
||
profile["updated_at"] = utcnow().isoformat()
|
||
self._validate_exposed_profile_ids_available_locked(
|
||
model_id,
|
||
profile,
|
||
exclude_profile_name=name,
|
||
reserved_model_ids=reserved_model_ids,
|
||
)
|
||
|
||
# Snapshot for rollback on write failure
|
||
profiles_snapshot = copy.deepcopy(self._profiles)
|
||
settings_snapshot = copy.deepcopy(self._settings)
|
||
|
||
# Also update ModelSettings.active_profile_name if renamed and it was active
|
||
old_active = None
|
||
if rename_mode:
|
||
old_active = self._settings.get(model_id)
|
||
if old_active is not None and old_active.active_profile_name == name:
|
||
old_active.active_profile_name = target_name
|
||
del per_model[name]
|
||
|
||
per_model[target_name] = profile
|
||
|
||
# Write profiles first; if this throws, rollback everything
|
||
try:
|
||
self._save_profiles()
|
||
if rename_mode and old_active is not None:
|
||
self._save()
|
||
except Exception:
|
||
self._profiles = profiles_snapshot
|
||
self._settings = settings_snapshot
|
||
raise
|
||
|
||
return dict(profile)
|
||
|
||
def delete_profile(self, model_id: str, name: str) -> bool:
|
||
with self._lock:
|
||
per_model = self._profiles.get(model_id, {})
|
||
if name not in per_model:
|
||
return False
|
||
|
||
profiles_snapshot = copy.deepcopy(self._profiles)
|
||
settings_snapshot = copy.deepcopy(self._settings)
|
||
|
||
del per_model[name]
|
||
if not per_model and model_id in self._profiles:
|
||
del self._profiles[model_id]
|
||
# Clear active_profile_name if it referenced this profile
|
||
old_active = self._settings.get(model_id)
|
||
if old_active is not None and old_active.active_profile_name == name:
|
||
old_active.active_profile_name = None
|
||
|
||
try:
|
||
self._save_profiles()
|
||
if old_active is not None and old_active.active_profile_name is None:
|
||
self._save()
|
||
except Exception:
|
||
self._profiles = profiles_snapshot
|
||
self._settings = settings_snapshot
|
||
raise
|
||
return True
|
||
|
||
def apply_profile(
|
||
self,
|
||
model_id: str,
|
||
name: str,
|
||
settings_sanitizer: Optional[Callable[[dict[str, Any]], None]] = None,
|
||
) -> Optional[ModelSettings]:
|
||
"""Merge profile settings into the model's live settings and persist."""
|
||
with self._lock:
|
||
per_model = self._profiles.get(model_id, {})
|
||
if name not in per_model:
|
||
return None
|
||
profile_settings = per_model[name].get("settings", {}) or {}
|
||
|
||
settings_snapshot = copy.deepcopy(self._settings)
|
||
|
||
current = self._settings.get(model_id)
|
||
if current is None:
|
||
current = ModelSettings()
|
||
merged = current.to_dict()
|
||
for k, v in profile_settings.items():
|
||
merged[k] = v
|
||
merged["active_profile_name"] = name
|
||
if settings_sanitizer is not None:
|
||
settings_sanitizer(merged)
|
||
new_settings = ModelSettings.from_dict(merged)
|
||
self._settings[model_id] = new_settings
|
||
try:
|
||
self._save()
|
||
except Exception:
|
||
self._settings = settings_snapshot
|
||
raise
|
||
return ModelSettings.from_dict(new_settings.to_dict())
|
||
|
||
# ==================== Templates ====================
|
||
|
||
def _load_templates(self) -> None:
|
||
# Built-in defaults ship inside the package (omlx/default_global_templates.json)
|
||
# and are merged in at read time — they are NEVER copied to disk and never
|
||
# appear in `self._templates`. The user file under <base_path> holds
|
||
# ONLY user-created templates; a missing/empty file is the legitimate
|
||
# initial state.
|
||
if not self.templates_file.exists():
|
||
self._templates = {}
|
||
return
|
||
try:
|
||
with open(self.templates_file, "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
version = data.get("version", 1)
|
||
if version != TEMPLATES_VERSION:
|
||
logger.warning(
|
||
f"Templates file version {version} differs from current {TEMPLATES_VERSION}"
|
||
)
|
||
self._templates = data.get("templates", {}) or {}
|
||
# Migration: strip ttl_seconds from existing template settings
|
||
for name, template in self._templates.items():
|
||
settings = template.get("settings")
|
||
if settings and "ttl_seconds" in settings:
|
||
del settings["ttl_seconds"]
|
||
except Exception as e:
|
||
logger.error(f"Failed to load templates file: {e}")
|
||
self._templates = {}
|
||
|
||
def _save_templates(self) -> None:
|
||
"""Must be called while holding the lock."""
|
||
data = {"version": TEMPLATES_VERSION, "templates": self._templates}
|
||
try:
|
||
temp_file = self.templates_file.with_suffix(".tmp")
|
||
with open(temp_file, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, indent=2, ensure_ascii=False, default=str)
|
||
temp_file.replace(self.templates_file)
|
||
except Exception as e:
|
||
logger.error(f"Failed to save templates file: {e}")
|
||
raise
|
||
|
||
def list_templates(self) -> list[dict]:
|
||
# Shipped JSON seeds were retired in favor of the client-side preset
|
||
# bundle (`omlx/admin/static/omlx_preset.json`); every entry on this
|
||
# surface is user-created. Callers that distinguish presets from
|
||
# user templates do so via the preset bundle, not an `is_builtin`
|
||
# flag on this response.
|
||
with self._lock:
|
||
return [dict(t) for t in self._templates.values()]
|
||
|
||
def get_template(self, name: str) -> Optional[dict]:
|
||
with self._lock:
|
||
u = self._templates.get(name)
|
||
return dict(u) if u is not None else None
|
||
|
||
def save_template(
|
||
self,
|
||
name: str,
|
||
display_name: str,
|
||
description: Optional[str],
|
||
settings: Dict[str, Any],
|
||
) -> dict:
|
||
validate_profile_name(name)
|
||
filtered = filter_universal_fields(settings or {})
|
||
with self._lock:
|
||
if name in self._templates:
|
||
raise ValueError(f"Template '{name}' already exists")
|
||
now = utcnow().isoformat()
|
||
self._templates[name] = {
|
||
"name": name,
|
||
"display_name": display_name or name,
|
||
"description": description,
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
"settings": filtered,
|
||
}
|
||
self._save_templates()
|
||
return dict(self._templates[name])
|
||
|
||
def upsert_template(
|
||
self,
|
||
name: str,
|
||
display_name: str,
|
||
description: Optional[str],
|
||
settings: Dict[str, Any],
|
||
) -> dict:
|
||
"""Create or replace a template with the given settings."""
|
||
validate_profile_name(name)
|
||
filtered = filter_universal_fields(settings or {})
|
||
with self._lock:
|
||
now = utcnow().isoformat()
|
||
existing = self._templates.get(name)
|
||
created_at = existing["created_at"] if existing else now
|
||
self._templates[name] = {
|
||
"name": name,
|
||
"display_name": display_name or name,
|
||
"description": description,
|
||
"created_at": created_at,
|
||
"updated_at": now,
|
||
"settings": filtered,
|
||
}
|
||
self._save_templates()
|
||
return dict(self._templates[name])
|
||
|
||
def update_template(
|
||
self,
|
||
name: str,
|
||
*,
|
||
new_name: Optional[str] = None,
|
||
display_name: Optional[str] = None,
|
||
description: Optional[str] = None,
|
||
settings: Optional[Dict[str, Any]] = None,
|
||
) -> Optional[dict]:
|
||
with self._lock:
|
||
if name not in self._templates:
|
||
return None
|
||
template = dict(self._templates[name])
|
||
target = name
|
||
if new_name is not None and new_name != name:
|
||
validate_profile_name(new_name)
|
||
if new_name in self._templates:
|
||
raise ValueError(f"Template '{new_name}' already exists")
|
||
target = new_name
|
||
template["name"] = new_name
|
||
if display_name is not None:
|
||
template["display_name"] = display_name
|
||
if description is not None:
|
||
template["description"] = description
|
||
if settings is not None:
|
||
template["settings"] = filter_universal_fields(settings)
|
||
template["updated_at"] = utcnow().isoformat()
|
||
if target != name:
|
||
del self._templates[name]
|
||
self._templates[target] = template
|
||
self._save_templates()
|
||
return dict(template)
|
||
|
||
def delete_template(self, name: str) -> bool:
|
||
with self._lock:
|
||
if name not in self._templates:
|
||
return False
|
||
del self._templates[name]
|
||
self._save_templates()
|
||
return True
|