Filter HuggingFace downloads by target model and adapter name (fixes #3)

src/granite_switch/composer/adapter_discovery.py

@@ -279,12 +279,182 @@ def is_adapter_library(path: str) -> bool:
     return p.is_dir() and not (p / "adapter_config.json").exists()
 
 
-def resolve_repo_path(path_or_repo: str) -> str:
+# ------------------------------------------------------------------ #
+# HuggingFace Hub metadata helpers (no file downloads)
+# ------------------------------------------------------------------ #
+
+
+def _list_repo_adapter_names(repo_id: str) -> List[str]:
+    """Get adapter folder names from a HF repo using metadata-only API calls.
+
+    Returns top-level directory names, skipping entries that start with ``_``
+    (e.g. ``_ollama``).
+    """
+    from huggingface_hub import list_repo_tree
+    from huggingface_hub.hf_api import RepoFolder
+
+    tree = list_repo_tree(repo_id, repo_type="model")
+    return [
+        item.path for item in tree
+        if isinstance(item, RepoFolder) and not item.path.startswith("_")
+    ]
+
+
+def _resolve_technology(
+    repo_id: str,
+    adapter_name: str,
+    target_model_name: str,
+) -> Optional[str]:
+    """Resolve preferred technology for an adapter via Hub metadata.
+
+    Prefers ``alora`` over ``lora``. Returns ``None`` if neither exists
+    for this adapter/model combination.
+    """
+    from huggingface_hub import list_repo_tree
+    from huggingface_hub.hf_api import RepoFolder
+    from huggingface_hub.errors import EntryNotFoundError
+
+    try:
+        subtree = list_repo_tree(
+            repo_id, repo_type="model",
+            path_in_repo=f"{adapter_name}/{target_model_name}",
+        )
+        technologies = {
+            item.path.split("/")[-1] for item in subtree
+            if isinstance(item, RepoFolder)
+            and item.path.split("/")[-1] in ("alora", "lora")
+        }
+    except EntryNotFoundError:
+        return None
+
+    if "alora" in technologies:
+        return "alora"
+    elif "lora" in technologies:
+        return "lora"
+    return None
+
+
+def _build_allow_patterns(
+    repo_id: str,
+    target_model_name: Optional[str] = None,
+    include_adapters: Optional[List[str]] = None,
+    exclude_adapters: Optional[List[str]] = None,
+) -> Optional[List[str]]:
+    """Build ``allow_patterns`` for selective ``snapshot_download``.
+
+    Uses lightweight Hub API calls to discover adapter names, then applies
+    fnmatch-based include/exclude filtering and target model constraints to
+    construct download patterns.  When both adapter names and target model
+    are known, also resolves the preferred technology (alora > lora) so
+    only the needed technology variant is downloaded.
+
+    Returns:
+        List of glob patterns, or ``None`` if no filtering is possible.
+    """
+    adapter_names = _list_repo_adapter_names(repo_id)
+
+    # Apply include filter
+    if include_adapters:
+        adapter_names = [
+            name for name in adapter_names
+            if any(fnmatch(name, pat) for pat in include_adapters)
+        ]
+
+    # Apply exclude filter
+    if exclude_adapters:
+        adapter_names = [
+            name for name in adapter_names
+            if not any(fnmatch(name, pat) for pat in exclude_adapters)
+        ]
+
+    # Construct patterns with technology resolution
+    if adapter_names and target_model_name:
+        patterns = []
+        for name in adapter_names:
+            tech = _resolve_technology(repo_id, name, target_model_name)
+            if tech:
+                patterns.append(f"{name}/{target_model_name}/{tech}/**")
+            else:
+                # Model not found for this adapter — include anyway so
+                # discover_adapters can report it as missing downstream
+                patterns.append(f"{name}/{target_model_name}/**")
+        return patterns if patterns else None
+    elif target_model_name:
+        return [f"*/{target_model_name}/**"]
+    elif adapter_names:
+        return [f"{name}/**" for name in adapter_names]
+    else:
+        return None
+
+
+def list_repo_adapters_remote(
+    repo_id: str,
+    target_model_name: str,
+) -> List[Dict[str, object]]:
+    """List adapters available in a remote HF repo without downloading.
+
+    Uses Hub metadata API calls to discover adapter names and their
+    available technologies for the given target model.
+
+    Args:
+        repo_id: HuggingFace repo ID (e.g., ``"ibm-granite/granitelib-rag-r1.0"``).
+        target_model_name: Target model name (e.g., ``"granite-4.1-3b"``).
+
+    Returns:
+        List of dicts ``{"name": str, "technologies": [str]}``, sorted
+        by adapter name.
+    """
+    from huggingface_hub import list_repo_tree
+    from huggingface_hub.hf_api import RepoFolder
+    from huggingface_hub.errors import EntryNotFoundError
+
+    adapter_names = _list_repo_adapter_names(repo_id)
+    results = []
+
+    for name in adapter_names:
+        try:
+            subtree = list_repo_tree(
+                repo_id, repo_type="model",
+                path_in_repo=f"{name}/{target_model_name}",
+            )
+            technologies = sorted(
+                item.path.split("/")[-1] for item in subtree
+                if isinstance(item, RepoFolder)
+                and item.path.split("/")[-1] in ("alora", "lora")
+            )
+            if technologies:
+                results.append({"name": name, "technologies": technologies})
+        except EntryNotFoundError:
+            # Adapter doesn't have this target model — skip
+            continue
+
+    return sorted(results, key=lambda x: x["name"])
+
+
+# ------------------------------------------------------------------ #
+# Path resolution
+# ------------------------------------------------------------------ #
+
+
+def resolve_repo_path(
+    path_or_repo: str,
+    target_model_name: Optional[str] = None,
+    include_adapters: Optional[List[str]] = None,
+    exclude_adapters: Optional[List[str]] = None,
+) -> str:
     """Resolve a local path or HuggingFace repo ID to a local directory.
 
+    For HuggingFace repos, applies selective downloading using
+    ``allow_patterns`` constructed from the provided filters.
+
     Args:
         path_or_repo: Either a local directory path or a HuggingFace repo ID
             (e.g., ``"ibm-granite/granite-lib-rag-r1.0"``).
+        target_model_name: Target model name to filter by (e.g.,
+            ``"granite-4.1-3b"``).
+        include_adapters: Only download adapters matching these fnmatch
+            patterns.
+        exclude_adapters: Skip adapters matching these fnmatch patterns.
 
     Returns:
         Absolute local path to the directory.
@@ -298,12 +468,30 @@ def resolve_repo_path(path_or_repo: str) -> str:
 
     if "/" in path_or_repo and not local.exists():
         print(f"  Detected HuggingFace repo: {path_or_repo}")
+
+        # Build selective download patterns
+        allow_patterns = None
+        if target_model_name or include_adapters or exclude_adapters:
+            try:
+                allow_patterns = _build_allow_patterns(
+                    path_or_repo,
+                    target_model_name=target_model_name,
+                    include_adapters=include_adapters,
+                    exclude_adapters=exclude_adapters,
+                )
+                if allow_patterns:
+                    print(f"  Selective download patterns: {allow_patterns}")
+            except Exception as e:
+                print(f"  WARNING: Failed to build download filters ({e}), "
+                      f"downloading full repo")
+                allow_patterns = None
+
         print(f"  Downloading from HuggingFace Hub...")
         try:
-            cache_dir = snapshot_download(
-                repo_id=path_or_repo,
-                repo_type="model",
-            )
+            kwargs = {"repo_id": path_or_repo, "repo_type": "model"}
+            if allow_patterns:
+                kwargs["allow_patterns"] = allow_patterns
+            cache_dir = snapshot_download(**kwargs)
             print(f"  Downloaded to: {cache_dir}")
             return cache_dir
         except Exception as e:

src/granite_switch/composer/compose_granite_switch.py

@@ -56,6 +56,7 @@
     filter_adapters,
     is_adapter_library,
     list_available_adapters,
+    list_repo_adapters_remote,
     resolve_repo_path,
 )
 from granite_switch.composer.tokenizer_setup import (
@@ -530,15 +531,30 @@ def build():
             print("ERROR: --list-adapters requires --adapters")
             return 1
         for entry in args.adapters:
-            try:
-                resolved_path = resolve_repo_path(entry)
-            except Exception as e:
-                print(f"Failed to resolve {entry}: {e}")
-                return 1
-            if not is_adapter_library(resolved_path):
-                print(f"\n{entry} is a single adapter, not a library.")
-                continue
-            available = list_available_adapters(resolved_path, args.target_model)
+            # For HF repos, use metadata-only listing (no download)
+            local = Path(entry)
+            if "/" in entry and not local.exists():
+                try:
+                    available = list_repo_adapters_remote(
+                        entry, args.target_model
+                    )
+                except Exception as e:
+                    print(f"Failed to list adapters from {entry}: {e}")
+                    return 1
+            else:
+                # Local path — resolve and scan
+                try:
+                    resolved_path = resolve_repo_path(entry)
+                except Exception as e:
+                    print(f"Failed to resolve {entry}: {e}")
+                    return 1
+                if not is_adapter_library(resolved_path):
+                    print(f"\n{entry} is a single adapter, not a library.")
+                    continue
+                available = list_available_adapters(
+                    resolved_path, args.target_model
+                )
+
             if not available:
                 print(f"\nNo adapters found in {entry} for target '{args.target_model}'")
                 continue
@@ -587,7 +603,12 @@ def build():
         for entry in args.adapters:
             print(f"\nResolving: {entry}")
             try:
-                resolved_path = resolve_repo_path(entry)
+                resolved_path = resolve_repo_path(
+                    entry,
+                    target_model_name=args.target_model,
+                    include_adapters=args.include_adapters,
+                    exclude_adapters=args.exclude_adapters,
+                )
             except Exception as e:
                 print(f"Failed to resolve {entry}: {e}")
                 return 1