admesh/agentic-intent-classifier
254
1"""2AdmeshIntentPipeline — transformers.Pipeline subclass for3admesh/agentic-intent-classifier.4 5Because config.json declares "pt": [] the transformers pipeline() loader6skips AutoModel.from_pretrained() entirely and passes model=None straight7to this class. All model loading is handled internally via combined_inference,8which resolves paths relative to __file__ so it works wherever HF downloads9the repo (Inference Endpoints, Spaces, local snapshot_download, etc.).10 11Supported HF deployment surfaces12---------------------------------131. transformers.pipeline() direct call (trust_remote_code=True):14 15 from transformers import pipeline16 clf = pipeline(17 "admesh-intent",18 model="admesh/agentic-intent-classifier",19 trust_remote_code=True,20 )21 result = clf("Which laptop should I buy for college?")22 232. HF Inference Endpoints — Standard (PyTorch, trust_remote_code=True):24 Deploy from https://ui.endpoints.huggingface.co — no custom container25 needed; HF loads this pipeline class automatically.26 273. HF Spaces (Gradio / Streamlit):28 29 import sys30 from huggingface_hub import snapshot_download31 local_dir = snapshot_download("admesh/agentic-intent-classifier", repo_type="model")32 sys.path.insert(0, local_dir)33 from pipeline import AdmeshIntentPipeline34 clf = AdmeshIntentPipeline()35 result = clf("I need a CRM for a 5-person startup")36 374. Anywhere via from_pretrained():38 39 from pipeline import AdmeshIntentPipeline40 clf = AdmeshIntentPipeline.from_pretrained("admesh/agentic-intent-classifier")41"""42 43from __future__ import annotations44 45import sys46from pathlib import Path47from typing import Union48 49# ── try to import transformers.Pipeline; fall back gracefully if absent ───────50try:51 from transformers import Pipeline as _HFPipeline52 _TRANSFORMERS_AVAILABLE = True53except ImportError:54 _HFPipeline = object # bare object as base when transformers is not installed55 _TRANSFORMERS_AVAILABLE = False56 57 58class AdmeshIntentPipeline(_HFPipeline):59 """60 Full intent + IAB classification pipeline.61 62 Inherits from ``transformers.Pipeline`` so it works natively with63 ``pipeline()``, HF Inference Endpoints (standard mode), and HF Spaces.64 65 When ``transformers`` is not installed it falls back to a plain callable66 class so the same code works in minimal environments too.67 68 Parameters69 ----------70 model:71 Ignored — we load all models internally. Present only to satisfy72 the ``transformers.Pipeline`` interface when HF calls73 ``PipelineClass(model=None, ...)``.74 **kwargs:75 Forwarded to ``transformers.Pipeline.__init__`` if transformers is76 available, otherwise ignored.77 """78 79 # ── init ──────────────────────────────────────────────────────────────────80 81 def __init__(self, model=None, tokenizer=None, **kwargs):82 # Ensure this repo's directory is on sys.path so all relative imports83 # in combined_inference / config / model_runtime resolve correctly.84 # Path(__file__) points to wherever HF cached the repo snapshot.85 _repo_dir = Path(__file__).resolve().parent86 if str(_repo_dir) not in sys.path:87 sys.path.insert(0, str(_repo_dir))88 89 if _TRANSFORMERS_AVAILABLE:90 import torch91 92 # transformers.Pipeline requires certain attributes to be set.93 # Because config.json has "pt": [] HF passes model=None here —94 # we satisfy the interface by setting the minimum required attrs95 # manually instead of calling super().__init__(model=None, ...)96 # which would raise inside infer_framework_load_model().97 self.task = kwargs.pop("task", "admesh-intent")98 self.model = model # None — unused, kept for interface compat99 self.tokenizer = tokenizer # None — unused100 self.feature_extractor = None101 self.image_processor = None102 self.modelcard = None103 self.framework = "pt"104 self.device = torch.device(kwargs.pop("device", "cpu"))105 self.binary_output = kwargs.pop("binary_output", False)106 self.call_count = 0107 self._batch_size = kwargs.pop("batch_size", 1)108 self._num_workers = kwargs.pop("num_workers", 0)109 self._preprocess_params: dict = {}110 self._forward_params: dict = {}111 self._postprocess_params: dict = {}112 # else: plain object, no init needed113 114 self._classify_fn = None # lazy-loaded on first __call__115 116 # ── transformers.Pipeline abstract methods ────────────────────────────────117 # These are required by the ABC but our __call__ override bypasses them.118 # They are still implemented in case a caller invokes them directly.119 120 def _sanitize_parameters(self, **kwargs):121 forward_kwargs = {}122 if "threshold_overrides" in kwargs:123 forward_kwargs["threshold_overrides"] = kwargs["threshold_overrides"]124 if "force_iab_placeholder" in kwargs:125 forward_kwargs["force_iab_placeholder"] = kwargs["force_iab_placeholder"]126 return {}, forward_kwargs, {}127 128 def preprocess(self, inputs):129 return {"text": inputs if isinstance(inputs, str) else str(inputs)}130 131 def _forward(self, model_inputs, threshold_overrides=None, force_iab_placeholder=False):132 self._ensure_loaded()133 return self._classify_fn(134 model_inputs["text"],135 threshold_overrides=threshold_overrides,136 force_iab_placeholder=force_iab_placeholder,137 )138 139 def postprocess(self, model_outputs):140 return model_outputs141 142 # ── __call__ override ─────────────────────────────────────────────────────143 # We bypass Pipeline's preprocess→_forward→postprocess chain entirely so144 # we never touch self.model and keep full control over batching logic.145 146 def __call__(147 self,148 inputs: Union[str, list[str]],149 *,150 threshold_overrides: dict[str, float] | None = None,151 force_iab_placeholder: bool = False,152 ) -> Union[dict, list[dict]]:153 """154 Classify one or more query strings.155 156 Parameters157 ----------158 inputs:159 A single query string or a list of query strings.160 threshold_overrides:161 Optional per-head confidence threshold overrides, e.g.162 ``{"intent_type": 0.5, "iab_content": 0.3}``.163 force_iab_placeholder:164 Skip IAB classifier and return placeholder values (faster,165 no IAB accuracy).166 167 Returns168 -------169 dict or list[dict]:170 Full classification payload matching the combined_inference schema.171 Returns a single dict for a string input, list of dicts for a list.172 173 Examples174 --------175 ::176 177 clf = pipeline("admesh-intent", model="admesh/agentic-intent-classifier",178 trust_remote_code=True)179 180 # single181 result = clf("Which laptop should I buy for college?")182 183 # batch184 results = clf(["Best running shoes", "How does TCP work?"])185 186 # custom thresholds187 result = clf("Buy headphones", threshold_overrides={"intent_type": 0.6})188 """189 self._ensure_loaded()190 191 single = isinstance(inputs, str)192 texts: list[str] = [inputs] if single else list(inputs)193 194 results = [195 self._classify_fn(196 text,197 threshold_overrides=threshold_overrides,198 force_iab_placeholder=force_iab_placeholder,199 )200 for text in texts201 ]202 return results[0] if single else results203 204 # ── warm-up / compile ─────────────────────────────────────────────────────205 206 def warm_up(self, compile: bool = False) -> "AdmeshIntentPipeline":207 """208 Pre-load all models and optionally compile them with torch.compile().209 210 Call once after instantiation so the first real request pays no211 model-load cost. HF Inference Endpoints automatically sends a212 warm-up probe before routing live traffic, so this is optional there.213 214 Parameters215 ----------216 compile:217 If ``True``, call ``torch.compile()`` on the DistilBERT encoder218 and IAB classifier (requires PyTorch >= 2.0). Gives ~15-30 %219 CPU speedup after the first traced call.220 """221 self._ensure_loaded()222 223 if compile:224 import torch # noqa: PLC0415225 if not hasattr(torch, "compile"):226 import warnings227 warnings.warn(228 "torch.compile() is not available (PyTorch >= 2.0 required). "229 "Skipping.",230 stacklevel=2,231 )232 else:233 try:234 from .multitask_runtime import get_multitask_runtime # type: ignore235 from .model_runtime import get_head # type: ignore236 except ImportError:237 from multitask_runtime import get_multitask_runtime238 from model_runtime import get_head239 240 rt = get_multitask_runtime()241 if rt._model is not None:242 rt._model = torch.compile(rt._model)243 iab_head = get_head("iab_content")244 if iab_head._model is not None:245 iab_head._model = torch.compile(iab_head._model)246 247 # Dry run — triggers any remaining lazy init (calibration JSON reads, etc.)248 self("warm up query for intent classification", force_iab_placeholder=True)249 return self250 251 # ── factory ───────────────────────────────────────────────────────────────252 253 @classmethod254 def from_pretrained(255 cls,256 repo_id: str = "admesh/agentic-intent-classifier",257 *,258 revision: str | None = None,259 token: str | None = None,260 ) -> "AdmeshIntentPipeline":261 """262 Download the model bundle from HF Hub and return a ready-to-use instance.263 264 Parameters265 ----------266 repo_id:267 HF Hub model id.268 revision:269 Optional git commit hash to pin a specific release.270 token:271 Optional HF auth token for private repos.272 273 Example274 -------275 ::276 277 from pipeline import AdmeshIntentPipeline278 clf = AdmeshIntentPipeline.from_pretrained("admesh/agentic-intent-classifier")279 print(clf("I need a CRM for a 5-person startup"))280 """281 try:282 from huggingface_hub import snapshot_download # noqa: PLC0415283 except ImportError as exc:284 raise ImportError(285 "huggingface_hub is required. Install: pip install huggingface_hub"286 ) from exc287 288 kwargs: dict = {"repo_type": "model"}289 if revision:290 kwargs["revision"] = revision291 if token:292 kwargs["token"] = token293 294 local_dir = snapshot_download(repo_id=repo_id, **kwargs)295 if str(local_dir) not in sys.path:296 sys.path.insert(0, str(local_dir))297 return cls()298 299 # ── internal ──────────────────────────────────────────────────────────────300 301 def _ensure_loaded(self) -> None:302 if self._classify_fn is None:303 try:304 from .combined_inference import classify_query # type: ignore305 except ImportError:306 from combined_inference import classify_query307 self._classify_fn = classify_query308 309 def __repr__(self) -> str:310 state = "loaded" if self._classify_fn is not None else "not yet loaded"311 return f"AdmeshIntentPipeline(classify_fn={state})"312