Spaces:
Sleeping
Sleeping
File size: 6,871 Bytes
06671aa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 | 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"])
|