import json from pathlib import Path from typing import Any def _to_bool(value: Any, default: bool) -> bool: if isinstance(value, bool): return value if isinstance(value, str): normalized = value.strip().lower() if normalized in {"true", "1", "yes", "y"}: return True if normalized in {"false", "0", "no", "n"}: return False return default def _to_int(value: Any, default: int) -> int: try: return int(value) except (TypeError, ValueError): return default def _to_float(value: Any, default: float) -> float: try: return float(value) except (TypeError, ValueError): return default def resolve_model_profile( catalog_path: str, requested_model: str | None, requested_tp: int ) -> tuple[str, dict[str, Any], dict[str, str]]: content = json.loads(Path(catalog_path).read_text(encoding="utf-8")) if not isinstance(content, dict): raise ValueError("config.json must be a JSON object") default_model = content.get("default_model") profiles = {k: v for k, v in content.items() if k != "default_model"} model_key = requested_model or default_model if not model_key or model_key not in profiles: raise ValueError(f"model profile '{model_key}' not found in config.json") profile = profiles[model_key] if not isinstance(profile, dict): raise ValueError(f"model profile '{model_key}' must be a JSON object") valid_tp_raw = profile.get("valid_tp", []) valid_tp = [_to_int(item, 0) for item in valid_tp_raw if _to_int(item, 0) > 0] resolved_tp = requested_tp if valid_tp and resolved_tp not in valid_tp: resolved_tp = valid_tp[0] updates = { "selected_model": model_key, "model_name": profile.get("hf_model_id", model_key), "served_model_name": profile.get("served_model_name", model_key), "max_model_len": _to_int(profile.get("ctx"), 8192), "max_num_seqs": _to_int(profile.get("max_num_seqs"), 64), "max_tokens": _to_int(profile.get("max_tokens"), 4096), "gpu_memory_utilization": _to_float(profile.get("gpu_util"), 0.92), "trust_remote_code": _to_bool(profile.get("trust_remote"), False), "enforce_eager": _to_bool(profile.get("enforce_eager"), False), "tensor_parallel_size": resolved_tp, "tool_call_parser": profile.get("tool_call_parser"), "enable_auto_tool_choice": _to_bool(profile.get("enable_auto_tool_choice"), False), } env_vars = {str(k): str(v) for k, v in dict(profile.get("env", {})).items()} return model_key, updates, env_vars