Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import hashlib | |
| import json | |
| import os | |
| from pathlib import Path | |
| from typing import Any | |
| try: | |
| import jsonschema | |
| except ImportError: # pragma: no cover - validated in requirements.txt | |
| jsonschema = None # type: ignore[assignment] | |
| class InferenceManifestValidationError(ValueError): | |
| def __init__(self, issues: list[str]) -> None: | |
| super().__init__("; ".join(issues)) | |
| self.issues = issues | |
| _PROVIDER_IDS = {"proxy-openai", "google-genai", "vertex-ai"} | |
| _LEGACY_MODEL_ALIASES = {"pro"} | |
| _cached_manifest: dict[str, Any] | None = None | |
| _cached_manifest_path: str | None = None | |
| _cached_manifest_sha256: str | None = None | |
| def resolve_manifest_path(custom_path: str | None = None) -> Path: | |
| if custom_path: | |
| return Path(custom_path) | |
| env_path = os.environ.get("INFERENCE_MANIFEST_PATH", "").strip() | |
| if env_path: | |
| return Path(env_path) | |
| return Path(__file__).resolve().parents[2] / "config" / "inference.manifest.json" | |
| def resolve_schema_path(manifest_path: Path) -> Path: | |
| return manifest_path.with_name("inference.manifest.schema.json") | |
| def validate_inference_manifest(raw: Any) -> dict[str, Any]: | |
| if not isinstance(raw, dict): | |
| raise InferenceManifestValidationError(["manifest root must be an object"]) | |
| issues: list[str] = [] | |
| defaults = raw.get("defaults") | |
| if not isinstance(defaults, dict): | |
| issues.append("defaults must be an object") | |
| elif defaults.get("provider") not in _PROVIDER_IDS: | |
| issues.append("defaults.provider must be a valid provider id") | |
| models = raw.get("models") | |
| if not isinstance(models, dict) or not models: | |
| issues.append("models must be a non-empty object") | |
| else: | |
| for model_id, entry in models.items(): | |
| if not isinstance(entry, dict): | |
| issues.append(f"models.{model_id} must be an object") | |
| continue | |
| providers = entry.get("providers") | |
| if not isinstance(providers, list) or not providers: | |
| issues.append(f"models.{model_id}.providers must be a non-empty array") | |
| else: | |
| for provider in providers: | |
| if provider not in _PROVIDER_IDS: | |
| issues.append( | |
| f"models.{model_id}.providers contains invalid provider {provider}" | |
| ) | |
| providers = raw.get("providers") | |
| if not isinstance(providers, dict): | |
| issues.append("providers must be an object") | |
| else: | |
| for provider_id in _PROVIDER_IDS: | |
| if provider_id not in providers: | |
| issues.append(f"providers.{provider_id} is required") | |
| if issues: | |
| raise InferenceManifestValidationError(issues) | |
| return raw | |
| def _validate_with_jsonschema(raw: dict[str, Any], schema_path: Path) -> None: | |
| if jsonschema is None: | |
| raise RuntimeError("jsonschema is required for inference manifest validation") | |
| if not schema_path.exists(): | |
| raise FileNotFoundError(f"Inference manifest schema not found: {schema_path}") | |
| schema = json.loads(schema_path.read_text(encoding="utf-8")) | |
| jsonschema.validate(instance=raw, schema=schema) | |
| def load_inference_manifest(manifest_path: Path | None = None) -> dict[str, Any]: | |
| path = manifest_path or resolve_manifest_path() | |
| raw = json.loads(path.read_text(encoding="utf-8")) | |
| validated = validate_inference_manifest(raw) | |
| _validate_with_jsonschema(validated, resolve_schema_path(path)) | |
| return validated | |
| def get_inference_manifest() -> dict[str, Any]: | |
| global _cached_manifest, _cached_manifest_path, _cached_manifest_sha256 | |
| path = resolve_manifest_path() | |
| path_str = str(path) | |
| if _cached_manifest is not None and _cached_manifest_path == path_str: | |
| return _cached_manifest | |
| _cached_manifest = load_inference_manifest(path) | |
| _cached_manifest_path = path_str | |
| _cached_manifest_sha256 = hashlib.sha256(path.read_bytes()).hexdigest() | |
| return _cached_manifest | |
| def reset_inference_manifest_cache() -> None: | |
| global _cached_manifest, _cached_manifest_path, _cached_manifest_sha256 | |
| _cached_manifest = None | |
| _cached_manifest_path = None | |
| _cached_manifest_sha256 = None | |
| def manifest_sha256(manifest_path: Path | None = None) -> str: | |
| global _cached_manifest_sha256 | |
| path = manifest_path or resolve_manifest_path() | |
| path_str = str(path) | |
| if _cached_manifest_sha256 is not None and _cached_manifest_path == path_str: | |
| return _cached_manifest_sha256 | |
| return hashlib.sha256(path.read_bytes()).hexdigest() | |
| def _feature_model(manifest: dict[str, Any], feature_id: str) -> str | None: | |
| features = manifest.get("features") or {} | |
| feature = features.get(feature_id) or {} | |
| model = feature.get("model") | |
| return str(model).strip() if model else None | |
| def resolve_model( | |
| *, | |
| request_model_id: str | None = None, | |
| feature_id: str = "python-review", | |
| ) -> str: | |
| manifest = get_inference_manifest() | |
| models = manifest.get("models") or {} | |
| candidate = (request_model_id or "").strip() | |
| if candidate and candidate not in _LEGACY_MODEL_ALIASES and candidate in models: | |
| return candidate | |
| env_model = os.environ.get("TABULAR_AGENT_MODEL", "").strip() | |
| if env_model and env_model in models: | |
| return env_model | |
| feature_model = _feature_model(manifest, feature_id) | |
| if feature_model and feature_model in models: | |
| return feature_model | |
| if feature_id == "python-chat": | |
| review_model = _feature_model(manifest, "python-review") | |
| if review_model and review_model in models: | |
| return review_model | |
| defaults = manifest.get("defaults") or {} | |
| for key in ("model", "mainModel", "titleModel", "tabularModel"): | |
| default_model = defaults.get(key) | |
| if isinstance(default_model, str) and default_model in models: | |
| return default_model | |
| raise InferenceManifestValidationError(["no resolvable model in inference manifest"]) | |
| def resolve_timeout_seconds(feature_id: str = "python-review") -> float: | |
| env_value = os.environ.get("TABULAR_AGENT_TIMEOUT_SECONDS", "").strip() | |
| if env_value: | |
| return float(env_value) | |
| manifest = get_inference_manifest() | |
| features = manifest.get("features") or {} | |
| feature = features.get(feature_id) or {} | |
| timeout = feature.get("timeoutSeconds") | |
| if isinstance(timeout, (int, float)) and timeout > 0: | |
| return float(timeout) | |
| return 180.0 | |
| def default_main_model() -> str: | |
| manifest = get_inference_manifest() | |
| defaults = manifest.get("defaults") or {} | |
| main_model = defaults.get("model") or defaults.get("mainModel") | |
| if isinstance(main_model, str) and main_model.strip(): | |
| return main_model.strip() | |
| raise InferenceManifestValidationError(["defaults.model is missing"]) | |