音声会話ループの構築
割り込み処理を備えた、録音 → 文字起こし → 推論 → 発話のサイクルを構築します。
「音声会話ループの構築」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン3/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
音声会話のループ
完全な音声会話のループでは、マイク入力とエージェント出力を継続的なサイクルで接続します。録音 → 文字起こし → エージェント処理 → TTS → 再生 → 繰り返しという流れです。
このレッスンでは各コンポーネントに加えて、テキストベースのエージェントとは異なる音声ループ特有の課題、つまり無音検出、割り込み処理、発話終了検出について学びます。
マイクからの録音
sounddeviceを使って、デフォルトのマイクから音声を取り込みます。一定時間録音するか、無音が検出されるまで録音します。Whisperが想定するサンプルレートである16kHzのモノラルで、必ず録音してください。
pip install sounddevice soundfile numpyでインストールします。
import sounddevice as sd
import numpy as np
import tempfile
import soundfile as sf
SAMPLE_RATE = 16000 # 16kHz mono — optimal for Whisper
DURATION = 5 # seconds
def record_audio(duration=DURATION, sample_rate=SAMPLE_RATE):
print(f'Recording for {duration} seconds...')
audio_data = sd.rec(
int(duration * sample_rate),
samplerate=sample_rate,
channels=1,
dtype='float32'
)
sd.wait() # block until recording finishes
print('Recording complete')
return audio_data, sample_rate
def audio_to_file(audio_data, sample_rate, filepath='/tmp/recording.wav'):
sf.write(filepath, audio_data, sample_rate)
return filepath発話終了のための無音検出
一定時間録音する方法は扱いにくいものです。ユーザーが2秒で話し終わっても、待たなければならないからです。無音検出を使うと、ユーザーが一定時間静かになったときに録音を自動的に停止できます。
import sounddevice as sd
import numpy as np
from collections import deque
SAMPLE_RATE = 16000
CHUNK_SIZE = 1024 # samples per chunk
SILENCE_THRESHOLD = 0.02 # RMS volume below this = silence
SILENCE_DURATION = 1.5 # seconds of silence to end recording
MAX_DURATION = 30 # max recording length in seconds
def record_until_silence():
chunks = []
silent_chunks = 0
max_silent = int(SILENCE_DURATION * SAMPLE_RATE / CHUNK_SIZE)
max_chunks = int(MAX_DURATION * SAMPLE_RATE / CHUNK_SIZE)
print('Listening... (speak now)')
with sd.InputStream(samplerate=SAMPLE_RATE, channels=1,
blocksize=CHUNK_SIZE, dtype='float32') as stream:
while True:
chunk, _ = stream.read(CHUNK_SIZE)
chunks.append(chunk.copy())
rms = np.sqrt(np.mean(chunk**2))
if rms < SILENCE_THRESHOLD:
silent_chunks += 1
else:
silent_chunks = 0 # speech detected — reset counter
if silent_chunks >= max_silent and len(chunks) > max_silent:
break
if len(chunks) >= max_chunks:
break
audio = np.concatenate(chunks, axis=0)
return audio, SAMPLE_RATE完全な音声ループ
すべてのコンポーネントを継続的なループに接続します。エージェントが応答したらすぐに再び聞き始めることで、自然な対話のやり取りを実現します。
import openai
import os
client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
conversation_history = []
def voice_conversation_loop():
print('Voice agent started. Say something (Ctrl+C to exit).')
while True:
# 1. Record user
audio_data, sample_rate = record_until_silence()
audio_path = audio_to_file(audio_data, sample_rate)
# 2. Transcribe
user_text = transcribe_file(audio_path)
print(f'You: {user_text}')
if not user_text.strip():
continue # empty — listen again
# 3. Run agent
conversation_history.append({'role': 'user', 'content': user_text})
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'system', 'content': 'You are a helpful voice assistant. Keep responses concise.'}] + conversation_history[-10:]
)
agent_text = response.choices[0].message.content
conversation_history.append({'role': 'assistant', 'content': agent_text})
print(f'Agent: {agent_text}')
# 4. TTS and play
say(agent_text) # speaks sentence by sentenceウェイクワード検出の概要
常時聞き取りは、継続的にWhisperを呼び出すためコストが高く、プライバシーの観点でも問題があります。ウェイクワード検出では、軽量なローカルモデルを実行し、特定のフレーズ（例：'Hey Agent'）が聞こえた場合にのみ、完全な処理を開始します。
# Wake word detection with pvporcupine (Picovoice)
# pip install pvporcupine sounddevice
import pvporcupine
import sounddevice as sd
import numpy as np
import os
def listen_for_wake_word(wake_word='computer'):
porcupine = pvporcupine.create(
access_key=os.getenv('PICOVOICE_KEY'),
keywords=[wake_word] # built-in: computer, hey barista, hey google, jarvis...
)
print(f'Listening for wake word: "{wake_word}"...')
with sd.InputStream(
samplerate=porcupine.sample_rate,
channels=1,
dtype='int16',
blocksize=porcupine.frame_length
) as stream:
while True:
frame, _ = stream.read(porcupine.frame_length)
pcm = frame.flatten().astype('int16')
result = porcupine.process(pcm)
if result >= 0:
print(f'Wake word detected!')
porcupine.delete()
return True割り込み処理
ユーザーは、エージェントが話している途中でも割り込める必要があります。再生を別スレッドで実行しながらマイクの音量を監視し、ユーザーが話し始めたら直ちに再生を停止することで、割り込み処理を実装します。
import threading
import sounddevice as sd
import numpy as np
interrupt_event = threading.Event()
def monitor_for_interrupt(threshold=0.03):
def audio_callback(indata, frames, time_info, status):
rms = np.sqrt(np.mean(indata**2))
if rms > threshold:
print('Interrupt detected!')
interrupt_event.set()
with sd.InputStream(callback=audio_callback, channels=1,
samplerate=16000, blocksize=1024):
while not interrupt_event.is_set():
sd.sleep(50)
def speak_with_interrupt(text, voice='nova'):
interrupt_event.clear()
# Start interrupt monitor in background
monitor_thread = threading.Thread(target=monitor_for_interrupt, daemon=True)
monitor_thread.start()
# Play audio sentence by sentence
for sentence in split_into_sentences(text):
if interrupt_event.is_set():
print('Playback interrupted')
break
audio_path = cached_tts(sentence, voice)
play_audio_file(audio_path)ノイズフィルタリング
マイクの音声には、ファンの音、キーボードの打鍵音、部屋の反響などの背景ノイズが含まれることがよくあります。単純なノイズゲートを適用してしきい値未満の音声を抑制し、必要に応じて音声活動検出器（VAD）を使用すると、精度を高められます。
import numpy as np
NOISE_GATE_THRESHOLD = 0.015 # RMS threshold
def apply_noise_gate(audio_data, threshold=NOISE_GATE_THRESHOLD):
rms = np.sqrt(np.mean(audio_data**2))
if rms < threshold:
return np.zeros_like(audio_data) # silence below threshold
return audio_data
def has_speech_content(audio_data, threshold=0.02):
rms_values = [
np.sqrt(np.mean(audio_data[i:i+1600]**2))
for i in range(0, len(audio_data), 1600)
]
speech_frames = sum(1 for r in rms_values if r > threshold)
speech_ratio = speech_frames / max(len(rms_values), 1)
return speech_ratio > 0.1 # at least 10% of frames have speech
# Use in loop before transcribing
audio, sr = record_until_silence()
if has_speech_content(audio):
text = transcribe_file(audio_to_file(audio, sr))
else:
print('No speech detected — listening again')音声会話ループのステートマシン
堅牢な音声ループでは、ステートマシンを使ってエージェントがその時点で何をしているかを追跡し、再生と聞き取りの間の競合状態を防ぎます。
from enum import Enum
class AgentState(Enum):
IDLE = 'idle'
LISTENING = 'listening'
TRANSCRIBING = 'transcribing'
THINKING = 'thinking'
SPEAKING = 'speaking'
current_state = AgentState.IDLE
def set_state(new_state):
global current_state
print(f'State: {current_state.value} -> {new_state.value}')
current_state = new_state
def voice_loop_with_state():
while True:
set_state(AgentState.LISTENING)
audio, sr = record_until_silence()
if not has_speech_content(audio):
continue
set_state(AgentState.TRANSCRIBING)
text = transcribe_file(audio_to_file(audio, sr))
if not text.strip():
continue
print(f'You: {text}')
set_state(AgentState.THINKING)
agent_response = run_agent(text)
print(f'Agent: {agent_response}')
set_state(AgentState.SPEAKING)
speak_with_interrupt(agent_response)会話コンテキストの管理
音声エージェントは会話履歴を保持するため、「それについて詳しく教えてください」や「価格はいくらだと言いましたか」といった追加質問に答えられます。コンテキストがあふれないように、履歴をメモリに保持し、最後のNターンに絞り込んでください。
import openai
import os
client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
MAX_HISTORY_TURNS = 10
SYSTEM_PROMPT = 'You are a concise voice assistant. Keep responses under 3 sentences unless asked for detail.'
class VoiceAgent:
def __init__(self):
self.history = []
def respond(self, user_text):
self.history.append({'role': 'user', 'content': user_text})
# Trim history to last N turns
recent = self.history[-(MAX_HISTORY_TURNS * 2):]
response = client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'system', 'content': SYSTEM_PROMPT}] + recent
)
agent_text = response.choices[0].message.content
self.history.append({'role': 'assistant', 'content': agent_text})
return agent_text
def reset(self):
self.history = []
print('Conversation history cleared')正常なシャットダウンとクリーンアップ
Ctrl+Cなどの終了シグナルを適切に処理します。終了前に音声ストリームを停止し、一時ファイルを削除して、別れの言葉を伝えてください。
import signal
import sys
import glob
import os
TMP_DIR = '/tmp'
agent = None
def cleanup_temp_files():
for f in glob.glob(os.path.join(TMP_DIR, 'recording_*.wav')):
os.unlink(f)
print('Temp files cleaned up')
def graceful_shutdown(signum, frame):
print('\nShutting down voice agent...')
if agent:
say('Goodbye! Have a great day.')
cleanup_temp_files()
sys.exit(0)
# Register signal handler
signal.signal(signal.SIGINT, graceful_shutdown)
signal.signal(signal.SIGTERM, graceful_shutdown)
# Main entry point
agent = VoiceAgent()
print('Voice agent ready. Press Ctrl+C to exit.')
voice_loop_with_state()会話のログ記録
タイムスタンプ、文字起こしテキスト、エージェントの応答、エラーなど、会話の永続的なログを保持してください。これは音声エージェントの問題をデバッグし、会話レビュー機能を構築するために不可欠です。
import json
import os
from datetime import datetime
LOG_DIR = '/tmp/voice_agent_logs'
os.makedirs(LOG_DIR, exist_ok=True)
conversation_log = []
def log_turn(speaker, text, latency_ms=None, error=None):
entry = {
'timestamp': datetime.now().isoformat(),
'speaker': speaker, # 'user' or 'agent'
'text': text,
'latency_ms': latency_ms
}
if error:
entry['error'] = error
conversation_log.append(entry)
def save_conversation_log(session_id=None):
if not session_id:
session_id = datetime.now().strftime('%Y%m%d_%H%M%S')
log_path = os.path.join(LOG_DIR, f'session_{session_id}.json')
with open(log_path, 'w') as f:
json.dump({
'session_id': session_id,
'turns': len(conversation_log),
'log': conversation_log
}, f, indent=2)
print(f'Log saved: {log_path}')
return log_path
# Usage in voice loop
log_turn('user', transcribed_text, latency_ms=420)
log_turn('agent', agent_response, latency_ms=850)知識チェック
音声会話のループにおける無音検出の目的は何ですか。
まとめ：音声会話ループの構築
音声会話ループは、無音検出付きでマイクから録音 → Whisperで文字起こし → 会話履歴を使ってエージェントを実行 → 文単位でTTS応答を生成 → 割り込みを監視しながら再生 → 繰り返し、という流れです。
主なコンポーネントは、競合状態を防ぐステートマシン、常時聞き取りのためのウェイクワード検出、音声品質を高めるノイズゲート、コンテキスト管理のための会話履歴の絞り込みです。Ctrl+Cはクリーンアップと別れのメッセージを伴って適切に処理してください。
よくある質問
「音声会話ループの構築」レッスンは無料ですか？
はい。「音声会話ループの構築」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「音声会話ループの構築」で何を学びますか？
割り込み処理を備えた、録音 → 文字起こし → 推論 → 発話のサイクルを構築します。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン3/4です。
「音声会話ループの構築」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。