BaseInferenceModel

class rapidtools.models.BaseInferenceModel

Bases: ABC

The top-level contract for all AI models (API or local).

Subclasses implement run_inference(); run_batch() provides a threaded default suitable for network-bound API models (local models override it to run sequentially).

Example

A minimal fake model useful in tests and pipelines:

>>> from rapidtools.models.base import BaseInferenceModel, ModelOutput
>>> class EchoModel(BaseInferenceModel):
...     model_id = 'echo'
...     def run_inference(self, image_inputs, prompt, **kwargs):
...         return ModelOutput(text=f'{len(image_inputs)} image(s): {prompt}')
>>> EchoModel().run_inference(['a.jpg'], 'hi').text
'1 image(s): hi'

Methods

run_batch	Run run_inference() over many assets with a thread pool.
run_inference	Execute a single inference pass.

abstractmethod run_inference(image_inputs: Any, prompt: str | Path, **kwargs) → ModelOutput | None

Execute a single inference pass.

Parameters:

• image_inputs – One image path/URL or a list of them.

• prompt – Prompt text or path to a prompt file.

• **kwargs – Provider-specific options (temperature, max_tokens, json_mode, …).

Returns:

ModelOutput | None – The standardized output, or None on failure.

run_batch(asset_inputs: Iterable[tuple[str, Any]], prompt: str | Path, **kwargs) → Generator[tuple[str, str, ModelOutput | str], None, None]

Run run_inference() over many assets with a thread pool.

Uses min(self.max_workers, 10) threads (default 10 when the model defines no max_workers).

Parameters:

• asset_inputs – (asset_id, image_inputs) pairs.

• prompt – Prompt text or file shared by every asset.

• **kwargs – Forwarded to run_inference().

Yields:

tuple[str, str, ModelOutput | str] – (asset_id, status, payload) where status is 'ok' (payload is a ModelOutput) or 'failed' (payload is an error message).

Example

>>> from rapidtools.models import GeminiInference
>>> model = GeminiInference(api_key='AIza...')
>>> results = dict(
...     (a, out) for a, status, out in model.run_batch(
...         [('b1', ['b1.jpg']), ('b2', ['b2.jpg'])], 'Rate damage 0-5.'
...     ) if status == 'ok'
... )
>>> sorted(results)
['b1', 'b2']