moss-audio-tn-intent

MOSS-Audio Tunisian Intent Classifier

A lightweight audio intent classifier for Tunisian Arabic built on top of the frozen OpenMOSS-Team/MOSS-Audio-8B-Instruct audio encoder and DeepStack feature merger.

The model achieves 67.4% accuracy on the SLURP-TN held-out evaluation set across 18 intent classes — well above majority-class baseline (~18%) and within striking distance of published Arabic intent-classification SOTA, despite using only ~0.79M trainable parameters (the rest of the 5.2B-parameter pipeline is frozen).

This validates the frozen-MOSS + DeepStack + lightweight head architecture for closed-set Arabic audio classification tasks where end-to-end MOSS generation falls short.

---

Architecture

                ┌─────────────────────────────────────────┐
   raw audio ──>│  MOSS-Audio audio encoder (FROZEN)      │
   16kHz mono   │  32 Whisper-style encoder layers, 1280-d│
                └─────┬─────────┬─────────┬───────────────┘
                      │ layer10 │ layer21 │ layer31
                      ▼         ▼         ▼
                ┌───────────────────────────────────────┐
                │  DeepStack mergers (FROZEN)           │
                │  3 × GatedMLP, each 1280 → 2560-d     │
                └─────┬─────────┬─────────┬─────────────┘
                      │         │         │
                      └─────sum─┴───sum───┘
                                ▼
                       ┌──────────────────┐
                       │  Mean-pool over  │
                       │  time dimension  │
                       └────────┬─────────┘
                                ▼
                          [2560-d vector]
                                ▼
                ┌───────────────────────────────────────┐
                │  Classifier head (TRAINABLE, 0.79M)   │
                │  LayerNorm → Linear(2560,512) → GELU  │
                │     → Linear(512,256) → GELU          │
                │     → Linear(256, num_classes)        │
                └────────────────┬──────────────────────┘
                                 ▼
                          intent label

Key design choices:

---

Performance

Overall

Per-class accuracy

Sorted by sample count (high-resource classes first):

Pattern: classes with ≥20 training samples consistently hit 60–85% accuracy. Low-sample classes (n<5) cluster near 0% — these are not "broken" but simply under-trained. The SLURP-TN dataset has overlapping label pairs (quirky/general_quirky, joke/general_joke, etc.) that further depress the per-class number for the rare variants; merging these in a future training run would likely raise overall accuracy to ~73%.

The full confusion matrix is in confusion_matrix.csv.

---

Quick Start

Install dependencies

# MOSS-Audio package
git clone https://github.com/OpenMOSS/MOSS-Audio.git
cd MOSS-Audio
pip install -e .

# Other deps
pip install torch transformers huggingface_hub librosa

Minimal inference

import torch, torch.nn as nn, librosa, sys
from huggingface_hub import hf_hub_download

# 1. Make MOSS-Audio importable
sys.path.insert(0, "/path/to/MOSS-Audio")
from src.modeling_moss_audio import MossAudioModel
from src.processing_moss_audio import MossAudioProcessor

# 2. Apply Whisper kwarg-stripping patch (idempotent)
import transformers.models.whisper.modeling_whisper as whisper_mod
if not getattr(whisper_mod.WhisperEncoderLayer.forward, "_moss_patched", False):
    _orig = whisper_mod.WhisperEncoderLayer.forward
    def _patched(self, *a, **kw):
        kw.pop("layer_head_mask", None); kw.pop("head_mask", None)
        return _orig(self, *a, **kw)
    _patched._moss_patched = True
    whisper_mod.WhisperEncoderLayer.forward = _patched

# 3. Load the frozen base
MOSS_REPO = "OpenMOSS-Team/MOSS-Audio-8B-Instruct"
moss_model = MossAudioModel.from_pretrained(
    MOSS_REPO, trust_remote_code=True, dtype="auto",
    device_map="cuda:0", attn_implementation="sdpa",
).eval()
processor = MossAudioProcessor.from_pretrained(
    MOSS_REPO, trust_remote_code=True, enable_time_marker=True,
)

# 4. Download the classifier head
head_path = hf_hub_download(repo_id="FatimahEmadEldin/moss-audio-tn-intent",
                            filename="best.pt")

# 5. Load head + label maps
state = torch.load(head_path, weights_only=False, map_location="cuda:0")
ID2LABEL = state["ID2LABEL"]
NUM_CLASSES = state["num_classes"]

class ClassifierHead(nn.Module):
    def __init__(self, in_dim, num_classes, hidden=512, dropout=0.15):
        super().__init__()
        self.net = nn.Sequential(
            nn.LayerNorm(in_dim),
            nn.Linear(in_dim, hidden), nn.GELU(), nn.Dropout(dropout),
            nn.Linear(hidden, hidden // 2), nn.GELU(), nn.Dropout(dropout),
            nn.Linear(hidden // 2, num_classes),
        )
    def forward(self, x):
        return self.net(x)

head = ClassifierHead(state["feature_dim"], NUM_CLASSES).to("cuda:0")
head.load_state_dict(state["head_state_dict"])
head.eval()

# 6. Predict
@torch.no_grad()
def predict(audio_path):
    wav, _ = librosa.load(audio_path, sr=16000, mono=True)
    mel = processor(text="<|audio|>", audios=[wav], return_tensors="pt")
    audio_data = mel["audio_data"].to(moss_model.device, dtype=moss_model.dtype)

    enc_out = moss_model.audio_encoder(audio_data, output_hidden_states=True,
                                        return_dict=True)
    projected = []
    for merger_idx, layer_id in enumerate([10, 21, 31]):
        h = enc_out.hidden_states[layer_id]
        projected.append(moss_model.deepstack_audio_merger_list[merger_idx](h))
    agg = torch.stack(projected, dim=0).sum(dim=0)
    feat = agg.mean(dim=1).squeeze(0).float()

    logits = head(feat.unsqueeze(0))
    probs = torch.softmax(logits, dim=-1).squeeze(0)
    pred_id = int(probs.argmax())
    return {
        "intent": ID2LABEL[pred_id],
        "confidence": float(probs[pred_id]),
        "top3": [(ID2LABEL[i], float(probs[i]))
                 for i in probs.topk(3).indices.cpu().tolist()],
    }

result = predict("my_audio.wav")
print(result)
# Example output:
# {'intent': 'weather_query',
#  'confidence': 0.84,
#  'top3': [('weather_query', 0.84), ('news_query', 0.09), ('general_quirky', 0.03)]}

---

Detailed Inference (production-quality wrapper)

A more complete wrapper with audio validation, multiple input types, and structured output:

import torch
import torch.nn as nn
import librosa
import numpy as np
import sys, os
from pathlib import Path
from typing import Union, List, Dict
from huggingface_hub import hf_hub_download


class MossAudioIntentClassifier:
    """Tunisian Arabic intent classifier using frozen MOSS-Audio + DeepStack + head."""

    def __init__(
        self,
        moss_audio_repo_path: str = "/content/MOSS-Audio",
        classifier_repo: str = "FatimahEmadEldin/moss-audio-tn-intent",
        moss_base_repo: str = "OpenMOSS-Team/MOSS-Audio-8B-Instruct",
        device: str = "cuda:0",
        moss_weights_dir: str = None,
    ):
        """
        Args:
            moss_audio_repo_path: Local path to a clone of github.com/OpenMOSS/MOSS-Audio
            classifier_repo: HF repo holding the trained classifier head (best.pt)
            moss_base_repo: HF repo for the MOSS-Audio base model
            device: 'cuda:0' / 'cuda' / 'cpu'
            moss_weights_dir: Optional local cache dir for MOSS weights
        """
        self.device = device

        # 1. Import MOSS-Audio modules from local clone
        if moss_audio_repo_path not in sys.path:
            sys.path.insert(0, moss_audio_repo_path)
        from src.modeling_moss_audio import MossAudioModel
        from src.processing_moss_audio import MossAudioProcessor

        # 2. Apply idempotent Whisper kwarg-stripping patch
        # (MOSS-Audio uses WhisperEncoderLayer but passes kwargs newer
        #  transformers versions don't accept)
        self._patch_whisper()

        # 3. Load (or cache) MOSS-Audio base model
        if moss_weights_dir and os.path.exists(os.path.join(moss_weights_dir, "config.json")):
            base_path = moss_weights_dir
        else:
            base_path = moss_base_repo
        print(f"📥 Loading frozen MOSS-Audio from {base_path}")
        self.moss_model = MossAudioModel.from_pretrained(
            base_path, trust_remote_code=True, dtype="auto",
            device_map=device, attn_implementation="sdpa",
        ).eval()
        for p in self.moss_model.parameters():
            p.requires_grad = False

        self.processor = MossAudioProcessor.from_pretrained(
            base_path, trust_remote_code=True, enable_time_marker=True,
        )

        # 4. Download + load classifier head
        print(f"📥 Loading classifier head from {classifier_repo}")
        head_path = hf_hub_download(repo_id=classifier_repo, filename="best.pt")
        state = torch.load(head_path, weights_only=False, map_location=device)

        self.feature_dim = state["feature_dim"]
        self.num_classes = state["num_classes"]
        self.ID2LABEL = state["ID2LABEL"]
        self.LABEL2ID = state["LABEL2ID"]
        self.deepstack_layer_ids = state["config"]["DEEPSTACK_LAYER_IDS"]
        self.aggregation = state["config"]["AGGREGATION"]

        self.head = self._build_head(self.feature_dim, self.num_classes).to(device)
        self.head.load_state_dict(state["head_state_dict"])
        self.head.eval()

        print(f"✅ Ready. {self.num_classes} classes, "
              f"DeepStack layers {self.deepstack_layer_ids}, agg={self.aggregation}")

    @staticmethod
    def _patch_whisper():
        import transformers.models.whisper.modeling_whisper as whisper_mod
        if getattr(whisper_mod.WhisperEncoderLayer.forward, "_moss_patched", False):
            return
        _orig = whisper_mod.WhisperEncoderLayer.forward
        def _patched(self, *args, **kwargs):
            kwargs.pop("layer_head_mask", None)
            kwargs.pop("head_mask", None)
            return _orig(self, *args, **kwargs)
        _patched._moss_patched = True
        whisper_mod.WhisperEncoderLayer.forward = _patched

    @staticmethod
    def _build_head(in_dim, num_classes, hidden=512, dropout=0.15):
        return nn.Sequential(
            nn.LayerNorm(in_dim),
            nn.Linear(in_dim, hidden), nn.GELU(), nn.Dropout(dropout),
            nn.Linear(hidden, hidden // 2), nn.GELU(), nn.Dropout(dropout),
            nn.Linear(hidden // 2, num_classes),
        )

    def _load_audio(self, audio_input: Union[str, np.ndarray, Dict]) -> np.ndarray:
        """Accepts file path / numpy array / HF Datasets Audio dict."""
        if isinstance(audio_input, (str, Path)):
            wav, _ = librosa.load(str(audio_input), sr=16000, mono=True)
            return wav.astype(np.float32)
        if isinstance(audio_input, dict) and "array" in audio_input:
            wav = np.asarray(audio_input["array"], dtype=np.float32)
            sr = audio_input["sampling_rate"]
            if wav.ndim > 1:
                wav = wav.mean(axis=0)
            if sr != 16000:
                import torchaudio
                wav = torchaudio.functional.resample(
                    torch.from_numpy(wav).unsqueeze(0),
                    orig_freq=sr, new_freq=16000,
                ).squeeze(0).numpy()
            return wav
        if isinstance(audio_input, np.ndarray):
            return audio_input.astype(np.float32)
        raise TypeError(f"Unsupported audio input: {type(audio_input)}")

    @torch.no_grad()
    def extract_features(self, audio_input) -> torch.Tensor:
        """Audio → 2560-d feature vector (the DeepStack representation)."""
        wav = self._load_audio(audio_input)
        mel = self.processor(text="<|audio|>", audios=[wav], return_tensors="pt")
        audio_data = mel["audio_data"].to(self.moss_model.device,
                                           dtype=self.moss_model.dtype)
        enc = self.moss_model.audio_encoder(
            audio_data, output_hidden_states=True, return_dict=True,
        )
        projected = []
        for merger_idx, layer_id in enumerate(self.deepstack_layer_ids):
            h = enc.hidden_states[layer_id]
            projected.append(self.moss_model.deepstack_audio_merger_list[merger_idx](h))

        stacked = torch.stack(projected, dim=0)
        if self.aggregation == "sum":
            agg = stacked.sum(dim=0)
        elif self.aggregation == "mean":
            agg = stacked.mean(dim=0)
        elif self.aggregation == "concat":
            agg = torch.cat(projected, dim=-1)
        else:
            raise ValueError(f"Unknown aggregation: {self.aggregation}")

        return agg.mean(dim=1).squeeze(0).float()

    @torch.no_grad()
    def predict(self, audio_input, top_k: int = 3) -> Dict:
        """Classify audio → intent label + confidence + top-K alternatives."""
        feat = self.extract_features(audio_input)
        logits = self.head(feat.unsqueeze(0))
        probs = torch.softmax(logits, dim=-1).squeeze(0)
        pred_id = int(probs.argmax())
        topk = probs.topk(min(top_k, self.num_classes))
        return {
            "intent": self.ID2LABEL[pred_id],
            "intent_id": pred_id,
            "confidence": float(probs[pred_id]),
            "top_k": [
                {"intent": self.ID2LABEL[i], "prob": float(probs[i])}
                for i in topk.indices.cpu().tolist()
            ],
        }

    @torch.no_grad()
    def predict_batch(self, audio_inputs: List, top_k: int = 1) -> List[Dict]:
        """Sequential batch prediction (MOSS encoder doesn't pad-batch cleanly)."""
        return [self.predict(a, top_k=top_k) for a in audio_inputs]

    def list_classes(self) -> List[str]:
        return [self.ID2LABEL[i] for i in range(self.num_classes)]

Example usage

clf = MossAudioIntentClassifier(
    moss_audio_repo_path="/content/MOSS-Audio",
    device="cuda:0",
)

# Show classes
print("Available intent classes:")
for c in clf.list_classes():
    print(f"  • {c}")

# Predict on a file
result = clf.predict("path/to/tunisian_audio.wav", top_k=3)
print(f"\nPredicted: {result['intent']}  (conf={result['confidence']:.3f})")
print("Top-3 candidates:")
for alt in result["top_k"]:
    print(f"  {alt['intent']:<25} {alt['prob']:.3f}")

# Evaluate on a held-out HF Datasets row
from datasets import load_dataset, Audio
ds = load_dataset("Elyadata/SLURP-TN", split="validation")
ds = ds.cast_column("audio", Audio(sampling_rate=16000))

print("\nFirst 5 validation samples:")
for i in range(5):
    row = ds[i]
    result = clf.predict(row["audio"], top_k=1)
    gt = row["intent"]
    mark = "✅" if result["intent"] == gt else "❌"
    print(f"{mark} GT={gt:<20} PRED={result['intent']:<20} "
          f"({result['confidence']:.2f})")

Just the feature extractor (use as a general-purpose Arabic audio embedding)

If you don't need the intent labels — just the 2560-d acoustic feature vector that captures Arabic prosodic / phonetic / paralinguistic content — you can use the feature extractor on its own:

clf = MossAudioIntentClassifier(...)

# Get the feature embedding only
embedding = clf.extract_features("audio.wav")
print(embedding.shape)  # torch.Size([2560])

# Use for similarity search, clustering, downstream tasks ...

This is useful if you want to build other Arabic audio classifiers (emotion, dialect, speaker, etc.) on top of the same frozen MOSS backbone — just train a different head on these embeddings.

---

Supported intents (18 classes)

alarm_query         email_query              general_greet     query
alarm_remove        email_querycontact       general_joke      quirky
alarm_set           email_sendemail          general_quirky    sendemail
email_addcontact    general_*                news_query        takeaway_order
                                              takeaway_query   weather_query

The authoritative LABEL2ID mapping is stored inside best.pt — load it for the exact ID ordering.

---

Training Details

Data

• Dataset: Elyadata/SLURP-TN
• Original size: 2,677 train + 595 val + 893 test clips of Tunisian Arabic
• Used: all 2,677 train rows after dropping 3 singleton classes (joke, addcontact, querycontact each had n=1)
• Final: 2,174 train / 500 eval (stratified split, random_state=42)

Pipeline

1. Audio → MOSS-Audio mel feature extractor → [1, 128, T]
2. Mel → MOSS audio encoder forward with output_hidden_states=True
3. Pull hidden states at layers [10, 21, 31] → three [1, T', 1280] tensors
4. Pass each through corresponding deepstack_audio_merger_list[i] (GatedMLP, 1280→2560)
5. Sum across the 3 depth dimensions → [1, T', 2560]
6. Mean-pool over time → [1, 2560]
7. Trainable head → [1, 18] logits

Hyperparameters

Hardware

• A100 40GB (Google Colab Pro+)
• Feature extraction: ~5 minutes (cached to Drive for re-runs)
• Training: ~1 minute total for 5 epochs

Loss trajectory

The model picked up most of its capability in the first two epochs; the OneCycleLR schedule then helped consolidate. Epoch 5 shows mild overfit, but load_best_model_at_end=True ensures we keep epoch 4.

---

Files in this Repository

best.pt                  # Classifier head weights + label maps + config (3.2 MB)
README.md                # This file
confusion_matrix.csv     # Full 18×18 confusion matrix

best.pt contents

{
  "head_state_dict": OrderedDict,      # PyTorch state dict for the head
  "feature_dim": 2560,                  # Input dim to the head
  "num_classes": 18,                    # Output dim of the head
  "LABEL2ID": dict[str, int],           # e.g. {"weather_query": 16, ...}
  "ID2LABEL": dict[int, str],           # inverse of LABEL2ID
  "config": {
    "DEEPSTACK_LAYER_IDS": [10, 21, 31],
    "AGGREGATION": "sum",
  },
  "epoch": 3,                           # zero-indexed (= epoch 4)
  "best_acc": 0.674,
}

---

Limitations

1. Dialect specificity: trained only on Tunisian Arabic. Performance on other Arabic varieties (MSA, Saudi, Egyptian, Levantine) is untested and likely degraded.

2. Domain narrowness: SLURP-TN is a smart-assistant intent corpus (alarms, email, weather). The model has no notion of intents outside this 18-class space.

3. Long-tail under-resourced classes: classes with n<5 training examples (e.g. quirky, sendemail) consistently get 0% accuracy. The acoustic classifier can't learn from one example any better than humans can.

4. Audio quality assumptions: trained on relatively clean studio-quality recordings. Noisy phone-call audio or low-bitrate compressed audio likely degrade performance.

5. No streaming / online inference: the architecture pools over the full clip's time dimension before classifying. Live partial-utterance classification would need re-architecting.

6. Hardware cost at inference: while the trainable head is tiny (0.79M), feature extraction requires loading the full 5.2B-parameter MOSS-Audio encoder (~18GB VRAM in bf16). For deployment, consider exporting the encoder to ONNX or using a smaller frozen encoder.

7. Label leakage potential: the train/eval split is random-stratified within class, not speaker-stratified. The reported 67.4% may overestimate true cross-speaker generalization.

---

Citation

If you use this model, please cite:

@misc{moss_audio_tn_intent_2026,
  title  = {MOSS-Audio DeepStack Classifier for Tunisian Arabic Intent},
  author = {Fatimah Emad El-din},
  year   = {2026},
  url    = {https://huggingface.co/FatimahEmadEldin/moss-audio-tn-intent}
}

Please also cite the base model and dataset:

@misc{moss_audio,
  title  = {MOSS-Audio: An Open-Source Audio Understanding Model},
  author = {OpenMOSS Team},
  year   = {2025},
  url    = {https://huggingface.co/OpenMOSS-Team/MOSS-Audio-8B-Instruct}
}

@misc{slurp_tn,
  title  = {SLURP-TN: Tunisian Arabic Spoken Language Understanding},
  author = {Elyadata},
  year   = {2024},
  url    = {https://huggingface.co/datasets/Elyadata/SLURP-TN}
}

---

Acknowledgments

• OpenMOSS Team for the MOSS-Audio base model and DeepStack architecture
• Elyadata for releasing the SLURP-TN dataset

---

License

Apache 2.0.

The base MOSS-Audio weights remain under their own license; this classifier head is a small, independent component that only consumes MOSS's frozen intermediate representations.