From afbde6c50b785892f86d32572ebf57a01aed43e5 Mon Sep 17 00:00:00 2001 From: Dhiraj Kumar Sah Date: Tue, 14 Jul 2026 14:56:36 +0530 Subject: [PATCH 1/2] Added a QEff library utility, wired it into core model and diffusers flows, and validated that dumps are created once per model even when later stages fail. This captures environment/runtime/library metadata in a robust, non-intrusive way for CausalLM, VLM, and diffusers CPU paths. - Add QEfficient/utils/library_dump.py with dump_qeff_library_once(model_architecture, model_name) that writes a JSON manifest under QEFF_HOME/// qeff_library-.json - Capture schema version, creation timestamp, model name/architecture, QEff package info (version/import path/editable flag), runtime (python/platform/pid), QAIC SDK versions (apps/platform from qaic-version-util), and a full pip-style dependency snapshot - Ensure the manifest is written only once per model per process via a process-level _written registry - Add dump_qeff_library_once into QEFFBaseModel.__init__ before any PyTorch transforms so dumps survive transform/export failures and never break model construction - Add unit tests in tests/unit_test/utils/test_library_dump.py to cover manifest shape, paths, deduplication, QAIC SDK parsing, and basic resilience - Add tests/unit_test/utils/test_library_dump_api_calls.py to validate CausalLM, VLM (qwen3_vl custom config), Diffusers (Flux pipeline) with this utility. Signed-off-by: Dhiraj Kumar Sah --- QEfficient/base/modeling_qeff.py | 9 + QEfficient/utils/library_dump.py | 228 ++++++++++++++++++ tests/unit_test/utils/test_library_dump.py | 180 ++++++++++++++ .../utils/test_library_dump_api_calls.py | 105 ++++++++ 4 files changed, 522 insertions(+) create mode 100644 QEfficient/utils/library_dump.py create mode 100644 tests/unit_test/utils/test_library_dump.py create mode 100644 tests/unit_test/utils/test_library_dump_api_calls.py diff --git a/QEfficient/base/modeling_qeff.py b/QEfficient/base/modeling_qeff.py index 0425173840..b27e92a57a 100755 --- a/QEfficient/base/modeling_qeff.py +++ b/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: diff --git a/QEfficient/utils/library_dump.py b/QEfficient/utils/library_dump.py new file mode 100644 index 0000000000..6a21b71114 --- /dev/null +++ b/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 / / / qeff_library-.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 / / / qeff_library-.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 diff --git a/tests/unit_test/utils/test_library_dump.py b/tests/unit_test/utils/test_library_dump.py new file mode 100644 index 0000000000..39c625ec02 --- /dev/null +++ b/tests/unit_test/utils/test_library_dump.py @@ -0,0 +1,180 @@ +# ----------------------------------------------------------------------------- +# +# Copyright (c) Qualcomm Technologies, Inc. and/or its subsidiaries. +# SPDX-License-Identifier: BSD-3-Clause +# +# ----------------------------------------------------------------------------- +""" +Lean unit tests for QEfficient/utils/library_dump.py. + +Focus: manifest shape, path behavior, deduplication, basic resilience. +""" + +import json +import os +from unittest.mock import MagicMock, patch + +import pytest +from transformers import LlamaConfig, LlamaForCausalLM + + +def make_tiny_llama(): + cfg = LlamaConfig( + num_hidden_layers=1, + num_attention_heads=2, + num_key_value_heads=2, + hidden_size=64, + intermediate_size=128, + vocab_size=500, + max_position_embeddings=64, + ) + return LlamaForCausalLM(cfg).eval() + + +@pytest.fixture(autouse=True) +def _reset_written_registry(): + import QEfficient.utils.library_dump as ld + + ld._written.clear() + yield + ld._written.clear() + + +def _read_manifest(path): + return json.loads(path.read_text()) + + +class TestManifestBasics: + def test_manifest_has_required_top_level_keys(self, tmp_path): + from QEfficient.utils import library_dump as ld + + with patch.object(ld, "QEFF_HOME", tmp_path): + path = ld.dump_qeff_library_once("LlamaForCausalLM", "LlamaForCausalLM") + + manifest = _read_manifest(path) + for key in ("schema_version", "created_at", "model", "qeff", "runtime", "qaic_sdk", "dependencies"): + assert key in manifest + + def test_schema_version_is_one(self, tmp_path): + from QEfficient.utils import library_dump as ld + + with patch.object(ld, "QEFF_HOME", tmp_path): + path = ld.dump_qeff_library_once(None, "LlamaForCausalLM") + + assert _read_manifest(path)["schema_version"] == 1 + + def test_created_at_is_timezone_aware(self, tmp_path): + from QEfficient.utils import library_dump as ld + + with patch.object(ld, "QEFF_HOME", tmp_path): + path = ld.dump_qeff_library_once(None, "LlamaForCausalLM") + + ts = _read_manifest(path)["created_at"] + assert "+" in ts or ts.endswith("Z") + + def test_model_block_fields(self, tmp_path): + from QEfficient.utils import library_dump as ld + + with patch.object(ld, "QEFF_HOME", tmp_path): + path = ld.dump_qeff_library_once("LlamaForCausalLM", "LlamaForCausalLM") + + model_block = _read_manifest(path)["model"] + assert model_block["model_name"] == "LlamaForCausalLM" + assert model_block["model_architecture"] == "LlamaForCausalLM" + + def test_runtime_block_has_pid(self, tmp_path): + from QEfficient.utils import library_dump as ld + + with patch.object(ld, "QEFF_HOME", tmp_path): + path = ld.dump_qeff_library_once(None, "LlamaForCausalLM") + + runtime = _read_manifest(path)["runtime"] + assert runtime["pid"] == os.getpid() + + def test_dependencies_is_dict_with_core_packages(self, tmp_path): + from QEfficient.utils import library_dump as ld + + with patch.object(ld, "QEFF_HOME", tmp_path): + path = ld.dump_qeff_library_once(None, "LlamaForCausalLM") + + deps = _read_manifest(path)["dependencies"] + assert isinstance(deps, dict) + for pkg in ("torch", "transformers", "numpy"): + assert pkg in deps + + +class TestPathAndDedup: + def test_path_uses_arch_or_model(self, tmp_path): + from QEfficient.utils import library_dump as ld + + with patch.object(ld, "QEFF_HOME", tmp_path): + p1 = ld.dump_qeff_library_once("LlamaForCausalLM", "LlamaForCausalLM") + p2 = ld.dump_qeff_library_once(None, "GPT2LMHeadModel") + + assert p1.parent == tmp_path / "LlamaForCausalLM" / "LlamaForCausalLM" + assert p2.parent == tmp_path / "GPT2LMHeadModel" / "GPT2LMHeadModel" + assert p1.name.startswith("qeff_library-") and p1.name.endswith(".json") + + def test_same_model_returns_same_path(self, tmp_path): + from QEfficient.utils import library_dump as ld + + with patch.object(ld, "QEFF_HOME", tmp_path): + p1 = ld.dump_qeff_library_once("LlamaForCausalLM", "LlamaForCausalLM") + p2 = ld.dump_qeff_library_once("LlamaForCausalLM", "LlamaForCausalLM") + + assert p1 == p2 + + +class TestResilience: + def test_unwritable_location_returns_none(self, tmp_path): + from QEfficient.utils import library_dump as ld + + locked = tmp_path / "locked" + locked.mkdir() + locked.chmod(0o444) + try: + with patch.object(ld, "QEFF_HOME", locked): + result = ld.dump_qeff_library_once("LlamaForCausalLM", "LlamaForCausalLM") + assert result is None + finally: + locked.chmod(0o755) + + +class TestIntegration: + def test_manifest_created_during_model_construction(self, tmp_path): + import QEfficient.utils.library_dump as ld + from QEfficient.transformers.models.modeling_auto import QEFFAutoModelForCausalLM + + with patch.object(ld, "QEFF_HOME", tmp_path): + QEFFAutoModelForCausalLM(make_tiny_llama()) + + manifests = list(tmp_path.rglob("qeff_library-*.json")) + assert len(manifests) == 1 + + +class TestQaicSdkVersions: + def test_qaic_sdk_parses_aic_strings_or_none(self): + from QEfficient.utils import library_dump as ld + + fake_output = "platform:AIC.1.23.0.18\napps:AIC.1.23.0.18\n" + mock_proc = MagicMock() + mock_proc.stdout = fake_output + + with patch("subprocess.run", return_value=mock_proc): + result = ld._qaic_sdk_versions() + + for key in ("apps", "platform"): + val = result[key] + assert val is None or val.startswith("AIC") + + def test_qaic_sdk_not_found_becomes_none(self): + from QEfficient.utils import library_dump as ld + + fake_output = "platform:not found\napps:not found\n" + mock_proc = MagicMock() + mock_proc.stdout = fake_output + + with patch("subprocess.run", return_value=mock_proc): + result = ld._qaic_sdk_versions() + + assert result == {"apps": None, "platform": None} diff --git a/tests/unit_test/utils/test_library_dump_api_calls.py b/tests/unit_test/utils/test_library_dump_api_calls.py new file mode 100644 index 0000000000..e1412a5496 --- /dev/null +++ b/tests/unit_test/utils/test_library_dump_api_calls.py @@ -0,0 +1,105 @@ +# ----------------------------------------------------------------------------- +# +# 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. +""" + +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[2] / "configs" / "image_text_model_configs.json" + + +def _read_manifest_paths(root: Path): + return list(root.rglob("qeff_library-*.json")) + + +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 + + with patch("QEfficient.utils.library_dump.QEFF_HOME", tmp_path): + pipeline, _ = _build_flux_pipeline(enable_first_block_cache=False) + + # 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) From 94967be8b29f1f337cd2f3baf811ae800987dfeb Mon Sep 17 00:00:00 2001 From: Dhiraj Kumar Sah Date: Wed, 15 Jul 2026 14:58:05 +0530 Subject: [PATCH 2/2] Moved efficient-transformers/tests/transformers/test_library_dump_api_calls.py to transformers test instead of unit_tests directory as these tests require model weights to test for different stages of API calls. Signed-off-by: Dhiraj Kumar Sah --- .../test_library_dump_api_calls.py | 25 ++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) rename tests/{unit_test/utils => transformers}/test_library_dump_api_calls.py (81%) diff --git a/tests/unit_test/utils/test_library_dump_api_calls.py b/tests/transformers/test_library_dump_api_calls.py similarity index 81% rename from tests/unit_test/utils/test_library_dump_api_calls.py rename to tests/transformers/test_library_dump_api_calls.py index e1412a5496..e24578dff8 100644 --- a/tests/unit_test/utils/test_library_dump_api_calls.py +++ b/tests/transformers/test_library_dump_api_calls.py @@ -7,6 +7,10 @@ """ 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 @@ -20,13 +24,22 @@ from QEfficient import QEFFAutoModelForCausalLM, QEFFAutoModelForImageTextToText from QEfficient.utils.test_utils import load_vlm_model_from_config -CONFIG_PATH = Path(__file__).resolve().parents[2] / "configs" / "image_text_model_configs.json" +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). @@ -86,8 +99,14 @@ def test_flux_pipeline_exports_only_one_dump_per_module(self, tmp_path, monkeypa from tests.diffusers.test_flux import _build_flux_pipeline - with patch("QEfficient.utils.library_dump.QEFF_HOME", tmp_path): - pipeline, _ = _build_flux_pipeline(enable_first_block_cache=False) + # 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):