QEff library utility to capture environment/runtime/library metadata right after loading a model before any transforms are applied.

QEfficient/base/modeling_qeff.py

@@ -53,6 +53,7 @@
     to_named_specializations,
 )
 from QEfficient.utils.export_utils import export_wrapper
+from QEfficient.utils.library_dump import dump_qeff_library_once
 from QEfficient.utils.torch_patches import layerwise_safe_onnx_export_patches
 
 logger = logging.getLogger(__name__)
@@ -136,6 +137,14 @@ def __init__(self, model: torch.nn.Module, **kwargs) -> None:
         self.is_transformed: bool = False
 
         self._normalize_torch_dtype()
+
+        # Capture library/environment manifest before any transforms run so
+        # the file survives transform or export failures.
+        try:
+            dump_qeff_library_once(self.model_architecture, self.model_name)
+        except Exception as _dump_exc:  # non-fatal
+            logger.warning("dump_qeff_library_once failed (non-fatal): %s", _dump_exc)
+
         # Apply the transformations
         any_transformed = False
         for transform in self._pytorch_transforms:

QEfficient/utils/library_dump.py

@@ -0,0 +1,228 @@
+# -----------------------------------------------------------------------------
+#
+# Copyright (c) Qualcomm Technologies, Inc. and/or its subsidiaries.
+# SPDX-License-Identifier: BSD-3-Clause
+#
+# -----------------------------------------------------------------------------
+"""
+Utility for writing a small JSON manifest that captures the installed
+QEfficient library, runtime environment, model wrapper, and key dependency
+versions at model-construction time.
+It'll dump the config for every new model that's run in the current session so multiple component based models would have multiple dump files for each component.
+
+Public API
+----------
+dump_qeff_library_once(model_architecture, model_name) -> Optional[Path]
+    Write (at most once per model per Python process) a manifest under:
+        QEFF_HOME / <arch_or_name> / <model_name> / qeff_library-<run_id>.json
+    Returns the Path on success, None on any failure.
+"""
+
+import json
+import logging
+import os
+import platform
+import subprocess
+from datetime import datetime
+from importlib import metadata as importlib_metadata
+from pathlib import Path
+from typing import Optional
+
+import QEfficient
+from QEfficient.utils.cache import QEFF_HOME
+
+logger = logging.getLogger(__name__)
+
+# Schema version — bump when the manifest layout changes in a breaking way.
+_SCHEMA_VERSION = 1
+
+
+# Process-level registry: model_name -> Path of the already-written manifest.
+# Ensures at most one manifest per model per Python process.
+_written: dict[str, Path] = {}
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+
+def _package_version(package_name: str) -> Optional[str]:
+    """Return the installed version of *package_name*, or None if not found."""
+    try:
+        return importlib_metadata.version(package_name)
+    except importlib_metadata.PackageNotFoundError:
+        return None
+
+
+def _all_installed_packages() -> dict:
+    """
+    Return a mapping of all installed package names to versions.
+
+    This is a lightweight, in-process equivalent of a `pip list` snapshot.
+    Names are normalized to the distribution metadata name.
+    """
+    packages = {}
+    for dist in importlib_metadata.distributions():
+        name = dist.metadata.get("Name") or dist.metadata.get("Summary") or dist.name
+        if not name:
+            continue
+        version = dist.version
+        packages[name] = version
+    # Stable ordering for deterministic JSON output
+    return dict(sorted(packages.items(), key=lambda kv: kv[0].lower()))
+
+
+def _qeff_version() -> Optional[str]:
+    return _package_version("QEfficient")
+
+
+def _qeff_import_path() -> str:
+    return str(Path(QEfficient.__file__).parent)
+
+
+def _qeff_editable() -> bool:
+    """
+    Best-effort check: True when QEfficient appears to be installed in
+    editable / local-source mode.
+
+    Strategy (in order):
+    1. Look for a direct_url.json dist-info entry that carries
+       ``{"dir_info": {"editable": true}}``.
+    2. Fall back to checking whether the import path sits outside the
+       standard site-packages tree (i.e. the source checkout is on sys.path
+       directly).
+    """
+    try:
+        dist = importlib_metadata.distribution("QEfficient")
+        # PEP 610 — direct_url.json is present for editable installs made with
+        # modern pip (>= 21.3).
+        direct_url_text = dist.read_text("direct_url.json")
+        if direct_url_text:
+            import json as _json
+
+            info = _json.loads(direct_url_text)
+            dir_info = info.get("dir_info", {})
+            if dir_info.get("editable", False):
+                return True
+    except Exception:
+        pass
+
+    # Heuristic fallback: editable installs typically live outside site-packages.
+    import_path = _qeff_import_path()
+    return "site-packages" not in import_path
+
+
+def _run_id() -> str:
+    """Build a run identifier from the current timestamp and process ID."""
+    ts = datetime.now().strftime("%Y%m%d-%H%M%S")
+    return f"{ts}-pid{os.getpid()}"
+
+
+_QAIC_VERSION_UTIL = "/opt/qti-aic/tools/qaic-version-util"
+
+
+def _qaic_sdk_versions() -> dict:
+    """
+    Run qaic-version-util and parse the apps/platform SDK version strings.
+
+    Returns a dict with keys ``apps`` and ``platform``, each set to the
+    version string (e.g. ``"AIC.1.23.0.18"``) or ``None`` when the tool is
+    unavailable, the entry is missing, or the value is ``"not found"``.
+    """
+    result = {"apps": None, "platform": None}
+    try:
+        proc = subprocess.run(
+            [_QAIC_VERSION_UTIL],
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+        for line in proc.stdout.splitlines():
+            line = line.strip()
+            if line.startswith("apps:"):
+                value = line.split(":", 1)[1].strip()
+                result["apps"] = value if value != "not found" else None
+            elif line.startswith("platform:"):
+                value = line.split(":", 1)[1].strip()
+                result["platform"] = value if value != "not found" else None
+    except Exception:
+        pass
+    return result
+
+
+def _build_manifest(model_architecture: Optional[str], model_name: str) -> dict:
+    """Assemble the full manifest dictionary."""
+    return {
+        "schema_version": _SCHEMA_VERSION,
+        "created_at": datetime.now().astimezone().isoformat(timespec="seconds"),
+        "model": {
+            "model_name": model_name,
+            "model_architecture": model_architecture,
+        },
+        "qeff": {
+            "name": "QEfficient",
+            "version": _qeff_version(),
+            "import_path": _qeff_import_path(),
+            "editable": _qeff_editable(),
+        },
+        "runtime": {
+            "python": platform.python_version(),
+            "platform": platform.platform(),
+            "pid": os.getpid(),
+        },
+        "qaic_sdk": _qaic_sdk_versions(),
+        "dependencies": _all_installed_packages(),
+    }
+
+
+def _manifest_path(model_architecture: Optional[str], model_name: str, run_id: str) -> Path:
+    """
+    Return the target path for the manifest file.
+
+    Layout: QEFF_HOME / <arch_or_name> / <model_name> / qeff_library-<run_id>.json
+    """
+    arch_dir = model_architecture if model_architecture else model_name
+    return QEFF_HOME / arch_dir / model_name / f"qeff_library-{run_id}.json"
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+
+def dump_qeff_library_once(model_architecture: Optional[str], model_name: str) -> Optional[Path]:
+    """
+    Write a QEff library manifest for *model_name* at most once per process.
+
+    If the same *model_name* has already been dumped in this Python run, the
+    previously written path is returned immediately without touching the
+    filesystem.  A different *model_name* always gets its own manifest under
+    its own cache sub-directory.
+
+    Args:
+        model_architecture: Architecture string from the model config
+            (e.g. ``"LlamaForCausalLM"``).  Pass ``None`` when unavailable.
+        model_name: The QEff wrapper model name (e.g. ``"LlamaForCausalLM"``).
+
+    Returns:
+        Path to the written (or previously written) manifest, or ``None`` if
+        the dump failed for any reason.
+    """
+    if model_name in _written:
+        return _written[model_name]
+
+    try:
+        run_id = _run_id()
+        path = _manifest_path(model_architecture, model_name, run_id)
+        path.parent.mkdir(parents=True, exist_ok=True)
+
+        manifest = _build_manifest(model_architecture, model_name)
+        path.write_text(json.dumps(manifest, indent=2), encoding="utf-8")
+
+        _written[model_name] = path
+        logger.debug("QEff library manifest written: %s", path)
+        return path
+    except Exception as exc:
+        logger.warning("dump_qeff_library_once failed (non-fatal): %s", exc)
+        return None

tests/transformers/test_library_dump_api_calls.py

@@ -0,0 +1,124 @@
+# -----------------------------------------------------------------------------
+#
+# Copyright (c) Qualcomm Technologies, Inc. and/or its subsidiaries.
+# SPDX-License-Identifier: BSD-3-Clause
+#
+# -----------------------------------------------------------------------------
+"""
+CPU-only tests that ensure library dumps are written for key API entry points
+(even when failures happen) and deduplication holds for multi-module pipelines.
+
+These tests live under ``tests/transformers`` because they rely on real model
+weights/configs (CausalLM, VLM, and Flux diffusers) rather than pure unit-level
+stubs.
+"""
+
+import json
+import os
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+from transformers import GPT2Config, GPT2LMHeadModel
+
+from QEfficient import QEFFAutoModelForCausalLM, QEFFAutoModelForImageTextToText
+from QEfficient.utils.test_utils import load_vlm_model_from_config
+
+CONFIG_PATH = Path(__file__).resolve().parents[1] / "configs" / "image_text_model_configs.json"
+
+
+def _read_manifest_paths(root: Path):
+    return list(root.rglob("qeff_library-*.json"))
+
+
+@pytest.fixture(autouse=True)
+def _reset_written_registry():
+    import QEfficient.utils.library_dump as ld
+
+    ld._written.clear()
+    yield
+    ld._written.clear()
+
+
+def _load_qwen3_0_6b_custom_vlm_config():
+    data = json.loads(CONFIG_PATH.read_text())
+    # Use qwen3_vl custom config from tests (no downloads).
+    model_entry = next(cfg for cfg in data["image_text_models"] if cfg["model_type"] == "qwen3_vl")
+    custom = model_entry.get("additional_params", {})
+    from transformers import AutoConfig
+
+    return AutoConfig.for_model("qwen3_vl", trust_remote_code=True, **custom)
+
+
+class TestCausalLmDumpFailure:
+    def test_causal_lm_dump_survives_transform_failure(self, tmp_path):
+        model = GPT2LMHeadModel(GPT2Config(n_layer=1, n_head=2, n_embd=64, vocab_size=256)).eval()
+
+        first_transform = QEFFAutoModelForCausalLM._pytorch_transforms[0]
+
+        def exploding_apply(cls, model, *args, **kwargs):
+            raise RuntimeError("forced transform failure")
+
+        with (
+            patch("QEfficient.utils.library_dump.QEFF_HOME", tmp_path),
+            patch.object(first_transform, "apply", classmethod(exploding_apply)),
+        ):
+            with pytest.raises(RuntimeError, match="forced transform failure"):
+                QEFFAutoModelForCausalLM(model)
+
+        assert _read_manifest_paths(tmp_path), "expected manifest to exist despite failure"
+
+
+class TestVlmDumpFailure:
+    def test_vlm_dump_survives_transform_failure(self, tmp_path):
+        config = _load_qwen3_0_6b_custom_vlm_config()
+        model_hf = load_vlm_model_from_config(config)
+
+        from QEfficient.transformers.models.modeling_auto import _QEFFAutoModelForImageTextToTextSingleQPC
+
+        first_transform = _QEFFAutoModelForImageTextToTextSingleQPC._pytorch_transforms[0]
+
+        def exploding_apply(cls, model, *args, **kwargs):
+            raise RuntimeError("forced transform failure")
+
+        with (
+            patch("QEfficient.utils.library_dump.QEFF_HOME", tmp_path),
+            patch.object(first_transform, "apply", classmethod(exploding_apply)),
+        ):
+            with pytest.raises(RuntimeError, match="forced transform failure"):
+                QEFFAutoModelForImageTextToText(model_hf, kv_offload=False)
+
+        assert _read_manifest_paths(tmp_path), "expected manifest to exist despite failure"
+
+
+class TestDiffusersDumpDedup:
+    def test_flux_pipeline_exports_only_one_dump_per_module(self, tmp_path, monkeypatch):
+        # Ensure HF downloads use the configured cache.
+        monkeypatch.setenv("HF_HUB_ENABLE_HF_TRANSFER", "1")
+        monkeypatch.setenv("HF_HUB_CACHE", os.environ.get("HF_HUB_CACHE", "/home/huggingface_hub"))
+
+        from tests.diffusers.test_flux import _build_flux_pipeline
+
+        # Build the Flux pipeline. If the base Flux configs or weights are
+        # unavailable in this environment (no internet / model not installed),
+        # gracefully skip instead of failing the entire suite.
+        try:
+            with patch("QEfficient.utils.library_dump.QEFF_HOME", tmp_path):
+                pipeline, _ = _build_flux_pipeline(enable_first_block_cache=False)
+        except OSError:
+            pytest.skip("Flux base model/configs not available; skipping dump dedup test.")
+
+        # Stub export calls to keep the test fast while exercising pipeline iteration.
+        def _noop_export(*args, **kwargs):
+            return str(tmp_path / "dummy.onnx")
+
+        for module in pipeline.modules.values():
+            monkeypatch.setattr(module, "export", _noop_export, raising=True)
+
+        with patch("QEfficient.utils.library_dump.QEFF_HOME", tmp_path):
+            pipeline.export(export_dir=str(tmp_path))
+            pipeline.export(export_dir=str(tmp_path))
+
+        manifests = _read_manifest_paths(tmp_path)
+        # One manifest per module, no duplicates across repeated exports.
+        assert len(manifests) == len(pipeline.modules)