はじめに

自作の流星自動検出システム meteo に、画像認識の誤検出フィルタリング機能を追加しようとしたのがきっかけです。最初は Ollama 経由で GGUF モデルを使っていましたが、推論速度に限界を感じ、Apple Silicon ネイティブの MLX フレームワークと Gemma 4（e2b） の組み合わせを試しました。

結果として、同じモデル・同じ Mac 上で Ollama/GGUF と比べて大幅な高速化が得られ、FaceTime カメラのリアルタイム認識（約 2 秒間隔）も実用レベルで動作しました。本記事ではその構成・手順・性能を整理します。

背景：meteo プロジェクトとは

meteo は、全天カメラの映像から流星を自動検出・記録する Python 製システムです。主な処理フローは次のとおりです。

RTSPカメラから映像をリアルタイム取得
フレーム間差分で動体を検出
検出イベントをクリップ保存 + composite.jpg（最大値合成画像）生成
VLM（Vision Language Model）で「本当に流星か」を二次判定 ← 今回の追加部分

従来の画像処理だけでは、虫・雨滴・照明フラッシュなどの誤検出を完全には除去できません。VLM を最終段に置くことで精度向上を狙っています。

Gemma 4（e2b）とは

Gemma 4 は Google DeepMind が 2026 年に公開したマルチモーダルモデルファミリーです。e2b（Efficient 2B）は軽量版で、テキストと画像を同時に扱える Vision Language Model（VLM）です。

項目	値
パラメータ数	約 2B
量子化	4bit（MLX 版）
モデルサイズ	約 3.4 GB
入力	テキスト + 画像（複数枚対応）
ライセンス	Gemma Terms of Use

Hugging Face の mlx-community/gemma-4-e2b-it-4bit として配布されており、mlx_vlm から 1 行でロードできます。

必要な環境とパッケージ

動作要件

Apple Silicon Mac（M1 / M2 / M3 / M4 系）
macOS 13 Ventura 以降
Python 3.11 以降（| 型アノテーション構文のため）
空き VRAM（ユニファイドメモリ）：最低 8 GB 推奨、16 GB 以上が快適

パッケージインストール

pip install mlx-vlm          # MLX ベースの VLM 推論フレームワーク
pip install Pillow            # 画像処理
pip install opencv-python     # カメラキャプチャ・表示（リアルタイム認識用）
pip install av                # 動画デコード（動画フレーム解析用）

mlx-vlm をインストールすると mlx・mlx-lm などの依存パッケージも自動でインストールされます。

モデルのダウンロード

初回ロード時に自動ダウンロードされます（約 3.4 GB）。事前に取得したい場合：

python -c "from mlx_vlm import load; load('mlx-community/gemma-4-e2b-it-4bit')"

キャッシュ先：~/.cache/huggingface/hub/models--mlx-community--gemma-4-e2b-it-4bit/

基本的な使い方

モデルのロードと推論

from mlx_vlm import load, generate
from mlx_vlm.prompt_utils import apply_chat_template
from PIL import Image
import time

MODEL_ID = "mlx-community/gemma-4-e2b-it-4bit"

# モデルロード（初回のみ時間がかかる）
model, processor = load(MODEL_ID)

# 画像を読み込む
image = Image.open("photo.jpg").convert("RGB")

# プロンプト準備
prompt = "この画像に何が見えますか？日本語で答えてください。"
formatted = apply_chat_template(processor, model.config, prompt, num_images=1)

# 推論
t0 = time.perf_counter()
result = generate(
    model, processor,
    prompt=formatted,
    image=[image],
    max_tokens=200,
    temperature=0.1,
    verbose=False,
)
print(f"{time.perf_counter()-t0:.2f}s: {result.text}")

モデルをメモリに保持したまま繰り返し推論できるため、バッチ処理やリアルタイム処理に向いています。

MLX vs Ollama/GGUF 性能比較

同じ gemma4:e2b モデルを Ollama（GGUF）と mlx_vlm（MLX 4bit）で比較しました。テスト環境は MacBook Pro M2 16GB、テスト画像は 1280×960 の夜空監視カメラ映像です。

指標	Ollama / GGUF	mlx-vlm / MLX 4bit
モデルロード時間	〜8 s（都度）	〜5 s（初回のみ）
推論時間（150 tokens）	〜12 s	〜2–3 s
生成速度	〜10 tok/s	〜50–80 tok/s
メモリ（統合）	〜4 GB	〜4 GB

MLX が速い理由

モデルをメモリに保持：Ollama はリクエストごとにロードが発生しやすいが、MLX はセッション中ロードしたまま使い回せる
Apple Silicon のユニファイドメモリを直接活用：CPU/GPU 間のコピーが不要
MLX フレームワーク自体が Apple Silicon 向けに最適化されており、Metal GPU 演算を効率よく使う

実際の流星判定では 1 枚あたり 2〜3 秒で応答が得られるため、meteo の検出後バッチ処理には十分な速度です。

meteo への組み込み：流星 VLM 判定

静止画（composite）による判定

meteo が生成する composite_original.jpg（検出区間の最大値合成画像）を Gemma4 に渡して流星を判定します。

PROMPT_METEOR = """この画像は晴れた夜空の監視カメラ映像です。
流星または光跡の有無のみを判定してJSONのみ返答してください（他のテキストは不要）:

{
  "is_meteor": true または false,
  "confidence": 0.0〜1.0,
  "description": "日本語で1〜2文の説明"
}

判定基準:
- is_meteor=true: 直線状の光跡（流星・人工衛星・飛行機の軌跡）が明確に見られる
- is_meteor=false: ノイズ・虫・雨・カメラアーティファクトなど
"""

JSON 形式で返答させることで後段の処理が容易になります。推論結果のパースは ```json ``` ブロックへの対応を含めて行います。

動画クリップによる判定

composite 画像だけでは判断が難しいケースには、動画クリップから均等サンプリングした複数フレームを渡す方法も有効です。mlx_vlm は image=[] に複数の PIL Image を渡すだけで対応できます。

# 8フレームを渡す例
result = generate(
    model, processor,
    prompt=formatted,
    image=frames,      # list[PIL.Image.Image]
    max_tokens=300,
    temperature=0.1,
)

リアルタイムカメラ認識

アーキテクチャ

VLM の推論は数秒かかるため、メインのカメラ表示ループとは別スレッドで実行します。

メインスレッド                バックグラウンドスレッド
─────────────────            ──────────────────────────
cap.read() → フレーム取得     ←── フレームを受け取る
↓                            推論（Gemma4）
overlay_text() でテキスト合成  ──→ 推論結果を書き込む
cv2.imshow()                 interval 秒待機 → 繰り返す

threading.Lock で共有データを保護しており、推論中もカメラ映像は途切れません。

動作の様子

起動するとまずカメラ一覧が表示されます：

利用可能なカメラ:
  [0] カメラ 0  (1280x720)   ← FaceTime
  [1] カメラ 1  (1920x1080)  ← 外部カメラ

カメラ番号を選択 [0]:

選択後にモデルをロードし、カメラウィンドウが開きます。画面下部に半透明のテキストボックスが表示され、Gemma4 の認識結果が約 2 秒ごとに更新されます。

ステータス行にはリアルタイムで以下が表示されます：

推論: 2.3s  |  カメラ: 29.8fps  |  回数: 15  |  間隔: 2.0s

実行方法

# デフォルト（起動時にカメラ選択）
python scripts/camera_realtime_gemma4.py

# カメラ番号と推論間隔を指定
python scripts/camera_realtime_gemma4.py --camera 0 --interval 3.0

# 別のモデルを使う
python scripts/camera_realtime_gemma4.py --model mlx-community/gemma-4-12b-it-4bit

q または ESC で終了します。

まとめ

項目	結果
推論速度	Ollama 比 4〜5 倍高速（約 50–80 tok/s）
モデルサイズ	4bit 量子化で 3.4 GB
リアルタイム認識	約 2 秒間隔で安定動作
日本語プロンプト	問題なく対応
JSON 応答	概ね安定（パースエラーは稀に発生）

Apple Silicon の統合メモリ構造と MLX フレームワークの相性は非常によく、ローカルで VLM をリアルタイムに動かすには現時点で最良の選択肢のひとつだと感じています。

meteo への組み込みとしては、composite 画像による VLM 二次判定が実用レベルの速度・精度で動作することを確認できました。今後は信頼度スコアを使った閾値チューニングと、誤検出パターンのプロンプト改善を進める予定です。

サンプルコード：camera_realtime_gemma4.py

meteoへ組み込む際に、お遊びで作ったFaceTime カメラ映像を Gemma4 でリアルタイム認識し、画面にスーパーインポーズ表示するスクリプトです。これだけでも結構遊べます。

#!/usr/bin/env python3
"""FaceTimeカメラ + Gemma4 リアルタイム認識スクリプト

MacBook の FaceTime カメラ映像を表示しながら、
mlx_vlm 経由で Gemma4 に画像を送り「何が見えているか」をリアルタイムで
カメラ映像上にスーパーインポーズ表示する。

使い方:
    python scripts/camera_realtime_gemma4.py
    python scripts/camera_realtime_gemma4.py --camera 0 --interval 2.0
    python scripts/camera_realtime_gemma4.py --model mlx-community/gemma-4-e2b-it-4bit

操作:
    q / ESC : 終了
"""

import argparse
import threading
import time
from pathlib import Path

import cv2
import numpy as np
from PIL import Image, ImageDraw, ImageFont

# ---------------------------------------------------------------------------
# 設定
# ---------------------------------------------------------------------------
MODEL_ID = "mlx-community/gemma-4-e2b-it-4bit"

PROMPT = """この画像に何が見えますか？
日本語で簡潔に答えてください（2〜3文以内）。
人物・物体・場所・状況など具体的に説明してください。"""

# テキスト描画設定
TEXT_BG_ALPHA = 0.65       # 背景の透明度
TEXT_COLOR = (255, 255, 255)
TEXT_BG_COLOR = (0, 0, 0)
FONT_SIZE = 22
LINE_SPACING = 6


# ---------------------------------------------------------------------------
# テキスト描画（PIL で日本語対応）
# ---------------------------------------------------------------------------

def get_font(size: int) -> ImageFont.FreeTypeFont | ImageFont.ImageFont:
    """日本語フォントを取得（なければデフォルト）"""
    candidates = [
        "/System/Library/Fonts/ヒラギノ角ゴシック W3.ttc",
        "/System/Library/Fonts/Hiragino Sans GB.ttc",
        "/System/Library/Fonts/AppleSDGothicNeo.ttc",
        "/Library/Fonts/Arial Unicode MS.ttf",
    ]
    for path in candidates:
        if Path(path).exists():
            try:
                return ImageFont.truetype(path, size)
            except Exception:
                continue
    return ImageFont.load_default()


def wrap_text(text: str, font: ImageFont.FreeTypeFont | ImageFont.ImageFont,
              max_width: int) -> list[str]:
    """テキストを max_width ピクセル内に折り返す"""
    lines = []
    for paragraph in text.split("\n"):
        if not paragraph.strip():
            lines.append("")
            continue
        words = list(paragraph)   # 日本語は1文字ずつ区切る
        line = ""
        for ch in words:
            test = line + ch
            bbox = font.getbbox(test)
            w = bbox[2] - bbox[0]
            if w > max_width and line:
                lines.append(line)
                line = ch
            else:
                line = test
        if line:
            lines.append(line)
    return lines


def overlay_text(frame_bgr: np.ndarray, text: str, elapsed: float,
                 font: ImageFont.FreeTypeFont | ImageFont.ImageFont,
                 status: str = "") -> np.ndarray:
    """
    OpenCV の BGR フレームに PIL でテキストをスーパーインポーズして返す。
    テキストは画面下部に半透明背景付きで表示する。
    """
    h, w = frame_bgr.shape[:2]
    frame_rgb = cv2.cvtColor(frame_bgr, cv2.COLOR_BGR2RGB)
    pil = Image.fromarray(frame_rgb)

    max_text_w = w - 20
    wrapped = wrap_text(text, font, max_text_w)

    line_h = FONT_SIZE + LINE_SPACING
    total_text_h = line_h * len(wrapped) + LINE_SPACING * 2
    status_h = FONT_SIZE + LINE_SPACING * 2 if status else 0
    box_h = total_text_h + status_h + 10

    # 半透明背景
    overlay = Image.new("RGBA", pil.size, (0, 0, 0, 0))
    draw_ov = ImageDraw.Draw(overlay)
    box_y = h - box_h - 10
    draw_ov.rectangle([(5, box_y), (w - 5, h - 5)],
                      fill=(*TEXT_BG_COLOR, int(255 * TEXT_BG_ALPHA)))

    pil = pil.convert("RGBA")
    pil = Image.alpha_composite(pil, overlay)
    draw = ImageDraw.Draw(pil)

    # テキスト描画
    y = box_y + LINE_SPACING
    for line in wrapped:
        draw.text((12, y), line, font=font, fill=(*TEXT_COLOR, 255))
        y += line_h

    # ステータス行
    if status:
        small_font_size = max(14, FONT_SIZE - 6)
        try:
            small_font = ImageFont.truetype(
                str(next((p for p in [
                    "/System/Library/Fonts/ヒラギノ角ゴシック W3.ttc",
                    "/System/Library/Fonts/Hiragino Sans GB.ttc",
                ] if Path(p).exists()), "")),
                small_font_size
            ) if any(Path(p).exists() for p in [
                "/System/Library/Fonts/ヒラギノ角ゴシック W3.ttc",
                "/System/Library/Fonts/Hiragino Sans GB.ttc",
            ]) else font
        except Exception:
            small_font = font
        draw.text((12, y + 4), status, font=small_font,
                  fill=(180, 255, 180, 255))

    result_rgb = np.array(pil.convert("RGB"))
    return cv2.cvtColor(result_rgb, cv2.COLOR_RGB2BGR)


# ---------------------------------------------------------------------------
# 推論スレッド
# ---------------------------------------------------------------------------

class InferenceWorker:
    """バックグラウンドスレッドで Gemma4 推論を実行する"""

    def __init__(self, model, processor, interval: float = 2.0):
        self.model = model
        self.processor = processor
        self.interval = interval

        self._lock = threading.Lock()
        self._latest_frame: np.ndarray | None = None
        self._result_text: str = "推論中..."
        self._infer_sec: float = 0.0
        self._infer_count: int = 0
        self._running = True
        self._thread = threading.Thread(target=self._loop, daemon=True)

    def start(self):
        self._thread.start()

    def stop(self):
        self._running = False

    def submit_frame(self, frame_bgr: np.ndarray):
        """表示中のフレームをセット（スレッドセーフ）"""
        with self._lock:
            self._latest_frame = frame_bgr.copy()

    def get_result(self) -> tuple[str, float, int]:
        """(テキスト, 推論秒数, 推論回数) を取得"""
        with self._lock:
            return self._result_text, self._infer_sec, self._infer_count

    def _loop(self):
        from mlx_vlm import generate
        from mlx_vlm.prompt_utils import apply_chat_template

        while self._running:
            with self._lock:
                frame = self._latest_frame

            if frame is None:
                time.sleep(0.1)
                continue

            frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
            pil_image = Image.fromarray(frame_rgb)

            try:
                formatted = apply_chat_template(
                    self.processor, self.model.config, PROMPT, num_images=1
                )
                t0 = time.perf_counter()
                result = generate(
                    self.model, self.processor,
                    prompt=formatted,
                    image=[pil_image],
                    max_tokens=150,
                    temperature=0.1,
                    verbose=False,
                )
                elapsed = time.perf_counter() - t0
                text = result.text if hasattr(result, "text") else str(result)

                with self._lock:
                    self._result_text = text.strip() or "（応答なし）"
                    self._infer_sec = elapsed
                    self._infer_count += 1

            except Exception as e:
                elapsed = 0.0
                with self._lock:
                    self._result_text = f"[エラー] {e}"

            wait = max(0.0, self.interval - elapsed)
            time.sleep(wait)


# ---------------------------------------------------------------------------
# カメラ選択
# ---------------------------------------------------------------------------

def list_cameras(max_index: int = 5) -> list[tuple[int, str]]:
    """利用可能なカメラを列挙して (index, name) リストを返す"""
    available = []
    for i in range(max_index):
        cap = cv2.VideoCapture(i)
        if cap.isOpened():
            w = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
            h = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
            cap.release()
            available.append((i, f"カメラ {i}  ({w}x{h})"))
        else:
            cap.release()
    return available


def select_camera(default: int | None) -> int:
    """利用可能なカメラを表示してユーザーに選択させる"""
    if default is not None:
        return default

    cameras = list_cameras()
    if not cameras:
        print("カメラが見つかりませんでした。index 0 を使用します。")
        return 0

    if len(cameras) == 1:
        idx, name = cameras[0]
        print(f"カメラ検出: {name}  → 自動選択")
        return idx

    print("\n利用可能なカメラ:")
    for idx, name in cameras:
        print(f"  [{idx}] {name}")

    while True:
        raw = input(f"\nカメラ番号を選択 [{cameras[0][0]}]: ").strip()
        if raw == "":
            return cameras[0][0]
        if raw.isdigit() and int(raw) in {c[0] for c in cameras}:
            return int(raw)
        print("  ※ 表示された番号を入力してください。")


# ---------------------------------------------------------------------------
# メインループ
# ---------------------------------------------------------------------------

def run(camera_index: int, interval: float, model_id: str):
    print(f"モデル読込中: {model_id}")
    from mlx_vlm import load
    t0 = time.perf_counter()
    model, processor = load(model_id)
    print(f"読込完了: {time.perf_counter()-t0:.2f}s\n")

    font = get_font(FONT_SIZE)

    cap = cv2.VideoCapture(camera_index)
    if not cap.isOpened():
        print(f"カメラ {camera_index} を開けませんでした")
        return

    cap.set(cv2.CAP_PROP_FRAME_WIDTH, 1280)
    cap.set(cv2.CAP_PROP_FRAME_HEIGHT, 720)

    worker = InferenceWorker(model, processor, interval=interval)
    worker.start()
    print("カメラ起動。q / ESC で終了。")

    fps_timer = time.perf_counter()
    fps_count = 0
    fps_disp = 0.0

    while True:
        ret, frame = cap.read()
        if not ret:
            print("フレーム取得失敗")
            break

        worker.submit_frame(frame)

        fps_count += 1
        now = time.perf_counter()
        if now - fps_timer >= 1.0:
            fps_disp = fps_count / (now - fps_timer)
            fps_count = 0
            fps_timer = now

        text, infer_sec, count = worker.get_result()
        status = (f"推論: {infer_sec:.1f}s  |  カメラ: {fps_disp:.1f}fps  "
                  f"|  回数: {count}  |  間隔: {interval:.1f}s")

        display = overlay_text(frame, text, infer_sec, font, status)
        cv2.imshow("Gemma4 リアルタイム認識 (q/ESC で終了)", display)

        key = cv2.waitKey(1) & 0xFF
        if key in (ord("q"), 27):
            break

    worker.stop()
    cap.release()
    cv2.destroyAllWindows()
    print("終了しました。")


# ---------------------------------------------------------------------------
# main
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description="FaceTimeカメラ + Gemma4 リアルタイム認識"
    )
    parser.add_argument("--camera", type=int, default=None,
                        help="カメラインデックス（省略時は起動時に選択）")
    parser.add_argument("--interval", type=float, default=2.0,
                        help="推論間隔（秒）(default: 2.0)")
    parser.add_argument("--model", default=MODEL_ID,
                        help=f"モデルID (default: {MODEL_ID})")
    args = parser.parse_args()

    camera_index = select_camera(args.camera)
    run(camera_index, args.interval, args.model)


if __name__ == "__main__":
    main()

サンプルコード２：ptt_voice_input.py

ついでなので、音声データの聞き取りプログラムも作ってみた。
コマンドラインでプロンプトを渡すことで、聞き取りでも評価でも可能にしてあります。
実験してみたら、複数か国語の聞き取りが可能でした。もちろん、１会話の中で混じっていてもちゃんと聞き取りしました。camera_realtime_gemma4.pyと違って音声のリアルタイム処理は色々と面倒なんでやめました。

追加で必要なパッケージは以下の3つ。

pip install miniaudio
pip install rich
pip install pynput

#!/usr/bin/env python3
"""PTT (Push-to-Talk) 音声入力ツール

録音した音声を直接 Gemma 4 (mlx_vlm) に渡して応答を得る。

必要なパッケージ（オプション）:
    pip install pynput   # PTT キー検出（なければトグルモードで動作）

使い方:
    python scripts/ptt_voice_input.py
    python scripts/ptt_voice_input.py --model mlx-community/gemma-4-e2b-it-4bit
    python scripts/ptt_voice_input.py --prompt "音声を文字起こししてください"

操作:
    SPACE (押しっぱなし): 録音（pynput があれば PTT、なければトグル）
    r                  : 録音トグル（pynput なし時）
    c                  : 会話履歴クリア
    q / ESC / Ctrl+C  : 終了
"""

from __future__ import annotations

import argparse
import sys
import tempfile
import threading
import time
import wave
from pathlib import Path

import miniaudio
from rich.console import Console
from rich.live import Live
from rich.panel import Panel
from rich.text import Text

# ---------------------------------------------------------------------------
# 設定
# ---------------------------------------------------------------------------
DEFAULT_MODEL = "mlx-community/gemma-4-e2b-it-4bit"
SAMPLE_RATE   = 16000
CHANNELS      = 1

DEFAULT_PROMPT = "音声の内容を聞いて、日本語で簡潔に応答してください。"


# ---------------------------------------------------------------------------
# 音声録音
# ---------------------------------------------------------------------------

class AudioRecorder:
    """miniaudio を使って録音する。start/stop で制御する。"""

    def __init__(self, sample_rate: int = SAMPLE_RATE, channels: int = CHANNELS):
        self.sample_rate = sample_rate
        self.channels    = channels
        self._chunks: list[bytes] = []
        self._device: miniaudio.CaptureDevice | None = None
        self._lock = threading.Lock()

    def start(self) -> None:
        with self._lock:
            self._chunks.clear()

        chunks_ref = self._chunks
        lock_ref   = self._lock

        def _capture_gen() -> miniaudio.CaptureCallbackGeneratorType:
            while True:
                chunk = yield
                if chunk:
                    with lock_ref:
                        chunks_ref.append(bytes(chunk))

        gen = _capture_gen()
        next(gen)

        self._device = miniaudio.CaptureDevice(
            input_format=miniaudio.SampleFormat.SIGNED16,
            nchannels=self.channels,
            sample_rate=self.sample_rate,
            buffersize_msec=50,
        )
        self._device.start(gen)

    def stop(self) -> bytes:
        """録音停止し PCM バイト列を返す（16bit signed LE）"""
        if self._device:
            self._device.stop()
            self._device.close()
            self._device = None
        with self._lock:
            return b"".join(self._chunks)

    def save_wav(self, pcm: bytes, path: str) -> None:
        with wave.open(path, "wb") as wf:
            wf.setnchannels(self.channels)
            wf.setsampwidth(2)
            wf.setframerate(self.sample_rate)
            wf.writeframes(pcm)

    @staticmethod
    def pcm_duration(pcm: bytes, sample_rate: int = SAMPLE_RATE,
                     channels: int = CHANNELS) -> float:
        return len(pcm) // (2 * channels) / sample_rate


# ---------------------------------------------------------------------------
# 推論ワーカー（音声 → Gemma 4 直接）
# ---------------------------------------------------------------------------

class InferenceWorker:
    """録音した WAV を Gemma 4 に直接渡して応答を生成する"""

    def __init__(self, model, processor, user_prompt: str):
        self._model      = model
        self._processor  = processor
        self._prompt     = user_prompt
        self._history: list[dict] = []   # {"role": "user"|"assistant", "content": str}
        self._status  = "待機中 [SPACE: 録音開始]"
        self._lock    = threading.Lock()
        self._queue: "queue.Queue[str]" = __import__("queue").Queue()
        self._running = True
        self._thread  = threading.Thread(target=self._loop, daemon=True)

    def start(self) -> None:
        self._thread.start()

    def stop(self) -> None:
        self._running = False

    def submit(self, wav_path: str) -> None:
        self._queue.put(wav_path)

    def get_status(self) -> str:
        with self._lock:
            return self._status

    def get_history(self) -> list[dict]:
        with self._lock:
            return list(self._history)

    def clear_history(self) -> None:
        with self._lock:
            self._history.clear()

    # -------------------------------------------------------------------------

    def _set_status(self, msg: str) -> None:
        with self._lock:
            self._status = msg

    def _add_message(self, role: str, content: str) -> None:
        with self._lock:
            self._history.append({"role": role, "content": content})

    def _loop(self) -> None:
        import queue as q_mod

        while self._running:
            try:
                wav_path = self._queue.get(timeout=0.2)
            except q_mod.Empty:
                continue

            self._set_status("Gemma 4 処理中...")
            try:
                answer = self._infer(wav_path)
            finally:
                Path(wav_path).unlink(missing_ok=True)

            self._add_message("assistant", answer)
            self._set_status("待機中 [SPACE: 録音開始]")

    def _infer(self, wav_path: str) -> str:
        try:
            from mlx_vlm import generate
            from mlx_vlm.prompt_utils import apply_chat_template

            formatted = apply_chat_template(
                self._processor,
                self._model.config,
                self._prompt,
                num_images=0,
                num_audios=1,
            )
            result = generate(
                self._model,
                self._processor,
                prompt=formatted,
                audio=wav_path,
                max_tokens=512,
                temperature=0.7,
                verbose=False,
            )
            text = result.text if hasattr(result, "text") else str(result)
            return text.strip() or "（応答なし）"

        except Exception as e:
            return f"[エラー: {e}]"


# ---------------------------------------------------------------------------
# PTT キーボードハンドラー
# ---------------------------------------------------------------------------

class PTTHandler:
    """pynput があれば SPACE ホールドで PTT、なければトグル"""

    def __init__(self, on_start, on_stop):
        self.on_start  = on_start
        self.on_stop   = on_stop
        self._pressing = False
        self._listener = None
        self.mode      = "toggle"

    def start(self) -> bool:
        try:
            from pynput import keyboard

            def on_press(key):
                if self._pressing:
                    return
                if key == keyboard.Key.space:
                    self._pressing = True
                    self.on_start()

            def on_release(key):
                if key == keyboard.Key.space and self._pressing:
                    self._pressing = False
                    self.on_stop()

            self._listener = keyboard.Listener(
                on_press=on_press, on_release=on_release
            )
            self._listener.start()
            self.mode = "ptt"
            return True

        except ImportError:
            return False

    def stop(self) -> None:
        if self._listener:
            self._listener.stop()


# ---------------------------------------------------------------------------
# ターミナル UI
# ---------------------------------------------------------------------------

def build_display(history: list[dict], status: str,
                  recording: bool, mode: str) -> Panel:
    lines = Text()

    if not history:
        lines.append("（まだ会話がありません）\n", style="dim")
    else:
        for msg in history[-20:]:
            style = "bold green"
            label = "AI"
            lines.append(f"[{label}] ", style=style)
            lines.append(msg["content"] + "\n")
            lines.append("\n")

    if recording:
        status_text = Text(status, style="bold red")
    else:
        status_text = Text(status, style="bold yellow")

    mode_label = ("PTT モード (SPACE を押している間録音)"
                  if mode == "ptt" else
                  "トグルモード (SPACE または r: 録音開始/停止)")

    footer = Text()
    footer.append(f"[{mode_label}]  ", style="dim")
    footer.append("c: 履歴クリア  q/ESC: 終了", style="dim")

    content = Text()
    content.append_text(lines)
    content.append("─" * 60 + "\n", style="dim")
    content.append_text(status_text)
    content.append("\n")
    content.append_text(footer)

    return Panel(content, title="[bold]PTT 音声チャット (Gemma 4)[/bold]",
                 border_style="blue")


# ---------------------------------------------------------------------------
# 起動処理
# ---------------------------------------------------------------------------

def _load_env() -> None:
    import os
    env_path = Path(__file__).parent.parent / ".env"
    if not env_path.exists():
        return
    for line in env_path.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, val = line.partition("=")
        os.environ.setdefault(key.strip(), val.strip())


def _load_model(model_id: str, console: Console):
    console.print(f"モデル読込中: {model_id}")
    from mlx_vlm import load
    model, processor = load(model_id)
    console.print(f"[green]✓ 読込完了: {model_id}[/green]")
    return model, processor


# ---------------------------------------------------------------------------
# メインループ
# ---------------------------------------------------------------------------

def run(model_id: str, user_prompt: str) -> None:
    _load_env()

    console = Console()
    console.print("[bold]PTT 音声チャット 起動中...[/bold]")

    model, processor = _load_model(model_id, console)

    recorder = AudioRecorder()
    worker   = InferenceWorker(model, processor, user_prompt)

    MAX_RECORD_SEC = 30.0

    recording      = False
    record_start_t = 0.0
    quit_flag      = threading.Event()

    def start_recording():
        nonlocal recording, record_start_t
        if not recording:
            recording      = True
            record_start_t = time.perf_counter()
            recorder.start()

    def stop_recording():
        nonlocal recording
        if recording:
            recording = False
            pcm = recorder.stop()
            if AudioRecorder.pcm_duration(pcm) < 0.3:
                return
            tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
            tmp.close()
            recorder.save_wav(pcm, tmp.name)
            worker.submit(tmp.name)

    ptt = PTTHandler(on_start=start_recording, on_stop=stop_recording)
    ptt_available = ptt.start()

    worker.start()

    import select as _select
    import tty as _tty
    import termios as _termios

    old_settings = None
    try:
        old_settings = _termios.tcgetattr(sys.stdin)
        _tty.setcbreak(sys.stdin.fileno())
    except Exception:
        pass

    try:
        with Live(console=console, refresh_per_second=8) as live:
            while not quit_flag.is_set():
                # 30 秒超えたら自動停止
                if recording and (time.perf_counter() - record_start_t) >= MAX_RECORD_SEC:
                    stop_recording()

                history = worker.get_history()
                status  = worker.get_status()
                if recording:
                    elapsed = time.perf_counter() - record_start_t
                    remaining = MAX_RECORD_SEC - elapsed
                    disp_status = f"● 録音中... ({remaining:.0f}s)"
                else:
                    disp_status = status
                live.update(build_display(history, disp_status, recording, ptt.mode))

                readable, _, _ = _select.select([sys.stdin], [], [], 0.05)
                if readable:
                    ch = sys.stdin.read(1)
                    if ch in ("q", "\x1b"):
                        quit_flag.set()
                    elif ch == "c":
                        worker.clear_history()
                    elif ch in ("r", " ") and not ptt_available:
                        if recording:
                            stop_recording()
                        else:
                            start_recording()

    except KeyboardInterrupt:
        pass
    finally:
        if old_settings is not None:
            try:
                _termios.tcsetattr(sys.stdin, _termios.TCSADRAIN, old_settings)
            except Exception:
                pass
        ptt.stop()
        worker.stop()
        if recording:
            recorder.stop()

    console.print("\n[bold]終了しました。[/bold]")


# ---------------------------------------------------------------------------
# main
# ---------------------------------------------------------------------------

def main() -> None:
    parser = argparse.ArgumentParser(
        description="PTT 音声入力ツール（音声を Gemma 4 に直接入力）"
    )
    parser.add_argument(
        "--model", default=DEFAULT_MODEL,
        help=f"モデル ID (default: {DEFAULT_MODEL})",
    )
    parser.add_argument(
        "--prompt", default=DEFAULT_PROMPT,
        help=f"音声に添えるプロンプト (default: {DEFAULT_PROMPT!r})",
    )
    args = parser.parse_args()
    run(args.model, args.prompt)


if __name__ == "__main__":
    main()

参考リンク