from __future__ import annotations import json import re import tempfile from pathlib import Path from typing import Any from huggingface_hub import HfApi, hf_hub_download MODEL_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*/[A-Za-z0-9][A-Za-z0-9._-]*$") SMALL_CONFIG_MAX_BYTES = 2_000_000 WEIGHT_EXTENSIONS = (".safetensors", ".bin", ".pt", ".pth", ".ckpt", ".onnx", ".gguf") RISKY_SERIALIZATION_EXTENSIONS = (".bin", ".pt", ".pth", ".ckpt", ".pkl", ".pickle", ".joblib") PYTHON_EXTENSIONS = (".py",) DIFFUSERS_EXAMPLE_RE = re.compile(r"(from\s+diffusers\s+import|DiffusionPipeline\.from_pretrained|AutoPipelineFor(?:Text2Image|Image2Image|Inpainting)\.from_pretrained|StableDiffusion(?:XL)?Pipeline\.from_pretrained|FluxPipeline\.from_pretrained|ZImagePipeline)", re.IGNORECASE) GENERIC_INFERENCE_EXAMPLE_RE = re.compile(r"(from_pretrained\s*\(|gradio_client|pipeline\s*\(|pipe\s*\(|predict\s*\()", re.IGNORECASE) DIFFUSERS_PIPELINE_CLASSES = { "DiffusionPipeline", "StableDiffusionPipeline", "StableDiffusionXLPipeline", "StableDiffusionImg2ImgPipeline", "StableDiffusionInpaintPipeline", "AutoPipelineForText2Image", "AutoPipelineForImage2Image", "AutoPipelineForInpainting", "FluxPipeline", "ZImagePipeline", } TASK_OUTPUT_TYPE_MAP = { "text-to-image": "image", "image-to-image": "image", "image-to-video": "video", "text-to-video": "video", "text-to-audio": "audio", "text-to-speech": "audio", "automatic-speech-recognition": "text", "audio-classification": "text", "text-generation": "text", "text2text-generation": "text", "summarization": "text", "translation": "text", "question-answering": "text", "fill-mask": "text", "sentence-similarity": "text", "token-classification": "text", "zero-shot-classification": "text", "image-classification": "text", "object-detection": "image", "image-segmentation": "image", } IMAGE_PIPELINE_HINTS = ( "text2image", "texttoimage", "image2image", "inpaint", "stableDiffusion".lower(), "fluxpipeline", "zimagepipeline", ) VIDEO_OUTPUT_HINT_RE = re.compile( r"(image-to-video|text-to-video|audio-to-video|audio\s+to\s+video|audio-driven\s+avatar|video\s+avatar|avatar-v|single_audio_to_video|multi_audio_to_video|to_video|export_to_video|generated video)", re.IGNORECASE, ) IMAGE_OUTPUT_HINT_RE = re.compile( r"(\.images\s*\[|generated image|text-to-image|image = pipe\()", re.IGNORECASE, ) INPUT_MODALITY_ORDER = ("text", "image", "audio", "video", "file") INPUT_MODALITY_HINTS = { "text": ( "text prompt", "input prompt", "prompt:", "prompt=", "gr.textbox", "textbox", ), "image": ( "image_path", "input image", "image input", "source image", "reference image", "avatar image", "person image", "gr.image", "pil.image", "upload image", ), "audio": ( "audio_path", "input audio", "audio input", "source audio", "speech audio", "voice audio", "single audio", "multi audio", "gr.audio", ".wav", ), "video": ( "video_path", "input video", "video input", "source video", "reference video", "gr.video", ".mp4", ), "file": ( "file_path", "input file", "file input", "upload file", "gr.file", ), } def _ordered_modalities(values: set[str]) -> list[str]: return [name for name in INPUT_MODALITY_ORDER if name in values] def infer_expected_inputs( *, pipeline_tag: str | None = None, tags: list[str] | None = None, readme: str | None = None, ) -> list[str]: modalities: set[str] = set() task = (pipeline_tag or "").strip().lower() tag_set = {str(t).lower() for t in (tags or [])} task_like = {task, *tag_set} for item in task_like: if item.startswith("text-to-") or item in { "text-generation", "text2text-generation", "summarization", "translation", "question-answering", "fill-mask", "sentence-similarity", "token-classification", "zero-shot-classification", }: modalities.add("text") if item.startswith("image-to-") or item in {"image-classification", "object-detection", "image-segmentation"}: modalities.add("image") if item.startswith("audio-to-") or item in {"automatic-speech-recognition", "audio-classification"}: modalities.add("audio") if item.startswith("video-to-"): modalities.add("video") text = (readme or "").lower() for modality, hints in INPUT_MODALITY_HINTS.items(): if any(hint in text for hint in hints): modalities.add(modality) if re.search(r"\bprompt\b", text) and any(word in text for word in ("pipe(", "generate", "demo", "gradio")): modalities.add("text") return _ordered_modalities(modalities) def infer_output_guess( *, pipeline_tag: str | None = None, library_name: str | None = None, tags: list[str] | None = None, pipeline_class: str | None = None, readme: str | None = None, ) -> dict[str, Any]: expected = infer_expected_output_type( pipeline_tag=pipeline_tag, library_name=library_name, tags=tags, pipeline_class=pipeline_class, readme=readme, ) task = (pipeline_tag or "").strip().lower() tag_set = {str(t).lower() for t in (tags or [])} cls = (pipeline_class or "").strip().lower() text = (readme or "").lower() library = (library_name or "").strip().lower() combined = " ".join([library, cls, " ".join(sorted(tag_set)), text]) evidence: list[str] = [] confidence = "low" source = "none" if task in TASK_OUTPUT_TYPE_MAP: confidence = "high" source = "pipeline_tag" evidence.append(f"pipeline_tag={task} maps to {TASK_OUTPUT_TYPE_MAP[task]}.") elif {"text-to-video", "image-to-video"} & tag_set: confidence = "high" source = "tags" evidence.append("Hub tags include text-to-video or image-to-video.") elif {"text-to-image", "image-to-image", "diffusers"} & tag_set and expected == "image": confidence = "medium" source = "tags" evidence.append("Hub tags indicate a Diffusers image workflow.") elif {"text-to-audio", "text-to-speech", "audio"} & tag_set and expected == "audio": confidence = "medium" source = "tags" evidence.append("Hub tags indicate an audio generation workflow.") elif any(hint in cls for hint in IMAGE_PIPELINE_HINTS) and expected == "image": confidence = "medium" source = "pipeline_class" evidence.append(f"Pipeline class {pipeline_class} looks image-oriented.") elif "video" in cls and expected == "video": confidence = "high" source = "pipeline_class" evidence.append(f"Pipeline class {pipeline_class} looks video-oriented.") elif "audio" in cls or "speech" in cls: confidence = "medium" source = "pipeline_class" evidence.append(f"Pipeline class {pipeline_class} looks audio-oriented.") if VIDEO_OUTPUT_HINT_RE.search(combined) or re.search(r"\.frames\s*\[", text): if expected == "video": confidence = "high" source = "model_card" evidence.append("Model card or library name mentions video/avatar generation markers.") elif IMAGE_OUTPUT_HINT_RE.search(text) and expected == "image": if confidence == "low": confidence = "medium" source = "model_card" evidence.append("Model card includes image output usage such as pipe(...).images.") elif re.search(r"\.audios?\s*\[|generated audio|text-to-speech|soundfile|\.wav", text) and expected == "audio": if confidence == "low": confidence = "medium" source = "model_card" evidence.append("Model card includes audio output markers.") elif re.search(r"generated_text|tokenizer\.decode|text-generation|response\s*=", text) and expected == "text": if confidence == "low": confidence = "medium" source = "model_card" evidence.append("Model card includes text output markers.") if expected == "any": confidence = "low" if not evidence: evidence.append("Custom or ambiguous library without a strong generated-output signal; keeping output flexible.") elif not expected: evidence.append("No strong generated-output signal found in pre-scan metadata.") elif not evidence: evidence.append(f"Pre-scan inferred {expected} from weak metadata signals.") return { "schema_version": "pre_scan_output_guess.v1", "expected_output_type": expected or "", "confidence": confidence, "source": source, "authority": "pre_scan_guess", "final_authority": "runtime_contract_or_gradio_schema", "evidence": evidence[:6], } def infer_expected_output_type( *, pipeline_tag: str | None = None, library_name: str | None = None, tags: list[str] | None = None, pipeline_class: str | None = None, readme: str | None = None, ) -> str | None: task = (pipeline_tag or "").strip().lower() if task in TASK_OUTPUT_TYPE_MAP: return TASK_OUTPUT_TYPE_MAP[task] tag_set = {str(t).lower() for t in (tags or [])} if {"text-to-image", "image-to-image", "diffusers"} & tag_set: if "text-to-video" not in tag_set and "image-to-video" not in tag_set: return "image" if {"text-to-video", "image-to-video"} & tag_set: return "video" if {"text-to-audio", "text-to-speech", "audio"} & tag_set and "automatic-speech-recognition" not in tag_set: return "audio" cls = (pipeline_class or "").strip().lower() if any(hint in cls for hint in IMAGE_PIPELINE_HINTS): return "image" if "video" in cls: return "video" if "audio" in cls or "speech" in cls: return "audio" text = (readme or "").lower() library = (library_name or "").strip().lower() combined = " ".join([library, cls, " ".join(sorted(tag_set)), text]) if VIDEO_OUTPUT_HINT_RE.search(combined) or re.search(r"\.frames\s*\[", text): return "video" if IMAGE_OUTPUT_HINT_RE.search(text): return "image" if re.search(r"\.audios?\s*\[|generated audio|text-to-speech|soundfile|\.wav", text): return "audio" if re.search(r"generated_text|tokenizer\.decode|text-generation|response\s*=", text): return "text" if library and library not in {"diffusers", "transformers", "sentence-transformers", "timm", "onnx", "keras", "pytorch"}: return "any" return None def normalize_model_id(value: str | None) -> str: cleaned = (value or "").strip() cleaned = cleaned.replace("https://huggingface.co/", "") cleaned = cleaned.split("?", 1)[0].split("#", 1)[0].strip("/") if not MODEL_ID_RE.match(cleaned): raise ValueError("Model ID must look like owner/name or a Hugging Face model URL.") return cleaned def _get(obj: Any, name: str, default: Any = None) -> Any: if isinstance(obj, dict): return obj.get(name, default) return getattr(obj, name, default) def _sibling_name(sibling: Any) -> str: return str(_get(sibling, "rfilename", _get(sibling, "path", _get(sibling, "name", ""))) or "") def _sibling_size(sibling: Any) -> int | None: value = _get(sibling, "size", None) try: return int(value) if value is not None else None except Exception: return None def _read_small_json(repo_id: str, filename: str, *, token: str | None, max_bytes: int = SMALL_CONFIG_MAX_BYTES) -> dict[str, Any]: try: with tempfile.TemporaryDirectory(prefix="asf-model-scan-") as tmp: path = hf_hub_download(repo_id=repo_id, filename=filename, repo_type="model", token=token, local_dir=tmp) p = Path(path) if p.stat().st_size > max_bytes: return {} return json.loads(p.read_text(encoding="utf-8")) except Exception: return {} def _read_small_text(repo_id: str, filename: str, *, token: str | None, max_bytes: int = SMALL_CONFIG_MAX_BYTES) -> str: try: with tempfile.TemporaryDirectory(prefix="asf-model-scan-") as tmp: path = hf_hub_download(repo_id=repo_id, filename=filename, repo_type="model", token=token, local_dir=tmp) p = Path(path) if p.stat().st_size > max_bytes: return "" return p.read_text(encoding="utf-8", errors="replace") except Exception: return "" def _model_index_pipeline_class(model_index: dict[str, Any]) -> str: value = model_index.get("_class_name") or model_index.get("pipeline_class") or "" return str(value or "") def _extract_model_card_signals(readme: str, model_index: dict[str, Any]) -> dict[str, Any]: text = readme or "" diffusers_example = bool(DIFFUSERS_EXAMPLE_RE.search(text)) inference_example = diffusers_example or bool(GENERIC_INFERENCE_EXAMPLE_RE.search(text)) pipeline_class = _model_index_pipeline_class(model_index) if not pipeline_class and text: for name in sorted(DIFFUSERS_PIPELINE_CLASSES, key=len, reverse=True): if name in text: pipeline_class = name break runtime_hints = [] lower = text.lower() for label, patterns in [ ("bfloat16", ("bfloat16", "bf16")), ("float16", ("float16", "fp16")), ("device_map", ("device_map",)), ("cuda", ("cuda", ".to(\"cuda\")", ".to('cuda')")), ("offload", ("offload", "enable_model_cpu_offload")), ("num_inference_steps", ("num_inference_steps", "inference steps")), ]: if any(p in lower for p in patterns): runtime_hints.append(label) return { "has_inference_example": inference_example, "has_diffusers_example": diffusers_example, "pipeline_class": pipeline_class, "runtime_hints": runtime_hints[:8], } NATIVE_KERNEL_PATTERNS: tuple[tuple[str, tuple[str, ...], str], ...] = ( ("flash_attn", ("flash-attn", "flash_attn", "flash attention", "flashattention", "flash-attention", "enable_flashattn", "flash_attention_2"), "Flash Attention dependency or runtime flag"), ("xformers", ("xformers", "memory_efficient_attention"), "xFormers attention dependency"), ("triton", ("triton", "triton kernel", "@triton", "triton.jit"), "Triton/fused kernel dependency"), ("custom_cuda", ("cuda extension", "custom cuda", "cpp_extension", "setup.py build_ext", "fused kernel", "fused ops", "custom kernel"), "Custom native/CUDA extension"), ("attention_interface", ("attentioninterface", "attention interface", "attn_implementation"), "Transformers attention backend hook"), ("hf_kernels", ("hf kernels", "kernel hub", "hugging face kernels", "kernels-community", "from kernels import", "pip install kernels"), "HF Kernels / Kernel Hub mention"), ) def build_kernel_strategy(*, readme: str | None = None, files: list[str] | None = None) -> dict[str, Any]: """Return a visibility-only mitigation plan for native kernel dependencies. This is metadata/readme based. It must not inject packages or mark a model terminally blocked by itself; it gives Pi and the UI a safer strategy than blindly adding source-built CUDA packages to requirements.txt. """ text = (readme or "").lower() file_list = [str(f).lower() for f in (files or [])] detected: list[str] = [] signals: list[str] = [] for key, patterns, label in NATIVE_KERNEL_PATTERNS: if any(pattern in text for pattern in patterns): detected.append(key) signals.append(label) if any(f.endswith((".cu", ".cuh")) or "/csrc/" in f or f.startswith("csrc/") for f in file_list): if "custom_cuda" not in detected: detected.append("custom_cuda") signals.append("Native CUDA/C++ source files in repository") if any(f.endswith((".cpp", ".cc")) and ("cuda" in f or "/csrc/" in f or f.startswith("csrc/")) for f in file_list): if "custom_cuda" not in detected: detected.append("custom_cuda") signals.append("Native C++ extension source files in repository") native_risk = any(k in detected for k in {"flash_attn", "xformers", "triton", "custom_cuda"}) candidate_backends: list[str] = [] if native_risk: candidate_backends.extend(["torch_sdpa", "hf_kernels", "xformers_wheel"]) if "attention_interface" in detected: candidate_backends.append("transformers_attention_interface") if "hf_kernels" in detected and "hf_kernels" not in candidate_backends: candidate_backends.append("hf_kernels") rejected_actions = [] if native_risk: rejected_actions.append({ "action": "blind_pip_install_native_cuda_package", "reason": "Source-built CUDA/native packages are fragile in Spaces and can mismatch the managed PyTorch/CUDA runtime.", }) if "flash_attn" in detected: rejected_actions.append({ "action": "pip_install_flash_attn_without_fallback", "reason": "Prefer PyTorch SDPA, Transformers AttentionInterface, HF Kernels/Kernel Hub, or a compatible wheel before forcing flash-attn source builds.", }) selected = "none" if native_risk: selected = "prefer_runtime_backends_before_source_builds" elif "hf_kernels" in detected: selected = "hf_kernels_available_if_model_uses_supported_kernel" return { "schema_version": "kernel_strategy.v1", "native_kernel_risk": bool(native_risk), "detected_dependencies": detected, "signals": signals[:10], "selected_strategy": selected, "candidate_backends": candidate_backends[:8], "rejected_actions": rejected_actions, "requires_manual_review": bool("custom_cuda" in detected), "pi_instruction": ( "Do not compile native CUDA packages blindly. Prefer PyTorch SDPA, compatible wheels, HF Kernels/Kernel Hub, " "Transformers AttentionInterface, or Diffusers attention processors when they match the required operation. " "Declare a technical blocker if a strict native extension has no plausible fallback." if native_risk else "No native kernel dependency was detected by the metadata scan." ), "visibility_only": True, } def _has_any(text: str, patterns: tuple[str, ...]) -> bool: lower = text.lower() return any(pattern in lower for pattern in patterns) def _build_complexity_assessment( *, pipeline_tag: str | None, library_name: str | None, tags: list[str], files: list[str], readme: str, expected_output_type: str | None, custom_code: bool, gated: Any, ) -> dict[str, Any]: """Estimate build/runtime risk without changing the existing pre-scan verdict. This is deliberately a lightweight metadata/model-card heuristic. It helps the UI warn users before long Jobs; it is not a hard launch gate. """ lower_readme = (readme or "").lower() lower_files = [str(f).lower() for f in files] tag_set = {str(t).lower() for t in tags} task = (pipeline_tag or "").lower() library = (library_name or "").lower() points = 0 signals: list[str] = [] mitigations: list[str] = [] def add(points_delta: int, signal: str, mitigation: str | None = None) -> None: nonlocal points points += points_delta if signal not in signals: signals.append(signal) if mitigation and mitigation not in mitigations: mitigations.append(mitigation) if expected_output_type == "video" or "video" in task or {"text-to-video", "image-to-video"} & tag_set: add(3, "video or image-to-video output", "Expect a long build/repair cycle and prefer strong fallback hardware.") if any(word in lower_readme for word in ("avatar", "audio-driven", "audio driven", "talking head", "lip sync", "lip-sync")): add(2, "audio/video avatar workflow", "Refresh sign-in before launch and expect larger validation payloads.") if custom_code or "trust_remote_code" in tag_set or "custom_code" in tag_set: add(2, "custom code or trust_remote_code required", "Review generated code and blockers before paid hardware attempts.") if gated in {True, "auto", "manual"} or str(gated).lower() in {"true", "auto", "manual"}: add(1, "gated model access", "Confirm the same HF account has accepted access terms.") if _has_any(lower_readme, ("torchrun", "nproc_per_node", "distributed", "init_process_group", "nccl", "context_parallel", "tensor parallel", "pipeline parallel", "multi-gpu", "multi gpu")): add(4, "multi-GPU / distributed runtime hints", "Treat ZeroGPU as unlikely unless Pi can prove a single-GPU refactor.") if _has_any(lower_readme, ("flash-attn", "flash_attn", "flash attention", "flashattention", "flash-attention")): add(2, "flash-attn or custom attention dependency", "Prefer PyTorch SDPA, xformers wheels, or HF Kernels before source builds.") if _has_any(lower_readme, ("xformers", "triton", "cuda extension", "fused kernel", "fused ops", "custom kernel")) or any(f.endswith((".cu", ".cpp")) or "/csrc/" in f for f in lower_files): add(2, "native CUDA/kernel dependency risk", "Prefer runtime-compatible wheels or HF Kernels over compiling during Space build.") if "ffmpeg" in lower_readme or any("ffmpeg" in f for f in lower_files): add(1, "ffmpeg or system media dependency", "Ensure generated app declares media/system requirements clearly.") if _has_any(lower_readme, ("conda ", "mamba ", "apt-get", "sudo apt", "pip install -e", "git clone")): add(1, "non-standard install instructions", "Pi should vendor required code or simplify requirements for Spaces.") weight_files = [f for f in lower_files if f.endswith(WEIGHT_EXTENSIONS)] safetensor_shards = [f for f in lower_files if f.endswith(".safetensors")] if len(weight_files) >= 20 or len(safetensor_shards) >= 12: add(2, "many weight shards", "Expect longer cold start and validation windows.") elif len(weight_files) >= 8 or len(safetensor_shards) >= 6: add(1, "multiple weight shards", "Allow extra boot time before validation.") vram_match = re.search(r"\b(?:vram|gpu memory|memory)\D{0,16}([3-9]\d|1\d{2})\s*(?:gb|gib)\b|\b([3-9]\d|1\d{2})\s*(?:gb|gib)\s*(?:vram|gpu)", lower_readme) if vram_match: add(3, "high VRAM mentioned in model card", "Prefer fixed GPU fallback and refresh sign-in before launch.") # Keep the assessment independent from the existing pass/fail scan score. if points >= 9: level = "very_high" label = "Very high risk" recommended_seconds = 180 * 60 elif points >= 6: level = "high" label = "High risk" recommended_seconds = 120 * 60 elif points >= 3: level = "medium" label = "Medium risk" recommended_seconds = 60 * 60 else: level = "low" label = "Low risk" recommended_seconds = 30 * 60 return { "schema_version": "model_build_risk.v1", "level": level, "label": label, "score": max(0, points), "recommended_session_seconds": recommended_seconds, "recommended_session_minutes": recommended_seconds // 60, "signals": signals[:10], "mitigations": mitigations[:6], "summary": f"{label}; recommended HF session remaining: {recommended_seconds // 60}m+.", "visibility_only": True, } def _classify(score: int, *, blocking: bool = False) -> str: if blocking: return "unsupported" if score >= 75: return "safe" if score >= 45: return "caution" return "risky" def analyze_model_metadata( *, model_id: str, pipeline_tag: str | None = None, library_name: str | None = None, tags: list[str] | None = None, siblings: list[Any] | None = None, gated: Any = None, private: Any = None, card_data: dict[str, Any] | None = None, config: dict[str, Any] | None = None, model_index: dict[str, Any] | None = None, readme: str | None = None, ) -> dict[str, Any]: """Pure heuristic model-card scan used by the API and tests. It intentionally does not claim to be a security scanner. It summarizes common signals that predict whether an autonomous Space build is likely to be safe, supported, and worth spending compute on. """ tags = [str(t) for t in (tags or []) if t] files = [_sibling_name(s) for s in (siblings or []) if _sibling_name(s)] lower_files = [f.lower() for f in files] config = config or {} model_index = model_index or {} readme = readme or "" card_signals = _extract_model_card_signals(readme, model_index) has_inference_example = bool(card_signals["has_inference_example"]) has_diffusers_example = bool(card_signals["has_diffusers_example"]) pipeline_class = str(card_signals["pipeline_class"] or "") runtime_hints = list(card_signals["runtime_hints"] or []) output_guess = infer_output_guess( pipeline_tag=pipeline_tag, library_name=library_name, tags=tags, pipeline_class=pipeline_class, readme=readme, ) expected_output_type = output_guess["expected_output_type"] or None expected_inputs = infer_expected_inputs( pipeline_tag=pipeline_tag, tags=tags, readme=readme, ) score = 50 good: list[str] = [] risk: list[str] = [] recommendations: list[str] = [] has_safetensors = any(f.endswith(".safetensors") for f in lower_files) has_risky_serialization = any(f.endswith(RISKY_SERIALIZATION_EXTENSIONS) for f in lower_files) has_weight_file = any(f.endswith(WEIGHT_EXTENSIONS) for f in lower_files) has_python = any(f.endswith(PYTHON_EXTENSIONS) for f in lower_files) has_config = "config.json" in lower_files has_model_index = "model_index.json" in lower_files has_readme = bool(readme.strip()) or any(f in {"readme.md", "README.md".lower()} for f in lower_files) custom_code = bool(config.get("auto_map")) or "custom_code" in tags or "trust_remote_code" in tags or has_python if pipeline_tag: score += 12 good.append(f"Pipeline tag detected: {pipeline_tag}.") else: score -= 12 risk.append("No pipeline tag detected; task type may be ambiguous for an autonomous builder.") if expected_inputs: good.append("Inputs detected by pre-scan: " + ", ".join(expected_inputs) + ".") if expected_output_type: good.append(f"Pre-scan output guess: {expected_output_type} (confidence {output_guess['confidence']}).") if library_name: score += 10 good.append(f"Library detected: {library_name}.") elif model_index: score += 6 good.append("Diffusers-style model_index.json detected.") else: score -= 8 risk.append("No library metadata detected.") if has_safetensors: score += 18 good.append("Safetensors weights are available.") elif has_weight_file: score -= 10 risk.append("No safetensors file detected; weights may use less safe or less portable formats.") else: score -= 18 risk.append("No obvious model weight file detected in the repository listing.") if has_risky_serialization: score -= 14 risk.append("Repository includes pickle-like or PyTorch pickle serialization files (.bin/.pt/.pth/.ckpt/.pkl).") if custom_code: score -= 22 risk.append("Custom code or trust_remote_code is likely required.") recommendations.append("Review code manually before using Strict inference or paid hardware.") if gated in {True, "auto", "manual"} or str(gated).lower() in {"true", "auto", "manual"}: score -= 18 risk.append("Model appears gated; the Job may fail unless the signed-in account has accepted access terms.") recommendations.append("Confirm model access with the same Hugging Face account before launching.") if private: score -= 8 risk.append("Model is private; ensure the OAuth token has access.") if has_config or has_model_index: score += 8 good.append("Standard config metadata is present.") else: score -= 8 risk.append("No config.json or model_index.json detected.") is_diffusers = (library_name or "").lower() == "diffusers" or "diffusers" in {t.lower() for t in tags} or bool(model_index) if is_diffusers and has_model_index: score += 8 good.append("Diffusers repository structure detected.") if pipeline_class: score += 5 good.append(f"Pipeline class detected: {pipeline_class}.") if has_diffusers_example: score += 12 good.append("Model card includes a runnable Diffusers example.") elif has_inference_example: score += 7 good.append("Model card includes an inference example.") elif has_readme: score -= 5 risk.append("No clear runnable inference example found in the model card.") if runtime_hints: score += 4 good.append("Model card provides runtime hints: " + ", ".join(runtime_hints) + ".") if has_readme and len(readme.strip()) > 400: score += 6 good.append("Model card documentation is present.") elif not has_readme: score -= 8 risk.append("README/model card appears missing.") else: score -= 4 risk.append("Model card documentation appears very short.") task = (pipeline_tag or "").lower() library = (library_name or "").lower() unsupported = False if task in {"reinforcement-learning", "robotics"}: unsupported = True risk.append(f"Task '{pipeline_tag}' is not a good fit for automatic Gradio Space generation.") if "gguf" in tags or any(f.endswith(".gguf") for f in lower_files): score -= 10 risk.append("GGUF assets may need a custom llama.cpp runtime rather than the standard Transformers/Diffusers path.") if library in {"adapter-transformers"}: score -= 8 risk.append(f"Library '{library_name}' may require custom integration.") # A clear Diffusers model card with safetensors and a runnable example is a strong # positive signal. Do not downgrade it just because the model may be large; size # affects hardware planning, not whether the card is healthy for autonomous build. if is_diffusers and has_safetensors and has_diffusers_example and has_model_index and not custom_code and not unsupported: score = max(score, 82) score = max(0, min(100, score)) verdict = _classify(score, blocking=unsupported) build_risk = _build_complexity_assessment( pipeline_tag=pipeline_tag, library_name=library_name, tags=tags, files=files, readme=readme, expected_output_type=expected_output_type, custom_code=custom_code, gated=gated, ) kernel_strategy = build_kernel_strategy(readme=readme, files=files) if kernel_strategy.get("native_kernel_risk"): recommendations.append("Native kernel risk detected; prefer PyTorch SDPA, HF Kernels/Kernel Hub, compatible wheels, or model-specific fallbacks before source builds.") if build_risk["level"] in {"high", "very_high"}: recommendations.append( f"Refresh HF sign-in before launch; this scan recommends {build_risk['recommended_session_minutes']}m+ remaining for this model." ) if not recommendations: if verdict == "safe": recommendations.append("Good candidate for Strict inference.") elif verdict == "caution": recommendations.append("Use Best effort inference if the model card lacks a clear runnable example.") elif verdict == "risky": recommendations.append("Prefer Best effort or Demo scaffold until the risky signals are reviewed.") else: recommendations.append("Do not launch automatically; inspect manually first.") return { "ok": True, "model_id": model_id, "verdict": verdict, "score": score, "summary": { "safe": "Looks like a good candidate for an autonomous build.", "caution": "Usable, but review the caution signals before spending compute.", "risky": "Risky for an autonomous paid build; review before launching.", "unsupported": "Not a good fit for the automated builder.", }[verdict], "good_signals": good[:8], "risk_signals": risk[:10], "recommendations": recommendations[:6], "expected_output_type": expected_output_type or "", "expected_inputs": expected_inputs, "output_confidence": output_guess["confidence"], "output_authority": output_guess["authority"], "output_evidence": output_guess["evidence"], "pre_scan_output": output_guess, "build_risk": build_risk, "kernel_strategy": kernel_strategy, "metadata": { "pipeline_tag": pipeline_tag or "", "library_name": library_name or "", "tags": tags[:30], "gated": gated, "private": bool(private), "file_count": len(files), "has_safetensors": has_safetensors, "has_risky_serialization": has_risky_serialization, "has_custom_code_signal": custom_code, "has_inference_example": has_inference_example, "has_diffusers_example": has_diffusers_example, "pipeline_class": pipeline_class, "runtime_hints": runtime_hints, "expected_output_type": expected_output_type or "", "expected_inputs": expected_inputs, "output_confidence": output_guess["confidence"], "output_authority": output_guess["authority"], "output_evidence": output_guess["evidence"], "pre_scan_output": output_guess, "diffusers_standard": bool(is_diffusers and has_model_index and has_safetensors), "build_risk_level": build_risk["level"], "recommended_session_minutes": build_risk["recommended_session_minutes"], "native_kernel_risk": bool(kernel_strategy.get("native_kernel_risk")), "kernel_strategy": kernel_strategy, }, } def scan_model_card(model_id_or_url: str, *, token: str | None = None) -> dict[str, Any]: model_id = normalize_model_id(model_id_or_url) api = HfApi(token=token) try: info = api.model_info(model_id, files_metadata=True, token=token) except TypeError: info = api.model_info(model_id, token=token) siblings = list(_get(info, "siblings", []) or []) files = {_sibling_name(s).lower(): s for s in siblings if _sibling_name(s)} config = _read_small_json(model_id, "config.json", token=token) if "config.json" in files else {} model_index = _read_small_json(model_id, "model_index.json", token=token) if "model_index.json" in files else {} readme = _read_small_text(model_id, "README.md", token=token) if "readme.md" in files else "" card_data = _get(info, "card_data", _get(info, "cardData", {})) or {} if not isinstance(card_data, dict): try: card_data = dict(card_data) except Exception: card_data = {} return analyze_model_metadata( model_id=model_id, pipeline_tag=_get(info, "pipeline_tag", None), library_name=_get(info, "library_name", None), tags=list(_get(info, "tags", []) or []), siblings=siblings, gated=_get(info, "gated", None), private=_get(info, "private", False), card_data=card_data, config=config, model_index=model_index, readme=readme, )