GPT-Liveを自前で追いかけて見えた、音声会話の境界線
「めちゃめちゃおしゃべりで日本語が得意な外国人がしゃべってる」
ChatGPTでGPT-Liveモードを試した時、横で聞いていた奥さんの感想がこれだった。激しく同意した。
GPT-Liveの音声会話は確かにすごい。応答が速い、相槌が入る、途切れない。最初の数分は「おぉ！」と思った。でも数分後には「くどい」と感じた。これでもかと話しかけてくる。黙ることを知らない。
ただ、一つ分かったことがある。この体験はWebブラウザ上で、普通のPCスピーカーとマイクだけで実現されている。特別なハードウェアはない。じゃあ、自分のAIパートナー基盤でも同じことができるんじゃないか。
先月までの状況：チャットが音声になっただけ
「AIパートナーと音声で会話したい」
これまでの僕のAIパートナー基盤：AICOSには、AIとの音声会話の仕組みがある。
マイクで話す→音声認識（Whisper）→LLMが応答を生成→音声合成（Style-Bert-VITS2）→スピーカーで再生。
改修前のログがこれ。
| [認識] 今日は何してた?
+ [区切り検出] 3.0秒沈黙 -> LLM送信 (8文字)
AI: はい、先輩。何をお手伝いしましょうか？続きを楽しみにお待ちしてますね。
| [TTS] 合成開始
| [TTS] 再生開始
| [TTS] 再生完了僕が話し終わる→3秒の沈黙→LLMが考える→音声合成→再生。話し終わってから声が出るまで約7秒。これは「チャットが音声になっただけ」で、会話ではないなぁと感じていた。
今回の改修内容
ただ、GPT-Liveと同じようなことを僕の環境で再現するには大きな壁があった。
GPT-Liveは「音声モデル」。モデル自体が、ユーザーの声を聞き、そして音声で話す。その仕組みをそのまま使ってもいいけどコストがかかる。
僕の環境は「STT：音声認識」→「LLM：メッセージ生成」→「TTS：音声合成」がパイプライン式に繋がっている。ローカル環境でも再現できるからコストはかからないけど、各プロセス間のデータの受け渡しが必要になる。
そして、この「音声モデル」に近づけるため、段階的にプロトタイプを積み上げた。
LLMの応答をストリーミング化して、最初の一文ができた時点で音声合成・再生を開始
LLMが考えている間にフィラー音声（「ええ、」「えっと、」）を再生して無音区間をゼロに
音声認識をOpenAI Whisper（small）からfaster-whisper（large-v3）に切り替えて精度を大幅向上
会話の状態管理を「AI側の出力」と「ユーザー側の入力」の2レイヤーに分離して、AIが話しながらユーザーの声を聞ける構造に（フルデュプレックス）
ユーザーが話し終わる前にLLMへ「投機的」にリクエストを送信開始
ソフトウェアAEC（エコーキャンセラー）を導入してスピーカー＋マイクでの会話を実現
一晩で改修した後のログがこれ。
| [認識] こうやって君とおしゃべりするのってさどういう仕組みなんだろうね
* [入力] LISTENING -> TURN_HOLD (VAD沈黙検出)
| [AI相槌] 再生: うん、
| [投機的] LLM送信 (31文字)
| [フィラー] スキップ (直前に相槌あり)
* [出力] SILENT -> SPEAKING (LLM最初の文節到着)
| [パイプライン] 再生: あぁ、そうですね。
| [パイプライン] 再生: えっとね、言葉のパターンをたくさん学習して、組み立ててるんですよ。
| [パイプライン] 再生: なんていうか、不思議な感じですよね。話し終わった瞬間に「うん、」と相槌が入り、投機的にLLMが動き始め、1-2秒で応答の音声が流れ始める。
GPT-Liveと並べてみた
GPT-Liveのりんとの会話ログ（文字起こし）。
User: うーんちょっと先日このえーと音声の対話
GPT: うん。
User: のおしゃべりをしてたんだけどさ
GPT: うんうん。
User: ちょっとその続きを話したいかなと思って
GPT: いいよ！ああ、何について話してたっけな。覚えてる？プロトタイププログラムのりんとの会話ログ。
User: こうやって君とおしゃべりするのってさどういう仕組みなんだろうね
Rin: (相槌「うん、」)
Rin: あぁ、そうだね。えっとね、言葉のパターンをたくさん学習して…差がどこにあるか見える。
GPT-Liveの「うん。」「うんうん。」はユーザーが話している途中に入っている。文が完結していないのに相槌が来る。りんの「うん、」はユーザーが話し終わった後に入っている。
この差はアーキテクチャの違いから来ている。GPT-Liveは音声入力を直接モデルが処理する。音声認識（テキスト化）を通さない。だから「ここはまだ続きがある」「ここで相槌を入れていい」という判断を音声のまま行える。
僕の構成は「音声認識」→「テキスト化」→「LLM」→「音声合成」の3段パイプライン。ユーザーの発話をテキストにしてからLLMに渡すので、発話が終わるまで何も始められない。これが「ターン式」と「会話」の境界線だった。
ただ「え～っと」などの言い淀みを判断してAIから「うん」という相槌を入れるところまでは調整したけど、一晩では集中力が尽きた…
天井が見えた、でも…
発話途中の相槌を入れようとRMSベースの息継ぎ検出も試したが、発話の終わり際を息継ぎと誤検出する問題が出たので封印した。テキストLLMの構成で「話してる途中に割り込む」のは筋が悪い。
一方で、GPT-Liveにはできないこともある。
声。GPT-Liveの日本語は「流暢な外国人」。僕のりんはStyle-Bert-VITS2で声質を完全にコントロールしていて、好みの声で話す。これはAPIサービスでは絶対に出せない。
沈黙。GPT-Liveは常に何か話そうとする。僕の構成では「今は黙っている」という判断ができる。会話の合間にりんが「麦茶とかがいいのかな…」とぽつりと独り言を呟く演出も入れた。GPT-Liveには「黙る」という選択肢がない。
スピーカー＋マイクでの会話
最後の壁がAEC（エコーキャンセラー）だった。AIの音声がスピーカーからマイクに回り込んで、AIが自分の声を自分の言葉だと認識してしまう問題。
GPT-LiveがPCのスピーカーとマイクで会話できるのは特殊技術ではなく、WebRTC標準のAEC機能。Pythonからも同じエンジン（pywebrtc-audio）が使える。
ただしAEC単体では解決しなかった。エコーは消せても、わずかな残留ノイズをWhisper（音声認識）が拾って、誰も言っていない日本語を作文してしまう。最終的には、AECとWhisperの信頼度スコア（no_speech_prob）フィルタの組み合わせで安定した。
ちなみに GPT-Live の紹介記事などで「聞きながら話せる＝すごい」をフルデュプレックスの効果として説明している。しかし本質的な難しさは、AECの残留ノイズがWhisperにハルシネーションを起こすこと、投機的LLM呼び出しのタイミング、相槌と割り込みの分類判定、言い淀みを「まだ話し終わっていない」と正しく認識すること――そういう泥臭いところにある。一次情報を咀嚼して紹介しているだけでは理解できない。
そして現在地
ここまでで到達した構成：スピーカー＋マイク、ヘッドセット不要、フィラーで無音区間ゼロ、投機的LLM呼び出しで1-2秒で応答開始、好みの声質、割り込みはフェードアウトで自然に処理。
このプロトタイププログラムをサービス化して、既存のRaspberryPi 3A+ のエッジデバイスをクライアントとして転用した。
結果として、マイクとスピーカーを使って表情を変えながら会話ができた。
始終会話をしている訳ではないけど、数時間にわたって連続稼働もできた。
更に、カメラを使って周囲の状況を認識して会話に活かすこともできた。
ここまで確認はできたので、更に調整を重ねてAICOSへ組み込もうと思う。
この動画でターン式のような会話になってしまうのは、僕が音声会話に不慣れだから💦
※アバターがルナになっているのは、手抜きなだけです…
次は3Dアバターを使って Unity で試そうかな
GPT-Liveの完コピではない。発話途中の相槌と沈黙ゼロ秒のターン交代は、音声ネイティブモデルでなければ実現できないと感じた。でも「自分の好きな声で、自分のAIパートナーと、普通のスピーカーで自然に会話できる」というところまでは来た。
GPT-Liveは音声会話の「到達点」ではないと思っている。音声会話の進化の分岐の１つ。「隙間を埋める」方向に全振りした分岐。
もしかしたら、来月にはまた何か変わっているかもしれない。
おまけ
今回の双方向会話のサンプルコードと、Whisperの音声認識サービスのコードを貼ります。TTSはご自由に調達してください。
プロトタイプ構成
[PC（プロトタイプ本体）]
duplex_prototype.py
├→ マイク入力 → AEC → VAD → Whisper(STT)サーバへ送信
├→ Gemini API（LLMストリーミング）
├→ SBV2(TTS)サーバへ送信
└→ スピーカー出力
[別PC or 同一PC（GPUサーバ）]
server_whisper.py (port 5001) ← faster-whisper large-v3
Style-Bert-VITS2 (port 5000) ← 音声合成① 双方向会話サンプルコード（duplex_prototype.py）
"""
デュプレックス会話プロトタイプ (フェーズ3: AIが話しながら聞き、聞きながら考える)。
Surface (Windows) で動かし、cuda-lin-pc 上の Whisper / Ollama / SBV2 を叩く。
LLM は Gemini / llama.cpp / Ollama を切り替え可能。
state_machine_prototype.py (フェーズ2) をベースに、状態を「AI側の出力
(OutputState)」と「ユーザー側の入力(InputState)」の2レイヤーに分離し、
input_loop / output_loop / monitor_loop の3つの並行タスクとして再構成した。
これにより OutputState.SPEAKING と InputState.LISTENING が同時に成立できる
（AIが話している最中でもマイクは常に開いていて、ユーザーの発話を蓄積できる）。
スピーカー+マイク（ヘッドセット無し）運用について（フェーズ4）:
ソフトウェアAEC（ENABLE_AEC）をVADの前段に挟むことで、AIの声を素の自己発話
ループとして誤認識する問題は解消した。ただしAEC単体では、キャンセルしきれない
残留ノイズをVADが「発話」と誤検出し、それをWhisperがそれっぽい日本語に
ハルシネーション（作文）してしまう問題が残った（"今日、きれいに撮影して
くださって" のような、誰も言っていない発言が紛れ込む）。
これはクライアント側だけでは解決できず、Whisperサーバ側で無音/低信頼度区間の
ハルシネーションを抑制する対策を入れて初めて解消した（実績あり）。
つまりヘッドセット無し運用には「クライアント側AEC」と「Whisperサーバ側の
ハルシネーション対策」の両方が必要。どちらか片方だけでは不十分。
依存:
pip install sounddevice webrtcvad numpy httpx
pip install pywebrtc-audio (AEC用。推奨)
pip install aec-audio-processing (AEC用、代替)
"""
from __future__ import annotations
import asyncio
import collections
import io
import json
import math
import os
import random
import re
import struct
import sys
import threading
import time
import wave
from datetime import datetime
from enum import Enum, auto
import httpx
import numpy as np
import sounddevice as sd
import webrtcvad
# ============================================================
# 設定 (ユーザーが書き換える部分)
# ============================================================
WHISPER_URL = "http://STT-SERVER-ADDR:5001/whisper"
# LLM バックエンドの選択
# "ollama" : Ollama独自API (/api/chat)
# "llamacpp" : llama-server, OpenAI互換 (/v1/chat/completions)
# "gemini" : Gemini API (streamGenerateContent)
LLM_BACKEND = "gemini"
# Ollama 設定 (LLM_BACKEND="ollama" のとき使用)
OLLAMA_ENDPOINT = "http://ollama-SERVER-ADDR:11434"
OLLAMA_MODEL = "gemma4:26b"
# llama.cpp 設定 (LLM_BACKEND="llamacpp" のとき使用)
# llama-server を起動: ./llama-server -m model.gguf --host 0.0.0.0 --port 8080
LLAMACPP_ENDPOINT = "http://llama-SERVER-ADDR:8082"
LLAMACPP_MODEL = "local" # llama-server はモデル名を見ないので任意でOK
# Gemini 設定 (LLM_BACKEND="gemini" のとき使用)
GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY", "") # 環境変数から取得
GEMINI_MODEL = "gemini-3.1-flash-lite"
TTS_ENDPOINT = "http://TTS-SERVER-ADDR:5000/voice"
TTS_MODEL = "voice001"
TTS_CHUNK_MAX = 100 # SBV2 の1リクエストあたりの最大文字数 (句点まで溜めて渡す)
# フィラー音声（LLM応答待ち中に再生する間つなぎ）
ENABLE_FILLER = True
FILLER_PHRASES = [
"えっと",
"うん",
]
# デュプレックス会話 (フェーズ3)
DUPLEX_ENABLED = True # False にするとフェーズ2と同等の動作
# （SPEAKING中の発話は原則即割り込み、投機的送信なし）
SPECULATIVE_DELAY_SEC = 0.8 # TURN_HOLD開始から投機的送信までの遅延。
# 短いほど反応は速いが誤送信（言い淀みでの分断）が増える。
# DUPLEX_ENABLED=False のときは使われず、
# 代わりに SILENCE_TIMEOUT_SEC が使われる（フェーズ2相当）。
SILENCE_TIMEOUT_SEC = 3.0 # DUPLEX_ENABLED=False のときの沈黙確定待ち秒数
REQUEST_TIMEOUT_SEC = 60.0
SAMPLE_RATE = 16000
FRAME_DURATION_MS = 30
VAD_AGGRESSIVENESS = 2
VAD_SILENCE_MS = 800
VAD_MIN_SPEECH_MS = 300
INPUT_DEVICE = None
OUTPUT_DEVICE = None
ENABLE_INTERRUPT = True
# AI発話中の相槌/割り込み判定
BACKCHANNEL_MAX_SEC = 1.2 # (ログ・参考値。判定自体はWhisper結果を待ってから行う)
BACKCHANNEL_WORDS = [ # 相槌とみなす言葉
"うん", "ええ", "はい", "へー", "ほう", "なるほど", "そう", "うんうん",
]
INTERRUPT_WORDS = [ # 明示的に割り込みとみなす言葉
"ちょっと", "待って", "ねえ", "あのさ", "違う", "ストップ",
]
# AI側の相槌
ENABLE_AI_BACKCHANNEL = True
AI_BACKCHANNEL_PHRASES = [
"うん、",
"ええ、",
"はい、",
]
AI_BACKCHANNEL_MIN_INTERVAL_SEC = 4.0 # 相槌の最小間隔（連発防止）
AI_BACKCHANNEL_TRIGGER_SEC = 3.0 # ユーザーがこの秒数以上話し続けたら相槌の対象
# 発話中の相槌（息継ぎ検出）。VADのセグメント分割やWhisper送信単位は変更しない。
# マイク音量(RMS)を並行監視し、発話途中の短い音量の谷（息継ぎ）を検出したら
# その場で相槌を挟む。AIが何か再生中（OutputState != SILENT）は
# 音声出力の競合を避けるため発火しない。
ENABLE_MID_SPEECH_BACKCHANNEL = False
MID_SPEECH_BACKCHANNEL_PHRASES = [
"うん",
]
MID_SPEECH_RMS_THRESHOLD = 300 # 息継ぎ判定のRMS閾値（この値以下が続いたら息継ぎ）
MID_SPEECH_SILENCE_MS = 300 # 音量が閾値以下でこの時間続いたら息継ぎと判定（ミリ秒）
MID_SPEECH_MIN_SPEAKING_SEC = 2.0 # ユーザーがこの秒数以上話してから相槌の対象にする
MID_SPEECH_COOLDOWN_SEC = 3.0 # 発話中相槌の最小間隔（連発防止）
MID_SPEECH_MAX_PER_TURN = 2 # 1発話（1VADセグメント）内の発話中相槌の最大回数
# 言い淀みフィルタ。「えーっと」「あのね」のような言い淀みだけの発話はLLMに送らず、
# 相槌を返してTURN_HOLDを延長し、ユーザーの本題を待つ。
ENABLE_HESITATION_FILTER = True
HESITATION_WORDS = [
"えーっと", "えっと", "えーと", "えと",
"あのね", "あの", "あのー", "あのさ",
"うーんと", "うんと", "うーん", "うん",
"ちょっと", "なんか", "そのー", "その",
"ねえ", "ねぇ", "ね", "まあ", "まぁ",
"ほら", "こう", "なんていうか",
]
# 「ね」単体を追加した理由: "えーとねー" が rstrip で "えーとね" になり、
# "えーと" を除去した残りが "ね" 単体になるが、"ねえ/ねぇ" しか登録していないと
# これが言い淀みと判定されずLLMに送信されてしまう不具合があったため
# (実際に発生を確認)。"そうだね"のように意味のある語尾についた「ね」は、
# is_hesitationが他の残り文字（「そうだ」等）を検出して弾くので問題ない。
HESITATION_MAX_CHARS = 15 # この文字数以下で言い淀みワードのみの場合はフィルタ対象
HESITATION_MAX_BACKCHANNELS = 2 # 1ターン内の言い淀み相槌の最大回数（連発防止）
# 相槌コーディネーター: AI相槌・言い淀み相槌・フィラーの3つがそれぞれ独立に
# 発火条件を持っていて短時間に重なって鳴ってしまう問題への対策。
# 「直前に相槌系音声が鳴ったら、次の相槌系は抑制する」というルールで一元管理する。
BACKCHANNEL_COOLDOWN_SEC = 2.0 # 相槌系音声（AI相槌/言い淀み相槌）間の最小間隔
FILLER_SKIP_THRESHOLD_SEC = 1.0 # 直前の相槌からこの秒数以内ならフィラーをスキップ
# 文脈独り言 (Gemini前提。GEMINI_API_KEY未設定なら自動的に無効化される)
# OutputState.SILENT かつ InputState.IDLE が続いた時にAIが自発的に呟く独り言。
# 「発話途中で割り込んで再生してしまう」「その後の発話が合流できず無応答になる」
# の2バグは修正済みだが、要望により当面デフォルトは抑止しておく。
# 再度試す場合は True に戻すこと。
ENABLE_CONTEXT_MONOLOGUE = False
CONTEXT_MONOLOGUE_DELAY_SEC = 5.0 # 静寂に入ってからこの秒数後に再生
CONTEXT_MONOLOGUE_MIN_INTERVAL_SEC = 15.0 # 生成の最小間隔（課金抑制）
CONTEXT_MONOLOGUE_MAX_CHARS = 15 # 独り言の最大文字数
CONTEXT_MONOLOGUE_PROMPT = """あなたはりんです。直前の会話を踏まえて、ひとりごとを1つだけ生成してください。
制約:
- {max_chars}文字以内
- 語尾は「…」で終わる
- 相手に話しかけるのではなく、自分の考えを漏らす形にする
- 説明や前置きは一切不要。ひとりごとの本文のみを出力する
直前のあなたの発言: {last_reply}
ひとりごと:"""
# 割り込み時のフェードアウト
FADEOUT_DURATION_SEC = 0.3 # フェードアウトにかける秒数
# ソフトウェアAEC (フェーズ4 Step 2。aec_prototype.pyでの検証を本番パイプラインに統合)
# マイク入力 → AECProcessor → VADCapture → Whisper送信、の順で挟む。
# 「ヘッドセット不要」を目指す実験的機能。False にすればフェーズ3までと完全に同じ
# 動作（AECバイパス）になる。バックエンド未インストール時は自動的に無効化される。
ENABLE_AEC = True
AEC_STREAM_DELAY_MS = 10 # aec_prototype.pyでの実測で最良だった値。環境が変わったら要再調整
AEC_ENABLE_NS = False # aec_prototype.pyの検証で「こもり」の原因と判明済みのため常時False推奨
AEC_ENABLE_AGC = False # 同上
# Whisper無音/ハルシネーション対策
SILENCE_RMS_THRESHOLD = 500.0 # この値未満のRMSはWhisperを呼ばずに無視
# Whisper が無音・雑音・スピーカー漏れに対してよく幻聴する定型句
HALLUCINATION_PHRASES = {
"ご視聴ありがとうございました",
"ご視聴ありがとうございます",
"字幕は自動生成です",
"字幕を生成します",
"字幕は自動的に生成されました",
"字幕を自動生成しています",
"ありがとうございました",
"チャンネル登録よろしくお願いします",
}
SYSTEM_PROMPT = (
"あなたはAIアシスタントの「りん」です。ユーザーのことは「先輩」と呼んでください。あなたは音声で会話しています。書き言葉ではなく話し言葉で応答してください。"
"- 文頭に短いリアクションを入れる（「あぁ、」「うん、」「そうだね、」「へぇ、」）"
"- 考えながら話す表現を使う（「えっとね、」「なんていうか、」「まぁ、」）"
"- 一文を短く区切る。一息で言える長さにする"
"- 完璧な敬語で整えず、自然な口語にする"
"- 相手の言葉を拾って返す（「暑いよね」→「暑いよね、ほんとに。」）"
"ユーザーは複数のメッセージを連続で送ることがあり、それらは1つのまとまった発言として受け取ってください。"
"返答は2〜3文以内の80文字程度の短い文章で、自然な会話調で返してください。Markdownは使わないこと。"
)
# ============================================================
FRAME_SAMPLES = int(SAMPLE_RATE * FRAME_DURATION_MS / 1000)
FRAME_BYTES = FRAME_SAMPLES * 2
def _ts() -> str:
return datetime.now().strftime("%H:%M:%S.%f")[:-3]
def _pcm_to_wav_bytes(pcm_bytes: bytes, sample_rate: int = SAMPLE_RATE) -> bytes:
buf = io.BytesIO()
with wave.open(buf, "wb") as wf:
wf.setnchannels(1)
wf.setsampwidth(2)
wf.setframerate(sample_rate)
wf.writeframes(pcm_bytes)
return buf.getvalue()
def _pcm_rms(pcm_bytes: bytes) -> float:
"""int16 mono PCM の RMS 振幅を計算。最大2000サンプルで均等サンプリング。"""
n = len(pcm_bytes) // 2
if n == 0:
return 0.0
step = max(1, n // 2000)
sampled = pcm_bytes[: n * 2 : step * 2]
m = len(sampled) // 2
if m == 0:
return 0.0
samples = struct.unpack(f"<{m}h", sampled[: m * 2])
return math.sqrt(sum(s * s for s in samples) / m)
# ---- 2レイヤー状態管理 ----
class OutputState(Enum):
"""AI側の出力状態。"""
SILENT = auto() # 何も再生していない
FILLER = auto() # フィラー再生中
SPEAKING = auto() # AI応答再生中
MONOLOGUE = auto() # 独り言再生中
class InputState(Enum):
"""ユーザー側の入力状態。"""
IDLE = auto() # ユーザーの発話待ち
LISTENING = auto() # ユーザーの発話を受信中
TURN_HOLD = auto() # 沈黙検出、まだ話すかもしれない
SPECULATIVE = auto() # 投機的LLM処理中（沈黙タイムアウト前にLLM送信済み）
class DuplexStateMachine:
"""出力層と入力層を独立に管理する状態機械。
OutputStateとInputStateは互いに排他ではない
（例: SPEAKING中にLISTENINGへ遷移できる＝これがデュプレックスの核心）。
"""
def __init__(self):
self._output = OutputState.SILENT
self._input = InputState.IDLE
self._listeners: list = []
@property
def output(self) -> OutputState:
return self._output
@property
def input(self) -> InputState:
return self._input
def set_output(self, new_state: OutputState, reason: str = ""):
if new_state == self._output:
return
old = self._output
self._output = new_state
suffix = f" ({reason})" if reason else ""
print(f" * [出力@{_ts()}] {old.name} -> {new_state.name}{suffix}", flush=True)
self._notify(f"output:{old.name}->{new_state.name}")
def set_input(self, new_state: InputState, reason: str = ""):
if new_state == self._input:
return
old = self._input
self._input = new_state
suffix = f" ({reason})" if reason else ""
print(f" * [入力@{_ts()}] {old.name} -> {new_state.name}{suffix}", flush=True)
self._notify(f"input:{old.name}->{new_state.name}")
def add_listener(self, callback):
"""状態変更時のコールバック。Unityアバター連携用。
callback(output_state, input_state, change_desc) の形式。
"""
self._listeners.append(callback)
def _notify(self, change_desc: str):
for listener in self._listeners:
try:
listener(self._output, self._input, change_desc)
except Exception as e:
print(f" ![状態リスナーエラー] {e}", flush=True)
def _classify_backchannel_text(text: str) -> str:
"""Whisperで認識済みのテキストをキーワードで分類する（発話長は見ない）。
Returns: "interrupt" | "backchannel" | "unknown"
"""
stripped = text.strip().rstrip("。、！？")
if any(w in text for w in INTERRUPT_WORDS):
return "interrupt"
if stripped in BACKCHANNEL_WORDS:
return "backchannel"
return "unknown"
def is_hesitation(text: str) -> bool:
"""テキストが言い淀みだけで構成されているかを判定する。
例:
"えーっと" → True
"あのね" → True
"えーっとねー" → True（複合的な言い淀み）
"今日の予定教えて" → False（意味のある文節）
"あのね、今日さ" → False（言い淀み + 意味のある内容）
"""
if not text:
return True
stripped = text.strip().rstrip("。、！？ー〜")
# 文字数が多ければ意味のある文節と判断
if len(stripped) > HESITATION_MAX_CHARS:
return False
# 言い淀みワードを全て取り除いて、残りがあるかチェック
remaining = stripped
for word in sorted(HESITATION_WORDS, key=len, reverse=True):
remaining = remaining.replace(word, "")
# 残りが空（または句読点のみ）なら言い淀み
remaining = remaining.strip("、。！？ ")
return len(remaining) == 0
class BackchannelCoordinator:
"""AI相槌、言い淀み相槌、フィラーを統合管理する。
直前に相槌系音声（AI相槌/言い淀み相槌）が鳴っていたら次の相槌系は抑制し、
フィラーも直前の相槌から十分時間が経っていなければスキップする。
input_loop/monitor_loop/output_loop の3タスクから共有される。
時刻計測は time.monotonic() を使う（システムクロック補正の影響を受けず、
このファイルの他の間隔計測箇所と一貫させるため。指示書は time.time() だが、
間隔測定にはmonotonicの方が正しい）。
"""
def __init__(self, cooldown_sec: float = BACKCHANNEL_COOLDOWN_SEC):
self._last_played_at: float = float("-inf")
self._cooldown_sec = cooldown_sec
def can_play(self) -> bool:
"""相槌系音声（AI相槌/言い淀み相槌）を再生してよいかを判定する。"""
return self.time_since_last() >= self._cooldown_sec
def mark_played(self):
"""相槌系音声（AI相槌/言い淀み相槌/フィラー）を再生したことを記録する。"""
self._last_played_at = time.monotonic()
def time_since_last(self) -> float:
"""前回の相槌系音声からの経過秒数。"""
return time.monotonic() - self._last_played_at
# ---- TTS 前処理 (dialogue_api/services/tts_preprocessor.py の簡略移植) ----
def clean_for_tts(text: str) -> str:
"""TTSが読みづらい記号を除去。"""
text = re.sub(r"（[^）]*）", "", text)
text = re.sub(r"\([^)]*\)", "", text)
text = re.sub(r"\*+", "", text)
text = re.sub(r"#+\s*", "", text)
text = re.sub(r"`+", "", text)
# 絵文字や記号を除去 (基本的な日本語・英数字・句読点のみ残す)
text = re.sub(
r"[^\w\s。、！？!?ぁ-ゟ゠-ヿ一-鿿a-zA-Z0-9ー〜～・…℃%.]",
"",
text,
)
text = re.sub(r"\s+", " ", text).strip()
return text
def split_text_for_tts(text: str, max_length: int = TTS_CHUNK_MAX) -> list[str]:
"""句読点で分割し、max_length文字以内のチャンクに分ける。"""
sentences = re.split(r"(?<=[。！？])", text)
chunks = []
for s in sentences:
s = s.strip()
if not s:
continue
if len(s) <= max_length:
chunks.append(s)
else:
for i in range(0, len(s), max_length):
chunks.append(s[i : i + max_length])
return chunks if chunks else [text]
def combine_wavs(wav_list: list[bytes]) -> bytes:
"""複数のWAVバイナリを1つに結合 (dialogue_api/services/tts.py から移植)。"""
if len(wav_list) == 1:
return wav_list[0]
output = io.BytesIO()
with wave.open(output, "wb") as out_wav:
with wave.open(io.BytesIO(wav_list[0]), "rb") as first:
out_wav.setparams(first.getparams())
for wav_bytes in wav_list:
with wave.open(io.BytesIO(wav_bytes), "rb") as w:
out_wav.writeframes(w.readframes(w.getnframes()))
return output.getvalue()
# ---- HTTP クライアント ----
async def whisper_transcribe(wav_bytes: bytes) -> str:
async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT_SEC) as client:
try:
r = await client.post(
WHISPER_URL,
files={"audio_file": ("audio.wav", wav_bytes, "audio/wav")},
)
r.raise_for_status()
return r.json().get("text", "").strip()
except httpx.HTTPError as e:
print(f" ![Whisperエラー] {e}", flush=True)
return ""
async def stream_llm(history: list[dict], user_text: str):
"""LLMストリーミング応答を文節単位でyieldする。"""
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
messages.extend(history)
messages.append({"role": "user", "content": user_text})
if LLM_BACKEND == "ollama":
async for sentence in _stream_ollama(messages):
yield sentence
elif LLM_BACKEND == "llamacpp":
async for sentence in _stream_llamacpp(messages):
yield sentence
elif LLM_BACKEND == "gemini":
async for sentence in _stream_gemini(messages):
yield sentence
else:
yield f"[設定エラー] 不明な LLM_BACKEND: {LLM_BACKEND}"
async def _stream_ollama(messages: list[dict]):
"""Ollamaのstreaming応答を文節単位でyieldする。"""
async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT_SEC) as client:
try:
async with client.stream(
"POST",
f"{OLLAMA_ENDPOINT}/api/chat",
json={
"model": OLLAMA_MODEL,
"messages": messages,
"stream": True,
},
) as response:
response.raise_for_status()
buffer = ""
async for line in response.aiter_lines():
if not line.strip():
continue
try:
chunk = json.loads(line)
delta = chunk.get("message", {}).get("content", "")
except json.JSONDecodeError:
continue
if not delta:
continue
buffer += delta
while True:
match = re.search(r"[。！？、]", buffer)
if match is None:
break
split_pos = match.end()
sentence = buffer[:split_pos]
buffer = buffer[split_pos:]
yield sentence
if buffer.strip():
yield buffer.strip()
except httpx.HTTPError as e:
print(f" ![Ollamaエラー] {e}", flush=True)
async def _stream_llamacpp(messages: list[dict]):
"""llama-serverのstreaming応答を文節単位でyieldする。"""
async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT_SEC) as client:
try:
async with client.stream(
"POST",
f"{LLAMACPP_ENDPOINT}/v1/chat/completions",
json={
"model": LLAMACPP_MODEL,
"messages": messages,
"stream": True,
},
) as response:
response.raise_for_status()
buffer = ""
async for line in response.aiter_lines():
if not line.startswith("data: "):
continue
data_str = line[len("data: "):]
if data_str.strip() == "[DONE]":
break
try:
chunk = json.loads(data_str)
delta = chunk["choices"][0]["delta"].get("content", "")
except (json.JSONDecodeError, KeyError, IndexError):
continue
if not delta:
continue
buffer += delta
while True:
match = re.search(r"[。！？、]", buffer)
if match is None:
break
split_pos = match.end()
sentence = buffer[:split_pos]
buffer = buffer[split_pos:]
yield sentence
if buffer.strip():
yield buffer.strip()
except httpx.HTTPError as e:
print(f" ![llama.cppエラー] {e}", flush=True)
async def _stream_gemini(messages: list[dict]):
"""Gemini APIのstreaming応答を文節単位でyieldする。
messages は OpenAI形式（role: system/user/assistant）で受け取り、
Gemini API形式（role: user/model、system_instructionは別）に変換する。
"""
system_text = ""
contents = []
for msg in messages:
if msg["role"] == "system":
system_text += msg["content"] + "\n"
elif msg["role"] == "user":
contents.append({"role": "user", "parts": [{"text": msg["content"]}]})
elif msg["role"] == "assistant":
contents.append({"role": "model", "parts": [{"text": msg["content"]}]})
payload = {
"contents": contents,
"generationConfig": {
"maxOutputTokens": 200,
},
}
if system_text.strip():
payload["system_instruction"] = {"parts": [{"text": system_text.strip()}]}
url = (
f"https://generativelanguage.googleapis.com/v1beta/models/{GEMINI_MODEL}"
f":streamGenerateContent?alt=sse&key={GEMINI_API_KEY}"
)
async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT_SEC) as client:
try:
async with client.stream("POST", url, json=payload) as response:
response.raise_for_status()
buffer = ""
async for line in response.aiter_lines():
if not line.startswith("data: "):
continue
data_str = line[len("data: "):]
try:
chunk = json.loads(data_str)
text = (
chunk.get("candidates", [{}])[0]
.get("content", {})
.get("parts", [{}])[0]
.get("text", "")
)
except (json.JSONDecodeError, KeyError, IndexError):
continue
if not text:
continue
buffer += text
while True:
match = re.search(r"[。！？、]", buffer)
if match is None:
break
split_pos = match.end()
sentence = buffer[:split_pos]
buffer = buffer[split_pos:]
yield sentence
if buffer.strip():
yield buffer.strip()
except httpx.HTTPError as e:
print(f" ![Geminiエラー] {e}", flush=True)
async def check_gemini_connection() -> bool:
"""起動時にGemini APIへの接続とAPIキー・モデル名の有効性を確認する。"""
url = (
f"https://generativelanguage.googleapis.com/v1beta/models/{GEMINI_MODEL}"
f"?key={GEMINI_API_KEY}"
)
async with httpx.AsyncClient(timeout=10.0) as client:
try:
r = await client.get(url)
r.raise_for_status()
print(f" | [Gemini接続確認@{_ts()}] OK (model={GEMINI_MODEL})", flush=True)
return True
except httpx.HTTPError as e:
print(f" ![Gemini接続確認エラー] {e}", flush=True)
return False
async def generate_context_monologue(last_reply: str) -> bytes | None:
"""直前のAI発言を踏まえた独り言を生成してTTS合成する。
生成失敗時はNoneを返す（呼び出し側は独り言なしにフォールバック）。
Gemini前提（他バックエンド使用時やAPIキー未設定時は呼び出し元でスキップする）。
"""
prompt = CONTEXT_MONOLOGUE_PROMPT.format(
max_chars=CONTEXT_MONOLOGUE_MAX_CHARS,
last_reply=last_reply,
)
messages = [{"role": "user", "content": prompt}]
text = ""
try:
async for chunk in _stream_gemini(messages):
text += chunk
except Exception as e:
print(f" ![独り言生成エラー] {e}", flush=True)
return None
text = text.strip()
if not text or len(text) > CONTEXT_MONOLOGUE_MAX_CHARS * 2:
return None
wav = await tts_synthesize_chunk(clean_for_tts(text))
if wav:
print(f" | [独り言@{_ts()}] 生成: {text}", flush=True)
return wav
async def classify_with_whisper(audio_segment: bytes, duration_sec: float) -> tuple[str, str]:
"""AI発話中に検出されたユーザー発話をWhisper結果で分類する。
Returns: (classification, text)。classification は
"noise" | "backchannel" | "interrupt" | "buffer"
DUPLEX_ENABLED=True の場合、割り込みワードを含まない意味のある発話は
即座に割り込みとはせず "buffer" として扱う（呼び出し側が speaking_buffer に
溜めて次のターンに合流させる＝デュプレックス会話の核）。
DUPLEX_ENABLED=False の場合はフェーズ2と同じ挙動で "interrupt" にする
（相槌ワードに一致しなければ即座に割り込む）。
"""
text = await whisper_transcribe(audio_segment)
if not text or not text.strip():
return "noise", text
if text in HALLUCINATION_PHRASES:
return "noise", text
classification = _classify_backchannel_text(text)
if classification == "unknown":
return ("buffer" if DUPLEX_ENABLED else "interrupt"), text
return classification, text
async def tts_synthesize_chunk(text: str) -> bytes | None:
"""1チャンクをSBV2に投げて WAV を取得。"""
async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT_SEC) as client:
try:
r = await client.get(
TTS_ENDPOINT,
params={"text": text, "model_name": TTS_MODEL},
)
r.raise_for_status()
return r.content
except httpx.HTTPError as e:
print(f" ![TTSチャンクエラー] text={text!r} err={e}", flush=True)
return None
async def tts_synthesize(text: str) -> bytes | None:
"""テキストを前処理 → チャンク分割 → 各チャンクをTTS → 結合して1つのWAVに。"""
cleaned = clean_for_tts(text)
if not cleaned:
return None
chunks = split_text_for_tts(cleaned)
print(f" | [TTS分割] {len(chunks)}チャンク: {[len(c) for c in chunks]}", flush=True)
wavs = []
for i, chunk in enumerate(chunks):
wav = await tts_synthesize_chunk(chunk)
if wav is None:
print(f" ![TTS] チャンク{i+1}失敗、スキップ", flush=True)
continue
wavs.append(wav)
if not wavs:
return None
return combine_wavs(wavs)
# ---- ソフトウェアAEC (フェーズ4 Step 2) ----
class AECBackendError(RuntimeError):
"""pywebrtc-audio / aec-audio-processing のどちらも見つからない場合。"""
class _AECEngine:
"""pywebrtc-audio と aec-audio-processing のどちらが入っていても
process(near_frame, far_frame) -> clean_frame という同じ形で使える薄いラッパー。
aec_prototype.py のAECEngineと同じ設計（フェーズ4 Step 1で検証済み）。
"""
def __init__(
self,
sample_rate: int,
enable_aec: bool,
enable_ns: bool,
enable_agc: bool,
stream_delay_ms: int,
):
self._backend_name: str
self._ap = None
try:
from pywebrtc_audio import AudioProcessor # type: ignore
self._ap = AudioProcessor(
sample_rate=sample_rate,
echo_cancellation=enable_aec,
noise_suppression=enable_ns,
auto_gain_control=enable_agc,
stream_delay_ms=stream_delay_ms,
)
self._backend_name = "pywebrtc-audio"
except ImportError:
try:
from aec_audio_processing import AudioProcessor # type: ignore
ap = AudioProcessor(
enable_aec=enable_aec,
enable_ns=enable_ns,
enable_agc=enable_agc,
)
ap.set_stream_format(
sample_rate_in=sample_rate,
channel_count_in=1,
sample_rate_out=sample_rate,
channel_count_out=1,
)
ap.set_reverse_stream_format(sample_rate, 1)
ap.set_stream_delay(stream_delay_ms)
self._ap = ap
self._backend_name = "aec-audio-processing"
except ImportError as e:
raise AECBackendError(
"pywebrtc-audio も aec-audio-processing も見つかりません。\n"
" pip install pywebrtc-audio\n"
" または\n"
" pip install aec-audio-processing"
) from e
@property
def backend_name(self) -> str:
return self._backend_name
def process(self, near_frame: np.ndarray, far_frame: np.ndarray) -> np.ndarray:
"""1フレーム分（10ms固定）のAEC処理を行う。"""
return self._ap.process(near_frame, far_frame)
def _resample_int16(audio: np.ndarray, orig_sr: int, target_sr: int) -> np.ndarray:
"""int16配列を単純な線形補間でリサンプルする（高音質は求めない、AEC参照信号用）。"""
if orig_sr == target_sr or len(audio) == 0:
return audio
duration = len(audio) / orig_sr
n_target = max(1, int(round(duration * target_sr)))
x_orig = np.linspace(0.0, duration, num=len(audio), endpoint=False)
x_target = np.linspace(0.0, duration, num=n_target, endpoint=False)
resampled = np.interp(x_target, x_orig, audio.astype(np.float64))
return resampled.astype(np.int16)
class AECProcessor:
"""マイク入力にAEC処理をかけるラッパー。play_wav_chunk が再生する音声を
参照信号として受け取り、マイク入力からその成分を差し引く。
VADCapture は30msフレーム単位で動くが、WebRTC AECは10ms固定フレームを
要求するため、内部で10msサブフレームに分割してAECにかけ、結合して返す。
feed_reference（play_wav_chunk呼び出し元のスレッド）と process_block
（オーディオコールバックスレッド）は別スレッドから呼ばれるため、
参照信号バッファは threading.Lock で保護する。
注意（フェーズ4 Step 1の検証結果を踏まえて）:
- NS/AGCは常にFalse推奨（こもりの原因と判明済み）
- フェードアウト中は実際の出力音量が下がっていくが、ここで供給する参照信号は
フェードアウト前の音量のままなので、フェードアウトの間だけAECの精度が
多少落ちる可能性がある（指示書で言及されている既知の制約、未対応）
"""
def __init__(self, sample_rate: int, stream_delay_ms: int, enable_ns: bool, enable_agc: bool):
self._engine = _AECEngine(sample_rate, True, enable_ns, enable_agc, stream_delay_ms)
self._sample_rate = sample_rate
self._frame_samples = sample_rate * 10 // 1000 # 10ms
self._reference_queue: collections.deque[np.ndarray] = collections.deque()
self._lock = threading.Lock()
@property
def backend_name(self) -> str:
return self._engine.backend_name
def feed_reference(self, wav_bytes: bytes, sample_rate: int, channels: int, sampwidth: int):
"""スピーカーに再生するWAVチャンクを参照信号として登録する。
play_wav_chunk の再生開始前に呼ぶ。サンプルレート・チャンネル数が
AEC側（sample_rate, モノラル）と異なる場合はここで変換する。
"""
if sampwidth != 2:
return # int16以外は非対応。参照信号を供給できないのでAECは効かないままになる
audio = np.frombuffer(wav_bytes, dtype=np.int16)
if channels > 1:
audio = audio[::channels] # 簡易ダウンミックス（先頭チャンネルのみ）
audio = _resample_int16(audio, sample_rate, self._sample_rate)
with self._lock:
for i in range(0, len(audio), self._frame_samples):
frame = audio[i : i + self._frame_samples]
if len(frame) < self._frame_samples:
frame = np.pad(frame, (0, self._frame_samples - len(frame)))
self._reference_queue.append(frame)
def process_block(self, pcm_bytes: bytes) -> bytes:
"""VADCaptureが受け取った生PCM（30msブロック、int16モノラル）にAECをかけて返す。"""
samples = np.frombuffer(pcm_bytes, dtype=np.int16)
out_chunks = []
for i in range(0, len(samples), self._frame_samples):
near = samples[i : i + self._frame_samples]
if len(near) < self._frame_samples:
out_chunks.append(near) # 端数はそのまま通す
continue
with self._lock:
far = (
self._reference_queue.popleft()
if self._reference_queue
else np.zeros(self._frame_samples, dtype=np.int16)
)
try:
clean = self._engine.process(near, far)
except Exception as e:
print(f" ![AEC処理エラー] {e}", file=sys.stderr, flush=True)
clean = near
out_chunks.append(np.asarray(clean, dtype=np.int16))
return np.concatenate(out_chunks).astype(np.int16).tobytes()
# ---- 再生制御 ----
class PlaybackController:
def __init__(self):
self.is_playing = False
self.interrupted = False
self._lock = threading.Lock()
# フェーズ4 Step 2: main()で設定されればplay_wav_chunkが再生音声を
# 参照信号として供給する。全呼び出し箇所に引数を追加するのを避けるため
# PlaybackControllerに持たせている
self.aec_processor: AECProcessor | None = None
def start_playback(self):
with self._lock:
self.is_playing = True
self.interrupted = False
def interrupt(self) -> bool:
"""割り込みを通知する。実際の停止（フェードアウト）は play_wav_chunk の
再生コールバックが self.interrupted を監視して行う。ここでは sd.stop() を
呼ばない（呼ぶと即座に無音になり、フェードアウトが機能しなくなるため）。
"""
with self._lock:
if self.is_playing and not self.interrupted:
self.interrupted = True
return True
return False
def end_playback(self):
with self._lock:
self.is_playing = False
def play_wav_chunk(wav_bytes: bytes, controller: PlaybackController) -> bool:
"""1チャンクのWAVを再生。割り込み時は FADEOUT_DURATION_SEC かけて
音量をゼロにフェードアウトしてから終了する（ぶつ切りを避けるため）。
割り込みがあればFalseを返す。
"""
with wave.open(io.BytesIO(wav_bytes), "rb") as wf:
sr = wf.getframerate()
channels = wf.getnchannels()
sampwidth = wf.getsampwidth()
raw = wf.readframes(wf.getnframes())
if controller.aec_processor is not None:
# 実際にスピーカーへ送る音声をAECの参照信号として供給する。
# フェードアウトで実際の出力音量が下がっても、ここで渡すのは
# フェードアウト前の音量のままになる（既知の制約、注意点参照）。
controller.aec_processor.feed_reference(raw, sr, channels, sampwidth)
if sampwidth == 2:
dtype = np.int16
elif sampwidth == 4:
dtype = np.int32
else:
print(f" ![TTS再生] 未対応 sampwidth={sampwidth}", flush=True)
return True
audio = np.frombuffer(raw, dtype=dtype).astype(np.float32)
normalizer = 32768.0 if dtype == np.int16 else 2147483648.0
audio = audio.reshape(-1, channels) if channels > 1 else audio.reshape(-1, 1)
total_frames = len(audio)
fadeout_frames = max(1, int(sr * FADEOUT_DURATION_SEC))
position = 0
fading = False
fade_position = 0
def callback(outdata, frames, time_info, status):
nonlocal position, fading, fade_position
if controller.interrupted and not fading:
fading = True
fade_position = 0
remaining = total_frames - position
if remaining <= 0:
outdata[:] = 0
raise sd.CallbackStop
n = min(frames, remaining)
chunk = audio[position:position + n].copy()
if fading:
fade_remaining = fadeout_frames - fade_position
if fade_remaining <= 0:
outdata[:] = 0
raise sd.CallbackStop
fade_n = min(n, fade_remaining)
fade_curve = np.linspace(
1.0 - fade_position / fadeout_frames,
1.0 - (fade_position + fade_n) / fadeout_frames,
fade_n,
).reshape(-1, 1)
chunk[:fade_n] *= fade_curve
if fade_n < n:
chunk[fade_n:] = 0
fade_position += fade_n
outdata[:n] = chunk / normalizer
if n < frames:
outdata[n:] = 0
position += n
controller.start_playback()
try:
with sd.OutputStream(
samplerate=sr,
channels=channels,
dtype="float32",
blocksize=1024,
device=OUTPUT_DEVICE,
callback=callback,
):
while True:
sd.sleep(50)
if position >= total_frames:
break
if fading and fade_position >= fadeout_frames:
break
except sd.CallbackStop:
pass
finally:
controller.end_playback()
return not controller.interrupted
async def preload_fillers() -> list[tuple[str, bytes]]:
"""起動時にフィラー音声をTTS合成してメモリに保持する。"""
fillers = []
for phrase in FILLER_PHRASES:
wav = await tts_synthesize_chunk(phrase)
if wav:
fillers.append((phrase, wav))
print(f" | [フィラー] プリロード完了: {phrase}", flush=True)
return fillers
async def preload_ai_backchannels() -> list[tuple[str, bytes]]:
"""起動時にAI相槌音声をTTS合成してメモリに保持する。"""
backchannels = []
for phrase in AI_BACKCHANNEL_PHRASES:
wav = await tts_synthesize_chunk(phrase)
if wav:
backchannels.append((phrase, wav))
print(f" | [AI相槌] プリロード完了: {phrase}", flush=True)
return backchannels
async def preload_mid_speech_backchannels() -> list[bytes]:
"""起動時に発話中相槌（息継ぎ検出時）の音声をTTS合成してメモリに保持する。"""
wavs = []
for phrase in MID_SPEECH_BACKCHANNEL_PHRASES:
wav = await tts_synthesize_chunk(phrase)
if wav:
wavs.append(wav)
print(f" | [発話中相槌] プリロード完了: {phrase}", flush=True)
return wavs
async def play_ai_backchannel(
ai_backchannels: list[tuple[str, bytes]],
playback: PlaybackController,
coordinator: BackchannelCoordinator | None = None,
):
"""ユーザーが長く話し続けている間に、AIの相槌を1つ再生する（非同期・結果を待たない）。
マイクは開いたままで、ユーザーの発話受信は並行して続く。
呼び出し側で coordinator.can_play() を確認してから呼ぶこと。
"""
if not ai_backchannels or playback.is_playing:
return
phrase, wav = random.choice(ai_backchannels)
print(f" | [AI相槌@{_ts()}] 再生: {phrase}", flush=True)
if coordinator is not None:
coordinator.mark_played()
await asyncio.to_thread(play_wav_chunk, wav, playback)
async def play_filler_loop(
fillers: list[tuple[str, bytes]],
playback: PlaybackController,
stop_event: asyncio.Event,
coordinator: BackchannelCoordinator | None = None,
):
"""LLMの最初の文節が到着するまで定型フィラー音声をループ再生する。
stop_event がセットされたら現在のフィラー再生完了後に終了する。
"""
if not fillers:
return
last_phrase = None
while not stop_event.is_set():
candidates = [f for f in fillers if f[0] != last_phrase] or fillers
phrase, wav = random.choice(candidates)
last_phrase = phrase
print(f" | [フィラー@{_ts()}] 再生: {phrase}", flush=True)
if coordinator is not None:
coordinator.mark_played()
ok = await asyncio.to_thread(play_wav_chunk, wav, playback)
if not ok:
break # ユーザー割り込みで停止された
try:
await asyncio.wait_for(stop_event.wait(), timeout=0.3)
break # stop_eventがセットされた
except asyncio.TimeoutError:
pass # まだLLM応答が来ていない → 次のフィラーへ
async def schedule_monologue(
last_reply: str,
state: DuplexStateMachine,
playback: PlaybackController,
capture: VADCapture,
):
"""独り言を生成し、CONTEXT_MONOLOGUE_DELAY_SEC後もSILENT+IDLEが継続していれば再生する。
フィラーとは独立して動作する。ユーザーが話しかける、あるいはAIが何か
再生し始めるなどして状態が変化していれば再生せず破棄する。
注意: InputState.IDLE/LISTENINGはVADが発話区切り（末尾無音）を検出して
初めて更新されるため、ユーザーが長く連続して話している最中は状態上まだ
IDLEのままになりうる。それだけを条件にすると発話の途中でモノローグが
割り込んでしまうため、VADCaptureが今まさに音声を検出中かどうか
（capture.in_speech）も合わせて確認する。
"""
wav = await generate_context_monologue(last_reply)
if wav is None:
return
await asyncio.sleep(CONTEXT_MONOLOGUE_DELAY_SEC)
if (
state.output == OutputState.SILENT
and state.input == InputState.IDLE
and not capture.in_speech
):
print(f" | [独り言@{_ts()}] 再生", flush=True)
state.set_output(OutputState.MONOLOGUE, "独り言再生")
await asyncio.to_thread(play_wav_chunk, wav, playback)
state.set_output(OutputState.SILENT, "独り言再生完了")
else:
print(f" | [独り言@{_ts()}] 破棄 (状態が変化)", flush=True)
async def stream_and_play(
llm_gen,
playback: PlaybackController,
state: DuplexStateMachine,
fillers: list[tuple[str, bytes]] | None = None,
coordinator: BackchannelCoordinator | None = None,
) -> tuple[str, bool]:
"""LLMストリーミング出力を文節単位でTTS合成→即再生するパイプライン。
fillers が渡された場合、LLM最初の文節到着まで定型フィラー音声をループ再生する
（OutputState.FILLER）。呼び出し側が相槌コーディネーターの判断で fillers=None
にすれば、フィラー自体をスキップできる（直前に相槌が鳴っていた場合）。
文節Aの再生中に文節BのTTS合成を先行実行し、文節間の間を最小化する
（tts_producer タスクが合成、再生ループが消費するプロデューサー/
コンシューマー方式）。実際に音声再生が始まった時点で OutputState.SPEAKING に、
割り込まれたら OutputState.SILENT に遷移する。
Returns:
(full_reply_text, was_interrupted)
"""
full_reply = ""
interrupted = False
first_sentence = True
filler_stop = asyncio.Event()
filler_task = None
if fillers and ENABLE_FILLER:
state.set_output(OutputState.FILLER, "フィラー開始")
filler_task = asyncio.create_task(play_filler_loop(fillers, playback, filler_stop, coordinator))
async def stop_filler():
filler_stop.set()
if filler_task and not filler_task.done():
if playback.is_playing:
playback.interrupt()
await asyncio.sleep(0.05) # 音声デバイス解放を待つ
try:
await asyncio.wait_for(filler_task, timeout=1.0)
except asyncio.TimeoutError:
filler_task.cancel()
# TTSプリフェッチ用キュー: LLMから来た文節を順次TTS合成し、合成済みWAVをキューに積む
tts_queue: asyncio.Queue[tuple[str, bytes] | None] = asyncio.Queue()
async def tts_producer():
"""LLMストリーミング → TTS合成 → キューに積むワーカー。
LLM側は読点でも区切ってくるが、TTSにはできるだけ句点（。！？）単位の
まとまった文をそのまま渡す。文全体のイントネーションをSBV2に整えさせるため。
"""
nonlocal full_reply, first_sentence
tts_buffer = ""
async def flush_to_tts(text: str):
for chunk_text in split_text_for_tts(text, max_length=TTS_CHUNK_MAX):
wav = await tts_synthesize_chunk(chunk_text)
if wav is None:
continue
await tts_queue.put((chunk_text, wav))
async for sentence in llm_gen:
full_reply += sentence
cleaned = clean_for_tts(sentence)
if not cleaned:
continue
if first_sentence:
await stop_filler()
first_sentence = False
tts_buffer += cleaned
while True:
match = re.search(r"[。！？]", tts_buffer)
if match is None:
break
split_pos = match.end()
complete_sentence = tts_buffer[:split_pos]
tts_buffer = tts_buffer[split_pos:]
await flush_to_tts(complete_sentence)
if tts_buffer.strip():
await flush_to_tts(tts_buffer.strip())
await tts_queue.put(None) # 全文節の合成完了を通知
producer_task = asyncio.create_task(tts_producer())
speaking_started = False
while True:
item = await tts_queue.get()
if item is None:
try:
await producer_task
except asyncio.CancelledError:
pass
break
chunk_text, wav = item
if not speaking_started:
state.set_output(OutputState.SPEAKING, "LLM最初の文節到着")
speaking_started = True
print(f" | [パイプライン@{_ts()}] 再生: {chunk_text}", flush=True)
ok = await asyncio.to_thread(play_wav_chunk, wav, playback)
if not ok:
interrupted = True
print(f" | [TTS@{_ts()}] 中断 (言葉による割り込み)", flush=True)
producer_task.cancel()
try:
await producer_task
except asyncio.CancelledError:
pass
async for rest in llm_gen:
full_reply += rest
break
await stop_filler()
state.set_output(OutputState.SILENT, "中断" if interrupted else "再生完了")
return full_reply, interrupted
# ---- 投機的LLM呼び出し ----
class SpeculativeLLM:
"""投機的LLM呼び出しを管理する。
送信のたびに新しい結果キューを作り直す（フェーズ3指示書のサンプル実装は
キューを使い回すため、キャンセル前に一部だけキューへ積まれていた場合に
古い応答の残骸と新しい応答が混ざる可能性があった。ここではsend_speculative
のたびに新規Queueを割り当てて、その混線を避けている）。
"""
def __init__(self):
self._current_task: asyncio.Task | None = None
self._result_queue: asyncio.Queue = asyncio.Queue()
self._ready_event = asyncio.Event()
async def send_speculative(self, history: list[dict], user_text: str):
"""投機的にLLMリクエストを送信する。前のリクエストがあればキャンセルする。"""
await self.cancel_and_wait()
queue: asyncio.Queue = asyncio.Queue()
self._result_queue = queue
self._current_task = asyncio.create_task(self._run(history, user_text, queue))
self._ready_event.set()
async def _run(self, history: list[dict], user_text: str, result_queue: asyncio.Queue):
"""LLMストリーミングを実行し、結果を指定されたキューに積む。"""
try:
async for sentence in stream_llm(history, user_text):
await result_queue.put(sentence)
finally:
await result_queue.put(None) # 完了シグナル（キャンセル時も含めて必ず送る）
def cancel(self):
"""現在のリクエストをキャンセルする（応答を待たない）。"""
if self._current_task and not self._current_task.done():
self._current_task.cancel()
async def cancel_and_wait(self):
"""現在のリクエストをキャンセルし、後始末が終わるまで待つ。"""
if self._current_task and not self._current_task.done():
self._current_task.cancel()
try:
await self._current_task
except asyncio.CancelledError:
pass
async def wait_for_next(self):
"""次の投機的応答の送信を待ち、そのストリームを読むasync generatorを返す。
send_speculative() が新しいキューを作った"後"にそのキューを掴む必要が
あるため、イベントで同期してからキューを取得する。
"""
await self._ready_event.wait()
self._ready_event.clear()
queue = self._result_queue
async def _stream():
while True:
item = await queue.get()
if item is None:
break
yield item
return _stream()
# ---- 発話中の相槌（息継ぎ検出） ----
class BreathDetector:
"""発話中の息継ぎ（音量の谷）を検出する。
VADのセグメント分割とは独立に、フレーム単位のRMSを監視する。
1つの発話（1 VADセグメント）内での相槌回数・間隔を自前で管理する。
"""
def __init__(self):
self._speaking_start: float | None = None # 発話開始時刻 (time.monotonic())
self._low_rms_start: float | None = None # 音量が閾値を下回った時刻
self._last_backchannel: float = 0.0 # 前回の相槌時刻
self._count_this_turn: int = 0 # 今の発話での相槌回数
def reset_turn(self):
"""新しい発話（VADセグメント）開始時にカウンタをリセットする。"""
self._speaking_start = None
self._low_rms_start = None
self._count_this_turn = 0
def on_speaking_start(self):
"""VADが発話開始を検出した時に呼ぶ。"""
if self._speaking_start is None:
self._speaking_start = time.monotonic()
def elapsed_speaking_sec(self) -> float:
if self._speaking_start is None:
return 0.0
return time.monotonic() - self._speaking_start
def feed_audio(self, rms: float) -> bool:
"""音声フレームのRMSを受け取り、相槌を入れるべきかを返す。
Returns: True なら相槌を再生すべき
"""
now = time.monotonic()
if self._speaking_start is None:
return False
if now - self._speaking_start < MID_SPEECH_MIN_SPEAKING_SEC:
return False
if self._count_this_turn >= MID_SPEECH_MAX_PER_TURN:
return False
if now - self._last_backchannel < MID_SPEECH_COOLDOWN_SEC:
return False
if rms <= MID_SPEECH_RMS_THRESHOLD:
if self._low_rms_start is None:
self._low_rms_start = now
elif (now - self._low_rms_start) * 1000 >= MID_SPEECH_SILENCE_MS:
# 息継ぎ検出 → 相槌トリガー
self._low_rms_start = None
self._last_backchannel = now
self._count_this_turn += 1
return True
else:
# 音量が戻った → リセット
self._low_rms_start = None
return False
# ---- VAD ----
class VADCapture:
def __init__(self):
self.vad = webrtcvad.Vad(VAD_AGGRESSIVENESS)
self.silence_frames_threshold = int(VAD_SILENCE_MS / FRAME_DURATION_MS)
self.min_speech_frames = int(VAD_MIN_SPEECH_MS / FRAME_DURATION_MS)
self.speech_frames: list[bytes] = []
self.silence_count = 0
self.in_speech = False
self.muted = False
# Whisper処理中の発話セグメント数。区切り判定のタイマーは
# この値が0でないとき発火させない (Whisper処理待ちで取りこぼし防止)
self.pending_count = 0
self.segment_queue: asyncio.Queue[bytes] = asyncio.Queue()
self.loop: asyncio.AbstractEventLoop | None = None
# フレーム単位のRMSを外部に通知するフック（発話中相槌向け）。
# callback(is_speech: bool, rms: float) の形式。オーディオコールバック
# スレッドから直接呼ばれるため、内部で例外を投げないこと・重い処理を
# しないこと（run_coroutine_threadsafeでの非同期再生予約程度に留める）。
self.frame_listener = None
# VADが1つの発話セグメントを閉じた瞬間（in_speech: True->False）に
# 呼ばれるフック。引数なし。BreathDetectorのリセットに使う。
# ここで確実に呼ぶことで、そのセグメントがどちらの経路
# （通常フロー/SPEAKING中の割り込み判定）で処理されたかに関わらず、
# 次の発話の計測がずれないようにする。
self.on_segment_end = None
# フェーズ4 Step 2: 設定されていれば、VADにかける前に生PCMへAECを適用する。
# マイク入力 → AECProcessor → VAD の順。
self.aec_processor: AECProcessor | None = None
def _sd_callback(self, indata, frames, time_info, status):
if status:
print(f" ![sd status] {status}", file=sys.stderr, flush=True)
if self.muted:
return
pcm = bytes(indata)
if self.aec_processor is not None:
try:
pcm = self.aec_processor.process_block(pcm)
except Exception as e:
print(f" ![AECブロック処理エラー] {e}", file=sys.stderr, flush=True)
offset = 0
while offset + FRAME_BYTES <= len(pcm):
frame = pcm[offset:offset + FRAME_BYTES]
offset += FRAME_BYTES
is_speech = self.vad.is_speech(frame, SAMPLE_RATE)
if self.frame_listener is not None:
try:
self.frame_listener(is_speech, _pcm_rms(frame))
except Exception as e:
print(f" ![frame_listenerエラー] {e}", file=sys.stderr, flush=True)
if is_speech:
self.speech_frames.append(frame)
self.silence_count = 0
if not self.in_speech:
self.in_speech = True
else:
if self.in_speech:
self.speech_frames.append(frame)
self.silence_count += 1
if self.silence_count >= self.silence_frames_threshold:
if len(self.speech_frames) >= self.min_speech_frames + self.silence_count:
segment = b"".join(self.speech_frames)
self.pending_count += 1 # Whisper処理待ちカウンタを上げる
if self.loop is not None:
asyncio.run_coroutine_threadsafe(
self.segment_queue.put(segment), self.loop
)
self.speech_frames = []
self.silence_count = 0
self.in_speech = False
if self.on_segment_end is not None:
try:
self.on_segment_end()
except Exception as e:
print(f" ![on_segment_endエラー] {e}", file=sys.stderr, flush=True)
async def get_segment(self) -> bytes:
return await self.segment_queue.get()
def attach_loop(self, loop: asyncio.AbstractEventLoop):
self.loop = loop
# ---- input_loop: ユーザー入力の処理 ----
async def _handle_speaking_utterance(
wav_bytes: bytes,
duration_sec: float,
playback: PlaybackController,
speculative: SpeculativeLLM,
capture: VADCapture,
speaking_buffer: list[str],
):
"""AI発話中（OutputState != SILENT）に検出された発話をclassify_with_whisperで判定する。
noise/backchannelは無視してAIの発話を継続させる。明示的な割り込みワードを
含む場合のみフェードアウトで停止し、進行中の投機的LLM呼び出しも打ち切る。
それ以外の意味のある発話は speaking_buffer に溜め、次のターンに含める
（割り込みでなければAIは話し続けるのがデュプレックス会話の核）。
"""
try:
classification, text = await classify_with_whisper(wav_bytes, duration_sec)
finally:
capture.pending_count = max(0, capture.pending_count - 1)
if classification == "noise":
if not text or not text.strip():
print(f" | [ノイズ@{_ts()}] Whisper空応答 -> 無視 (SPEAKING継続)", flush=True)
else:
print(f" | [ノイズ@{_ts()}] \"{text}\" -> ハルシネーション -> 無視 (SPEAKING継続)", flush=True)
return
if classification == "backchannel":
print(f" | [相槌@{_ts()}] {duration_sec:.1f}秒 \"{text}\" -> 無視 (SPEAKING継続)", flush=True)
return
if classification == "interrupt":
if playback.interrupt():
print(f" > [割り込み@{_ts()}] {duration_sec:.1f}秒 \"{text}\" -> フェードアウト停止", flush=True)
speculative.cancel() # 生成中だった応答があれば打ち切る（コスト・レイテンシ削減）
speaking_buffer.append(text)
return
# classification == "buffer": 割り込みではない意味のある発話。次のターンに合流させる。
print(f" | [認識@{_ts()}] {text} (SPEAKING継続、次のターンに合流)", flush=True)
speaking_buffer.append(text)
async def input_loop(
state: DuplexStateMachine,
speculative: SpeculativeLLM,
playback: PlaybackController,
capture: VADCapture,
history: list[dict],
status: dict,
mid_speech_backchannels: list[bytes],
ai_backchannels: list[tuple[str, bytes]],
coordinator: BackchannelCoordinator,
):
"""VAD発話セグメントを処理し、投機的LLM送信までを管理する。
OutputStateに関係なく常時動く。AI発話中（output != SILENT）は
classify_with_whisperでバックグラウンド判定し、割り込みでなければ
speaking_bufferに溜めて次のターンに合流させる（機能1）。
AI非発話中は、発話→沈黙検出→ SPECULATIVE_DELAY_SEC 待って投機的LLM送信、
という流れでターンを組み立てる（機能2）。待機中に追加発話が来たら
タイマーを仕切り直す。蓄積テキストが「えーっと」等の言い淀みだけの場合は
LLMに送らず、相槌でTURN_HOLDを延長して本題を待つ（言い淀みフィルタ）。
また、VADCaptureのフレーム単位RMSを監視するBreathDetectorを併用し、
ユーザーが長く話し続けている最中の息継ぎ（音量の谷）でも相槌を挟む
（発話中相槌）。VADのセグメント分割・Whisper送信単位はこれとは無関係で、
そのまま変更しない。
"""
turn_buffer: list[tuple[str, float]] = [] # 現在組み立て中のターンの (text, duration)
speaking_buffer: list[str] = [] # AI発話中に溜まったテキスト（次のターンに合流）
hold_task: asyncio.Task | None = None # TURN_HOLD中の投機的送信タイマー
hesitation_count = 0 # 現在のターンで言い淀み相槌を鳴らした回数
breath_detector = BreathDetector()
def on_frame(is_speech: bool, rms: float):
"""VADCaptureのオーディオコールバックスレッドから同期的に呼ばれる。
重い処理はせず、判定して非同期再生を予約するだけに留める。
"""
if is_speech:
breath_detector.on_speaking_start()
if not ENABLE_MID_SPEECH_BACKCHANNEL or not mid_speech_backchannels:
return
# AIが何か再生中（フィラー/応答/独り言）は音声出力の競合を避けるため発火しない
if state.output != OutputState.SILENT:
return
if not breath_detector.feed_audio(rms):
return
wav = random.choice(mid_speech_backchannels)
elapsed = breath_detector.elapsed_speaking_sec()
print(f" | [発話中相槌@{_ts()}] 再生: うん (息継ぎ検出, 発話{elapsed:.1f}秒経過)", flush=True)
if capture.loop is not None:
asyncio.run_coroutine_threadsafe(
asyncio.to_thread(play_wav_chunk, wav, playback), capture.loop
)
capture.frame_listener = on_frame
# VADが発話セグメントを閉じるたびにBreathDetectorをリセットする。
# SPEAKING中の割り込み判定（_handle_speaking_utterance）経由で処理された
# セグメントも含めて、必ずここで一律にリセットされる
# （start_hold()側だけに頼ると、その経路を通らないセグメントでリセット漏れが起きていた）。
capture.on_segment_end = breath_detector.reset_turn
async def commit_and_send():
nonlocal turn_buffer, hesitation_count
accumulated = "\n".join(text for text, _ in turn_buffer)
turn_buffer = []
hesitation_count = 0 # ターンを確定させたので言い淀みカウンタもリセット
status["turn_speech_duration"] = 0.0
print(f" | [投機的@{_ts()}] LLM送信 ({len(accumulated)}文字)", flush=True)
await speculative.send_speculative(list(history), accumulated)
history.append({"role": "user", "content": accumulated})
state.set_input(InputState.SPECULATIVE, "投機的LLM送信")
state.set_input(InputState.IDLE, "送信完了、次の発話待ち")
def reschedule_hold(delay: float, force_commit: bool = False):
"""新しい発話を検知したわけではないが、TURN_HOLDの待機を仕切り直す
（言い淀みで相槌を返した後や、相槌上限に達した後の延長に使う）。
"""
nonlocal hold_task
hold_task = asyncio.create_task(hold_then_send(delay, force_commit))
async def hold_then_send(delay: float, force_commit: bool = False):
nonlocal hesitation_count
try:
await asyncio.sleep(delay)
# Whisper処理待ちのセグメントがあれば、それが片付くまで待つ
# (直近の発話のテキストがまだ確定していない状態で確定させないため)
while capture.pending_count > 0:
await asyncio.sleep(0.1)
except asyncio.CancelledError:
return
accumulated_preview = "\n".join(text for text, _ in turn_buffer)
if not force_commit and ENABLE_HESITATION_FILTER and is_hesitation(accumulated_preview):
if hesitation_count < HESITATION_MAX_BACKCHANNELS:
if coordinator.can_play():
hesitation_count += 1
print(
f" | [言い淀み@{_ts()}] \"{accumulated_preview}\" -> 相槌で応答、続きを待つ",
flush=True,
)
if ai_backchannels:
asyncio.create_task(play_ai_backchannel(ai_backchannels, playback, coordinator))
else:
# 直前にAI相槌等が鳴ったばかり → 相槌の回数は消費せず黙って待つ
print(
f" | [言い淀み@{_ts()}] \"{accumulated_preview}\" -> 黙って続きを待つ",
flush=True,
)
# 蓄積テキストはそのまま保持し、続きの発話を待つ
reschedule_hold(SPECULATIVE_DELAY_SEC if DUPLEX_ENABLED else SILENCE_TIMEOUT_SEC)
return
elif DUPLEX_ENABLED:
# 相槌の回数を使い切った → もう相槌は鳴らさず、沈黙タイムアウトまで
# もう少しだけ長く待つ。それでも言い淀みのままなら諦めて送信する
print(
f" | [言い淀み@{_ts()}] \"{accumulated_preview}\" -> 相槌上限、沈黙タイムアウトまで待機",
flush=True,
)
reschedule_hold(SILENCE_TIMEOUT_SEC, force_commit=True)
return
# DUPLEX_ENABLED=Falseの場合は既にSILENCE_TIMEOUT_SEC分待っているので、
# 言い淀みのままでも素直に確定させる（下のcommit_and_sendに流れる）
await commit_and_send()
def start_hold(reason_listening: str, reason_hold: str):
"""発話（または蓄積済みバッファ）を検知した際、投機的送信タイマーを（再）開始する。"""
nonlocal hold_task
state.set_input(InputState.LISTENING, reason_listening)
if hold_task and not hold_task.done():
hold_task.cancel()
state.set_input(InputState.TURN_HOLD, reason_hold)
# BreathDetectorのリセットはVADCapture.on_segment_end（VADが実際に
# セグメントを閉じた瞬間）で行うため、ここでは呼ばない。
delay = SPECULATIVE_DELAY_SEC if DUPLEX_ENABLED else SILENCE_TIMEOUT_SEC
hold_task = asyncio.create_task(hold_then_send(delay))
def on_state_change(output_state, input_state, desc):
"""AIの発話がSILENTに戻った瞬間、speaking_bufferに溜まっていた発話があれば
新しいターンとして投機送信タイマーを開始する。
これをしないと、SPEAKING/MONOLOGUE中に「割り込みではない」と判定されて
speaking_bufferに蓄積されたユーザー発話が、ユーザーが追加で何か話すまで
ずっと放置され、AIがそれ以降ずっと応答しなくなる
（ユーザーは自分の質問には既に答え終えたつもりで黙っているため）。
"""
if output_state != OutputState.SILENT or not speaking_buffer:
return
if state.input != InputState.IDLE:
return # 既に別のターンを組み立て中
for buffered_text in speaking_buffer:
turn_buffer.append((buffered_text, 0.0))
speaking_buffer.clear()
start_hold("AI発話終了 (蓄積分あり)", "VAD沈黙検出")
state.add_listener(on_state_change)
while True:
segment_pcm = await capture.get_segment()
wav_bytes = _pcm_to_wav_bytes(segment_pcm)
duration_sec = len(segment_pcm) / 2 / SAMPLE_RATE
print(f" | [発話検出@{_ts()}] {duration_sec:.1f}秒 -> Whisper送信", flush=True)
rms = _pcm_rms(segment_pcm)
if rms < SILENCE_RMS_THRESHOLD:
print(f" | [無音スキップ@{_ts()}] rms={rms:.0f} < {SILENCE_RMS_THRESHOLD:.0f}", flush=True)
capture.pending_count = max(0, capture.pending_count - 1)
continue
if ENABLE_INTERRUPT and state.output != OutputState.SILENT:
# AI発話中（フィラー/応答/独り言） → 相槌/割り込み/バッファ判定（機能1）
asyncio.create_task(
_handle_speaking_utterance(wav_bytes, duration_sec, playback, speculative, capture, speaking_buffer)
)
continue
# AI非発話中の通常フロー
try:
text = await whisper_transcribe(wav_bytes)
finally:
capture.pending_count = max(0, capture.pending_count - 1)
if not text or not text.strip():
print(f" | [Whisper空応答] 言葉ではないので無視", flush=True)
continue
if text in HALLUCINATION_PHRASES:
print(f" | [ハルシネーション除外@{_ts()}] {text!r}", flush=True)
continue
print(f" | [認識@{_ts()}] {text}", flush=True)
# 直前がIDLE（＝新しいターンの開始）だったかをバッファ合流より前に判定しておく
is_new_turn = state.input == InputState.IDLE
# AI発話中に溜まっていたバッファがあれば今回のターンに合流させる
if speaking_buffer:
for buffered_text in speaking_buffer:
turn_buffer.append((buffered_text, 0.0))
speaking_buffer.clear()
turn_buffer.append((text, duration_sec))
status["turn_speech_duration"] = status.get("turn_speech_duration", 0.0) + duration_sec
start_hold(
"発話検出" if is_new_turn else "追加発話検出",
"VAD沈黙検出",
)
# ---- output_loop: AI出力の処理 ----
async def output_loop(
state: DuplexStateMachine,
speculative: SpeculativeLLM,
playback: PlaybackController,
fillers: list[tuple[str, bytes]],
history: list[dict],
coordinator: BackchannelCoordinator,
):
"""投機的LLM応答が用意され次第、フィラー→TTS再生を行う。input_loopとは独立に常時動く。"""
while True:
llm_gen = await speculative.wait_for_next()
# 直前にAI相槌/言い淀み相槌が鳴っていたら、フィラーは重ねずスキップする
if fillers and coordinator.time_since_last() < FILLER_SKIP_THRESHOLD_SEC:
print(f" | [フィラー@{_ts()}] スキップ (直前に相槌あり)", flush=True)
fillers_to_use = None
else:
fillers_to_use = fillers
reply, was_interrupted = await stream_and_play(llm_gen, playback, state, fillers_to_use, coordinator)
if reply:
history.append({"role": "assistant", "content": reply})
print(f"\nAI: {reply}\n", flush=True)
if was_interrupted:
print(f" | [TTS@{_ts()}] 中断 (言葉による割り込み)", flush=True)
elif reply:
print(f" | [TTS@{_ts()}] 再生完了", flush=True)
# ---- monitor_loop: 独り言・AI相槌のスケジューリング ----
async def monitor_loop(
state: DuplexStateMachine,
playback: PlaybackController,
capture: VADCapture,
ai_backchannels: list[tuple[str, bytes]],
history: list[dict],
status: dict,
coordinator: BackchannelCoordinator,
):
"""独り言とAI相槌のスケジューリングを行う補助ループ。"""
last_ai_backchannel_time = 0.0
last_monologue_time = 0.0
monologue_task: asyncio.Task | None = None
was_idle = False
while True:
await asyncio.sleep(0.2)
# --- AI相槌: ユーザーがターンの途中（LISTENING/TURN_HOLD）で一定時間話し続けたら ---
# LISTENINGはセグメント到着時に一瞬だけ立ってすぐTURN_HOLDに遷移するため
# （input_loop側でawaitを挟まず連続してset_inputするため）、0.2秒間隔の
# ポーリングではLISTENING単体はほぼ観測できない。TURN_HOLDも含めて判定する。
if (
ENABLE_AI_BACKCHANNEL
and ai_backchannels
and state.input in (InputState.LISTENING, InputState.TURN_HOLD)
and status.get("turn_speech_duration", 0.0) >= AI_BACKCHANNEL_TRIGGER_SEC
and (time.monotonic() - last_ai_backchannel_time) >= AI_BACKCHANNEL_MIN_INTERVAL_SEC
):
# 試行した時点でクールダウンを消費する（毎ティック連続でスキップログが
# 出続けるのを防ぐ）。実際に鳴らせるかは相槌コーディネーターが判断する。
last_ai_backchannel_time = time.monotonic()
if coordinator.can_play():
asyncio.create_task(play_ai_backchannel(ai_backchannels, playback, coordinator))
else:
print(
f" | [AI相槌@{_ts()}] スキップ (前回の相槌から{coordinator.time_since_last():.1f}秒)",
flush=True,
)
# --- 独り言: SILENT かつ IDLE に入った直後にスケジュールする ---
# capture.in_speechも見て、ユーザーがまだ話している最中（VADが区切りを
# 検出する前）には無駄な生成すら始めないようにする。
is_idle_now = (
state.output == OutputState.SILENT
and state.input == InputState.IDLE
and not capture.in_speech
)
if is_idle_now and not was_idle:
if (
ENABLE_CONTEXT_MONOLOGUE
and GEMINI_API_KEY
and history
and history[-1]["role"] == "assistant"
and (time.monotonic() - last_monologue_time) >= CONTEXT_MONOLOGUE_MIN_INTERVAL_SEC
):
last_monologue_time = time.monotonic()
monologue_task = asyncio.create_task(
schedule_monologue(history[-1]["content"], state, playback, capture)
)
elif not is_idle_now and was_idle:
if monologue_task and not monologue_task.done():
monologue_task.cancel()
was_idle = is_idle_now
def _pick_input_device():
if INPUT_DEVICE is None:
return None
devs = sd.query_devices()
for i, d in enumerate(devs):
if d["max_input_channels"] > 0 and INPUT_DEVICE in d["name"]:
print(f" | [INFO] input device: [{i}] {d['name']}", flush=True)
return i
print(f" ![WARN] '{INPUT_DEVICE}' に一致する入力デバイスが見つからない、デフォルト使用", flush=True)
return None
async def main():
print("=== デュプレックス会話プロトタイプ (フェーズ3+4) ===", flush=True)
if ENABLE_AEC:
print(
" スピーカー+マイク運用にはAEC(クライアント側)に加えてWhisperサーバ側の"
"ハルシネーション対策が必須です（AEC単体では残留ノイズの誤認識が残る）。",
flush=True,
)
else:
print("!!! ENABLE_AEC=False: ヘッドセット使用が前提です（AIの声がマイクに回り込む環境では誤動作します） !!!", flush=True)
print(f"Whisper: {WHISPER_URL}", flush=True)
if LLM_BACKEND == "ollama":
print(f"LLM (Ollama): {OLLAMA_ENDPOINT} Model: {OLLAMA_MODEL}", flush=True)
elif LLM_BACKEND == "llamacpp":
print(f"LLM (llama.cpp): {LLAMACPP_ENDPOINT} Model: {LLAMACPP_MODEL}", flush=True)
elif LLM_BACKEND == "gemini":
print(f"LLM (Gemini): Model: {GEMINI_MODEL}", flush=True)
else:
print(f"LLM (UNKNOWN): {LLM_BACKEND}", flush=True)
if LLM_BACKEND == "gemini":
if not GEMINI_API_KEY:
print(" ![ERROR] GEMINI_API_KEY が設定されていません。環境変数をセットしてください。", flush=True)
sys.exit(1)
if not await check_gemini_connection():
print(" ![ERROR] Gemini APIに接続できません。終了します。", flush=True)
sys.exit(1)
print(f"TTS: {TTS_ENDPOINT} Model: {TTS_MODEL} (1チャンク最大{TTS_CHUNK_MAX}文字)", flush=True)
print(
f"デュプレックス: {'有効' if DUPLEX_ENABLED else '無効 (フェーズ2相当)'}"
f" (投機的送信遅延 {SPECULATIVE_DELAY_SEC if DUPLEX_ENABLED else SILENCE_TIMEOUT_SEC}秒)",
flush=True,
)
print(
f"割り込み: {'有効 (相槌/割り込み分離)' if ENABLE_INTERRUPT else '無効'}"
f" (フェードアウト{FADEOUT_DURATION_SEC}秒)",
flush=True,
)
print(f"VAD: aggressiveness={VAD_AGGRESSIVENESS}, silence={VAD_SILENCE_MS}ms", flush=True)
print(f"フィラー: {'有効' if ENABLE_FILLER else '無効'} ({len(FILLER_PHRASES)}種)", flush=True)
print(f"AI相槌: {'有効' if ENABLE_AI_BACKCHANNEL else '無効'} (トリガー{AI_BACKCHANNEL_TRIGGER_SEC}秒/間隔{AI_BACKCHANNEL_MIN_INTERVAL_SEC}秒)", flush=True)
print(
f"発話中相槌: {'有効' if ENABLE_MID_SPEECH_BACKCHANNEL else '無効'}"
f" (RMS閾値{MID_SPEECH_RMS_THRESHOLD}/{MID_SPEECH_SILENCE_MS}ms,"
f" 発話{MID_SPEECH_MIN_SPEAKING_SEC}秒経過後, 間隔{MID_SPEECH_COOLDOWN_SEC}秒,"
f" 最大{MID_SPEECH_MAX_PER_TURN}回/発話)",
flush=True,
)
# 独り言は課金が発生しうる機能のため、意図しない有効化を防ぐためログに明示する
monologue_active = ENABLE_CONTEXT_MONOLOGUE and bool(GEMINI_API_KEY)
print(
f"独り言: {'有効' if monologue_active else '無効'}"
f" (完了{CONTEXT_MONOLOGUE_DELAY_SEC}秒後, 生成間隔{CONTEXT_MONOLOGUE_MIN_INTERVAL_SEC}秒,"
f" 最大{CONTEXT_MONOLOGUE_MAX_CHARS}文字)",
flush=True,
)
print(
f"AEC: {'有効' if ENABLE_AEC else '無効'}"
f" (delay={AEC_STREAM_DELAY_MS}ms, NS={AEC_ENABLE_NS}, AGC={AEC_ENABLE_AGC})",
flush=True,
)
print(
f"言い淀みフィルタ: {'有効' if ENABLE_HESITATION_FILTER else '無効'}"
f" (最大{HESITATION_MAX_CHARS}文字, 相槌上限{HESITATION_MAX_BACKCHANNELS}回/ターン)",
flush=True,
)
print(
f"相槌コーディネーター: 相槌間隔{BACKCHANNEL_COOLDOWN_SEC}秒,"
f" フィラースキップ閾値{FILLER_SKIP_THRESHOLD_SEC}秒",
flush=True,
)
print("-" * 50, flush=True)
fillers = await preload_fillers() if ENABLE_FILLER else []
ai_backchannels = await preload_ai_backchannels() if ENABLE_AI_BACKCHANNEL else []
mid_speech_backchannels = (
await preload_mid_speech_backchannels() if ENABLE_MID_SPEECH_BACKCHANNEL else []
)
playback = PlaybackController()
capture = VADCapture()
capture.attach_loop(asyncio.get_event_loop())
if ENABLE_AEC:
try:
aec_processor = AECProcessor(
sample_rate=SAMPLE_RATE,
stream_delay_ms=AEC_STREAM_DELAY_MS,
enable_ns=AEC_ENABLE_NS,
enable_agc=AEC_ENABLE_AGC,
)
capture.aec_processor = aec_processor
playback.aec_processor = aec_processor
print(f" | AECプロセッサ初期化完了 (backend={aec_processor.backend_name})", flush=True)
except AECBackendError as e:
print(f" ![AEC無効化] {e}", flush=True)
print(" ![AEC無効化] AECなしで続行します（フェーズ3と同じ動作）。", flush=True)
state = DuplexStateMachine()
speculative = SpeculativeLLM()
coordinator = BackchannelCoordinator()
history: list[dict] = []
status: dict = {"turn_speech_duration": 0.0}
device = _pick_input_device()
stream = sd.RawInputStream(
samplerate=SAMPLE_RATE,
blocksize=FRAME_SAMPLES,
device=device,
channels=1,
dtype="int16",
callback=capture._sd_callback,
)
with stream:
print(" | [マイク開始] 話しかけてください", flush=True)
await asyncio.gather(
input_loop(state, speculative, playback, capture, history, status, mid_speech_backchannels, ai_backchannels, coordinator),
output_loop(state, speculative, playback, fillers, history, coordinator),
monitor_loop(state, playback, capture, ai_backchannels, history, status, coordinator),
)
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\n[終了]", flush=True)
② Whisperサーバー（server_whisper.py ）
import os
import tempfile
from flask import Flask, request, jsonify
from flask_cors import CORS
from faster_whisper import WhisperModel
app = Flask(__name__)
CORS(app)
# faster-whisper + large-v3 で精度重視
# compute_type: float16(高精度) / int8_float16(速度寄り) / int8(最速)
MODEL_SIZE = "large-v3"
DEVICE = "cuda"
COMPUTE_TYPE = "float16"
# no_speech_prob フィルタ閾値
# この値以上のセグメントは「人間の発話ではない」として除外する
# AEC残留ノイズのハルシネーション対策
# 0.7 がスタート値。効きすぎたら上げる（0.8-0.9）、弱ければ下げる（0.5-0.6）
NO_SPEECH_PROB_THRESHOLD = 0.7
print(f"Loading faster-whisper model: {MODEL_SIZE} (device={DEVICE}, compute_type={COMPUTE_TYPE})...")
model = WhisperModel(MODEL_SIZE, device=DEVICE, compute_type=COMPUTE_TYPE)
print(f"Model loaded. Ready to serve. (no_speech_prob threshold={NO_SPEECH_PROB_THRESHOLD})")
@app.route('/whisper', methods=['POST'])
def transcribe_audio():
if 'audio_file' not in request.files:
return jsonify({"error": "No audio file provided"}), 400
audio_file = request.files['audio_file']
# 一時ファイルとして保存
with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as temp_wav:
temp_path = temp_wav.name
audio_file.save(temp_path)
try:
# 文字起こし実行
segments, info = model.transcribe(
temp_path,
language="ja",
beam_size=5,
vad_filter=True,
vad_parameters=dict(
min_silence_duration_ms=500,
),
)
accepted = []
rejected = []
for seg in segments:
if seg.no_speech_prob >= NO_SPEECH_PROB_THRESHOLD:
rejected.append(f"[p={seg.no_speech_prob:.2f}] {seg.text.strip()}")
else:
accepted.append(seg.text)
text = "".join(accepted).strip()
# ログ出力
if rejected:
for r in rejected:
print(f"🚫 Rejected (no_speech): {r}")
if text:
print(f"🎤 Received Audio -> Text: {text}")
else:
if rejected:
print(f"🔇 All segments rejected as non-speech")
else:
print(f"🔇 No speech detected")
return jsonify({"text": text})
except Exception as e:
print(f"❌ Error: {e}")
return jsonify({"error": str(e)}), 500
finally:
# 掃除
if os.path.exists(temp_path):
os.remove(temp_path)
if __name__ == '__main__':
# 外部からのアクセスを許可するために 0.0.0.0 を指定
app.run(host='0.0.0.0', port=5001)
