From dd1dfc4a1a5dfc98306bc78dc88a2aad133f7671 Mon Sep 17 00:00:00 2001 From: hualxie Date: Mon, 22 Jun 2026 15:55:51 +0800 Subject: [PATCH] clean up unused codes --- src/winml/modelkit/utils/__init__.py | 16 - src/winml/modelkit/utils/hub_utils.py | 348 --------------------- src/winml/modelkit/utils/optimum_loader.py | 160 ---------- 3 files changed, 524 deletions(-) delete mode 100644 src/winml/modelkit/utils/hub_utils.py delete mode 100644 src/winml/modelkit/utils/optimum_loader.py diff --git a/src/winml/modelkit/utils/__init__.py b/src/winml/modelkit/utils/__init__.py index 093c8cada..373270989 100644 --- a/src/winml/modelkit/utils/__init__.py +++ b/src/winml/modelkit/utils/__init__.py @@ -6,25 +6,9 @@ from .config_utils import merge_config from .constants import normalize_ep_name -from .hub_utils import ( - inject_hub_metadata, - is_hub_model, - load_hf_components_from_onnx, - save_local_model_configs, -) -from .optimum_loader import ( - OptimumONNXModel, - load_optimum_model, -) __all__ = [ - "OptimumONNXModel", - "inject_hub_metadata", - "is_hub_model", - "load_hf_components_from_onnx", - "load_optimum_model", "merge_config", "normalize_ep_name", - "save_local_model_configs", ] diff --git a/src/winml/modelkit/utils/hub_utils.py b/src/winml/modelkit/utils/hub_utils.py deleted file mode 100644 index 3da385a3b..000000000 --- a/src/winml/modelkit/utils/hub_utils.py +++ /dev/null @@ -1,348 +0,0 @@ -# ------------------------------------------------------------------------- -# Copyright (c) Microsoft Corporation. All rights reserved. -# Licensed under the MIT License. -# -------------------------------------------------------------------------- -"""HuggingFace Hub utilities for model detection and configuration loading. - -This module provides intelligent detection of HuggingFace Hub models vs local models, -and handles the appropriate metadata storage and configuration loading strategies. -""" - -import logging -import re -from pathlib import Path -from typing import Any - - -logger = logging.getLogger(__name__) - - -def is_hub_model(model_name_or_path: str) -> tuple[bool, dict]: - """Comprehensive Hub model detection with metadata extraction. - - Args: - model_name_or_path: Model identifier or path - - Returns: - Tuple of (is_hub_model, metadata_dict) - """ - # Quick rejection for obvious local paths - if Path(model_name_or_path).exists(): - return False, {"type": "local", "path": model_name_or_path} - - # Check for local path indicators - if any(model_name_or_path.startswith(prefix) for prefix in ["./", "../", "/", "~/"]): - return False, {"type": "local", "path": model_name_or_path} - - # Check for Windows absolute paths - if re.match(r"^[A-Za-z]:[\\/]", model_name_or_path): - return False, {"type": "local", "path": model_name_or_path} - - # Parse potential Hub model format - # Supports: model-name, org/model, org/model@revision - hub_pattern = r"^(?:([^/@]+)/)?([^/@]+)(?:@(.+))?$" - match = re.match(hub_pattern, model_name_or_path) - - if not match: - return False, {"type": "invalid"} - - org, model, revision = match.groups() - full_model_id = f"{org}/{model}" if org else model - - # Try to verify with Hub API - try: - from huggingface_hub import HfApi - - api = HfApi() - model_info = api.model_info(full_model_id, revision=revision) - - # Extract comprehensive metadata - metadata = { - "type": "hub", - "model_id": model_info.id, - "sha": model_info.sha, - "revision": revision or "main", - "tags": model_info.tags if hasattr(model_info, "tags") else [], - "pipeline_tag": model_info.pipeline_tag - if hasattr(model_info, "pipeline_tag") - else None, - "library_name": model_info.library_name - if hasattr(model_info, "library_name") - else None, - "author": model_info.author if hasattr(model_info, "author") else None, - "last_modified": str(model_info.lastModified) - if hasattr(model_info, "lastModified") - else None, - "private": model_info.private if hasattr(model_info, "private") else False, - "gated": model_info.gated if hasattr(model_info, "gated") else False, - } - - # Try to get model card info if available - try: - from huggingface_hub import ModelCard - - card = ModelCard.load(full_model_id) - if hasattr(card.data, "base_model"): - metadata["base_model"] = card.data.base_model - if hasattr(card.data, "license"): - metadata["license"] = card.data.license - if hasattr(card.data, "language"): - metadata["language"] = card.data.language - if hasattr(card.data, "task_categories"): - metadata["task_categories"] = card.data.task_categories - except Exception: - pass - - return True, metadata - - except Exception as e: - # Could not verify with Hub - might be private or offline - # Use heuristics to guess - if len(model_name_or_path.split("/")) <= 2 and "\\" not in model_name_or_path: - return True, { - "type": "hub_unverified", - "model_id": full_model_id, - "revision": revision or "main", - "error": str(e), - } - return False, {"type": "local", "path": model_name_or_path} - - -def inject_hub_metadata(onnx_model: Any, model_name_or_path: str, metadata: dict) -> None: - """Inject HuggingFace Hub metadata into ONNX model. - - Args: - onnx_model: ONNX model proto - model_name_or_path: Original model identifier - metadata: Hub metadata dictionary - """ - from datetime import datetime, timezone - - # Clear any existing HF metadata - # We need to remove items by filtering, not reassigning - hf_props_to_remove = [] - for i, prop in enumerate(onnx_model.metadata_props): - if prop.key.startswith("hf_"): - hf_props_to_remove.append(i) - - # Remove in reverse order to maintain indices - for i in reversed(hf_props_to_remove): - del onnx_model.metadata_props[i] - - # Add required metadata - def add_prop(key: str, value: Any) -> None: - if value is not None: - prop = onnx_model.metadata_props.add() - prop.key = key - prop.value = str(value) - - # Required fields - add_prop("hf_hub_id", metadata.get("model_id")) - add_prop("hf_hub_revision", metadata.get("sha", "")[:8]) - add_prop("hf_model_type", "hub") - - # Get ModelExport version - try: - from .. import __version__ - - export_version = __version__ - except ImportError: - export_version = "unknown" - - add_prop("hf_export_version", export_version) - add_prop("hf_export_timestamp", datetime.now(timezone.utc).isoformat()) - - # Optional fields - for key in ["pipeline_tag", "library_name", "base_model", "private", "gated"]: - if key in metadata: - add_prop(f"hf_{key}", metadata[key]) - - # Producer information - onnx_model.producer_name = "ModelExport-HTP" - onnx_model.producer_version = export_version - onnx_model.domain = "com.modelexport.htp" - - # Add doc string for human readability - onnx_model.doc_string = ( - f"Exported from HuggingFace model: {metadata.get('model_id')}\n" - f"Revision: {metadata.get('sha', 'unknown')[:8]}\n" - f"Export timestamp: {datetime.now(timezone.utc).isoformat()}\n" - f"ModelExport version: {export_version}" - ) - - -def save_local_model_configs(model_name_or_path: str, output_dir: Path, metadata: dict) -> None: - """Save configuration files for local/in-house models. - - Args: - model_name_or_path: Path to local model - output_dir: Directory to save configs - metadata: Local model metadata - """ - # Check if the path exists first - if not Path(model_name_or_path).exists(): - logger.info(f"Local model path {model_name_or_path} does not exist, skipping config copy") - return - - try: - from transformers import AutoConfig - - # Save config - config = AutoConfig.from_pretrained(model_name_or_path) - config.save_pretrained(output_dir) - logger.info(f"Saved config.json to {output_dir}") - - # Track what components were saved - components_saved = [] - - # Try AutoProcessor (for multimodal) - try: - from transformers import AutoProcessor - - processor = AutoProcessor.from_pretrained(model_name_or_path) - processor.save_pretrained(output_dir) - components_saved.append("processor") - except Exception: - pass - - # Try AutoTokenizer (for text models) - only if processor wasn't saved - if "processor" not in components_saved: - try: - from transformers import AutoTokenizer - - tokenizer = AutoTokenizer.from_pretrained(model_name_or_path) - tokenizer.save_pretrained(output_dir) - components_saved.append("tokenizer") - except Exception: - pass - - # Try AutoImageProcessor (for vision) - try: - from transformers import AutoImageProcessor - - image_processor = AutoImageProcessor.from_pretrained(model_name_or_path) - image_processor.save_pretrained(output_dir) - components_saved.append("image_processor") - except Exception: - pass - - # Try AutoFeatureExtractor (for audio) - try: - from transformers import AutoFeatureExtractor - - feature_extractor = AutoFeatureExtractor.from_pretrained(model_name_or_path) - feature_extractor.save_pretrained(output_dir) - components_saved.append("feature_extractor") - except Exception: - pass - - if components_saved: - logger.info(f"Saved preprocessing components: {', '.join(components_saved)}") - - except Exception as e: - logger.warning(f"Could not save config for local model: {e}") - logger.warning("User will need to provide config manually for inference") - - -def load_hf_components_from_onnx(onnx_path: str) -> tuple[Any, Any]: - """Load HuggingFace config and preprocessing components from ONNX. - - Handles both: - 1. Hub models - loads from HF Hub using metadata - 2. Local models - loads from co-located config files - - Args: - onnx_path: Path to ONNX model - - Returns: - Tuple of (config, preprocessor) - """ - from pathlib import Path - - from transformers import ( - AutoConfig, - AutoFeatureExtractor, - AutoImageProcessor, - AutoProcessor, - AutoTokenizer, - ) - - from ..onnx import load_onnx - - # Load ONNX model and extract metadata - onnx_model = load_onnx(onnx_path, validate=False) - onnx_dir = Path(onnx_path).parent - - # Extract metadata - metadata = {} - for prop in onnx_model.metadata_props: - metadata[prop.key] = prop.value - - model_type = metadata.get("hf_model_type", "unknown") - - if model_type == "hub": - # Hub model: Load from HuggingFace Hub - hf_hub_id = metadata.get("hf_hub_id") - hf_revision = metadata.get("hf_hub_revision") - - if not hf_hub_id: - raise ValueError("ONNX model marked as Hub model but missing hf_hub_id metadata") - - # Load config from Hub - config = AutoConfig.from_pretrained(hf_hub_id, revision=hf_revision) - - # Try to load preprocessor from Hub - preprocessor = None - # Heterogeneous transformers Auto-loaders sharing a from_pretrained classmethod. - hub_loaders: list[Any] = [ - AutoProcessor, - AutoTokenizer, - AutoImageProcessor, - AutoFeatureExtractor, - ] - for loader_cls in hub_loaders: - try: - preprocessor = loader_cls.from_pretrained(hf_hub_id, revision=hf_revision) - break - except Exception: - continue - - return config, preprocessor - - if model_type == "local": - # Local model: Load from co-located files - config_path = onnx_dir / "config.json" - - if not config_path.exists(): - raise ValueError( - f"Local model but config.json not found at {config_path}. " - "The model may have been moved without its config files." - ) - - # Load config from local file - config = AutoConfig.from_pretrained(onnx_dir) - - # Try to load preprocessor from local files - preprocessor = None - # Heterogeneous transformers Auto-loaders sharing a from_pretrained classmethod. - local_loaders: list[Any] = [ - AutoProcessor, - AutoTokenizer, - AutoImageProcessor, - AutoFeatureExtractor, - ] - for loader_cls in local_loaders: - try: - preprocessor = loader_cls.from_pretrained(onnx_dir) - break - except Exception: - continue - - return config, preprocessor - - # Unknown or legacy model - raise ValueError( - f"ONNX model has unknown type '{model_type}'. " - "Was it exported with an older version of ModelExport? " - "Please re-export the model." - ) diff --git a/src/winml/modelkit/utils/optimum_loader.py b/src/winml/modelkit/utils/optimum_loader.py deleted file mode 100644 index cf74dd36c..000000000 --- a/src/winml/modelkit/utils/optimum_loader.py +++ /dev/null @@ -1,160 +0,0 @@ -# ------------------------------------------------------------------------- -# Copyright (c) Microsoft Corporation. All rights reserved. -# Licensed under the MIT License. -# -------------------------------------------------------------------------- -"""Optimum integration utilities for loading ONNX models with HuggingFace configurations. - -This module provides seamless integration with HuggingFace Optimum for inference -using exported ONNX models with preserved hierarchy and Hub metadata. -""" - -import logging -import shutil -import tempfile -from pathlib import Path -from typing import Any, cast - -from .hub_utils import load_hf_components_from_onnx - - -logger = logging.getLogger(__name__) - - -class OptimumONNXModel: - """Wrapper for seamless Optimum integration with Hub metadata.""" - - @classmethod - def from_onnx( - cls, onnx_path: str, task: str = "auto", device: str = "cpu", **kwargs: Any - ) -> tuple[Any, Any]: - """Load Optimum model from ONNX with Hub metadata. - - Args: - onnx_path: Path to ONNX model - task: Task type or "auto" to detect - device: Device to run on - **kwargs: Additional arguments for ORTModel - - Returns: - Tuple of (model, preprocessor) - """ - # Load config and preprocessor - config, preprocessor = load_hf_components_from_onnx(onnx_path) - - # Auto-detect task if needed - if task == "auto": - task = cls._detect_task(config, onnx_path) - - # Get appropriate ORTModel class - ort_model_class = cls._get_ort_model_class(task) - - # Create temporary directory with required files - with tempfile.TemporaryDirectory() as temp_dir: - temp_path = Path(temp_dir) - - # Save config - config.save_pretrained(temp_path) - - # Save preprocessor if available - if preprocessor: - preprocessor.save_pretrained(temp_path) - - # Copy ONNX model - shutil.copy(onnx_path, temp_path / "model.onnx") - - # Load with Optimum - model = ort_model_class.from_pretrained( - temp_path, - provider="CPUExecutionProvider" if device == "cpu" else "CUDAExecutionProvider", - **kwargs, - ) - - return model, preprocessor - - @staticmethod - def _detect_task(config: Any, onnx_path: str) -> str: - """Detect task from config and metadata.""" - from ..onnx import load_onnx - - # Try to get task from metadata - try: - onnx_model = load_onnx(onnx_path, validate=False) - for prop in onnx_model.metadata_props: - if prop.key == "hf_pipeline_tag": - return prop.value - except Exception: - pass - - # Check architectures - if hasattr(config, "architectures"): - arch = config.architectures[0] if config.architectures else "" - - task_mapping = { - "ForSequenceClassification": "text-classification", - "ForTokenClassification": "token-classification", - "ForQuestionAnswering": "question-answering", - "ForCausalLM": "text-generation", - "ForConditionalGeneration": "text2text-generation", - "ForImageClassification": "image-classification", - "ForObjectDetection": "object-detection", - "ForAudioClassification": "audio-classification", - } - - for pattern, task in task_mapping.items(): - if pattern in arch: - return task - - # Default - return "feature-extraction" - - @staticmethod - def _get_ort_model_class(task: str) -> type[Any]: - """Get appropriate ORTModel class for task.""" - from optimum.onnxruntime import ( - ORTModel, - ORTModelForAudioClassification, - ORTModelForCausalLM, - ORTModelForFeatureExtraction, - ORTModelForImageClassification, - ORTModelForQuestionAnswering, - ORTModelForSeq2SeqLM, - ORTModelForSequenceClassification, - ORTModelForTokenClassification, - ) - - task_to_model = { - "text-classification": ORTModelForSequenceClassification, - "token-classification": ORTModelForTokenClassification, - "question-answering": ORTModelForQuestionAnswering, - "text-generation": ORTModelForCausalLM, - "text2text-generation": ORTModelForSeq2SeqLM, - "translation": ORTModelForSeq2SeqLM, - "summarization": ORTModelForSeq2SeqLM, - "image-classification": ORTModelForImageClassification, - "audio-classification": ORTModelForAudioClassification, - "feature-extraction": ORTModelForFeatureExtraction, - } - - return cast("type[Any]", task_to_model.get(task, ORTModel)) - - -def load_optimum_model( - onnx_path: str, task: str = "auto", device: str = "cpu", **kwargs: Any -) -> tuple[Any, Any]: - """Convenience function to load an ONNX model for Optimum inference. - - Args: - onnx_path: Path to ONNX model exported with ModelExport - task: Task type (auto-detected if not specified) - device: Device to run on ('cpu' or 'cuda') - **kwargs: Additional arguments for ORTModel - - Returns: - Tuple of (model, preprocessor) - - Example: - >>> model, tokenizer = load_optimum_model("bert.onnx") - >>> inputs = tokenizer("Hello world!", return_tensors="pt") - >>> outputs = model(**inputs) - """ - return OptimumONNXModel.from_onnx(onnx_path, task, device, **kwargs)