はじめに
自作の流星自動検出システム 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()