# Copyright (c) 2025 The University of Washington
#
# This file is part of rapidtools.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
#
# 1. Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
#
# 2. Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
#
# 3. Neither the name of the copyright holder nor the names of its contributors
# may be used to endorse or promote products derived from this software without
# specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
# You should have received a copy of the BSD 3-Clause License along with
# rapidtools. If not, see <http://www.opensource.org/licenses/>.
#
# Contributors:
# Barbaros Cetiner
#
# Last updated:
# 09-22-2026
"""
Pipeline components that describe assets with vision-language models.
:class:`AssetAnalyzer` is the single analysis step: give it any model from
:mod:`rapidtools.models` (hosted API or local checkpoint) and a prompt, and it
sends each asset's images to the model, parses the reply into key/value
attributes (``<prefix>_<key>``) and returns the mutated collection. Hosted
models run in a thread pool with a shared rate-limit cooldown; local models
run sequentially or in batches to protect GPU memory.
The provider-specific classes (``GeminiAssetAnalyzer``, ``Gemma4AssetAnalyzer``,
...) remain as thin, deprecated wrappers that build the model for you.
Example:
>>> from rapidtools import Pipeline
>>> from rapidtools.models import load
>>> from rapidtools.processing import AssetAnalyzer
>>> analyzer = AssetAnalyzer(
... load('gemini', api_key='AIza...'), prompt='Rate the damage 0-5 as JSON.'
... )
>>> collection = Pipeline([analyzer]).run(collection)
>>> collection.get('bldg_01').attributes['gemini_damage']
'3'
"""
from __future__ import annotations
import json
import logging
import math
import re
import threading
import time
import warnings
from collections.abc import Callable
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from tqdm import tqdm
from rapidtools.core import (
ImageAsset,
PhysicalAsset,
PhysicalAssetCollection,
is_cancelled,
raise_if_cancelled,
)
from rapidtools.models import (
ClaudeInference,
GeminiInference,
Gemma4Inference,
GenerationConfig,
HFVisionInference,
LlamaVisionInference,
MuseGlimmerInference,
MuseSparkInference,
OpenAIInference,
QwenInference,
QwenVisionInference,
)
from rapidtools.models.base import BaseInferenceModel, resolve_prompt_text
from rapidtools.models.catalogs import (
ANTHROPIC_DEFAULT_MODEL,
DEFAULT_OPEN_VLM,
GEMINI_DEFAULT_LITE_MODEL,
GEMMA4_DEFAULT_MODEL,
LLAMA_DEFAULT_MODEL,
MUSE_GLIMMER_DEFAULT_MODEL,
MUSE_SPARK_DEFAULT_MODEL,
OPENAI_DEFAULT_MODEL,
QWEN_DEFAULT_MODEL,
QWEN_VL_DEFAULT_MODEL,
)
from rapidtools.models.local_base import BaseLocalInferenceModel
from .step import Stage
logger = logging.getLogger(__name__)
# Analyzers default to the cheapest current Gemini tier because they may run
# over thousands of assets; pass ``model_id`` for a stronger model.
GEMINI_ANALYZER_DEFAULT_MODEL = GEMINI_DEFAULT_LITE_MODEL
def parse_structured_results(text: str) -> dict[str, Any]:
"""
Extract key-value pairs from a vision-language model response.
Attempts to parse a JSON block (wrapped in ```json ... ``` or bare). If no
JSON object is found, falls back to line-by-line ``Key: Value`` parsing,
normalising keys to ``snake_case``.
Args:
text: The raw text generated by the model.
Returns:
dict[str, Any]: The parsed attributes (empty if nothing matched).
Example:
>>> parse_structured_results('```json\\n{"level": 3}\\n```')
{'level': 3}
>>> parse_structured_results('CHS Level: 3\\nNote - fine')
{'chs_level': '3', 'note': 'fine'}
"""
results: dict[str, Any] = {}
json_match = re.search(r'```json\s*(.*?)\s*```', text, re.DOTALL)
json_text = json_match.group(1) if json_match else text
try:
parsed = json.loads(json_text)
if isinstance(parsed, dict):
return parsed
except json.JSONDecodeError:
pass
for line in text.split('\n'):
match = re.match(r'^([\w\s]+)[:\-]\s*(.*)$', line.strip())
if match:
key = match.group(1).strip().lower().replace(' ', '_')
results[key] = match.group(2).strip()
return results
[docs]
@dataclass
class RateLimitPolicy:
"""
Shared back-off applied by :class:`AssetAnalyzer` to hosted APIs.
When a request fails (rate limit, transient error, safety block), every
worker pauses for ``cooldown_seconds`` multiplied by the number of
consecutive failures; the first success resets the multiplier.
Assets whose request failed are retried in additional passes once the
first pass is complete (``max_asset_retries``), so a burst of 429s or
timeouts does not leave holes in the results.
Attributes:
cooldown_seconds: Base pause after a failure. ``0`` disables pausing.
max_asset_retries: Extra passes over the assets that failed.
``0`` disables retries.
Example:
>>> RateLimitPolicy(cooldown_seconds=10).wait_for(3)
30
>>> RateLimitPolicy(max_asset_retries=1).max_asset_retries
1
"""
cooldown_seconds: float = 30.0
max_asset_retries: int = 2
[docs]
def wait_for(self, consecutive_errors: int) -> float:
"""Return the pause (seconds) after ``consecutive_errors`` failures."""
return self.cooldown_seconds * consecutive_errors
[docs]
class AssetAnalyzer:
"""
Describe every asset in a collection with a vision-language model.
Args:
model (BaseInferenceModel):
Any rapidtools model wrapper (see :func:`rapidtools.models.load`).
prompt (str | Path):
Prompt text or path to a prompt file applied to every asset.
attribute_prefix (str | None):
Prefix of the attributes written to each asset. Defaults to the
provider's prefix (``'gemini'``, ``'qwen_vl'``, ...).
max_workers (int | None):
Concurrent requests for hosted models. Defaults to 5 for APIs and
1 for local models (local inference is never threaded).
batch_size (int):
Assets per batch when the model offers ``run_inference_batch``.
Defaults to 4.
max_images_per_asset (int | None):
Cap on the images sent per asset; ``None`` sends them all.
image_filter (Callable[[ImageAsset], bool] | None):
Optional predicate selecting which images to analyze.
cancel_event (threading.Event | None):
Cooperative cancellation flag checked between assets/batches.
rate_limit (RateLimitPolicy | None):
Back-off and retry policy for hosted models; ``None`` uses a
30-second cooldown with two retry passes for APIs and no cooldown
for local models.
generation (GenerationConfig | None):
Per-call generation options (``json_mode``, ``temperature``, ...)
forwarded to the model.
mode (str | None):
Force ``'api'`` (threaded) or ``'local'`` (sequential/batched)
execution; inferred from the model when ``None``.
Example:
>>> from rapidtools.models import GenerationConfig, load
>>> from rapidtools.processing import AssetAnalyzer
>>> analyzer = AssetAnalyzer(
... load('qwen_vl', model_id='Qwen/Qwen3.5-9B', load_in_4bit=True),
... prompt='Return JSON with a damage_level key.',
... batch_size=4,
... generation=GenerationConfig(json_mode=True, temperature=0.0),
... )
>>> collection = analyzer(collection)
>>> collection.get('bldg_01').attributes['qwen_vl_damage_level']
'moderate'
"""
stage = Stage.ANALYZE
#: Default attribute prefix when neither the model nor the caller gives one.
ATTRIBUTE_PREFIX: str | None = None
PROVIDER_NAME: str | None = None
def __init__(
self,
model: BaseInferenceModel,
prompt: str | Path = '',
attribute_prefix: str | None = None,
max_workers: int | None = None,
batch_size: int = 4,
max_images_per_asset: int | None = None,
image_filter: Callable[[ImageAsset], bool] | None = None,
cancel_event: threading.Event | None = None,
rate_limit: RateLimitPolicy | None = None,
generation: GenerationConfig | None = None,
mode: str | None = None,
) -> None:
"""Wrap ``model`` for use as the analysis step of a pipeline."""
self.model = model
self.prompt = resolve_prompt_text(prompt)
self.image_filter = image_filter
self.cancel_event = cancel_event
self.generation = generation
self.max_images_per_asset = max_images_per_asset
self.batch_size = max(1, batch_size)
info = getattr(model, 'INFO', None)
if mode is None:
if info is not None and info.kind in ('api', 'local'):
mode = info.kind
elif isinstance(model, BaseLocalInferenceModel):
mode = 'local'
else:
mode = 'api'
if mode not in ('api', 'local'):
raise ValueError(f"mode must be 'api' or 'local', got {mode!r}.")
self.mode = mode
prefix = attribute_prefix or type(self).ATTRIBUTE_PREFIX
if not prefix:
prefix = info.attribute_prefix if info is not None else 'vlm'
self.ATTRIBUTE_PREFIX = prefix or 'vlm'
self.PROVIDER_NAME = type(self).PROVIDER_NAME or (
info.label if info is not None else 'VLM'
)
if max_workers is None:
max_workers = 5 if self.mode == 'api' else 1
self.max_workers = max(1, max_workers)
if rate_limit is None:
rate_limit = RateLimitPolicy() if self.mode == 'api' else RateLimitPolicy(0)
self.rate_limit = rate_limit
self._lock = threading.Lock()
self._global_cooldown_until = 0.0
self._consecutive_error_count = 0
# IDs of assets the provider rejected outright (refusal, 4xx, safety
# block): they are reported as failed but neither pause the workers
# nor get another pass.
self._permanent_failures: set[str] = set()
self.model_id = getattr(model, 'model_id', None)
# ------------------------------------------------------------ properties
@property
def attribute_prefix(self) -> str:
"""Prefix of the attributes written to each asset."""
# ``__init__`` always stores a non-empty prefix; the class attribute
# is Optional only so subclasses may leave it unset.
return self.ATTRIBUTE_PREFIX or 'vlm'
@property
def cooldown_duration(self) -> float:
"""Base cooldown in seconds (from :attr:`rate_limit`)."""
return self.rate_limit.cooldown_seconds
# --------------------------------------------------------------- helpers
def _parse_structured_results(self, text: str) -> dict[str, Any]:
"""Extract key-value pairs from the model response."""
return parse_structured_results(text)
def _image_paths(self, asset: PhysicalAsset) -> list[Path]:
"""Return the downloaded image paths of an asset after filtering/capping."""
target_images = asset.image_assets
if self.image_filter is not None:
target_images = target_images.filter(self.image_filter)
paths = [img.path for img in target_images if img.is_downloaded]
if self.max_images_per_asset is not None:
paths = paths[: self.max_images_per_asset]
return paths
def _infer(self, image_paths: list[Path], prompt: str) -> Any:
"""Call the model, forwarding the generation config when one is set."""
if self.generation is not None:
return self.model.run_inference(
image_inputs=image_paths, prompt=prompt, config=self.generation
)
return self.model.run_inference(image_inputs=image_paths, prompt=prompt)
def _apply_result(self, asset: PhysicalAsset, img_paths: list, result: Any) -> bool:
"""Write a model response onto the asset; return ``True`` on success."""
if result is None or not result.text:
return False
structured_data = self._parse_structured_results(result.text)
prefix = self.ATTRIBUTE_PREFIX
if structured_data:
for key, val in structured_data.items():
asset.attributes[f'{prefix}_{key}'] = val
else:
asset.attributes[f'{prefix}_analysis_raw'] = result.text
asset.attributes['ai_model_used'] = getattr(self.model, 'model_id', None)
asset.attributes['images_analyzed_count'] = len(img_paths)
return True
# ------------------------------------------------------------- API mode
def _process_single_asset(self, asset: PhysicalAsset) -> bool:
"""
Analyze one asset, honouring the shared rate-limit cooldown.
Returns:
bool: ``True`` when attributes were written.
"""
if is_cancelled(self.cancel_event):
return False
with self._lock:
wait_time = self._global_cooldown_until - time.time()
if wait_time > 0:
logger.debug(
f'Asset {asset.id} delayed: global cooldown active for '
f'{int(wait_time)}s...'
)
time.sleep(wait_time)
image_paths = self._image_paths(asset)
if not image_paths:
return False
result = self._infer(image_paths, self.prompt)
if (
result is not None
and getattr(result, 'failed', False)
and not getattr(result, 'retryable', True)
):
# A rejection is not a rate limit: do not stall the other workers
# or re-send the same request later.
logger.error(
f'Asset {asset.id} was rejected by the provider ({result.error}); '
'it will not be retried.'
)
with self._lock:
self._permanent_failures.add(asset.id)
return False
if result is None or not result.text:
with self._lock:
self._consecutive_error_count += 1
dynamic_wait = self.rate_limit.wait_for(self._consecutive_error_count)
if dynamic_wait > 0:
self._global_cooldown_until = time.time() + dynamic_wait
logger.warning(
f'Request failed for asset {asset.id} (consecutive failures: '
f'{self._consecutive_error_count}); pausing all workers for '
f'{dynamic_wait:g}s. The asset will be retried later.'
)
return False
with self._lock:
if self._consecutive_error_count > 0:
logger.info('Successful response received. Resetting error multiplier.')
self._consecutive_error_count = 0
return self._apply_result(asset, image_paths, result)
def _run_pass(
self, assets: list[PhysicalAsset], description: str
) -> list[PhysicalAsset]:
"""Analyze ``assets`` concurrently and return the ones that failed."""
workers = min(self.max_workers, max(1, len(assets)))
failed: list[PhysicalAsset] = []
with ThreadPoolExecutor(max_workers=workers) as executor:
future_to_asset = {
executor.submit(self._process_single_asset, asset): asset
for asset in assets
}
for future in tqdm(
as_completed(future_to_asset), total=len(assets), desc=description
):
asset = future_to_asset[future]
if is_cancelled(self.cancel_event):
executor.shutdown(wait=False, cancel_futures=True)
break
try:
if not future.result() and asset.id not in self._permanent_failures:
failed.append(asset)
except Exception as e: # noqa: BLE001 - reported per asset
logger.error(f'Error on asset {asset.id}: {e}')
failed.append(asset)
return failed
def _run_api(self, assets: list[PhysicalAsset]) -> int:
"""
Analyze assets concurrently with retry passes; return the failures.
Assets without downloaded images are skipped up front. Assets whose
request failed transiently (rate limit, timeout, network error) are
retried in up to ``rate_limit.max_asset_retries`` further passes;
assets the provider rejected (refusal, safety block, 4xx) are counted
as failed without a retry.
"""
self._permanent_failures = set()
pending = [asset for asset in assets if self._image_paths(asset)]
skipped = len(assets) - len(pending)
if skipped:
logger.info(f'{skipped} assets have no downloaded images and are skipped.')
workers = min(self.max_workers, max(1, len(pending)))
logger.info(
f'{self.PROVIDER_NAME}: analyzing {len(pending)} assets with {workers} '
f'workers (base cooldown: {self.cooldown_duration:g}s, up to '
f'{self.rate_limit.max_asset_retries} retry passes).'
)
failed: list[PhysicalAsset] = []
for attempt in range(1 + max(0, self.rate_limit.max_asset_retries)):
if not pending or is_cancelled(self.cancel_event):
break
description = (
'Analyzing Assets' if attempt == 0 else f'Retry pass {attempt}'
)
if attempt > 0:
logger.info(
f'Retrying {len(pending)} assets that failed (pass {attempt} of '
f'{self.rate_limit.max_asset_retries})...'
)
failed = self._run_pass(pending, description)
pending = failed
if failed:
logger.error(
f'{len(failed)} assets still failed after all retries: '
f'{", ".join(a.id for a in failed[:10])}'
f'{"..." if len(failed) > 10 else ""}'
)
if self._permanent_failures:
logger.error(
f'{len(self._permanent_failures)} assets were rejected by the '
'provider and not retried.'
)
return len(failed) + len(self._permanent_failures) + skipped
# ----------------------------------------------------------- local mode
def _run_local(self, assets: list[PhysicalAsset]) -> int:
"""Analyze assets sequentially or in batches; return the failures."""
run_inference_batch = getattr(self.model, 'run_inference_batch', None)
batched = run_inference_batch is not None
batch_size = self.batch_size if batched else 1
total_batches = math.ceil(len(assets) / batch_size)
logger.info(
f'{self.PROVIDER_NAME}: analyzing {len(assets)} assets in batches of '
f'{batch_size}...'
)
failed = 0
for i in tqdm(
range(0, len(assets), batch_size),
total=total_batches,
desc='Batch Analysis',
):
raise_if_cancelled(self.cancel_event, f'{self.PROVIDER_NAME} analysis')
batch_assets = assets[i : i + batch_size]
batch_images = [self._image_paths(a) for a in batch_assets]
if run_inference_batch is not None:
kwargs = {'config': self.generation} if self.generation else {}
results = run_inference_batch(
batch_images, [self.prompt] * len(batch_assets), **kwargs
)
else:
results = [
self._infer(paths, self.prompt) if paths else None
for paths in batch_images
]
for asset, img_paths, result in zip(
batch_assets, batch_images, results, strict=True
):
if not self._apply_result(asset, img_paths, result):
failed += 1
self._release_gpu_memory()
return failed
@staticmethod
def _release_gpu_memory() -> None:
"""Clear the CUDA cache between batches when PyTorch is available."""
try:
import torch
except ImportError: # pragma: no cover - torch is a dependency
return
if torch.cuda.is_available():
torch.cuda.empty_cache()
# -------------------------------------------------------------- __call__
def __call__(
self, asset_collection: PhysicalAssetCollection
) -> PhysicalAssetCollection:
"""
Analyze every asset that has downloaded images.
Args:
asset_collection: The assets to describe.
Returns:
PhysicalAssetCollection: The same collection with new attributes.
Raises:
OperationCancelled: If ``cancel_event`` is set during the run.
"""
assets_with_images = [a for a in asset_collection if len(a.image_assets) > 0]
if not assets_with_images:
logger.warning('No assets with images found to analyze. Skipping.')
return asset_collection
if self.mode == 'api':
failed = self._run_api(assets_with_images)
else:
failed = self._run_local(assets_with_images)
raise_if_cancelled(self.cancel_event, f'{self.PROVIDER_NAME} analysis')
if failed > 0:
logger.error(f'{self.PROVIDER_NAME}: {failed} assets failed to process.')
else:
logger.info(f'{self.PROVIDER_NAME}: All assets processed successfully.')
return asset_collection
# ======================================================================
# Deprecated provider-specific wrappers (kept for backwards compatibility)
# ======================================================================
def _warn_deprecated(old: str, provider: str) -> None:
"""Emit the standard deprecation warning for a provider-specific analyzer."""
warnings.warn(
f'{old} is deprecated and will be removed in rapidtools 0.2; use '
f"AssetAnalyzer(rapidtools.models.load('{provider}', ...), prompt=...) "
'instead.',
DeprecationWarning,
stacklevel=3,
)
class BaseAPIAssetAnalyzer(AssetAnalyzer):
"""
Deprecated base for hosted-API analyzers; builds the model from ``MODEL_CLASS``.
Prefer :class:`AssetAnalyzer` with :func:`rapidtools.models.load`.
Example:
>>> analyzer = GeminiAssetAnalyzer(api_key='...', prompt='Rate it 0-5.')
>>> collection = analyzer(collection)
"""
MODEL_CLASS: type = GeminiInference
ATTRIBUTE_PREFIX = 'gemini'
PROVIDER_NAME = 'Gemini API'
DEFAULT_MODEL_ID: str | None = None
def __init__(
self,
api_key: str | Path | None = None,
prompt: str | Path = '',
model_id: str | None = None,
max_workers: int | None = 5,
max_retries: int | None = 3,
cooldown_duration: int | None = 30,
system_instruction: str | None = None,
temperature: float = 0.4,
max_tokens: int = 2048,
image_filter: Callable[[ImageAsset], bool] | None = None,
cancel_event: threading.Event | None = None,
):
"""
Build the provider model and wrap it in an :class:`AssetAnalyzer`.
Args:
api_key: Provider API key (or path to a file containing it).
prompt: Text prompt or path to a prompt file.
model_id: Provider model ID. Defaults to the provider's default.
max_workers: Concurrent threads. Defaults to 5.
max_retries: Retries per individual request. Defaults to 3.
cooldown_duration: Base seconds to wait on rate limit. Defaults to 30.
system_instruction: Optional model persona/constraints.
temperature: Randomness (0.0 to 2.0).
max_tokens: Maximum output length.
image_filter: Optional function to filter ImageAssets.
cancel_event: Optional ``threading.Event``; when set, the analyzer
stops submitting requests and raises ``OperationCancelled``.
"""
_warn_deprecated(type(self).__name__, type(self).ATTRIBUTE_PREFIX)
self.max_retries = max_retries
model_kwargs: dict[str, Any] = dict(
api_key=api_key,
max_workers=max_workers,
max_retries=max_retries,
system_instruction=system_instruction,
temperature=temperature,
max_tokens=max_tokens,
)
chosen = model_id or self.DEFAULT_MODEL_ID
if chosen:
model_kwargs['model_id'] = chosen
model = self.MODEL_CLASS(**model_kwargs)
super().__init__(
model,
prompt=prompt,
attribute_prefix=type(self).ATTRIBUTE_PREFIX,
max_workers=max_workers,
image_filter=image_filter,
cancel_event=cancel_event,
rate_limit=RateLimitPolicy(cooldown_duration or 0),
mode='api',
)
[docs]
class GeminiAssetAnalyzer(BaseAPIAssetAnalyzer):
"""Deprecated: use ``AssetAnalyzer(load('gemini', ...))``. Prefix ``gemini_``."""
MODEL_CLASS = GeminiInference
ATTRIBUTE_PREFIX = 'gemini'
PROVIDER_NAME = 'Gemini API'
DEFAULT_MODEL_ID = GEMINI_ANALYZER_DEFAULT_MODEL
[docs]
class ClaudeAssetAnalyzer(BaseAPIAssetAnalyzer):
"""Deprecated: use ``AssetAnalyzer(load('claude', ...))``. Prefix ``claude_``."""
MODEL_CLASS = ClaudeInference
ATTRIBUTE_PREFIX = 'claude'
PROVIDER_NAME = 'Claude API'
DEFAULT_MODEL_ID = ANTHROPIC_DEFAULT_MODEL
[docs]
class OpenAIAssetAnalyzer(BaseAPIAssetAnalyzer):
"""Deprecated: use ``AssetAnalyzer(load('openai', ...))``. Prefix ``openai_``."""
MODEL_CLASS = OpenAIInference
ATTRIBUTE_PREFIX = 'openai'
PROVIDER_NAME = 'OpenAI API'
DEFAULT_MODEL_ID = OPENAI_DEFAULT_MODEL
[docs]
class MuseSparkAssetAnalyzer(BaseAPIAssetAnalyzer):
"""Deprecated: use ``AssetAnalyzer(load('muse_spark', ...))``."""
MODEL_CLASS = MuseSparkInference
ATTRIBUTE_PREFIX = 'muse_spark'
PROVIDER_NAME = 'Meta Model API'
DEFAULT_MODEL_ID = MUSE_SPARK_DEFAULT_MODEL
[docs]
class QwenAssetAnalyzer(BaseAPIAssetAnalyzer):
"""Deprecated: use ``AssetAnalyzer(load('qwen', ...))``. Prefix ``qwen_``."""
MODEL_CLASS = QwenInference
ATTRIBUTE_PREFIX = 'qwen'
PROVIDER_NAME = 'Qwen API'
DEFAULT_MODEL_ID = QWEN_DEFAULT_MODEL
class BaseLocalAssetAnalyzer(AssetAnalyzer):
"""
Compatibility alias for :class:`AssetAnalyzer` in local (sequential) mode.
Example:
>>> analyzer = BaseLocalAssetAnalyzer(Gemma4Inference(), prompt='Rate it.')
>>> collection = analyzer(collection)
"""
ATTRIBUTE_PREFIX = 'vlm'
PROVIDER_NAME = 'Local VLM'
def __init__(
self,
model: Any,
prompt: str | Path = '',
max_images_per_asset: int = 2,
batch_size: int = 4,
image_filter: Callable[[ImageAsset], bool] | None = None,
cancel_event: threading.Event | None = None,
attribute_prefix: str | None = None,
) -> None:
"""
Wrap an already-loaded local model.
Args:
model: A local inference model exposing ``run_inference`` (and
optionally ``run_inference_batch``) plus a ``model_id``.
prompt: Text instruction or path to a prompt file.
max_images_per_asset: Safeguard limit on images passed per asset.
batch_size: Assets processed simultaneously when the model
supports batching.
image_filter: Optional function to filter ``ImageAsset`` objects.
cancel_event: Optional ``threading.Event`` checked between batches.
attribute_prefix: Overrides the class-level ``ATTRIBUTE_PREFIX``.
"""
super().__init__(
model,
prompt=prompt,
attribute_prefix=attribute_prefix or type(self).ATTRIBUTE_PREFIX,
batch_size=batch_size,
max_images_per_asset=max_images_per_asset,
image_filter=image_filter,
cancel_event=cancel_event,
mode='local',
)
[docs]
class Gemma4AssetAnalyzer(BaseLocalAssetAnalyzer):
"""Deprecated: use ``AssetAnalyzer(load('gemma4', ...))``. Prefix ``gemma4_``."""
ATTRIBUTE_PREFIX = 'gemma4'
PROVIDER_NAME = 'Gemma-4'
def __init__(
self,
prompt: str | Path = '',
model_id: str = GEMMA4_DEFAULT_MODEL,
device: str = 'auto',
temperature: float = 0.4,
max_tokens: int = 512,
max_images_per_asset: int = 2,
batch_size: int = 4,
image_filter: Callable[[ImageAsset], bool] | None = None,
cancel_event: threading.Event | None = None,
) -> None:
"""Build a :class:`~rapidtools.models.Gemma4Inference` and wrap it."""
_warn_deprecated('Gemma4AssetAnalyzer', 'gemma4')
model = Gemma4Inference(
model_id=model_id,
device=device,
temperature=temperature,
max_tokens=max_tokens,
)
super().__init__(
model,
prompt=prompt,
max_images_per_asset=max_images_per_asset,
batch_size=batch_size,
image_filter=image_filter,
cancel_event=cancel_event,
)
[docs]
class LlamaVisionAssetAnalyzer(BaseLocalAssetAnalyzer):
"""Deprecated: use ``AssetAnalyzer(load('llama', ...))``. Prefix ``llama_``."""
ATTRIBUTE_PREFIX = 'llama'
PROVIDER_NAME = 'Llama Vision'
def __init__(
self,
prompt: str | Path = '',
model_id: str = LLAMA_DEFAULT_MODEL,
device: str = 'auto',
load_in_4bit: bool = True,
temperature: float = 0.4,
max_tokens: int = 512,
max_images_per_asset: int = 2,
image_filter: Callable[[ImageAsset], bool] | None = None,
cancel_event: threading.Event | None = None,
) -> None:
"""Build a :class:`~rapidtools.models.LlamaVisionInference` and wrap it."""
_warn_deprecated('LlamaVisionAssetAnalyzer', 'llama')
model = LlamaVisionInference(
model_id=model_id,
device=device,
load_in_4bit=load_in_4bit,
temperature=temperature,
max_tokens=max_tokens,
)
super().__init__(
model,
prompt=prompt,
max_images_per_asset=max_images_per_asset,
batch_size=1,
image_filter=image_filter,
cancel_event=cancel_event,
)
[docs]
class HFVisionAssetAnalyzer(BaseLocalAssetAnalyzer):
"""
Deprecated: use ``AssetAnalyzer(load('hf', ...))``. Prefix ``vlm_``.
Subclasses swap ``MODEL_CLASS`` for a family-specific wrapper.
"""
MODEL_CLASS: type[HFVisionInference] = HFVisionInference
ATTRIBUTE_PREFIX = 'vlm'
PROVIDER_NAME = 'Hugging Face VLM'
def __init__(
self,
prompt: str | Path = '',
model_id: str | None = DEFAULT_OPEN_VLM,
device: str = 'auto',
load_in_4bit: bool = False,
temperature: float = 0.4,
max_tokens: int = 512,
max_images_per_asset: int = 2,
batch_size: int = 1,
image_filter: Callable[[ImageAsset], bool] | None = None,
cancel_event: threading.Event | None = None,
attribute_prefix: str | None = None,
trust_remote_code: bool = False,
) -> None:
"""Build ``MODEL_CLASS`` and wrap it."""
info = getattr(self.MODEL_CLASS, 'INFO', None)
_warn_deprecated(type(self).__name__, info.key if info else 'hf')
model = self.MODEL_CLASS(
model_id=model_id or self.MODEL_CLASS.DEFAULT_MODEL,
device=device,
load_in_4bit=load_in_4bit,
temperature=temperature,
max_tokens=max_tokens,
trust_remote_code=trust_remote_code,
)
super().__init__(
model,
prompt=prompt,
max_images_per_asset=max_images_per_asset,
batch_size=batch_size,
image_filter=image_filter,
cancel_event=cancel_event,
attribute_prefix=attribute_prefix,
)
[docs]
class MuseGlimmerAssetAnalyzer(HFVisionAssetAnalyzer):
"""Deprecated: use ``AssetAnalyzer(load('muse_glimmer', ...))``."""
MODEL_CLASS = MuseGlimmerInference
ATTRIBUTE_PREFIX = 'muse_glimmer'
PROVIDER_NAME = 'Muse Glimmer'
def __init__(
self,
prompt: str | Path = '',
model_id: str = MUSE_GLIMMER_DEFAULT_MODEL,
device: str = 'auto',
load_in_4bit: bool = True,
temperature: float = 0.4,
max_tokens: int = 512,
max_images_per_asset: int = 2,
batch_size: int = 1,
image_filter: Callable[[ImageAsset], bool] | None = None,
cancel_event: threading.Event | None = None,
attribute_prefix: str | None = None,
) -> None:
"""Build a :class:`~rapidtools.models.MuseGlimmerInference` and wrap it."""
super().__init__(
prompt=prompt,
model_id=model_id,
device=device,
load_in_4bit=load_in_4bit,
temperature=temperature,
max_tokens=max_tokens,
max_images_per_asset=max_images_per_asset,
batch_size=batch_size,
image_filter=image_filter,
cancel_event=cancel_event,
attribute_prefix=attribute_prefix,
)
[docs]
class QwenVisionAssetAnalyzer(HFVisionAssetAnalyzer):
"""Deprecated: use ``AssetAnalyzer(load('qwen_vl', ...))``. Prefix ``qwen_vl_``."""
MODEL_CLASS = QwenVisionInference
ATTRIBUTE_PREFIX = 'qwen_vl'
PROVIDER_NAME = 'Qwen (local)'
def __init__(
self,
prompt: str | Path = '',
model_id: str = QWEN_VL_DEFAULT_MODEL,
device: str = 'auto',
load_in_4bit: bool = False,
temperature: float = 0.4,
max_tokens: int = 512,
max_images_per_asset: int = 2,
batch_size: int = 1,
image_filter: Callable[[ImageAsset], bool] | None = None,
cancel_event: threading.Event | None = None,
attribute_prefix: str | None = None,
) -> None:
"""Build a :class:`~rapidtools.models.QwenVisionInference` and wrap it."""
super().__init__(
prompt=prompt,
model_id=model_id,
device=device,
load_in_4bit=load_in_4bit,
temperature=temperature,
max_tokens=max_tokens,
max_images_per_asset=max_images_per_asset,
batch_size=batch_size,
image_filter=image_filter,
cancel_event=cancel_event,
attribute_prefix=attribute_prefix,
)
