Backends

Backend plugin implementations.

@domain backends @see [[knowledge/architecture/backends]]

Mock

Mock backend exports.

@domain backends @see [[knowledge/architecture/backends]] @tested-by [[tests/backends/test_mock.py]] @tested-by [[tests/backends/test_mock_audio.py]]

Deterministic mock backend for CI-safe E2E tests.

@domain backends @see [[knowledge/architecture/backends]] @tested-by [[tests/backends/test_mock.py]]

class mllm_shap.backends.mock.backend.MockBackend(context: PluginContext | None = None, *, model_id: str = 'mock')

Bases: object

Hash-based generative backend with no external model dependencies.

options_model

Accepts no explanation.plugin_options.

alias of PluginOptions

property model_id: str

Return the backend model identifier.

async generate(snapshot: ConversationSnapshot, generation: GenerationConfig) → GenerationRecord

Return deterministic generation output for snapshot.

Deterministic mock backend that hears the voiced runs of audio parts.

@domain backends @see [[knowledge/architecture/backends]] @tested-by [[tests/backends/test_mock_audio.py]]

mllm_shap.backends.mock.audio.SILENCE = 0.01

Largest absolute amplitude (of full scale 1) that counts as silence.

mllm_shap.backends.mock.audio.MIN_GAP_S = 0.01

Shortest silence that separates two voiced runs; shorter dips (zero crossings) stay inside one run.

class mllm_shap.backends.mock.audio.MockAudioBackend(context: PluginContext | None = None, *, model_id: str = 'mock')

Bases: MockBackend

Echo text words and answer one token per voiced run of every audio part.

A run’s token hashes its 16-bit samples, so silencing or cutting a run drops exactly its token and leaves the others unchanged. 16-bit PCM WAV decodes with the standard library; other formats need the audio extra.

async generate(snapshot: ConversationSnapshot, generation: GenerationConfig) → GenerationRecord

Return the heard words of snapshot, at most max_new_tokens.

OpenAI-compatible HTTP API

OpenAI-compatible HTTP API backend plugin.

@domain backends @see [[knowledge/architecture/backends]] @tested-by [[tests/backends/test_api.py]]

OpenAI-compatible HTTP generative backend.

@domain backends @see [[knowledge/architecture/backends]] @tested-by [[tests/backends/test_api.py]]

mllm_shap.backends.api.backend.RESPONSE_CACHE_SIZE = 1024

Maximum deterministic responses kept in memory per backend instance.

mllm_shap.backends.api.backend.RETRYABLE_STATUS = frozenset({408, 429, 500, 502, 503, 504})

HTTP status codes treated as transient and retried.

mllm_shap.backends.api.backend.MAX_RETRY_DELAY_S = 8.0

Upper bound of the exponential backoff between retries.

mllm_shap.backends.api.backend.OPENAI_API_HOST = 'api.openai.com'

The only host that receives OPENAI_API_KEY.

class mllm_shap.backends.api.backend.ApiBackend(context: PluginContext, transport: Any | None = None)

Bases: object

Generate text through an OpenAI-compatible /chat/completions endpoint.

options_model

Accepts no explanation.plugin_options.

alias of PluginOptions

property model_id: str

Return the configured remote model identifier.

property returns_logprobs: bool

Always true; requests ask for logprobs, servers that omit them give None.

async generate(snapshot: ConversationSnapshot, generation: GenerationConfig) → GenerationRecord

Generate assistant text for snapshot via the HTTP API.

async aclose() → None

Close the underlying HTTP client.

Chat-completions request and response helpers.

@domain backends @see [[knowledge/architecture/backends]] @tested-by [[tests/backends/test_api.py]]

mllm_shap.backends.api.payload.snapshot_to_api_messages(snapshot: ConversationSnapshot) → list[dict[str, Any]]

Convert snapshot to chat messages; audio becomes input_audio parts.

A message with audio sends its text part first, then one base64 WAV part per AudioPart; non-WAV audio is re-encoded (audio extra).

mllm_shap.backends.api.payload.build_chat_payload(model_id: str, messages: list[dict[str, Any]], generation: GenerationConfig) → dict[str, Any]

Build an OpenAI-style chat-completions JSON body that asks for logprobs.

mllm_shap.backends.api.payload.payload_cache_key(payload: dict[str, Any]) → str

Return a stable SHA256 key for deterministic request deduplication.

mllm_shap.backends.api.payload.extract_message_content(data: dict[str, Any]) → str

Parse assistant text from a chat-completions JSON response.

mllm_shap.backends.api.payload.extract_logprobs(data: dict[str, Any]) → tuple[float, ...] | None

Return choices[0].logprobs.content[*].logprob, or None when absent.

Servers that ignore the logprobs request field (or send another shape) give None; the logprob value function then refuses the record.

LM Studio

LM Studio backend plugin.

@domain backends @see [[knowledge/architecture/backends]] @tested-by [[tests/backends/test_lmstudio.py]]

LM Studio generative backend.

@domain backends @see [[knowledge/architecture/backends]] @tested-by [[tests/backends/test_lmstudio.py]]

class mllm_shap.backends.lmstudio.backend.LmStudioBackend(context: PluginContext)

Bases: object

Generate text through a local LM Studio API server.

options_model

Accepts no explanation.plugin_options.

alias of PluginOptions

property options: LmStudioOptions

Return the validated LM Studio options.

property model_id: str

Return the configured LM Studio model identifier.

async generate(snapshot: ConversationSnapshot, generation: GenerationConfig) → GenerationRecord

Generate assistant text for snapshot via LM Studio.

async aclose() → None

Close the underlying LM Studio client connection.

Convert conversation snapshots to LM Studio chat contexts.

@domain backends @see [[knowledge/architecture/backends]] @tested-by [[tests/backends/test_lmstudio_chat.py]]

mllm_shap.backends.lmstudio.chat.snapshot_to_chat(snapshot: ConversationSnapshot, chat_factory: Any) → Any

Build an LM Studio chat object from snapshot turns.

Resolve and download LM Studio models through the lmstudio SDK.

@domain backends @see [[knowledge/architecture/backends]] @tested-by [[tests/backends/test_lmstudio_models.py]]

mllm_shap.backends.lmstudio.models.normalize_model_token(value: str) → str

Normalize a model identifier for fuzzy comparison.

mllm_shap.backends.lmstudio.models.match_downloaded_model_key(downloaded: Sequence[Any], model_id: str) → str | None

Return the downloaded model_key for model_id, if any.

Candidates are tried in three stages, each on normalized names: an exact match, then a key whose last / segment starts with model_id, then a key containing model_id. The first stage with a single match wins.

Raises:

BackendUnavailableError – If a stage matches more than one key.

mllm_shap.backends.lmstudio.models.pick_search_result(results: Sequence[Any], model_id: str) → Any

Choose the best repository search hit for model_id.

Raises:

BackendUnavailableError – If there are no hits or none is flagged exact or has a name matching model_id.

mllm_shap.backends.lmstudio.models.pick_download_option(options: Sequence[Any], quantization: str) → Any

Choose a repository download option for the requested quantization.

async mllm_shap.backends.lmstudio.models.resolve_model_key(client: Any, model_id: str, options: LmStudioOptions) → str

Return a loadable LM Studio model key, downloading when configured.

Settings of the LM Studio backend read from BackendConfig.options.

@domain backends @see [[knowledge/architecture/backends]] @tested-by [[tests/backends/test_lmstudio_options.py]]

class mllm_shap.backends.lmstudio.options.LmStudioOptions(*, hub_search: str | None = None, quantization: str = 'Q4_K_M', auto_download: bool = True, serialize_generate: bool = True, audio: Literal['refuse', 'transcript'] = 'refuse')

Bases: BaseModel

LM Studio settings accepted under backend.options.

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

hub_search: str | None

Repository search term used to download the model; defaults to model_id.

quantization: str

GGUF quantization to download when the model is not already local.

auto_download: bool

Download missing models through the LM Studio repository API.

serialize_generate: bool

Serialize generate calls (needed for MLX models in LM Studio).

audio: Literal['refuse', 'transcript']

What to do with audio parts, which the LM Studio SDK cannot send. refuse raises ValueError; transcript sends the text parts only.

mllm_shap.backends.lmstudio.options.parse_lmstudio_options(options: Mapping[str, object]) → LmStudioOptions

Validate backend.options for the LM Studio backend.

Raises:

ConfigError – If an option is unknown or has the wrong type.

Hugging Face transformers

Hugging Face transformers text backend.

@domain backends @see [[knowledge/architecture/backends]]

Hugging Face language model backend for text and, with a processor, audio.

@domain backends @see [[knowledge/architecture/backends]] @tested-by [[tests/backends/test_transformers_text.py]]

class mllm_shap.backends.transformers.text.TransformersTextBackend(context: PluginContext)

Bases: object

Generate text with a local Hugging Face language model.

A checkpoint AutoModelForCausalLM cannot load (or any backend.options.model_class) loads its own class with AutoProcessor and also takes audio parts (see audio.processor_inputs).

options_model

Accepts no explanation.plugin_options.

alias of PluginOptions

property model_id: str

Return the configured Hugging Face model identifier.

property returns_logprobs: bool

Always true; every record carries its generated tokens’ logprobs.

property dtype: Literal['auto', 'float32', 'float16', 'bfloat16']

Return the configured weight dtype (backend.options.dtype).

property kv_cache_hits: int

Return the number of prefix-cache hits during this backend lifetime.

property kv_cache_reused_tokens: int

Return the prompt tokens served from cached prefixes, summed over hits.

set_kv_cache_enabled(enabled: bool, max_entries: int) → None

Enable or disable prefix-cache reuse, keeping at most max_entries.

async generate(snapshot: ConversationSnapshot, generation: GenerationConfig) → GenerationRecord

Generate assistant text for snapshot with optional prefix caching.

Raises:

ValueError – If snapshot has audio and the model has no processor.

async aclose() → None

Release loaded model references.

Model inputs for audio language models through a Hugging Face processor.

@domain backends @see [[knowledge/architecture/backends]] @tested-by [[tests/backends/test_transformers_audio.py]]

mllm_shap.backends.transformers.audio.processor_inputs(processor: Any, snapshot: ConversationSnapshot) → Any

Encode snapshot with its audio for processor.

A processor whose chat template mentions audio gets structured audio and text parts; otherwise each audio part becomes processor.audio_token in the rendered prompt and the waveforms go to processor(text=..., audio=...). Audio parts come before the message text, and an audio part left empty by masking is dropped.

Conversation snapshot helpers for Hugging Face chat templates.

@domain backends @see [[knowledge/architecture/backends]] @tested-by [[tests/backends/test_transformers_chat.py]]

mllm_shap.backends.transformers.chat.render_prompt(tokenizer: Any, messages: list[dict[str, str]]) → str

Render chat messages to a model prompt string.

Embedding helpers for the transformers text backend.

@domain backends @see [[knowledge/architecture/backends]]

mllm_shap.backends.transformers.embeddings.static_embedding(model: Any, token_ids: list[int]) → tuple[float, ...]

Mean-pool input embeddings for generated token ids.

mllm_shap.backends.transformers.embeddings.contextual_embedding(model: Any, token_ids: list[int]) → tuple[float, ...]

Mean-pool the final hidden state for generated token ids.

Settings of the transformers backend read from BackendConfig.options.

@domain backends @see [[knowledge/architecture/backends]]

mllm_shap.backends.transformers.options.TorchDtype

Weight dtypes accepted by backend.options.dtype.

alias of Literal[‘auto’, ‘float32’, ‘float16’, ‘bfloat16’]

class mllm_shap.backends.transformers.options.TransformersOptions(*, revision: str | None = None, dtype: Literal['auto', 'float32', 'float16', 'bfloat16'] = 'auto', model_class: str | None = None, attn_implementation: str | None = None)

Bases: BaseModel

Transformers settings accepted under backend.options.

model_config = {'extra': 'forbid', 'frozen': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

revision: str | None

Hub revision to load (branch, tag or commit hash); pin a commit hash for reproducible runs. None loads the default branch.

dtype: Literal['auto', 'float32', 'float16', 'bfloat16']

Weight dtype; auto picks bfloat16 on accelerators that support it, float16 on other accelerators and float32 on CPU.

model_class: str | None

transformers class to load with the checkpoint’s processor, for example Gemma3nForConditionalGeneration. None uses AutoModelForCausalLM when it knows the checkpoint, else the checkpoint’s architecture.

attn_implementation: str | None

Attention kernel passed to from_pretrained (eager, sdpa, …); None keeps the model’s default. Granite Speech on Apple MPS needs eager: its projector fails under the default kernel.

mllm_shap.backends.transformers.options.parse_transformers_options(options: Mapping[str, object]) → TransformersOptions

Validate backend.options for the transformers backend.

Raises:

ConfigError – If an option is unknown or has the wrong value.