Amazonのアソシエイトとして、ラズパイダ(raspida.com)は適格販売により収入を得ています。詳しくは当サイトの プライバシーポリシーをご覧ください。
何度か試行錯誤した文字起こし（speech to text＝STT）が完成しました。コード作成にあたり、ChatGPT、Claude、Geminiをそれぞれ使ってみたところ、Geminiが最も良いものを速く作ってくれました。
GeminiもAPIに対応し始めたみたいですが、APIは文章なのでClaudeを使ってみました。
AIに指示した内容と結果も含め、最後にpythonコードも公開します。どうぞ修正して使ってみてください。
マイク選びをしないと苦労することが分かりましたね。
今回までに2記事を公開しています。同じようなセットアップや試行錯誤を確認してください。
完成したのでまとめてみると、次のようなことが分かりました。前回までの記事と被る部分があります。
スライド・ウィンドウ方式の限界と王道VAD処理
最初は15秒を”ウィンドウ”としてずらしていく方式で推論へ投げていました。重複を前提に推論して、重複処理を入れていくわけです。
結局、音声活動検出=VAD（Voice activity detection）で区切って一気に推論する方法が表記揺れを防ぐ最大の手段でした。まぁ、王道ってヤツみたいですね。
一定時間（2秒）の無音を検知した瞬間に「一文が完成した」とみなしてAIに処理を渡します。
これなら指定した秒数ではないため、文脈が途切れた状態で推論させません。ハルシネーションを劇的に減らせて、かつ文章の重複処理はさせないことにしました。VADが最も速く処理できました。最初だけ何も出力されないのはリアルタイム感が薄れますけどね。
リアルタイムを重視して、音声をズラして推論していくのではなく、人間が喋り終えた後の沈黙をトリガーにすれば一度推論した音声は二度と使いません。重複は100%発生しなくなり、認識精度も向上しました。
スライド・ウィンドウ方式
最初に採ったスライド・ウィンドウ方式は、音声の塊ごとに処理させ、それをずらしていくことで重複を排除させるために使いました。
| ▲CPU処理が多い | ○CPU処理が軽い |
|---|---|
| 【音声5秒】を塊にして、【最後の2秒】をタイムレコードで取得 → 次の5秒と冒頭を比較して重複を判断 | 【音声】 + 【2秒の無音】 → 確定・推論 |
これは不採用にしました。
Pi 5の性能を使う
使用したPi 5（16GB）モデルがメモリーをたくさん積んでいますからこれをを贅沢に使い、録音を止めずに全データをメモリに溜め続けて、同時にMoonshineも動かすことにしました。
それぞれをCPU４コアに1コアずつ割り振った並列処理が快適に動作するポイントでした。
+-------------------------------------------------------+
| Raspberry Pi 5 (Broadcom BCM2712 / 4 Cores) |
| |
| +-------------+ +-------------+ +-------------+ |
| | [Core 0] | | [Core 1] | | [Core 2-3] | |
| | | | | | | |
| | 録音処理 | | 推論Moonshine| | OS/その他 | |
| | | | | | | |
| | | | | | | |
| +------+------+ +------+------+ +-------------+ |
| | ^ |
| v | |
| [ メモリ (16GB) ] <------+ |
| (全音声データを保持) |
| |
+-------------------------------------------------------+
この録音音声データはメモリ上に保持させます。コアを分けた録音スレッドと推論スレッドで、AIが計算（推論）にしている間もマイクからの入力を止めません。
ハイブリッドで仕上げる
基本は外部へ音声を送ることなく、Moonshineによってラズパイ内で即座にテキスト化させます。ただ、どうしてもCPUの処理問題もあって完璧には認識できません。
そこで、最後にまとめてクラウド型AI（今回はClaude）で清書させることにしました。
これが時間もかからず、内容によっては精度はバッチリでした。
全てをラズパイで完結させようとせず、ハイブリッドな設計にしました。
完成コードには、OpenAI APIも使えるように両方の処理を追加しました。今回はClaude4.5を使いましたが、API利用はどちらも有料です。最初は最低限の5ドルでもかなり使えますよ。
もし無料がお望みなら、外部API処理はさせず、出力されたmdファイルをブラウザのChatGPTなどにコピペで投げても良いかもしれません。
■Pi 5は8GBモデルがオススメ
マイクの選定
そもそもマイク入力で音声認識による推論がしやすいように、マイクの選定が重要です。
最初に試したマイクは、ピンマイク式の安価な製品AK5371でした。以前の記事の通り、サンプリング44.1kHz固定だったので、きれいな音声が取得できるであろう16kHzには対応しておらず、リサンプリングさせても強制的に11025Hzにされていました。
このリサンプリングはffmpegで処理させていたこともあり、そうとうな負荷でネックした。始めから16kHzで録音できれば、そもそもこの処理は必要ありませんからすぐに別の処理が実行できます。
TSK95K
改めて対応しているだろうと思われる手に入れたマイクはTSK95Kです。これは対応していましたので、とてもクリアな音声で取得できました。リサンプリング処理も削除できます。
多摩電子工業「TSK95K」
＜マイク部＞
●指向性：無指向性マイク2個搭載（モノラル）
●周波数特性：100〜5,000Hz
●集音範囲：半径約2m程度/360°※
※使用状況などにより異なります。
販売価格帯の目安としては3000円前後くらいです。（同じ商品は在庫切れみたい）
- Pi 5にはアナログの音声入力端子がないため基本はUSB接続
- 周波数特性は50Hz～15kHz以上
調べると、人間の声の成分はほとんどが300Hz〜3,000Hzの間に集中しているそうです。だから5,000Hz までカバーしていれば、サ行などの高い音（摩擦音）も最低限判別できるため、Moonshineのような強力なAIなら文脈で補完してくれます。
逆に安物で「20Hz〜20kHz」と書いてある方が、実際は品質が悪い場合があります。
また、STT（音声認識）においては、「モノラル」の方が圧倒的に扱いやすいですし、そもそもMoonshineはモノラルでしか動作しません。
今回みたいに、100〜5,000Hz＋モノラルは音として頼りないと感じますが、人の声をキャプチャーするなら実用的なんです。
価格だけでは言えませんが、**3,000〜5,000円 (国内メーカー製)なら心配なさそうな製品が多いです。**仮に10kHzまでカバーしていたり、高感度（-39dB等）ならささやき声も拾えそうです。
1,000円以下だと、そもそもスペック表（仕様）が書いていない製品が多いのでギャンブルですね。
マイク実力診断スクリプト
マイクが16kHzに対応しているのか、なかなか分かりません。購入した後にはなりますが、対応診断プログラムもAIに書いてもらいました。
デバイス番号なども分かるため一度実行してチェックをオススメします。
実行後に3秒話しかけてチェックします。
import sounddevice as sd
import numpy as np
import time
def check_microphone(device_id, target_rate=16000):
print(f"--- マイク診断開始 (Device ID: {device_id}) ---")
# 1. ハードウェアの対応状況を確認
try:
dev_info = sd.query_devices(device_id)
print(f"【1】デバイス名: {dev_info['name']}")
print(f" デフォルト・レート: {dev_info['default_samplerate']} Hz")
except Exception as e:
print(f"エラー: デバイスが見つかりません。 {e}")
return
# 2. 16kHz/モノラルでの録音テスト
print(f"\n【2】16kHz/モノラルでの動作テスト中...", end="", flush=True)
duration = 3 # 3秒間テスト
try:
# 実際にそのレートで録音を試みる
recording = sd.rec(int(duration * target_rate), samplerate=target_rate, channels=1, dtype='float32', device=device_id)
sd.wait()
print(" [成功]")
print(" → このマイクは16kHz入力をハードウェアでサポートしています。")
except Exception as e:
print(" [失敗]")
print(f" → 16kHzは非対応のようです。エラー内容: {e}")
return
# 3. 音量（感度）のチェック
rms = np.sqrt(np.mean(recording**2))
peak = np.max(np.abs(recording))
print(f"\n【3】入力レベル確認 (3秒間の平均)")
print(f" 平均音量 (RMS): {rms:.4f}")
print(f" 最大音量 (Peak): {peak:.4f}")
if peak < 0.01:
print(" [警告] 音が非常に小さいです。マイクがミュートされているか、感度が低すぎます。")
elif peak > 0.98:
print(" [警告] 音が割れています（クリッピング）。マイクのゲインを下げてください。")
else:
print(" [良好] 適切な入力レベルです。")
print("\n--- 診断完了 ---")
if target_rate == 16000:
print("判定: このマイクはMoonshine STTに最適な設定で動作可能です！")
if __name__ == "__main__":
# デバイスリストを表示して、'USB Audio' という名前が含まれるものを探す
devices = sd.query_devices()
print("--- 認識されているデバイス一覧 ---")
target_id = None
for i, dev in enumerate(devices):
print(f"ID {i}: {dev['name']} (入力: {dev['max_input_channels']}ch)")
# 新しいマイクが届いたら、名前に 'USB' や 'PR-SK95CK' が入っているものを自動選択
if dev['max_input_channels'] > 0 and 'USB' in dev['name']:
target_id = i
if target_id is not None:
print(f"\n自動選択されたデバイス ID {target_id} でテストを開始します。\n")
check_microphone(target_id)
else:
print("\nマイク（入力デバイス）が見つかりませんでした。")
２つのマイクで試した結果はこうでした。
--- 認識されているデバイス一覧 ---
ID 0: AK5371: USB Audio (hw:0,0) (入力: 2ch)
ID 1: vc4-hdmi-0: MAI PCM i2s-hifi-0 (hw:1,0) (入力: 0ch)
ID 2: sysdefault (入力: 128ch)
ID 3: spdif (入力: 2ch)
ID 4: default (入力: 128ch)
自動選択されたデバイス ID 0 でテストを開始します。
--- マイク診断開始 (Device ID: 0) ---
【1】デバイス名: AK5371: USB Audio (hw:0,0)
デフォルト・レート: 44100.0 Hz
【2】16kHz/モノラルでの動作テスト中...Expression 'paInvalidSampleRate' failed in 'src/hostapi/alsa/pa_linux_alsa.c', line: 2048
Expression 'PaAlsaStreamComponent_InitialConfigure( &self->capture, inParams, self->primeBuffers, hwParamsCapture, &realSr )' failed in 'src/hostapi/alsa/pa_linux_alsa.c', line: 2718
Expression 'PaAlsaStream_Configure( stream, inputParameters, outputParameters, sampleRate, framesPerBuffer, &inputLatency, &outputLatency, &hostBufferSizeMode )' failed in 'src/hostapi/alsa/pa_linux_alsa.c', line: 2842
[失敗]
→ 16kHzは非対応のようです。エラー内容: Error opening InputStream: Invalid sample rate [PaErrorCode -9997]
--- 認識されているデバイス一覧 ---
ID 0: TSK95K: USB Audio (hw:0,0) (入力: 2ch)
ID 1: vc4-hdmi-0: MAI PCM i2s-hifi-0 (hw:1,0) (入力: 0ch)
ID 2: sysdefault (入力: 128ch)
ID 3: front (入力: 0ch)
ID 4: surround40 (入力: 0ch)
ID 5: iec958 (入力: 0ch)
ID 6: spdif (入力: 2ch)
ID 7: default (入力: 128ch)
ID 8: dmix (入力: 0ch)
自動選択されたデバイス ID 0 でテストを開始します。
--- マイク診断開始 (Device ID: 0) ---
【1】デバイス名: TSK95K: USB Audio (hw:0,0)
デフォルト・レート: 48000.0 Hz
【2】16kHz/モノラルでの動作テスト中... [成功]
→ このマイクは16kHz入力をハードウェアでサポートしています。
【3】入力レベル確認 (3秒間の平均)
平均音量 (RMS): 0.0353
最大音量 (Peak): 0.4380
[良好] 適切な入力レベルです。
--- 診断完了 ---
判定: このマイクはMoonshine STTに最適な設定で動作可能です！
最終的なコードと結果
例文に桃太郎の冒頭部分をテキトーに3行くらい読み上げました。
以下の出力されたmdファイルを確認すると、生ログはだいぶ心許ない結果でしたが、Claude APIで整形することでほぼ間違いなく奇麗な文章になっています。柴刈りが芝刈りになってしまうのは現代だと仕方ないと思います。他は完璧と言えます。
LLM学習材料でもあるような有名な物語の影響もありますから、全く個人的な内容ならもう少しおかしくなるかもしれません。でも、十分ですね。
# 最終議事録
## 📋 整形済み内容
# 整形後の議事録
昔々あるところに、おじいさんとおばあさんが住んでいました。
おじいさんは山へ芝刈りに、おばあさんは川へ洗濯に行きました。
すると川上から、大きな桃がどんぶらこ、どんぶらこと流れてきました。
めでたし、めでたし。
終了です。
---
## 📝 生ログ
昔々あるところに。
おじいさんとおばあさんがすんでいました。
おじいさんは山へ芝
おばあさんは川へせんたくに行きました。
すると川上から。
大きなももがどんぶらこうどんぶらこうとながれてきました。
めでたしめでた
終了です。
PythonでAPIを扱うにはClaude API Key（有料）、OpenAI API Key（有料）と、それぞれのパッケージをインストールする必要があります。
pip install anthropic
pip install openai
どちらかのAPI KeyがあればOKです。
ChatGPTとOpenAI APIはどちらもOpenAIですが別々の課金です。Claude のAPIも同様です。 APIは使ったトークン分だけ減っていくチャージ方式です。短い文章なら1円以下レベルでした。
例： 入力トークン212、出力トークン124の場合 Claude Sonnet 4.6：入力 $3 / 出力 $15（それぞれ約100万トークンあたり） = 約$0.0025（0.25セント：0.4円）
日本語の場合、漢字・ひらがな・カタカナ、句読点はそれぞれ1〜2トークン程度です。2バイト文字なのでアルファベットよりかかりますね。上記の例だとざっくり日本語100文字くらい送受信したことになります。
最低課金の5ドルなら日本語100文字程度のやりとりを数千回できる計算ですね。80万トークンくらい？ 出力トークンの方が高いので生成量が多かったり、文字以外の入出力があれば変わります。
ハイライトした6、7行目("OPENAI_API_KEY", "")の""にKeyを入れてください。
Claude API最新モデル対応Pythonコードを開く
import sys, time, datetime, re, threading, queue, wave, os
import numpy as np
import sounddevice as sd
from moonshine_voice.transcriber import Transcriber
# --- API設定 (どちらか一方、または両方設定可能) ---
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
CLAUDE_API_KEY = os.getenv("ANTHROPIC_API_KEY", "") # Claudeの場合はこちら
# --- Config ---
DEVICE = 0
RATE = 16000
CHANNELS = 1
VAD_THRESHOLD = 0.01
SILENCE_LIMIT = 2.0
audio_q = queue.Queue()
all_recording_data = []
timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
MD_FILENAME = f"raw_log_{timestamp}.md"
WAV_FILENAME = f"audio_{timestamp}.wav"
FINAL_FILENAME = f"final_minutes_{timestamp}.md"
def clean_text(text):
text = re.sub(r"\[\d+\.\d+s\]", "", text)
return text.replace(" ", "").strip()
def run_ai_refinement(raw_text):
"""終了後に外部API(Claude or OpenAI)を叩いて整形する"""
print(f"\n[System] AIによる整形処理を開始します...")
prompt = f"""
以下のテキストは音声認識によって生成された「生の議事録」です。
誤字、言い直し、重複、フィラーを適切に修正し、読みやすい議事録に整形してください。
重複した発言や「間違えた」等のメタ発言は削除してください。
# 生の議事録
{raw_text}
"""
# 1. Claude API
if CLAUDE_API_KEY:
try:
import anthropic
client = anthropic.Anthropic(api_key=CLAUDE_API_KEY)
# 最新のエイリアス候補を順に試す
model_candidates = ["claude-sonnet-4-5", "claude-3-7-sonnet-latest", "claude-3-5-sonnet-latest"]
for model_name in model_candidates:
try:
print(f"[System] Claude ({model_name}) を試行中...")
message = client.messages.create(
model=model_name,
max_tokens=2048,
messages=[{"role": "user", "content": prompt}]
)
return message.content[0].text
except Exception as e:
if "not_found_error" in str(e):
continue # 次のモデル名を試す
else:
raise e # それ以外のエラーは上に投げる
except Exception as e:
print(f"Claude API実行中に致命的なエラーが発生しました: {e}")
# 2. OpenAI API (Claudeが全滅した場合のフォールバック)
if OPENAI_API_KEY:
try:
import openai
print("[System] OpenAI (gpt-4o) を使用して整形中...")
client = openai.OpenAI(api_key=OPENAI_API_KEY)
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
except Exception as e:
print(f"OpenAIエラー: {e}")
return "APIの実行に失敗しました。モデル名またはAPIキーを確認してください。"
def callback(indata, frames, time, status):
d = indata.copy()
audio_q.put(d)
all_recording_data.append(d)
def run_stt():
print(f"【自律型】ユニバーサル議事録システム起動")
engine = Transcriber(model_path="/home/raspida/.cache/moonshine_voice/download.moonshine.ai/model/base-ja/quantized/base-ja")
current_sentence_buffer = []
silence_start = None
is_speaking = False
full_raw_text = ""
with open(MD_FILENAME, "w", encoding="utf-8") as f:
f.write(f"# 会議・生ログ\n- 開始: {datetime.datetime.now()}\n---\n")
with sd.InputStream(device=DEVICE, channels=CHANNELS, samplerate=RATE, callback=callback):
print("認識中... (Ctrl+Cで終了)")
try:
while True:
try:
data = audio_q.get(timeout=0.1).flatten()
except queue.Empty: continue
current_sentence_buffer.append(data)
volume = np.abs(data).mean()
if volume > VAD_THRESHOLD:
is_speaking = True
silence_start = None
else:
if is_speaking:
if silence_start is None: silence_start = time.time()
if time.time() - silence_start > SILENCE_LIMIT:
audio_segment = np.concatenate(current_sentence_buffer)
results = engine.transcribe_without_streaming(audio_segment)
if results:
res_obj = results[0] if isinstance(results, list) else results
text = clean_text(getattr(res_obj, 'text', str(res_obj)))
if text:
entry = f"[{datetime.datetime.now().strftime('%H:%M:%S')}] {text}\n"
sys.stdout.write(entry); sys.stdout.flush()
with open(MD_FILENAME, "a", encoding="utf-8") as f: f.write(entry)
full_raw_text += text + "\n"
current_sentence_buffer = []; is_speaking = False; silence_start = None
except KeyboardInterrupt:
print("\n[System] 終了処理中...")
if all_recording_data:
combined = np.concatenate(all_recording_data)
int16_data = (combined * 32767).astype(np.int16)
with wave.open(WAV_FILENAME, 'wb') as wf:
wf.setnchannels(CHANNELS); wf.setsampwidth(2); wf.setframerate(RATE)
wf.writeframes(int16_data.tobytes())
refined_text = run_ai_refinement(full_raw_text)
with open(FINAL_FILENAME, "w", encoding="utf-8") as f:
f.write(f"# 最終議事録\n\n## 📋 整形済み内容\n{refined_text}\n\n---\n## 📝 生ログ\n{full_raw_text}")
print(f"全て完了！\n- 音声: {WAV_FILENAME}\n- 最終議事録: {FINAL_FILENAME}")
if __name__ == "__main__":
run_stt()
エラーなく実行されれば、全体をWAV形式の音声ファイルとして出力し、議事録としてmdファイルを出力できます。聞き直したりテキストから確認や修正をすることができます。
[System] 終了処理中...
[System] AIによる整形処理を開始します...
[System] Claude (claude-3-7-sonnet-latest) を試行中...
/home/raspida/stt-meeting/universal_meeting_logger.py:53: DeprecationWarning: The model 'claude-3-7-sonnet-latest' is deprecated and will reach end-of-life on February 19th, 2026.
Please migrate to a newer model. Visit https://docs.anthropic.com/en/docs/resources/model-deprecations for more information.
message = client.messages.create(
[System] Claude (claude-sonnet-4-5) を試行中...
全て完了！
- 音声: audio_20260312_025428.wav
- 最終議事録: final_minutes_20260312_025428.md
これとは別に、VADで処理するコードも載せておきます。
VAD処理で音声認識して出力
import sys, time, datetime, re, threading, queue, wave
import numpy as np
import sounddevice as sd
from moonshine_voice.transcriber import Transcriber
# --- Config ---
DEVICE = 0
RATE = 16000
CHANNELS = 1
VAD_THRESHOLD = 0.01 # 音量のしきい値（環境に合わせて微調整）
SILENCE_LIMIT = 2.0 # 何秒無音が続いたら「区切り」とするか
audio_q = queue.Queue()
all_recording_data = []
timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
MD_FILENAME = f"meeting_log_{timestamp}.md"
WAV_FILENAME = f"meeting_audio_{timestamp}.wav"
def clean_text(text):
text = re.sub(r"\[\d+\.\d+s\]", "", text)
return text.replace(" ", "").strip()
def callback(indata, frames, time, status):
d = indata.copy()
audio_q.put(d)
all_recording_data.append(d)
def run_stt():
print(f"【VAD搭載・最終形態】議事録システム起動")
engine = Transcriber(model_path="/home/raspida/.cache/moonshine_voice/download.moonshine.ai/model/base-ja/quantized/base-ja")
current_sentence_buffer = []
silence_start = None
is_speaking = False
with open(MD_FILENAME, "w", encoding="utf-8") as f:
f.write(f"# 会議議事録 (VAD Mode)\n- 開始: {datetime.datetime.now()}\n---\n")
with sd.InputStream(device=DEVICE, channels=CHANNELS, samplerate=RATE, callback=callback):
print("認識中... 発言の終わりを検知して解析します。")
while True:
# キューからデータを取得
try:
data = audio_q.get(timeout=0.1).flatten()
except queue.Empty:
continue
current_sentence_buffer.append(data)
# 簡易VAD：音量の絶対値の平均で判断
volume = np.abs(data).mean()
if volume > VAD_THRESHOLD:
is_speaking = True
silence_start = None # 無音カウントをリセット
else:
if is_speaking:
if silence_start is None:
silence_start = time.time()
# 指定時間以上の無音が続いたら「発言終了」とみなす
if time.time() - silence_start > SILENCE_LIMIT:
# 推論へ
audio_segment = np.concatenate(current_sentence_buffer)
results = engine.transcribe_without_streaming(audio_segment)
if results:
res_obj = results[0] if isinstance(results, list) else results
text = clean_text(getattr(res_obj, 'text', str(res_obj)))
if text:
now_str = datetime.datetime.now().strftime('%H:%M:%S')
output = f"[{now_str}] {text}\n"
sys.stdout.write(output)
sys.stdout.flush()
with open(MD_FILENAME, "a", encoding="utf-8") as f:
f.write(output)
# バッファをリセット
current_sentence_buffer = []
is_speaking = False
silence_start = None
if __name__ == "__main__":
try:
run_stt()
except KeyboardInterrupt:
if all_recording_data:
print(f"\n[System] WAV保存中...")
combined = np.concatenate(all_recording_data)
int16_data = (combined * 32767).astype(np.int16)
with wave.open(WAV_FILENAME, 'wb') as wf:
wf.setnchannels(CHANNELS); wf.setsampwidth(2); wf.setframerate(RATE)
wf.writeframes(int16_data.tobytes())
print(f"完了: {MD_FILENAME}")
STTは既に製品やサービス化されていますので目新しくはありません。それでもPi 5だけで実行できたのは感動します。結構、実用的ですよ。
出来上がり体験よりも、AIが作成してくれたコードが上手く動くことに驚きますかね？