cjm-transcription-plugin-voxtral-hf 0.0.45

Mistral Voxtral plugin for the cjm-transcription-plugin-system library - provides local speech-to-text transcription through 🤗 Transformers with configurable model selection and parameter control.

cjm-transcription-plugin-voxtral-hf

Install

pip install cjm_transcription_plugin_voxtral_hf

Project Structure

nbs/
└── plugin.ipynb # Plugin implementation for Mistral Voxtral transcription through Hugging Face Transformers

Total: 1 notebook

Module Dependencies

graph LR
    plugin["plugin<br/>Voxtral HF Plugin"]

No cross-module dependencies detected.

CLI Reference

No CLI commands found in this project.

Module Overview

Detailed documentation for each module in the project:

Voxtral HF Plugin (plugin.ipynb)

Plugin implementation for Mistral Voxtral transcription through Hugging Face Transformers

Import

from cjm_transcription_plugin_voxtral_hf.plugin import (
    VoxtralHFPluginConfig,
    VoxtralHFPlugin
)

Functions

@patch
def _apply_config(
    self:VoxtralHFPlugin,
    config: Optional[Any] = None # Configuration dataclass, dict, or None
) -> None
    """
    CR-4: apply config + derive config-dependent state (device, dtype). No
    heavy-resource work. Called by initialize (first-time) and the substrate's
    reconfigure delta path. Model release on a model_id/device/dtype/quantization
    change is handled declaratively via RELOAD_TRIGGER -> _release_model.
    """

@patch
def _release_model(self:VoxtralHFPlugin) -> None:
    """Unload the current model + processor and free GPU memory.

    Delegates to cjm-torch-plugin-utils' `release_model` (move-to-CPU / del / gc /
    empty_cache / synchronize) -- the single source of truth across torch GPU plugins."""
    if self.model is None and self.processor is None
    """
    Unload the current model + processor and free GPU memory.
    
    Delegates to cjm-torch-plugin-utils' `release_model` (move-to-CPU / del / gc /
    empty_cache / synchronize) -- the single source of truth across torch GPU plugins.
    """

@patch
def _load_model(self:VoxtralHFPlugin) -> None:
    """Load the Voxtral model + processor (lazy).

    The heartbeat wraps BOTH the (potentially long, often quiet) snapshot download
    AND the silent from_pretrained build, so the substrate's prefetch stall detector
    always sees the (progress, message) tuple advance. snapshot_download_with_progress
    layers real per-file download % on top when the HF Hub tqdm callback fires.
    CUDA OOM on load surfaces as a typed PluginResourceError for CR-7 reactive retry."""
    if self.model is not None and self.processor is not None
    """
    Load the Voxtral model + processor (lazy).
    
    The heartbeat wraps BOTH the (potentially long, often quiet) snapshot download
    AND the silent from_pretrained build, so the substrate's prefetch stall detector
    always sees the (progress, message) tuple advance. snapshot_download_with_progress
    layers real per-file download % on top when the HF Hub tqdm callback fires.
    CUDA OOM on load surfaces as a typed PluginResourceError for CR-7 reactive retry.
    """

@patch
def _prepare_audio(
    self:VoxtralHFPlugin,
    audio: Union[str, Path] # Path to a decodable audio file
) -> str: # The audio file path
    """
    Validate the audio input and return it as a path string.
    
    The caller (orchestration / proxy) guarantees a model-ready audio file;
    in-memory preparation is no longer a plugin responsibility.
    """

@patch
def is_available(self:VoxtralHFPlugin) -> bool: # True if Voxtral and its dependencies are available
    "Check if Voxtral is available."

@patch
def prefetch(self:VoxtralHFPlugin) -> None
    """
    CR-4 (SG-19): eagerly load the model + processor so the first execute()
    doesn't pay the download/load cost. Idempotent via _load_model's None-guard.
    """

@patch
def on_disable(self:VoxtralHFPlugin) -> None
    """
    CR-2: release the GPU model + processor when the operator disables the
    plugin (the worker stays alive); lazy reload on the next execute.
    """

@patch
def cleanup(self:VoxtralHFPlugin) -> None
    "Release the model + processor (CR-4: delegates to `_release_model`)."

Classes

@dataclass
class VoxtralHFPluginConfig(HFCacheConfig):
    "Configuration for Voxtral HF transcription plugin."
    
    model_id: str = field(...)
    device: str = field(...)
    dtype: str = field(...)
    language: Optional[str] = field(...)
    max_new_tokens: int = field(...)
    do_sample: bool = field(...)
    temperature: float = field(...)
    top_p: float = field(...)
    compile_model: bool = field(...)
    load_in_8bit: bool = field(...)
    load_in_4bit: bool = field(...)

class VoxtralHFPlugin:
    def __init__(self):
        """Initialize the Voxtral HF plugin with default configuration."""
        self.logger = logging.getLogger(f"{__name__}.{type(self).__name__}")
        self.config: VoxtralHFPluginConfig = None
    """
    Mistral Voxtral transcription plugin via Hugging Face Transformers (stage 8: pure-compute tool capability).
    
    Native-surface model (PILLAR 1c): this tool is PURE COMPUTE — `transcribe`
    loads the model, runs inference, and builds the typed `TranscriptionResult`.
    The cache-check + persistence bookends + the per-call `force` control live in
    the generic transcription adapter (cjm-transcription-adapter-interface); the
    result DTO lives in cjm-capability-primitives; identity is derived from the
    installed distribution. No `get_plugin_metadata`, no `self.storage`.
    """
    
    def __init__(self):
            """Initialize the Voxtral HF plugin with default configuration."""
            self.logger = logging.getLogger(f"{__name__}.{type(self).__name__}")
            self.config: VoxtralHFPluginConfig = None
        "Initialize the Voxtral HF plugin with default configuration."
    
    def name(self) -> str: # Plugin name identifier
            """Plugin identity, derived from the installed distribution (PILLAR 1c).
    
            Runtime-derived: in the worker / in-env introspection `__package__`
            resolves; the manifest records the same value independently (the
            dual-mode generator reads it from the distribution)."""
            from importlib.metadata import metadata, packages_distributions
            dist = (packages_distributions().get(__package__) or [__package__.replace("_", "-")])[0]
            return metadata(dist)["Name"]
    
        @property
        def version(self) -> str: # Plugin version string
        "Plugin identity, derived from the installed distribution (PILLAR 1c).

Runtime-derived: in the worker / in-env introspection `__package__`
resolves; the manifest records the same value independently (the
dual-mode generator reads it from the distribution)."
    
    def version(self) -> str: # Plugin version string
            """Get the plugin version string."""
            from cjm_transcription_plugin_voxtral_hf import __version__
            return __version__
    
        def get_current_config(self) -> Dict[str, Any]: # Current configuration as dictionary
        "Get the plugin version string."
    
    def get_current_config(self) -> Dict[str, Any]: # Current configuration as dictionary
            """Return current configuration state."""
            if not self.config
        "Return current configuration state."
    
    def get_config_schema(self) -> Dict[str, Any]: # JSON Schema for configuration
            """Return JSON Schema for UI generation."""
            return dataclass_to_jsonschema(VoxtralHFPluginConfig)
    
        @staticmethod
        def get_config_dataclass() -> VoxtralHFPluginConfig: # Configuration dataclass
        "Return JSON Schema for UI generation."
    
    def get_config_dataclass() -> VoxtralHFPluginConfig: # Configuration dataclass
            """Return dataclass describing the plugin's configuration options."""
            return VoxtralHFPluginConfig
    
        def initialize(
            self,
            config: Optional[Any] = None # Configuration dataclass, dict, or None
        ) -> None
        "Return dataclass describing the plugin's configuration options."
    
    def initialize(
            self,
            config: Optional[Any] = None # Configuration dataclass, dict, or None
        ) -> None
        "First-time setup. CR-4: the manual model/device/dtype/quantization
diff-and-reload is replaced by declarative RELOAD_TRIGGER metadata; the
substrate's reconfigure path fires _release_model then re-applies config."
    
    def transcribe(
            self,
            audio: Union[str, Path], # Path to MODEL-READY audio (converted upstream)
            **kwargs # Provenance (source_start_time/source_end_time) stamped into metadata
        ) -> TranscriptionResult: # Typed transcription output
        "Transcribe model-ready audio using Voxtral — PURE COMPUTE.

Stage 8 / PILLAR 1c: the cache-check + persistence bookends moved to the
generic transcription adapter; this method loads the model, runs
inference, and builds the typed result. Model params come from
`self.config` (the CR-15 per-call override path is gone — the tool runs
its effective config, no metadata lie); `source_start_time` /
`source_end_time` ride the provenance kwarg channel into metadata."