| 1 | import os |
| 2 | from copy import deepcopy |
| 3 | |
| 4 | import models |
| 5 | from helpers import defer, plugins, files |
| 6 | from helpers.extension import call_extensions_async |
| 7 | from helpers import yaml as yaml_helper |
| 8 | from helpers.providers import get_provider_config, get_providers |
| 9 | |
| 10 | PRESETS_FILE = "presets.yaml" |
| 11 | FALLBACK_PRESETS_FILE = "mode_presets_fallback.yaml" |
| 12 | PROVIDER_METADATA_FILE = "provider_metadata.yaml" |
| 13 | DEFAULT_PRESET_NAME = "Default" |
| 14 | DEFAULT_VISION_TIMEOUT_SECONDS = 300 |
| 15 | DEFAULT_VISION_MAX_TOKENS = 2000 |
| 16 | MODEL_PRESET_CONFIG_KEY = "model_preset" |
| 17 | PRESET_SCOPE_GLOBAL = "global" |
| 18 | PRESET_SCOPE_PROJECT = "project" |
| 19 | PRESET_SLOT_CONFIG_SECTIONS = { |
| 20 | "chat": "chat_model", |
| 21 | "vision": "vision_model", |
| 22 | "utility": "utility_model", |
| 23 | "embedding": "embedding_model", |
| 24 | } |
| 25 | MODEL_SLOT_PRESET_REPLACE_FIELDS = {"kwargs"} |
| 26 | IMPLICIT_PRESET_SLOT_DEFAULTS = { |
| 27 | "vision": { |
| 28 | "vision": True, |
| 29 | "max_embeds": 10, |
| 30 | "timeout": DEFAULT_VISION_TIMEOUT_SECONDS, |
| 31 | "max_tokens": DEFAULT_VISION_MAX_TOKENS, |
| 32 | "override_main": False, |
| 33 | "rl_requests": 0, |
| 34 | "rl_input": 0, |
| 35 | "rl_output": 0, |
| 36 | "kwargs": {}, |
| 37 | }, |
| 38 | "utility": { |
| 39 | "ctx_length": 128000, |
| 40 | "ctx_input": 0.7, |
| 41 | "rl_requests": 0, |
| 42 | "rl_input": 0, |
| 43 | "rl_output": 0, |
| 44 | "kwargs": {}, |
| 45 | }, |
| 46 | "embedding": { |
| 47 | "rl_requests": 0, |
| 48 | "rl_input": 0, |
| 49 | "kwargs": {}, |
| 50 | }, |
| 51 | } |
| 52 | LOCAL_PROVIDERS = {"ollama", "lm_studio", "llama_cpp", "omlx", "vllm"} |
| 53 | LOCAL_EMBEDDING = {"huggingface"} |
| 54 | _PROVIDER_METADATA_CACHE: dict | None = None |
| 55 | |
| 56 | |
| 57 | def _get_provider_metadata_path() -> str: |
| 58 | plugin_dir = plugins.find_plugin_dir("_model_config") |
| 59 | return files.get_abs_path(plugin_dir, PROVIDER_METADATA_FILE) if plugin_dir else "" |
| 60 | |
| 61 | |
| 62 | def get_provider_metadata(model_type: str = "chat", provider: str = "") -> dict: |
| 63 | """Get plugin-owned provider metadata that does not belong in conf/model_providers.yaml.""" |
| 64 | global _PROVIDER_METADATA_CACHE |
| 65 | if _PROVIDER_METADATA_CACHE is None: |
| 66 | path = _get_provider_metadata_path() |
| 67 | if path and files.exists(path): |
| 68 | data = yaml_helper.loads(files.read_file(path)) |
| 69 | _PROVIDER_METADATA_CACHE = data if isinstance(data, dict) else {} |
| 70 | else: |
| 71 | _PROVIDER_METADATA_CACHE = {} |
| 72 | |
| 73 | section = _PROVIDER_METADATA_CACHE.get(model_type, {}) |
| 74 | if not isinstance(section, dict): |
| 75 | return {} |
| 76 | meta = section.get(str(provider or "").strip().lower(), {}) |
| 77 | return meta if isinstance(meta, dict) else {} |
| 78 | |
| 79 | |
| 80 | def _model_type_for_label(label: str) -> str: |
| 81 | return "embedding" if label == "Embedding Model" else "chat" |
| 82 | |
| 83 | |
| 84 | def provider_requires_api_key(provider: str, model_type: str = "chat") -> bool: |
| 85 | provider_id = str(provider or "").strip().lower() |
| 86 | if not provider_id: |
| 87 | return False |
| 88 | cfg = get_provider_config(model_type, provider_id) or get_provider_config("chat", provider_id) or {} |
| 89 | meta = get_provider_metadata(model_type, provider_id) or get_provider_metadata("chat", provider_id) |
| 90 | mode = str(meta.get("api_key_mode") or cfg.get("api_key_mode") or "required").strip().lower() |
| 91 | return mode not in {"none", "optional", "oauth"} |
| 92 | |
| 93 | |
| 94 | def _get_presets_path(project_name: str | None = None) -> str: |
| 95 | """Return the user-editable presets path for the requested scope.""" |
| 96 | if project_name: |
| 97 | return plugins.determine_plugin_asset_path( |
| 98 | "_model_config", project_name, "", PRESETS_FILE |
| 99 | ) |
| 100 | return files.get_abs_path(files.USER_DIR, files.PLUGINS_DIR, "_model_config", PRESETS_FILE) |
| 101 | |
| 102 | |
| 103 | def _get_fallback_presets_path() -> str: |
| 104 | """Return the plugin-local fallback used when no saved presets exist.""" |
| 105 | plugin_dir = plugins.find_plugin_dir("_model_config") |
| 106 | return files.get_abs_path(plugin_dir, FALLBACK_PRESETS_FILE) if plugin_dir else "" |
| 107 | |
| 108 | |
| 109 | def get_config(agent=None, project_name=None, agent_profile=None): |
| 110 | """Get the resolved model config for an agent or selected scope.""" |
| 111 | config = plugins.get_plugin_config( |
| 112 | "_model_config", |
| 113 | agent=agent, |
| 114 | project_name=project_name, |
| 115 | agent_profile=agent_profile, |
| 116 | ) or {} |
| 117 | # The plugin hook resolves selection-only config. Keep this boundary robust |
| 118 | # when hooks are disabled by tests or embedding applications. |
| 119 | if any(section in config for section in PRESET_SLOT_CONFIG_SECTIONS.values()): |
| 120 | return config |
| 121 | return resolve_config_settings(config) |
| 122 | |
| 123 | |
| 124 | def get_configured_preset_name(agent=None, project_name=None, agent_profile=None) -> str: |
| 125 | """Return the valid scoped preset selection, falling back to Default.""" |
| 126 | config = plugins.get_plugin_config( |
| 127 | "_model_config", |
| 128 | agent=agent, |
| 129 | project_name=project_name, |
| 130 | agent_profile=agent_profile, |
| 131 | ) or {} |
| 132 | name = str(config.get(MODEL_PRESET_CONFIG_KEY) or DEFAULT_PRESET_NAME).strip() |
| 133 | return name if resolve_preset(name) else DEFAULT_PRESET_NAME |
| 134 | |
| 135 | |
| 136 | def preset_to_config(preset: dict) -> dict: |
| 137 | """Convert a complete preset into the legacy runtime model-config shape.""" |
| 138 | config: dict = {} |
| 139 | for slot, section in PRESET_SLOT_CONFIG_SECTIONS.items(): |
| 140 | slot_config = preset.get(slot) if isinstance(preset, dict) else None |
| 141 | config[section] = ( |
| 142 | _strip_ui_fields(slot_config, strip_api_key=False) |
| 143 | if isinstance(slot_config, dict) |
| 144 | else {} |
| 145 | ) |
| 146 | return config |
| 147 | |
| 148 | |
| 149 | def config_to_preset(config: dict, name: str = DEFAULT_PRESET_NAME) -> dict: |
| 150 | """Convert legacy full model config into a preset without UI/API-key fields.""" |
| 151 | preset = {"name": str(name or "").strip()} |
| 152 | for slot, section in PRESET_SLOT_CONFIG_SECTIONS.items(): |
| 153 | slot_config = config.get(section) if isinstance(config, dict) else None |
| 154 | if isinstance(slot_config, dict): |
| 155 | preset[slot] = _strip_ui_fields(slot_config, strip_api_key=True) |
| 156 | return preset |
| 157 | |
| 158 | |
| 159 | def resolve_config_settings(settings: dict | None) -> dict: |
| 160 | """Resolve selection-only settings to the complete runtime config shape.""" |
| 161 | raw = settings if isinstance(settings, dict) else {} |
| 162 | selected_name = str(raw.get(MODEL_PRESET_CONFIG_KEY) or DEFAULT_PRESET_NAME).strip() |
| 163 | default_preset = resolve_preset(DEFAULT_PRESET_NAME) or {"name": DEFAULT_PRESET_NAME} |
| 164 | selected = resolve_preset(selected_name) or default_preset |
| 165 | config = preset_to_config(default_preset) |
| 166 | if selected.get("name") != DEFAULT_PRESET_NAME: |
| 167 | config = build_config_from_preset(selected, config, strip_api_key=False) |
| 168 | config[MODEL_PRESET_CONFIG_KEY] = str(selected.get("name") or DEFAULT_PRESET_NAME) |
| 169 | # Retained as a read-only compatibility flag for integrations that still |
| 170 | # inspect it. The switcher is always available in the unified preset model. |
| 171 | config["allow_chat_override"] = True |
| 172 | return config |
| 173 | |
| 174 | |
| 175 | def has_project_config(project_name: str) -> bool: |
| 176 | path = plugins.determine_plugin_asset_path( |
| 177 | "_model_config", project_name, "", plugins.CONFIG_FILE_NAME |
| 178 | ) |
| 179 | return files.exists(path) |
| 180 | |
| 181 | |
| 182 | def load_project_llm_data(project_name: str) -> dict: |
| 183 | """Build the preset-selection payload shown in Project Settings.""" |
| 184 | project_config_exists = has_project_config(project_name) |
| 185 | preset_name = get_configured_preset_name(project_name=project_name) |
| 186 | return { |
| 187 | "has_project_config": project_config_exists, |
| 188 | "selected_preset": { |
| 189 | "scope": PRESET_SCOPE_GLOBAL, |
| 190 | "project_name": "", |
| 191 | "name": preset_name, |
| 192 | }, |
| 193 | "presets": get_combined_presets(), |
| 194 | "global_presets": get_presets(), |
| 195 | "project_presets": [], |
| 196 | } |
| 197 | |
| 198 | |
| 199 | def save_project_llm_settings(project_name: str, llm_data: object) -> None: |
| 200 | """Persist only a project preset selection from Project Settings.""" |
| 201 | if not isinstance(llm_data, dict): |
| 202 | return |
| 203 | selected_preset = llm_data.get("selected_preset") |
| 204 | if not isinstance(selected_preset, dict): |
| 205 | return |
| 206 | name = str(selected_preset.get("name") or "").strip() |
| 207 | if resolve_preset(name): |
| 208 | if not has_project_config(project_name) and name == get_configured_preset_name(): |
| 209 | return |
| 210 | previous_embedding = get_config(project_name=project_name).get( |
| 211 | "embedding_model", |
| 212 | {}, |
| 213 | ) |
| 214 | plugins.save_plugin_config( |
| 215 | "_model_config", |
| 216 | project_name, |
| 217 | "", |
| 218 | {MODEL_PRESET_CONFIG_KEY: name}, |
| 219 | ) |
| 220 | current_embedding = get_config(project_name=project_name).get( |
| 221 | "embedding_model", |
| 222 | {}, |
| 223 | ) |
| 224 | if previous_embedding != current_embedding: |
| 225 | defer.DeferredTask().start_task( |
| 226 | call_extensions_async, |
| 227 | "embedding_model_changed", |
| 228 | ) |
| 229 | |
| 230 | |
| 231 | def _load_presets_from_path(path: str) -> list | None: |
| 232 | if files.exists(path): |
| 233 | try: |
| 234 | data = yaml_helper.loads(files.read_file(path)) |
| 235 | except Exception: |
| 236 | return None |
| 237 | if isinstance(data, list): |
| 238 | return data |
| 239 | return None |
| 240 | |
| 241 | |
| 242 | def _strip_ui_fields(value: dict, *, strip_api_key: bool) -> dict: |
| 243 | cleaned = deepcopy(value) |
| 244 | for key in list(cleaned.keys()): |
| 245 | if key.startswith("_"): |
| 246 | cleaned.pop(key, None) |
| 247 | if strip_api_key: |
| 248 | cleaned.pop("api_key", None) |
| 249 | return cleaned |
| 250 | |
| 251 | |
| 252 | def _preset_default_values_equal(value, default) -> bool: |
| 253 | if isinstance(default, float): |
| 254 | try: |
| 255 | return float(value) == default |
| 256 | except (TypeError, ValueError): |
| 257 | return False |
| 258 | return value == default |
| 259 | |
| 260 | |
| 261 | def _strip_implicit_preset_defaults(slot: str, slot_config: dict) -> dict: |
| 262 | cleaned = deepcopy(slot_config) |
| 263 | defaults = IMPLICIT_PRESET_SLOT_DEFAULTS.get(slot, {}) |
| 264 | for key, default in defaults.items(): |
| 265 | if key in cleaned and _preset_default_values_equal(cleaned[key], default): |
| 266 | cleaned.pop(key, None) |
| 267 | return cleaned |
| 268 | |
| 269 | |
| 270 | def _clean_preset_for_file(preset: dict) -> dict: |
| 271 | name = str(preset.get("name", "") or "").strip() |
| 272 | if name.casefold() == DEFAULT_PRESET_NAME.casefold(): |
| 273 | name = DEFAULT_PRESET_NAME |
| 274 | cleaned = { |
| 275 | "name": name, |
| 276 | } |
| 277 | has_named_slots = any( |
| 278 | isinstance(preset.get(slot), dict) for slot in PRESET_SLOT_CONFIG_SECTIONS |
| 279 | ) |
| 280 | for slot in PRESET_SLOT_CONFIG_SECTIONS: |
| 281 | slot_config = preset.get(slot) |
| 282 | if isinstance(slot_config, dict): |
| 283 | slot_clean = _strip_ui_fields(slot_config, strip_api_key=True) |
| 284 | cleaned[slot] = ( |
| 285 | slot_clean |
| 286 | if name == DEFAULT_PRESET_NAME |
| 287 | else _strip_implicit_preset_defaults(slot, slot_clean) |
| 288 | ) |
| 289 | # Very old presets stored the main model directly beside ``name``. Preserve |
| 290 | # those definitions while bringing them into the canonical slot schema. |
| 291 | if not has_named_slots and _slot_has_identity(preset): |
| 292 | raw_chat = { |
| 293 | key: value |
| 294 | for key, value in preset.items() |
| 295 | if key not in {"name", "scope", "project_name"} |
| 296 | } |
| 297 | raw_chat["name"] = name |
| 298 | cleaned["chat"] = _strip_ui_fields(raw_chat, strip_api_key=True) |
| 299 | return cleaned |
| 300 | |
| 301 | |
| 302 | def clean_presets_for_file(presets: list) -> list: |
| 303 | """Return presets without API/UI metadata, preserving the plain YAML schema.""" |
| 304 | cleaned = [] |
| 305 | for preset in presets: |
| 306 | if isinstance(preset, dict): |
| 307 | cleaned.append(_clean_preset_for_file(preset)) |
| 308 | return cleaned |
| 309 | |
| 310 | |
| 311 | def validate_presets(presets: list, *, require_default: bool = True) -> list: |
| 312 | """Validate and clean the durable global preset collection.""" |
| 313 | if not isinstance(presets, list): |
| 314 | raise ValueError("Presets must be a list.") |
| 315 | |
| 316 | cleaned: list[dict] = [] |
| 317 | seen: set[str] = set() |
| 318 | for raw in presets: |
| 319 | if not isinstance(raw, dict): |
| 320 | raise ValueError("Every preset must be an object.") |
| 321 | preset = _clean_preset_for_file(raw) |
| 322 | name = str(preset.get("name") or "").strip() |
| 323 | if not name: |
| 324 | raise ValueError("Preset names cannot be empty.") |
| 325 | normalized = name.casefold() |
| 326 | if normalized in seen: |
| 327 | raise ValueError(f"Preset names must be unique: '{name}'.") |
| 328 | if normalized == DEFAULT_PRESET_NAME.casefold(): |
| 329 | preset["name"] = DEFAULT_PRESET_NAME |
| 330 | for slot, label in ( |
| 331 | ("chat", "main"), |
| 332 | ("utility", "utility"), |
| 333 | ("embedding", "embedding"), |
| 334 | ): |
| 335 | if not _slot_has_identity(preset.get(slot) or {}): |
| 336 | raise ValueError( |
| 337 | f"The Default preset requires a {label} model." |
| 338 | ) |
| 339 | seen.add(normalized) |
| 340 | cleaned.append(preset) |
| 341 | |
| 342 | default_index = next( |
| 343 | (i for i, preset in enumerate(cleaned) if preset["name"] == DEFAULT_PRESET_NAME), |
| 344 | None, |
| 345 | ) |
| 346 | if require_default and default_index is None: |
| 347 | raise ValueError("The Default preset cannot be deleted or renamed.") |
| 348 | if default_index not in (None, 0): |
| 349 | cleaned.insert(0, cleaned.pop(default_index)) |
| 350 | return cleaned |
| 351 | |
| 352 | |
| 353 | def normalize_config_for_save(config: dict) -> dict: |
| 354 | """Remove UI-only fields and inline API keys before storing scoped config.""" |
| 355 | cleaned = deepcopy(config or {}) |
| 356 | for section_name in ( |
| 357 | "chat_model", |
| 358 | "vision_model", |
| 359 | "utility_model", |
| 360 | "embedding_model", |
| 361 | ): |
| 362 | section = cleaned.get(section_name) |
| 363 | if isinstance(section, dict): |
| 364 | cleaned[section_name] = _strip_ui_fields(section, strip_api_key=True) |
| 365 | return cleaned |
| 366 | |
| 367 | |
| 368 | def _legacy_default_preset() -> dict | None: |
| 369 | """Build Default from a pre-v2 global config when startup migration has not run.""" |
| 370 | path = plugins.determine_plugin_asset_path( |
| 371 | "_model_config", "", "", plugins.CONFIG_FILE_NAME |
| 372 | ) |
| 373 | if not files.exists(path): |
| 374 | return None |
| 375 | try: |
| 376 | raw = files.read_file_json(path) |
| 377 | except Exception: |
| 378 | return None |
| 379 | if not isinstance(raw, dict) or not any( |
| 380 | section in raw for section in PRESET_SLOT_CONFIG_SECTIONS.values() |
| 381 | ): |
| 382 | return None |
| 383 | return config_to_preset(raw, DEFAULT_PRESET_NAME) |
| 384 | |
| 385 | |
| 386 | def parse_preset_collection(text: str) -> list: |
| 387 | """Parse, validate, and sanitize a preset YAML document.""" |
| 388 | return validate_presets(yaml_helper.loads(text)) |
| 389 | |
| 390 | |
| 391 | def _fallback_presets() -> list: |
| 392 | path = _get_fallback_presets_path() |
| 393 | if not files.exists(path): |
| 394 | return [] |
| 395 | try: |
| 396 | return parse_preset_collection(files.read_file(path)) |
| 397 | except Exception: |
| 398 | return [] |
| 399 | |
| 400 | |
| 401 | def _ensure_default_preset(presets: list) -> list: |
| 402 | result = [deepcopy(preset) for preset in presets if isinstance(preset, dict)] |
| 403 | legacy_default = _legacy_default_preset() |
| 404 | bundled_default = next( |
| 405 | ( |
| 406 | deepcopy(preset) |
| 407 | for preset in _fallback_presets() |
| 408 | if isinstance(preset, dict) |
| 409 | and str(preset.get("name") or "").strip().casefold() |
| 410 | == DEFAULT_PRESET_NAME.casefold() |
| 411 | ), |
| 412 | None, |
| 413 | ) |
| 414 | fallback_default = bundled_default or {"name": DEFAULT_PRESET_NAME} |
| 415 | if legacy_default: |
| 416 | for slot in PRESET_SLOT_CONFIG_SECTIONS: |
| 417 | legacy_slot = legacy_default.get(slot) |
| 418 | if _slot_has_identity(legacy_slot or {}): |
| 419 | fallback_default[slot] = deepcopy(legacy_slot) |
| 420 | default_index = next( |
| 421 | ( |
| 422 | i |
| 423 | for i, preset in enumerate(result) |
| 424 | if str(preset.get("name") or "").strip().casefold() |
| 425 | == DEFAULT_PRESET_NAME.casefold() |
| 426 | ), |
| 427 | None, |
| 428 | ) |
| 429 | if default_index is not None: |
| 430 | result[default_index]["name"] = DEFAULT_PRESET_NAME |
| 431 | for slot in PRESET_SLOT_CONFIG_SECTIONS: |
| 432 | if not _slot_has_identity(result[default_index].get(slot) or {}): |
| 433 | fallback_slot = fallback_default.get(slot) |
| 434 | if isinstance(fallback_slot, dict): |
| 435 | result[default_index][slot] = deepcopy(fallback_slot) |
| 436 | if default_index: |
| 437 | result.insert(0, result.pop(default_index)) |
| 438 | return result |
| 439 | |
| 440 | result.insert(0, fallback_default) |
| 441 | return result |
| 442 | |
| 443 | |
| 444 | def get_presets(project_name: str | None = None) -> list: |
| 445 | """Get global presets with the required Default preset first.""" |
| 446 | if project_name: |
| 447 | return get_project_presets(project_name) |
| 448 | |
| 449 | path = _get_presets_path() |
| 450 | presets = _load_presets_from_path(path) |
| 451 | if presets is not None: |
| 452 | return _ensure_default_preset(presets) |
| 453 | |
| 454 | # Fall back to the repository-shipped offline collection. |
| 455 | return _ensure_default_preset(_fallback_presets()) |
| 456 | |
| 457 | |
| 458 | def get_project_presets(project_name: str) -> list: |
| 459 | """Load legacy project presets for migration/compatibility only.""" |
| 460 | return _load_presets_from_path(_get_presets_path(project_name)) or [] |
| 461 | |
| 462 | |
| 463 | def _with_preset_metadata(preset: dict, scope: str, project_name: str = "") -> dict: |
| 464 | item = deepcopy(preset) |
| 465 | item["scope"] = scope |
| 466 | item["project_name"] = project_name if scope == PRESET_SCOPE_PROJECT else "" |
| 467 | item["name"] = str(item.get("name", "") or "") |
| 468 | return item |
| 469 | |
| 470 | |
| 471 | def get_combined_presets(project_name: str | None = None) -> list: |
| 472 | """Get global presets with API metadata (project definitions are retired).""" |
| 473 | return [ |
| 474 | _with_preset_metadata(preset, PRESET_SCOPE_GLOBAL) |
| 475 | for preset in get_presets() |
| 476 | if isinstance(preset, dict) |
| 477 | ] |
| 478 | |
| 479 | |
| 480 | def save_presets(presets: list, project_name: str | None = None) -> None: |
| 481 | """Save global presets while enforcing the immutable Default identity.""" |
| 482 | if project_name: |
| 483 | raise ValueError("Project-specific preset definitions are no longer supported.") |
| 484 | cleaned = validate_presets(presets) |
| 485 | path = _get_presets_path(project_name) |
| 486 | files.write_file(path, yaml_helper.dumps(cleaned)) |
| 487 | |
| 488 | |
| 489 | def update_preset_from_config(name: str, config: dict) -> dict: |
| 490 | """Replace one global preset's model slots from a legacy config payload.""" |
| 491 | target = resolve_preset(name) |
| 492 | if not target: |
| 493 | raise ValueError(f"Preset '{name}' was not found.") |
| 494 | canonical_name = str(target.get("name") or DEFAULT_PRESET_NAME) |
| 495 | replacement = config_to_preset(config, canonical_name) |
| 496 | if canonical_name == DEFAULT_PRESET_NAME: |
| 497 | for slot in PRESET_SLOT_CONFIG_SECTIONS: |
| 498 | if not _slot_has_identity(replacement.get(slot) or {}): |
| 499 | current_slot = target.get(slot) |
| 500 | if isinstance(current_slot, dict): |
| 501 | replacement[slot] = deepcopy(current_slot) |
| 502 | presets = get_presets() |
| 503 | updated = False |
| 504 | for index, preset in enumerate(presets): |
| 505 | if str(preset.get("name") or "").casefold() == canonical_name.casefold(): |
| 506 | presets[index] = replacement |
| 507 | updated = True |
| 508 | break |
| 509 | if not updated: |
| 510 | raise ValueError(f"Preset '{name}' was not found.") |
| 511 | save_presets(presets) |
| 512 | return replacement |
| 513 | |
| 514 | |
| 515 | def reset_presets(project_name: str | None = None) -> list: |
| 516 | """Delete user presets for the scope. Global reset falls back to bundled defaults.""" |
| 517 | if project_name: |
| 518 | raise ValueError("Project-specific preset definitions are no longer supported.") |
| 519 | path = _get_presets_path(project_name) |
| 520 | if os.path.exists(path): |
| 521 | os.remove(path) |
| 522 | return get_presets() |
| 523 | |
| 524 | |
| 525 | def resolve_preset( |
| 526 | name: str, |
| 527 | *, |
| 528 | scope: str = PRESET_SCOPE_GLOBAL, |
| 529 | project_name: str | None = None, |
| 530 | ) -> dict | None: |
| 531 | """Resolve a preset by explicit scope so same-name presets are unambiguous.""" |
| 532 | if scope == PRESET_SCOPE_PROJECT: |
| 533 | return None |
| 534 | presets = get_presets() |
| 535 | |
| 536 | for p in presets: |
| 537 | if str(p.get("name") or "").casefold() == str(name or "").strip().casefold(): |
| 538 | return p |
| 539 | return None |
| 540 | |
| 541 | |
| 542 | def resolve_preset_selection(selection: dict | str, project_name: str | None = None) -> dict | None: |
| 543 | """Resolve a UI/API preset selection payload to a preset dict.""" |
| 544 | if isinstance(selection, str): |
| 545 | return resolve_preset(selection) |
| 546 | if not isinstance(selection, dict): |
| 547 | return None |
| 548 | |
| 549 | scope = str(selection.get("scope") or PRESET_SCOPE_GLOBAL) |
| 550 | if scope == "current": |
| 551 | return None |
| 552 | name = str(selection.get("name") or "") |
| 553 | selected_project = str(selection.get("project_name") or project_name or "") |
| 554 | return resolve_preset(name, scope=scope, project_name=selected_project or None) |
| 555 | |
| 556 | |
| 557 | def get_preset_by_name( |
| 558 | name: str, |
| 559 | *, |
| 560 | scope: str = PRESET_SCOPE_GLOBAL, |
| 561 | project_name: str | None = None, |
| 562 | ) -> dict | None: |
| 563 | """Find a preset by name. Defaults to global presets for legacy callers.""" |
| 564 | return resolve_preset(name, scope=scope, project_name=project_name) |
| 565 | |
| 566 | |
| 567 | def _deep_merge_dict(base: dict, override: dict) -> dict: |
| 568 | """Recursively overlay override onto base without mutating either input.""" |
| 569 | result = deepcopy(base) if isinstance(base, dict) else {} |
| 570 | for key, value in override.items(): |
| 571 | if ( |
| 572 | isinstance(value, dict) |
| 573 | and isinstance(result.get(key), dict) |
| 574 | ): |
| 575 | result[key] = _deep_merge_dict(result[key], value) |
| 576 | else: |
| 577 | result[key] = deepcopy(value) |
| 578 | return result |
| 579 | |
| 580 | |
| 581 | def _replace_preset_model_slot_fields(base: dict, override: dict, result: dict) -> dict: |
| 582 | """Clear or replace provider-specific fields that must not leak across presets.""" |
| 583 | for key in MODEL_SLOT_PRESET_REPLACE_FIELDS: |
| 584 | if key in override: |
| 585 | value = override.get(key) |
| 586 | result[key] = deepcopy(value) if isinstance(value, dict) else {} |
| 587 | elif key in base: |
| 588 | result[key] = {} |
| 589 | return result |
| 590 | |
| 591 | |
| 592 | def _slot_has_identity(slot_config: dict) -> bool: |
| 593 | return bool(slot_config.get("provider") or slot_config.get("name")) |
| 594 | |
| 595 | |
| 596 | def _get_preset_slot_config(preset: dict, slot: str) -> dict | None: |
| 597 | """Return the preset payload for a slot. |
| 598 | |
| 599 | Legacy raw overrides store the main/chat model directly at the top level, |
| 600 | while named presets store it under the "chat" key. |
| 601 | """ |
| 602 | if not isinstance(preset, dict): |
| 603 | return None |
| 604 | |
| 605 | slot_config = preset.get(slot) |
| 606 | if isinstance(slot_config, dict): |
| 607 | return slot_config |
| 608 | |
| 609 | if slot == "chat" and not any(key in preset for key in PRESET_SLOT_CONFIG_SECTIONS): |
| 610 | if _slot_has_identity(preset): |
| 611 | return preset |
| 612 | |
| 613 | return None |
| 614 | |
| 615 | |
| 616 | def _should_apply_preset_slot(slot: str, slot_config: dict | None) -> bool: |
| 617 | if not isinstance(slot_config, dict): |
| 618 | return False |
| 619 | |
| 620 | cleaned = _strip_implicit_preset_defaults( |
| 621 | slot, |
| 622 | _strip_ui_fields(slot_config, strip_api_key=False), |
| 623 | ) |
| 624 | meaningful = { |
| 625 | key: value |
| 626 | for key, value in cleaned.items() |
| 627 | if key != "api_key" |
| 628 | } |
| 629 | if not meaningful: |
| 630 | return False |
| 631 | |
| 632 | # Slots inherit the configured model unless the preset declares a model |
| 633 | # identity for that slot. This keeps empty UI placeholders from accidentally |
| 634 | # overriding context/rate-limit settings. |
| 635 | return _slot_has_identity(cleaned) |
| 636 | |
| 637 | |
| 638 | def _merge_model_slot( |
| 639 | slot: str, |
| 640 | base_slot: dict, |
| 641 | preset_slot: dict, |
| 642 | *, |
| 643 | strip_api_key: bool, |
| 644 | ) -> dict: |
| 645 | cleaned = _strip_implicit_preset_defaults( |
| 646 | slot, |
| 647 | _strip_ui_fields(preset_slot, strip_api_key=strip_api_key), |
| 648 | ) |
| 649 | if not strip_api_key and not str(cleaned.get("api_key") or "").strip(): |
| 650 | cleaned.pop("api_key", None) |
| 651 | base = base_slot if isinstance(base_slot, dict) else {} |
| 652 | return _replace_preset_model_slot_fields(base, cleaned, _deep_merge_dict(base, cleaned)) |
| 653 | |
| 654 | |
| 655 | def build_config_from_preset( |
| 656 | preset: dict, |
| 657 | base_config: dict, |
| 658 | *, |
| 659 | strip_api_key: bool = True, |
| 660 | slots: tuple[str, ...] | None = None, |
| 661 | ) -> dict: |
| 662 | """Overlay preset settings onto a standalone model config. |
| 663 | |
| 664 | Presets are intentionally partial: omitted fields inherit from the current |
| 665 | config, so selecting a preset does not reset tuned values such as context |
| 666 | windows or rate limits. Provider-specific kwargs are replaced when present |
| 667 | and cleared when omitted so stale params do not leak between providers. |
| 668 | """ |
| 669 | config = ( |
| 670 | normalize_config_for_save(base_config) |
| 671 | if strip_api_key |
| 672 | else deepcopy(base_config or {}) |
| 673 | ) |
| 674 | |
| 675 | for slot in slots or tuple(PRESET_SLOT_CONFIG_SECTIONS): |
| 676 | section = PRESET_SLOT_CONFIG_SECTIONS.get(slot) |
| 677 | if not section: |
| 678 | continue |
| 679 | slot_config = _get_preset_slot_config(preset, slot) |
| 680 | if not _should_apply_preset_slot(slot, slot_config): |
| 681 | if slot == "vision": |
| 682 | config[section] = {} |
| 683 | continue |
| 684 | config[section] = _merge_model_slot( |
| 685 | slot, |
| 686 | {} if slot == "vision" else config.get(section, {}), |
| 687 | slot_config, |
| 688 | strip_api_key=strip_api_key, |
| 689 | ) |
| 690 | |
| 691 | return config |
| 692 | |
| 693 | |
| 694 | def _resolve_override(agent) -> dict | None: |
| 695 | """Resolve the active per-chat override config dict. |
| 696 | Supports both raw override dicts and preset-based overrides. |
| 697 | Returns None if no override is active or if override is not allowed.""" |
| 698 | if not agent: |
| 699 | return None |
| 700 | if not is_chat_override_allowed(agent): |
| 701 | return None |
| 702 | override = agent.context.get_data("chat_model_override") |
| 703 | if not override: |
| 704 | return None |
| 705 | |
| 706 | # If this is a preset reference, resolve it |
| 707 | if "preset_name" in override: |
| 708 | preset = get_preset_by_name(override["preset_name"]) |
| 709 | if not preset: |
| 710 | return None |
| 711 | return preset |
| 712 | |
| 713 | return override |
| 714 | |
| 715 | |
| 716 | def get_effective_preset_name(agent=None) -> str: |
| 717 | """Return the valid preset used by a chat, including its explicit override.""" |
| 718 | if agent: |
| 719 | override = getattr(agent, "context", None) |
| 720 | override = override.get_data("chat_model_override") if override else None |
| 721 | if isinstance(override, dict): |
| 722 | name = str(override.get("preset_name") or "").strip() |
| 723 | preset = resolve_preset(name) if name else None |
| 724 | if preset: |
| 725 | return str(preset.get("name") or DEFAULT_PRESET_NAME) |
| 726 | config = get_config(agent) |
| 727 | return str(config.get(MODEL_PRESET_CONFIG_KEY) or DEFAULT_PRESET_NAME) |
| 728 | |
| 729 | |
| 730 | def get_effective_config(agent=None) -> dict: |
| 731 | """Resolve the complete model config, including a per-chat preset selection.""" |
| 732 | config = get_config(agent) |
| 733 | raw_override = None |
| 734 | if agent and getattr(agent, "context", None): |
| 735 | raw_override = agent.context.get_data("chat_model_override") |
| 736 | uses_named_preset = isinstance(raw_override, dict) and bool( |
| 737 | raw_override.get("preset_name") |
| 738 | ) |
| 739 | override = _resolve_override(agent) |
| 740 | if override: |
| 741 | base = ( |
| 742 | preset_to_config(resolve_preset(DEFAULT_PRESET_NAME) or {}) |
| 743 | if uses_named_preset |
| 744 | else config |
| 745 | ) |
| 746 | config = build_config_from_preset( |
| 747 | override, |
| 748 | base, |
| 749 | strip_api_key=False, |
| 750 | ) |
| 751 | if uses_named_preset: |
| 752 | config[MODEL_PRESET_CONFIG_KEY] = get_effective_preset_name(agent) |
| 753 | config["allow_chat_override"] = True |
| 754 | return config |
| 755 | |
| 756 | |
| 757 | def get_chat_model_config(agent=None) -> dict: |
| 758 | """Get chat model config, with per-chat override if active.""" |
| 759 | return get_effective_config(agent).get("chat_model", {}) |
| 760 | |
| 761 | |
| 762 | def get_vision_model_config(agent=None) -> dict: |
| 763 | """Get the active Vision Model config after applying Main-first routing.""" |
| 764 | cfg = get_effective_config(agent) |
| 765 | vision_cfg = cfg.get("vision_model", {}) |
| 766 | if not all( |
| 767 | str(vision_cfg.get(key) or "").strip() for key in ("provider", "name") |
| 768 | ): |
| 769 | return {} |
| 770 | chat_cfg = cfg.get("chat_model", {}) |
| 771 | return ( |
| 772 | vision_cfg |
| 773 | if not chat_cfg.get("vision") or vision_cfg.get("override_main") |
| 774 | else {} |
| 775 | ) |
| 776 | |
| 777 | |
| 778 | def get_utility_model_config(agent=None) -> dict: |
| 779 | """Get utility model config, with per-chat override if active.""" |
| 780 | return get_effective_config(agent).get("utility_model", {}) |
| 781 | |
| 782 | |
| 783 | def get_embedding_model_config(agent=None) -> dict: |
| 784 | """Get embedding model config from the effective preset.""" |
| 785 | cfg = get_effective_config(agent) |
| 786 | model_cfg = deepcopy(cfg.get("embedding_model", {})) |
| 787 | provider = str(model_cfg.get("provider") or "").strip().lower() |
| 788 | name = str(model_cfg.get("name") or "").strip().strip('"').strip("'") |
| 789 | |
| 790 | if provider: |
| 791 | model_cfg["provider"] = provider |
| 792 | if name: |
| 793 | model_cfg["name"] = name |
| 794 | |
| 795 | if name.startswith("huggingface/sentence-transformers/"): |
| 796 | model_cfg["provider"] = "huggingface" |
| 797 | model_cfg["name"] = name.removeprefix("huggingface/") |
| 798 | elif name.startswith("sentence-transformers/") and provider in {"", "openai", "other"}: |
| 799 | model_cfg["provider"] = "huggingface" |
| 800 | elif provider == "huggingface" and name == "all-MiniLM-L6-v2": |
| 801 | model_cfg["name"] = "sentence-transformers/all-MiniLM-L6-v2" |
| 802 | |
| 803 | return model_cfg |
| 804 | |
| 805 | |
| 806 | def is_chat_override_allowed(agent=None) -> bool: |
| 807 | """The unified preset switcher is always enabled.""" |
| 808 | return True |
| 809 | |
| 810 | |
| 811 | def get_ctx_history(agent=None) -> float: |
| 812 | """Get the chat model context history ratio.""" |
| 813 | cfg = get_chat_model_config(agent) |
| 814 | return float(cfg.get("ctx_history", 0.7)) |
| 815 | |
| 816 | |
| 817 | def get_ctx_input(agent=None) -> float: |
| 818 | """Get the utility model context input ratio.""" |
| 819 | cfg = get_utility_model_config(agent) |
| 820 | return float(cfg.get("ctx_input", 0.7)) |
| 821 | |
| 822 | |
| 823 | def _normalize_kwargs(kwargs: dict) -> dict: |
| 824 | """Convert string values that are valid numbers to numeric types.""" |
| 825 | result = {} |
| 826 | for key, value in kwargs.items(): |
| 827 | if isinstance(value, str): |
| 828 | try: |
| 829 | result[key] = int(value) |
| 830 | except ValueError: |
| 831 | try: |
| 832 | result[key] = float(value) |
| 833 | except ValueError: |
| 834 | result[key] = value |
| 835 | else: |
| 836 | result[key] = value |
| 837 | return result |
| 838 | |
| 839 | |
| 840 | def build_model_config(cfg: dict, model_type: models.ModelType) -> models.ModelConfig: |
| 841 | """Build a ModelConfig from a config dict section.""" |
| 842 | return models.ModelConfig( |
| 843 | type=model_type, |
| 844 | provider=cfg.get("provider", ""), |
| 845 | name=cfg.get("name", ""), |
| 846 | api_key=cfg.get("api_key", ""), |
| 847 | api_base=cfg.get("api_base", ""), |
| 848 | ctx_length=int(cfg.get("ctx_length", 0)), |
| 849 | vision=bool(cfg.get("vision", False)), |
| 850 | limit_requests=int(cfg.get("rl_requests", 0)), |
| 851 | limit_input=int(cfg.get("rl_input", 0)), |
| 852 | limit_output=int(cfg.get("rl_output", 0)), |
| 853 | kwargs=_normalize_kwargs(cfg.get("kwargs", {})), |
| 854 | ) |
| 855 | |
| 856 | |
| 857 | def build_chat_model(agent=None): |
| 858 | """Build and return a LiteLLMChatWrapper from config.""" |
| 859 | cfg = get_chat_model_config(agent) |
| 860 | mc = build_model_config(cfg, models.ModelType.CHAT) |
| 861 | return models.get_chat_model( |
| 862 | mc.provider, mc.name, model_config=mc, **mc.build_kwargs() |
| 863 | ) |
| 864 | |
| 865 | |
| 866 | def build_utility_model(agent=None): |
| 867 | """Build and return a LiteLLMChatWrapper for utility tasks.""" |
| 868 | cfg = get_utility_model_config(agent) |
| 869 | mc = build_model_config(cfg, models.ModelType.CHAT) |
| 870 | return models.get_chat_model( |
| 871 | mc.provider, mc.name, model_config=mc, **mc.build_kwargs() |
| 872 | ) |
| 873 | |
| 874 | |
| 875 | def build_vision_model(agent=None): |
| 876 | """Build the optional Vision Model selected by the effective preset.""" |
| 877 | cfg = get_vision_model_config(agent) |
| 878 | mc = build_model_config(cfg, models.ModelType.CHAT) |
| 879 | mc.vision = True |
| 880 | kwargs = mc.build_kwargs() |
| 881 | for key, default in ( |
| 882 | ("timeout", DEFAULT_VISION_TIMEOUT_SECONDS), |
| 883 | ("max_tokens", DEFAULT_VISION_MAX_TOKENS), |
| 884 | ): |
| 885 | value = cfg.get(key) |
| 886 | if value not in (None, ""): |
| 887 | kwargs[key] = _normalize_kwargs({key: value})[key] |
| 888 | else: |
| 889 | kwargs.setdefault(key, default) |
| 890 | return models.get_chat_model( |
| 891 | mc.provider, mc.name, model_config=mc, **kwargs |
| 892 | ) |
| 893 | |
| 894 | |
| 895 | def build_embedding_model(agent=None): |
| 896 | """Build and return an embedding model wrapper.""" |
| 897 | cfg = get_embedding_model_config(agent) |
| 898 | mc = build_model_config(cfg, models.ModelType.EMBEDDING) |
| 899 | return models.get_embedding_model( |
| 900 | mc.provider, mc.name, model_config=mc, **mc.build_kwargs() |
| 901 | ) |
| 902 | |
| 903 | |
| 904 | def get_embedding_model_config_object(agent=None) -> models.ModelConfig: |
| 905 | """Get a ModelConfig object for embeddings (needed by memory plugin).""" |
| 906 | cfg = get_embedding_model_config(agent) |
| 907 | return build_model_config(cfg, models.ModelType.EMBEDDING) |
| 908 | |
| 909 | |
| 910 | def get_chat_providers(): |
| 911 | """Get list of chat providers for UI dropdowns.""" |
| 912 | return get_providers("chat") |
| 913 | |
| 914 | |
| 915 | def get_embedding_providers(): |
| 916 | """Get list of embedding providers for UI dropdowns.""" |
| 917 | return get_providers("embedding") |
| 918 | |
| 919 | |
| 920 | def has_provider_api_key(provider: str, configured_api_key: str = "", model_type: str = "chat") -> bool: |
| 921 | if not provider_requires_api_key(provider, model_type): |
| 922 | return True |
| 923 | configured_value = (configured_api_key or "").strip() |
| 924 | if configured_value and configured_value != "None": |
| 925 | return True |
| 926 | |
| 927 | api_key = models.get_api_key(provider.lower()) |
| 928 | return bool(api_key and api_key.strip() and api_key != "None") |
| 929 | |
| 930 | |
| 931 | def get_missing_api_key_providers(agent=None) -> list[dict]: |
| 932 | """Check which configured providers are missing API keys.""" |
| 933 | cfg = get_effective_config(agent) |
| 934 | missing = [] |
| 935 | |
| 936 | checks = [ |
| 937 | ("Chat Model", cfg.get("chat_model", {})), |
| 938 | ("Utility Model", cfg.get("utility_model", {})), |
| 939 | ("Embedding Model", get_embedding_model_config(agent)), |
| 940 | ] |
| 941 | vision_cfg = get_vision_model_config(agent) |
| 942 | if vision_cfg: |
| 943 | checks.insert(1, ("Vision Model", vision_cfg)) |
| 944 | |
| 945 | for label, model_cfg in checks: |
| 946 | provider = model_cfg.get("provider", "") |
| 947 | if not provider: |
| 948 | continue |
| 949 | provider_lower = provider.lower() |
| 950 | if provider_lower in LOCAL_PROVIDERS: |
| 951 | continue |
| 952 | if label == "Embedding Model" and provider_lower in LOCAL_EMBEDDING: |
| 953 | continue |
| 954 | |
| 955 | if not has_provider_api_key(provider_lower, model_cfg.get("api_key", ""), _model_type_for_label(label)): |
| 956 | missing.append({"model_type": label, "provider": provider}) |
| 957 | |
| 958 | return missing |
| 959 | |
| 960 | |
| 961 | def is_chat_model_configured(config: dict | None = None) -> bool: |
| 962 | cfg = config if isinstance(config, dict) else get_config() |
| 963 | chat_cfg = cfg.get("chat_model", {}) if isinstance(cfg, dict) else {} |
| 964 | provider = str(chat_cfg.get("provider") or "").strip() |
| 965 | name = str(chat_cfg.get("name") or "").strip() |
| 966 | if not provider or not name: |
| 967 | return False |
| 968 | return has_provider_api_key(provider.lower(), chat_cfg.get("api_key", ""), "chat") |