Views
No views yet
LiquidAI/LFM2.5-Audio-1.5B using mlx-audio version 0.3.0.pip install -U mlx-audiopip install mlx-audio1import mlx.core as mx
2from mlx_audio.sts.models.lfm_audio import (
3 LFM2AudioModel,
4 LFM2AudioProcessor,
5 ChatState,
6 LFMModality,
7)
8from mlx_audio.sts.models.lfm_audio.model import AUDIO_EOS_TOKEN
9
10# Load model and processor
11model = LFM2AudioModel.from_pretrained("mlx-community/LFM2.5-Audio-1.5B-5bit")
12processor = LFM2AudioProcessor.from_pretrained("mlx-community/LFM2.5-Audio-1.5B-5bit")
13
14# Create chat state
15chat = ChatState(processor)
16chat.new_turn("system")
17chat.add_text("Perform TTS. Use a UK male voice.")
18chat.end_turn()
19chat.new_turn("user")
20chat.add_text("Hello, welcome to MLX Audio!")
21chat.end_turn()
22chat.new_turn("assistant")
23
24# Generate with interleaved text and audio
25audio_codes = []
26for token, modality in model.generate_sequential(
27 **dict(chat),
28 max_new_tokens=2048,
29 temperature=0.8,
30
31):
32 mx.eval(token)
33 if modality == LFMModality.AUDIO_OUT:
34 if token[0].item() == AUDIO_EOS_TOKEN:
35 break
36 audio_codes.append(token)
37
38# Decode audio
39audio_codes = mx.stack(audio_codes, axis=0)[None, :].transpose(0, 2, 1)
40waveform = processor.decode_audio(audio_codes)
41
42# Save audio (24kHz sample rate)
43import soundfile as sf
44sf.write("output.wav", waveform[0].tolist(), model.sample_rate)1import mlx.core as mx
2import numpy as np
3import soundfile as sf
4from mlx_audio.sts.models.lfm_audio import (
5 LFM2AudioModel,
6 LFM2AudioProcessor,
7 ChatState,
8 LFMModality,
9)
10
11# Load model and processor
12model = LFM2AudioModel.from_pretrained("mlx-community/LFM2.5-Audio-1.5B-5bit")
13processor = LFM2AudioProcessor.from_pretrained("mlx-community/LFM2.5-Audio-1.5B-5bit")
14
15# Load audio (must be 24kHz for audio input)
16audio, sr = sf.read("input.wav")
17audio = mx.array(audio.astype(np.float32))
18
19# Create chat state with audio input
20chat = ChatState(processor)
21chat.new_turn("user")
22chat.add_audio(audio, sample_rate=sr)
23chat.add_text("Transcribe the audio.")
24chat.end_turn()
25chat.new_turn("assistant")
26
27# Generate text response
28text_out = []
29for token, modality in model.generate_interleaved(**dict(chat), max_new_tokens=512):
30 mx.eval(token)
31 if modality == LFMModality.TEXT:
32 text_out.append(token)
33 print(processor.decode_text(token[None]), end="", flush=True)1import mlx.core as mx
2import numpy as np
3import soundfile as sf
4from mlx_audio.sts.models.lfm_audio import (
5 LFM2AudioModel,
6 LFM2AudioProcessor,
7 ChatState,
8 LFMModality,
9)
10
11# Load model and processor
12model = LFM2AudioModel.from_pretrained("mlx-community/LFM2.5-Audio-1.5B-5bit")
13processor = LFM2AudioProcessor.from_pretrained("mlx-community/LFM2.5-Audio-1.5B-5bit")
14
15# Load input audio (24kHz)
16audio, sr = sf.read("input.wav")
17audio = mx.array(audio.astype(np.float32))
18
19# Create chat state with audio input
20chat = ChatState(processor)
21chat.new_turn("system")
22chat.add_text("Respond with interleaved text and audio.")
23chat.end_turn()
24chat.new_turn("user")
25chat.add_audio(audio, sample_rate=sr)
26chat.end_turn()
27chat.new_turn("assistant")
28
29# Generate response with both text and audio
30text_out, audio_out = [], []
31for token, modality in model.generate_interleaved(**dict(chat), max_new_tokens=2048):
32 mx.eval(token)
33 if modality == LFMModality.TEXT:
34 text_out.append(token)
35 print(processor.decode_text(token[None]), end="", flush=True)
36 else:
37 audio_out.append(token)
38
39# Decode audio response
40if audio_out:
41 audio_codes = mx.stack(audio_out[:-1], axis=1)[None, :] # (1, 8, T)
42 waveform = processor.decode_with_detokenizer(audio_codes)
43 sf.write("response.wav", waveform[0].tolist(), 24000)generate_interleaved for mixed text and audio output. The model can respond with text, audio, or both interleaved together.generate_interleaved is a complete frame of shape (8,) containing all 8 codebook values:1from mlx_audio.sts.models.lfm_audio import LFMModality
2
3text_out, audio_out = [], []
4for token, modality in model.generate_interleaved(**dict(chat), max_new_tokens=2048):
5 mx.eval(token)
6 if modality == LFMModality.TEXT:
7 text_out.append(token)
8 # Stream text output
9 print(processor.decode_text(token[None]), end="", flush=True)
10 else: # LFMModality.AUDIO_OUT
11 audio_out.append(token) # token shape: (8,)
12
13# Stack audio frames: list of (8,) -> (8, T)
14if audio_out:
15 audio_codes = mx.stack(audio_out[:-1], axis=1)[None, :] # (1, 8, T)
16 waveform = processor.decode_with_detokenizer(audio_codes)1# Decode using detokenizer
2audio = processor.decode_with_detokenizer(codes[None]) # (1, T_audio)1# Decode using Mimi codec
2audio = processor.decode_audio(codes) # (1, 1, T_audio)1from mlx_audio.sts.models.lfm_audio import GenerationConfig
2
3config = GenerationConfig(
4 max_new_tokens=2048, # Maximum tokens to generate
5 temperature=0.9, # Text sampling temperature
6 top_k=50, # Text top-k sampling
7 top_p=1.0, # Text nucleus sampling
8 audio_temperature=0.7, # Audio sampling temperature
9 audio_top_k=30, # Audio top-k sampling
10)1from mlx_audio.sts.models.lfm_audio import LFMModality
2
3FRAMES_PER_CHUNK = 10 # Decode every 10 audio frames
4
5audio_buffer = []
6for token, modality in model.generate_interleaved(**dict(chat), max_new_tokens=2048):
7 mx.eval(token)
8 if modality == LFMModality.AUDIO_OUT:
9 audio_buffer.append(token)
10
11 # Decode when we have enough frames
12 if len(audio_buffer) >= FRAMES_PER_CHUNK:
13 codes = mx.stack(audio_buffer, axis=1)[None, :] # (1, 8, T)
14 chunk = processor.decode_with_detokenizer(codes)
15 # Play chunk with your audio library...
16 audio_buffer = []
17
18 elif modality == LFMModality.TEXT:
19 # Stream text output
20 print(processor.decode_text(token[None]), end="", flush=True)1class LFM2AudioModel:
2 @classmethod
3 def from_pretrained(cls, model_name: str) -> "LFM2AudioModel":
4 """Load pretrained model from HuggingFace Hub."""
5
6 def generate_interleaved(
7 self,
8 text_tokens: mx.array,
9 audio_features: mx.array,
10 modalities: mx.array,
11 max_new_tokens: int = 512,
12 temperature: float = 0.9,
13 audio_temperature: float = 0.7,
14 audio_top_k: int = 30,
15 ) -> Generator[Tuple[mx.array, LFMModality], None, None]:
16 """Generate interleaved text and audio tokens.
17
18 Yields:
19 (token, modality) tuples where:
20 - For TEXT: token is scalar, modality is LFMModality.TEXT
21 - For AUDIO_OUT: token is (8,) array, modality is LFMModality.AUDIO_OUT
22 """1class LFM2AudioProcessor:
2 @classmethod
3 def from_pretrained(cls, model_name: str) -> "LFM2AudioProcessor":
4 """Load pretrained processor from HuggingFace Hub."""
5
6 def preprocess_audio(self, audio: mx.array, sample_rate: int) -> mx.array:
7 """Convert audio to mel spectrogram features."""
8
9 def tokenize_audio(self, audio: mx.array, sample_rate: int) -> mx.array:
10 """Tokenize audio using Mimi codec."""
11
12 def decode_audio(self, codes: mx.array) -> mx.array:
13 """Decode audio codes using Mimi codec."""
14
15 def decode_with_detokenizer(self, codes: mx.array) -> mx.array:
16 """Decode audio codes using neural detokenizer."""
17
18 def tokenize_text(self, text: str) -> mx.array:
19 """Tokenize text."""
20
21 def decode_text(self, tokens: mx.array) -> str:
22 """Decode text tokens."""1class ChatState:
2 def __init__(self, processor: LFM2AudioProcessor):
3 """Initialize chat state."""
4
5 def new_turn(self, role: str):
6 """Start a new turn (user/assistant/system)."""
7
8 def end_turn(self):
9 """End the current turn."""
10
11 def add_text(self, text: str):
12 """Add text to current turn."""
13
14 def add_audio(self, audio: mx.array, sample_rate: int):
15 """Add audio to current turn."""