Views
No views yet
OpenMOSS-Team/MOSS-Audio-8B-Instruct
audio encoder and DeepStack feature merger. ┌─────────────────────────────────────────┐
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| Component | Choice | Rationale |
|---|---|---|
| Base encoder | MOSS-Audio 8B, frozen | Strong acoustic representations; no need to retrain a massive model |
| Feature extraction | DeepStack merge across layers 10/21/31 | Captures low/mid/high-level acoustic features; matches MOSS's 3-merger design |
| Aggregation across depths | Sum | Preserves all depth information; outperformed concat in early tests |
| Aggregation across time | Mean-pool | Simple, robust to variable clip length |
| Head capacity | 0.79M params, 3-layer MLP with dropout | Small enough to train in seconds; large enough to learn 18-class boundary |
| Metric | Value |
|---|---|
| Best eval accuracy | 0.674 (337/500) |
| Number of classes | 18 |
| Majority-class baseline | ~0.188 |
| Random baseline | ~0.056 |
| Training samples | 2,174 |
| Evaluation samples | 500 |
| Training time | ~1 minute (after feature extraction) |
| Trainable params | 789,330 |
| Frozen params | 5,222,891,520 (5.22B) |
| Class | Eval samples | Accuracy |
|---|---|---|
news_query | 94 | 0.777 |
weather_query | 94 | 0.819 |
general_quirky | 80 | 0.487 |
email_query | 41 | 0.854 |
alarm_set | 34 | 0.735 |
email_sendemail | 34 | 0.588 |
takeaway_order | 25 | 0.680 |
alarm_query | 24 | 0.417 |
takeaway_query | 23 | 0.826 |
alarm_remove | 15 | 0.467 |
email_querycontact | 12 | 0.333 |
general_joke | 10 | 0.800 |
email_addcontact | 5 | 0.400 |
general_greet | 4 | 0.250 |
query | 3 | 0.000 |
quirky | 1 | 0.000 |
sendemail | 1 | 0.000 |
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%.confusion_matrix.csv.1# MOSS-Audio package
2git clone https://github.com/OpenMOSS/MOSS-Audio.git
3cd MOSS-Audio
4pip install -e .
5
6# Other deps
7pip install torch transformers huggingface_hub librosa1import torch, torch.nn as nn, librosa, sys
2from huggingface_hub import hf_hub_download
3
4# 1. Make MOSS-Audio importable
5sys.path.insert(0, "/path/to/MOSS-Audio")
6from src.modeling_moss_audio import MossAudioModel
7from src.processing_moss_audio import MossAudioProcessor
8
9# 2. Apply Whisper kwarg-stripping patch (idempotent)
10import transformers.models.whisper.modeling_whisper as whisper_mod
11if not getattr(whisper_mod.WhisperEncoderLayer.forward, "_moss_patched", False):
12 _orig = whisper_mod.WhisperEncoderLayer.forward
13 def _patched(self, *a, **kw):
14 kw.pop("layer_head_mask", None); kw.pop("head_mask", None)
15 return _orig(self, *a, **kw)
16 _patched._moss_patched = True
17 whisper_mod.WhisperEncoderLayer.forward = _patched
18
19# 3. Load the frozen base
20MOSS_REPO = "OpenMOSS-Team/MOSS-Audio-8B-Instruct"
21moss_model = MossAudioModel.from_pretrained(
22 MOSS_REPO, trust_remote_code=True, dtype="auto",
23 device_map="cuda:0", attn_implementation="sdpa",
24).eval()
25processor = MossAudioProcessor.from_pretrained(
26 MOSS_REPO, trust_remote_code=True, enable_time_marker=True,
27)
28
29# 4. Download the classifier head
30head_path = hf_hub_download(repo_id="FatimahEmadEldin/moss-audio-tn-intent",
31 filename="best.pt")
32
33# 5. Load head + label maps
34state = torch.load(head_path, weights_only=False, map_location="cuda:0")
35ID2LABEL = state["ID2LABEL"]
36NUM_CLASSES = state["num_classes"]
37
38class ClassifierHead(nn.Module):
39 def __init__(self, in_dim, num_classes, hidden=512, dropout=0.15):
40 super().__init__()
41 self.net = nn.Sequential(
42 nn.LayerNorm(in_dim),
43 nn.Linear(in_dim, hidden), nn.GELU(), nn.Dropout(dropout),
44 nn.Linear(hidden, hidden // 2), nn.GELU(), nn.Dropout(dropout),
45 nn.Linear(hidden // 2, num_classes),
46 )
47 def forward(self, x):
48 return self.net(x)
49
50head = ClassifierHead(state["feature_dim"], NUM_CLASSES).to("cuda:0")
51head.load_state_dict(state["head_state_dict"])
52head.eval()
53
54# 6. Predict
55@torch.no_grad()
56def predict(audio_path):
57 wav, _ = librosa.load(audio_path, sr=16000, mono=True)
58 mel = processor(text="<|audio|>", audios=[wav], return_tensors="pt")
59 audio_data = mel["audio_data"].to(moss_model.device, dtype=moss_model.dtype)
60
61 enc_out = moss_model.audio_encoder(audio_data, output_hidden_states=True,
62 return_dict=True)
63 projected = []
64 for merger_idx, layer_id in enumerate([10, 21, 31]):
65 h = enc_out.hidden_states[layer_id]
66 projected.append(moss_model.deepstack_audio_merger_list[merger_idx](h))
67 agg = torch.stack(projected, dim=0).sum(dim=0)
68 feat = agg.mean(dim=1).squeeze(0).float()
69
70 logits = head(feat.unsqueeze(0))
71 probs = torch.softmax(logits, dim=-1).squeeze(0)
72 pred_id = int(probs.argmax())
73 return {
74 "intent": ID2LABEL[pred_id],
75 "confidence": float(probs[pred_id]),
76 "top3": [(ID2LABEL[i], float(probs[i]))
77 for i in probs.topk(3).indices.cpu().tolist()],
78 }
79
80result = predict("my_audio.wav")
81print(result)
82# Example output:
83# {'intent': 'weather_query',
84# 'confidence': 0.84,
85# 'top3': [('weather_query', 0.84), ('news_query', 0.09), ('general_quirky', 0.03)]}1import torch
2import torch.nn as nn
3import librosa
4import numpy as np
5import sys, os
6from pathlib import Path
7from typing import Union, List, Dict
8from huggingface_hub import hf_hub_download
9
10
11class MossAudioIntentClassifier:
12 """Tunisian Arabic intent classifier using frozen MOSS-Audio + DeepStack + head."""
13
14 def __init__(
15 self,
16 moss_audio_repo_path: str = "/content/MOSS-Audio",
17 classifier_repo: str = "FatimahEmadEldin/moss-audio-tn-intent",
18 moss_base_repo: str = "OpenMOSS-Team/MOSS-Audio-8B-Instruct",
19 device: str = "cuda:0",
20 moss_weights_dir: str = None,
21 ):
22 """
23 Args:
24 moss_audio_repo_path: Local path to a clone of github.com/OpenMOSS/MOSS-Audio
25 classifier_repo: HF repo holding the trained classifier head (best.pt)
26 moss_base_repo: HF repo for the MOSS-Audio base model
27 device: 'cuda:0' / 'cuda' / 'cpu'
28 moss_weights_dir: Optional local cache dir for MOSS weights
29 """
30 self.device = device
31
32 # 1. Import MOSS-Audio modules from local clone
33 if moss_audio_repo_path not in sys.path:
34 sys.path.insert(0, moss_audio_repo_path)
35 from src.modeling_moss_audio import MossAudioModel
36 from src.processing_moss_audio import MossAudioProcessor
37
38 # 2. Apply idempotent Whisper kwarg-stripping patch
39 # (MOSS-Audio uses WhisperEncoderLayer but passes kwargs newer
40 # transformers versions don't accept)
41 self._patch_whisper()
42
43 # 3. Load (or cache) MOSS-Audio base model
44 if moss_weights_dir and os.path.exists(os.path.join(moss_weights_dir, "config.json")):
45 base_path = moss_weights_dir
46 else:
47 base_path = moss_base_repo
48 print(f"📥 Loading frozen MOSS-Audio from {base_path}")
49 self.moss_model = MossAudioModel.from_pretrained(
50 base_path, trust_remote_code=True, dtype="auto",
51 device_map=device, attn_implementation="sdpa",
52 ).eval()
53 for p in self.moss_model.parameters():
54 p.requires_grad = False
55
56 self.processor = MossAudioProcessor.from_pretrained(
57 base_path, trust_remote_code=True, enable_time_marker=True,
58 )
59
60 # 4. Download + load classifier head
61 print(f"📥 Loading classifier head from {classifier_repo}")
62 head_path = hf_hub_download(repo_id=classifier_repo, filename="best.pt")
63 state = torch.load(head_path, weights_only=False, map_location=device)
64
65 self.feature_dim = state["feature_dim"]
66 self.num_classes = state["num_classes"]
67 self.ID2LABEL = state["ID2LABEL"]
68 self.LABEL2ID = state["LABEL2ID"]
69 self.deepstack_layer_ids = state["config"]["DEEPSTACK_LAYER_IDS"]
70 self.aggregation = state["config"]["AGGREGATION"]
71
72 self.head = self._build_head(self.feature_dim, self.num_classes).to(device)
73 self.head.load_state_dict(state["head_state_dict"])
74 self.head.eval()
75
76 print(f"✅ Ready. {self.num_classes} classes, "
77 f"DeepStack layers {self.deepstack_layer_ids}, agg={self.aggregation}")
78
79 @staticmethod
80 def _patch_whisper():
81 import transformers.models.whisper.modeling_whisper as whisper_mod
82 if getattr(whisper_mod.WhisperEncoderLayer.forward, "_moss_patched", False):
83 return
84 _orig = whisper_mod.WhisperEncoderLayer.forward
85 def _patched(self, *args, **kwargs):
86 kwargs.pop("layer_head_mask", None)
87 kwargs.pop("head_mask", None)
88 return _orig(self, *args, **kwargs)
89 _patched._moss_patched = True
90 whisper_mod.WhisperEncoderLayer.forward = _patched
91
92 @staticmethod
93 def _build_head(in_dim, num_classes, hidden=512, dropout=0.15):
94 return nn.Sequential(
95 nn.LayerNorm(in_dim),
96 nn.Linear(in_dim, hidden), nn.GELU(), nn.Dropout(dropout),
97 nn.Linear(hidden, hidden // 2), nn.GELU(), nn.Dropout(dropout),
98 nn.Linear(hidden // 2, num_classes),
99 )
100
101 def _load_audio(self, audio_input: Union[str, np.ndarray, Dict]) -> np.ndarray:
102 """Accepts file path / numpy array / HF Datasets Audio dict."""
103 if isinstance(audio_input, (str, Path)):
104 wav, _ = librosa.load(str(audio_input), sr=16000, mono=True)
105 return wav.astype(np.float32)
106 if isinstance(audio_input, dict) and "array" in audio_input:
107 wav = np.asarray(audio_input["array"], dtype=np.float32)
108 sr = audio_input["sampling_rate"]
109 if wav.ndim > 1:
110 wav = wav.mean(axis=0)
111 if sr != 16000:
112 import torchaudio
113 wav = torchaudio.functional.resample(
114 torch.from_numpy(wav).unsqueeze(0),
115 orig_freq=sr, new_freq=16000,
116 ).squeeze(0).numpy()
117 return wav
118 if isinstance(audio_input, np.ndarray):
119 return audio_input.astype(np.float32)
120 raise TypeError(f"Unsupported audio input: {type(audio_input)}")
121
122 @torch.no_grad()
123 def extract_features(self, audio_input) -> torch.Tensor:
124 """Audio → 2560-d feature vector (the DeepStack representation)."""
125 wav = self._load_audio(audio_input)
126 mel = self.processor(text="<|audio|>", audios=[wav], return_tensors="pt")
127 audio_data = mel["audio_data"].to(self.moss_model.device,
128 dtype=self.moss_model.dtype)
129 enc = self.moss_model.audio_encoder(
130 audio_data, output_hidden_states=True, return_dict=True,
131 )
132 projected = []
133 for merger_idx, layer_id in enumerate(self.deepstack_layer_ids):
134 h = enc.hidden_states[layer_id]
135 projected.append(self.moss_model.deepstack_audio_merger_list[merger_idx](h))
136
137 stacked = torch.stack(projected, dim=0)
138 if self.aggregation == "sum":
139 agg = stacked.sum(dim=0)
140 elif self.aggregation == "mean":
141 agg = stacked.mean(dim=0)
142 elif self.aggregation == "concat":
143 agg = torch.cat(projected, dim=-1)
144 else:
145 raise ValueError(f"Unknown aggregation: {self.aggregation}")
146
147 return agg.mean(dim=1).squeeze(0).float()
148
149 @torch.no_grad()
150 def predict(self, audio_input, top_k: int = 3) -> Dict:
151 """Classify audio → intent label + confidence + top-K alternatives."""
152 feat = self.extract_features(audio_input)
153 logits = self.head(feat.unsqueeze(0))
154 probs = torch.softmax(logits, dim=-1).squeeze(0)
155 pred_id = int(probs.argmax())
156 topk = probs.topk(min(top_k, self.num_classes))
157 return {
158 "intent": self.ID2LABEL[pred_id],
159 "intent_id": pred_id,
160 "confidence": float(probs[pred_id]),
161 "top_k": [
162 {"intent": self.ID2LABEL[i], "prob": float(probs[i])}
163 for i in topk.indices.cpu().tolist()
164 ],
165 }
166
167 @torch.no_grad()
168 def predict_batch(self, audio_inputs: List, top_k: int = 1) -> List[Dict]:
169 """Sequential batch prediction (MOSS encoder doesn't pad-batch cleanly)."""
170 return [self.predict(a, top_k=top_k) for a in audio_inputs]
171
172 def list_classes(self) -> List[str]:
173 return [self.ID2LABEL[i] for i in range(self.num_classes)]1clf = MossAudioIntentClassifier(
2 moss_audio_repo_path="/content/MOSS-Audio",
3 device="cuda:0",
4)
5
6# Show classes
7print("Available intent classes:")
8for c in clf.list_classes():
9 print(f" • {c}")
10
11# Predict on a file
12result = clf.predict("path/to/tunisian_audio.wav", top_k=3)
13print(f"\nPredicted: {result['intent']} (conf={result['confidence']:.3f})")
14print("Top-3 candidates:")
15for alt in result["top_k"]:
16 print(f" {alt['intent']:<25} {alt['prob']:.3f}")
17
18# Evaluate on a held-out HF Datasets row
19from datasets import load_dataset, Audio
20ds = load_dataset("Elyadata/SLURP-TN", split="validation")
21ds = ds.cast_column("audio", Audio(sampling_rate=16000))
22
23print("\nFirst 5 validation samples:")
24for i in range(5):
25 row = ds[i]
26 result = clf.predict(row["audio"], top_k=1)
27 gt = row["intent"]
28 mark = "✅" if result["intent"] == gt else "❌"
29 print(f"{mark} GT={gt:<20} PRED={result['intent']:<20} "
30 f"({result['confidence']:.2f})")1clf = MossAudioIntentClassifier(...)
2
3# Get the feature embedding only
4embedding = clf.extract_features("audio.wav")
5print(embedding.shape) # torch.Size([2560])
6
7# Use for similarity search, clustering, downstream tasks ...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_queryLABEL2ID mapping is stored inside best.pt — load it
for the exact ID ordering.Elyadata/SLURP-TNjoke, addcontact, querycontact each had n=1)[1, 128, T]output_hidden_states=True[10, 21, 31] → three [1, T', 1280] tensorsdeepstack_audio_merger_list[i] (GatedMLP, 1280→2560)[1, T', 2560][1, 2560][1, 18] logits| Param | Value |
|---|---|
| Optimizer | AdamW |
| Learning rate | 5e-4 (OneCycleLR, warmup_ratio=0.05) |
| Weight decay | 0.01 |
| Batch size | 16 |
| Epochs | 5 |
| Loss | Cross-entropy with inverse-frequency class weights |
| Gradient clipping | 1.0 |
| Head dropout | 0.15 |
| Head hidden dim | 512 → 256 → 18 |
| Seed | 42 |
| Epoch | Train loss | Eval acc |
|---|---|---|
| 1 | 2.6899 | 0.408 |
| 2 | 2.0387 | 0.604 |
| 3 | 1.5904 | 0.590 |
| 4 | 1.3198 | 0.674 ← best |
| 5 | 1.2009 | 0.656 |
load_best_model_at_end=True ensures we keep epoch 4.best.pt # Classifier head weights + label maps + config (3.2 MB)
README.md # This file
confusion_matrix.csv # Full 18×18 confusion matrixbest.pt contents1{
2 "head_state_dict": OrderedDict, # PyTorch state dict for the head
3 "feature_dim": 2560, # Input dim to the head
4 "num_classes": 18, # Output dim of the head
5 "LABEL2ID": dict[str, int], # e.g. {"weather_query": 16, ...}
6 "ID2LABEL": dict[int, str], # inverse of LABEL2ID
7 "config": {
8 "DEEPSTACK_LAYER_IDS": [10, 21, 31],
9 "AGGREGATION": "sum",
10 },
11 "epoch": 3, # zero-indexed (= epoch 4)
12 "best_acc": 0.674,
13}quirky, sendemail) consistently get 0% accuracy. The acoustic
classifier can't learn from one example any better than humans can.1@misc{moss_audio_tn_intent_2026,
2 title = {MOSS-Audio DeepStack Classifier for Tunisian Arabic Intent},
3 author = {Fatimah Emad El-din},
4 year = {2026},
5 url = {https://huggingface.co/FatimahEmadEldin/moss-audio-tn-intent}
6}1@misc{moss_audio,
2 title = {MOSS-Audio: An Open-Source Audio Understanding Model},
3 author = {OpenMOSS Team},
4 year = {2025},
5 url = {https://huggingface.co/OpenMOSS-Team/MOSS-Audio-8B-Instruct}
6}
7
8@misc{slurp_tn,
9 title = {SLURP-TN: Tunisian Arabic Spoken Language Understanding},
10 author = {Elyadata},
11 year = {2024},
12 url = {https://huggingface.co/datasets/Elyadata/SLURP-TN}
13}