Update project to use MLX for Apple Silicon support
Update dependencies for MLX models Add automatic model selection based on available RAM Rewrite README with MLX-specific documentation Update .gitignore for MLX project structure
This commit is contained in:
3
src/audio_summary/__init__.py
Normal file
3
src/audio_summary/__init__.py
Normal file
@@ -0,0 +1,3 @@
|
||||
"""Audio Summary with local MLX models on Apple Silicon."""
|
||||
|
||||
__version__ = "0.2.0"
|
||||
102
src/audio_summary/cli.py
Normal file
102
src/audio_summary/cli.py
Normal file
@@ -0,0 +1,102 @@
|
||||
"""Command-line entrypoint.
|
||||
|
||||
Download (optional), transcribe, and summarize audio/video files fully on
|
||||
Apple Silicon via MLX. Models are selected automatically from the detected
|
||||
unified memory (see ``config`` and ``device``).
|
||||
"""
|
||||
|
||||
import argparse
|
||||
from pathlib import Path
|
||||
|
||||
from .device import select_models
|
||||
from .download import download_from_youtube
|
||||
from .summarization import summarize_text
|
||||
from .transcription import transcribe_file
|
||||
|
||||
DEFAULT_LANGUAGE = "en" # Use "auto" on the CLI for automatic detection.
|
||||
|
||||
|
||||
def _resolve_language(arg_language: str | None) -> str | None:
|
||||
"""Map the CLI ``--language`` value to a code or ``None`` (auto-detect)."""
|
||||
language = arg_language if arg_language else DEFAULT_LANGUAGE
|
||||
if language and language.lower() == "auto":
|
||||
return None
|
||||
return language
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Download, transcribe, and summarize audio or video files "
|
||||
"on Apple Silicon (MLX)."
|
||||
)
|
||||
group = parser.add_mutually_exclusive_group(required=True)
|
||||
group.add_argument("--from-youtube", type=str, help="YouTube URL to download.")
|
||||
group.add_argument("--from-local", type=str, help="Path to the local audio file.")
|
||||
parser.add_argument(
|
||||
"--output",
|
||||
type=str,
|
||||
default="./summary.md",
|
||||
help="Output markdown file path.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--transcript-only",
|
||||
action="store_true",
|
||||
help="Only transcribe the file, do not summarize.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--language",
|
||||
type=str,
|
||||
help="Language code for transcription (e.g. 'en', 'fr', 'es', "
|
||||
"or 'auto' for detection).",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
language = _resolve_language(args.language)
|
||||
|
||||
# --- Automatic model selection based on detected unified memory ---
|
||||
ram_gb, tier = select_models()
|
||||
print("=" * 60)
|
||||
print(f"Detected unified memory : {ram_gb:.1f} GB")
|
||||
print(f"Selected hardware tier : >= {tier.min_ram_gb} GB")
|
||||
print(f"STT model : {tier.stt.repo} ({tier.stt.engine})")
|
||||
print(f"Summarization model : {tier.summarization_repo}")
|
||||
print("=" * 60)
|
||||
|
||||
# --- Working directory for intermediate files ---
|
||||
data_directory = Path("tmp")
|
||||
if not data_directory.exists():
|
||||
data_directory.mkdir(parents=True)
|
||||
print(f"Created directory: {data_directory}")
|
||||
|
||||
# --- Resolve input file ---
|
||||
if args.from_youtube:
|
||||
print(f"Downloading YouTube video from {args.from_youtube}")
|
||||
file_path = download_from_youtube(args.from_youtube, str(data_directory))
|
||||
else:
|
||||
file_path = Path(args.from_local)
|
||||
|
||||
# --- Transcription ---
|
||||
print(f"Transcribing file: {file_path}")
|
||||
transcript = transcribe_file(
|
||||
str(file_path),
|
||||
str(data_directory / "transcript.txt"),
|
||||
tier.stt,
|
||||
language,
|
||||
)
|
||||
|
||||
if args.transcript_only:
|
||||
print("Transcription complete. Skipping summary generation.")
|
||||
return
|
||||
|
||||
# --- Summarization ---
|
||||
print("Generating summary...")
|
||||
summary = summarize_text(transcript, tier.summarization_repo)
|
||||
|
||||
with open(args.output, "w") as md_file:
|
||||
md_file.write("# Summary\n\n")
|
||||
md_file.write(summary)
|
||||
print(f"Summary written to {args.output}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
58
src/audio_summary/config.py
Normal file
58
src/audio_summary/config.py
Normal file
@@ -0,0 +1,58 @@
|
||||
"""Model configuration for Apple Silicon (MLX).
|
||||
|
||||
This table maps the amount of unified memory detected at runtime to the best
|
||||
Speech-to-Text (STT) and summarization models. It is intentionally kept in its
|
||||
own module so the model line-up can be tuned without touching the core logic.
|
||||
|
||||
Each tier declares a ``min_ram_gb`` threshold. At runtime the highest tier whose
|
||||
threshold is *less than or equal to* the detected memory is selected (see
|
||||
``audio_summary.device.select_models``).
|
||||
|
||||
STT engines:
|
||||
* ``"voxtral"`` -> Mistral Voxtral via ``mlx-audio`` (streaming, long audio).
|
||||
* ``"whisper"`` -> ``mlx-whisper`` (lightweight fallback for low-RAM Macs).
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class STTModel:
|
||||
"""A Speech-to-Text model definition."""
|
||||
|
||||
engine: str # "voxtral" or "whisper"
|
||||
repo: str # Hugging Face repository id
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Tier:
|
||||
"""A hardware tier: minimum RAM and the models it unlocks."""
|
||||
|
||||
min_ram_gb: int
|
||||
stt: STTModel
|
||||
summarization_repo: str
|
||||
|
||||
|
||||
# Ordered from the smallest to the largest hardware tier.
|
||||
TIERS: list[Tier] = [
|
||||
Tier(
|
||||
min_ram_gb=8,
|
||||
stt=STTModel("whisper", "mlx-community/whisper-large-v3-turbo"),
|
||||
summarization_repo="mlx-community/Qwen3-4B-4bit",
|
||||
),
|
||||
Tier(
|
||||
min_ram_gb=16,
|
||||
stt=STTModel("voxtral", "mlx-community/Voxtral-Mini-4B-Realtime-2602-4bit"),
|
||||
summarization_repo="mlx-community/Qwen3-8B-4bit",
|
||||
),
|
||||
Tier(
|
||||
min_ram_gb=24,
|
||||
stt=STTModel("voxtral", "mlx-community/Voxtral-Mini-4B-Realtime-2602-fp16"),
|
||||
summarization_repo="mlx-community/Qwen3-8B-8bit",
|
||||
),
|
||||
Tier(
|
||||
min_ram_gb=32,
|
||||
stt=STTModel("voxtral", "mlx-community/Voxtral-Mini-4B-Realtime-2602-fp16"),
|
||||
summarization_repo="mlx-community/Qwen3-30B-A3B-4bit",
|
||||
),
|
||||
]
|
||||
32
src/audio_summary/device.py
Normal file
32
src/audio_summary/device.py
Normal file
@@ -0,0 +1,32 @@
|
||||
"""Hardware detection and automatic model selection.
|
||||
|
||||
Detects the amount of unified memory available on the Apple Silicon machine and
|
||||
picks the most capable model tier that fits (see ``config.TIERS``).
|
||||
"""
|
||||
|
||||
import psutil
|
||||
|
||||
from .config import TIERS, Tier
|
||||
|
||||
|
||||
def detect_ram_gb() -> float:
|
||||
"""Return the total unified memory in gibibytes."""
|
||||
return psutil.virtual_memory().total / (1024**3)
|
||||
|
||||
|
||||
def select_tier(ram_gb: float) -> Tier:
|
||||
"""Select the highest tier whose RAM requirement fits ``ram_gb``.
|
||||
|
||||
Falls back to the smallest tier if the machine has less memory than any
|
||||
declared threshold (the models simply run slower / may swap).
|
||||
"""
|
||||
eligible = [tier for tier in TIERS if tier.min_ram_gb <= ram_gb]
|
||||
if eligible:
|
||||
return max(eligible, key=lambda tier: tier.min_ram_gb)
|
||||
return min(TIERS, key=lambda tier: tier.min_ram_gb)
|
||||
|
||||
|
||||
def select_models() -> tuple[float, Tier]:
|
||||
"""Detect memory and return ``(ram_gb, selected_tier)``."""
|
||||
ram_gb = detect_ram_gb()
|
||||
return ram_gb, select_tier(ram_gb)
|
||||
27
src/audio_summary/download.py
Normal file
27
src/audio_summary/download.py
Normal file
@@ -0,0 +1,27 @@
|
||||
"""YouTube audio download helper (yt-dlp + ffmpeg)."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import yt_dlp
|
||||
|
||||
|
||||
def download_from_youtube(url: str, path: str) -> Path:
|
||||
"""Download the best audio track from ``url`` as an mp3 into ``path``.
|
||||
|
||||
Returns the path to the resulting ``to_transcribe.mp3`` file.
|
||||
"""
|
||||
ydl_opts = {
|
||||
"format": "bestaudio/best",
|
||||
"outtmpl": str(Path(path) / "to_transcribe.%(ext)s"),
|
||||
"postprocessors": [
|
||||
{
|
||||
"key": "FFmpegExtractAudio",
|
||||
"preferredcodec": "mp3",
|
||||
"preferredquality": "192",
|
||||
}
|
||||
],
|
||||
}
|
||||
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
||||
ydl.download([url])
|
||||
|
||||
return Path(path) / "to_transcribe.mp3"
|
||||
39
src/audio_summary/summarization.py
Normal file
39
src/audio_summary/summarization.py
Normal file
@@ -0,0 +1,39 @@
|
||||
"""Text summarization with Qwen3 on Apple Silicon via mlx-lm."""
|
||||
|
||||
from mlx_lm import generate, load
|
||||
|
||||
SYSTEM_PROMPT = "I would like for you to assume the role of a Technical Expert."
|
||||
|
||||
USER_PROMPT_TEMPLATE = """Generate a concise summary of the text below.
|
||||
Text : {text}
|
||||
Add a title to the summary.
|
||||
Make sure your summary has useful and true information about the main points of the topic.
|
||||
Begin with a short introduction explaining the topic. If you can, use bullet points to list important details,
|
||||
and finish your summary with a concluding sentence."""
|
||||
|
||||
|
||||
def summarize_text(text: str, model_repo: str, max_tokens: int = 2048) -> str:
|
||||
"""Summarize ``text`` using the Qwen3 model at ``model_repo``."""
|
||||
print(f"Loading summarization model: {model_repo}")
|
||||
model, tokenizer = load(model_repo)
|
||||
|
||||
messages = [
|
||||
{"role": "system", "content": SYSTEM_PROMPT},
|
||||
{"role": "user", "content": USER_PROMPT_TEMPLATE.format(text=text)},
|
||||
]
|
||||
|
||||
# Qwen3 supports a "thinking" mode; disable it for clean, direct summaries.
|
||||
prompt = tokenizer.apply_chat_template(
|
||||
messages,
|
||||
add_generation_prompt=True,
|
||||
enable_thinking=False,
|
||||
)
|
||||
|
||||
summary = generate(
|
||||
model,
|
||||
tokenizer,
|
||||
prompt=prompt,
|
||||
max_tokens=max_tokens,
|
||||
verbose=False,
|
||||
)
|
||||
return summary.strip()
|
||||
68
src/audio_summary/transcription.py
Normal file
68
src/audio_summary/transcription.py
Normal file
@@ -0,0 +1,68 @@
|
||||
"""Speech-to-Text on Apple Silicon.
|
||||
|
||||
Two engines are supported, selected through ``config.STTModel.engine``:
|
||||
|
||||
* ``voxtral`` -> Mistral Voxtral via ``mlx-audio``. Outperforms Whisper
|
||||
large-v3 on most benchmarks and natively handles long audio with streaming.
|
||||
* ``whisper`` -> ``mlx-whisper`` fallback for memory-constrained Macs.
|
||||
"""
|
||||
|
||||
from .config import STTModel
|
||||
|
||||
|
||||
def _transcribe_voxtral(file_path: str, model_repo: str, language: str | None) -> str:
|
||||
"""Transcribe using Mistral Voxtral through mlx-audio."""
|
||||
from mlx_audio.stt.utils import load
|
||||
|
||||
model = load(model_repo)
|
||||
# Voxtral handles long audio natively; ``generate`` returns an object with a
|
||||
# ``.text`` attribute. ``language`` is optional (auto-detected when None).
|
||||
kwargs: dict = {}
|
||||
if language:
|
||||
kwargs["language"] = language
|
||||
result = model.generate(file_path, **kwargs)
|
||||
return result.text.strip()
|
||||
|
||||
|
||||
def _transcribe_whisper(file_path: str, model_repo: str, language: str | None) -> str:
|
||||
"""Transcribe using the mlx-whisper fallback engine."""
|
||||
import mlx_whisper
|
||||
|
||||
result = mlx_whisper.transcribe(
|
||||
file_path,
|
||||
path_or_hf_repo=model_repo,
|
||||
language=language, # None triggers auto-detection
|
||||
)
|
||||
return result["text"].strip()
|
||||
|
||||
|
||||
def transcribe_file(
|
||||
file_path: str,
|
||||
output_file: str,
|
||||
stt_model: STTModel,
|
||||
language: str | None = None,
|
||||
) -> str:
|
||||
"""Transcribe ``file_path`` and persist the transcript to ``output_file``.
|
||||
|
||||
``language`` may be a language code (e.g. ``"en"``, ``"fr"``) or ``None`` for
|
||||
automatic detection. Returns the full transcript text.
|
||||
"""
|
||||
print(f"Transcribing with {stt_model.engine} model: {stt_model.repo}")
|
||||
if language:
|
||||
print(f"Transcription language: {language}")
|
||||
else:
|
||||
print("Using automatic language detection")
|
||||
print("Starting transcription (this may take a while for longer files)...")
|
||||
|
||||
if stt_model.engine == "voxtral":
|
||||
text = _transcribe_voxtral(file_path, stt_model.repo, language)
|
||||
elif stt_model.engine == "whisper":
|
||||
text = _transcribe_whisper(file_path, stt_model.repo, language)
|
||||
else:
|
||||
raise ValueError(f"Unknown STT engine: {stt_model.engine!r}")
|
||||
|
||||
with open(output_file, "w") as tmp_file:
|
||||
tmp_file.write(text)
|
||||
print(f"Transcription saved to file: {output_file}")
|
||||
|
||||
return text
|
||||
Reference in New Issue
Block a user