0Pricing
AI Agents · レッスン

音声エージェントのレイテンシ最適化

ストリーミングTTS、応答の分割、最初の発話までのレイテンシ短縮を学びます。

「音声エージェントのレイテンシ最適化」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン4/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。

音声でレイテンシーが重要な理由

テキストチャットでは、3秒の遅延は許容できます。しかし音声会話では、1.5秒を超えると不自然に感じられ、会話の流れが途切れます。

音声のレイテンシーには、文字起こし時間（STT）、LLMの処理時間（TTFT＋生成）、TTSの合成時間という3つの主な要素があります。それぞれを最適化することで、ユーザー体験を大幅に改善できます。

レイテンシー予算の測定

最適化する前に、各コンポーネントを測定します。タイミング計測をループに組み込み、実際にどこで時間が使われているかを把握してください。

import time

def voice_loop_timed():
    timing = {}

    # 1. Record
    t0 = time.perf_counter()
    audio, sr = record_until_silence()
    timing['record'] = time.perf_counter() - t0

    # 2. Transcribe
    t0 = time.perf_counter()
    audio_path = audio_to_file(audio, sr)
    user_text = transcribe_file(audio_path)
    timing['transcription'] = time.perf_counter() - t0

    # 3. LLM
    t0 = time.perf_counter()
    agent_text = agent.respond(user_text)
    timing['llm'] = time.perf_counter() - t0

    # 4. TTS
    t0 = time.perf_counter()
    say(agent_text)
    timing['tts'] = time.perf_counter() - t0

    print('Latency breakdown:')
    total = sum(timing.values())
    for step, duration in timing.items():
        print(f'  {step:15} {duration*1000:.0f}ms ({duration/total*100:.0f}%)')

ストリーミングTTS：生成しながら再生

レイテンシーを最も大きく短縮できる方法は、ストリーミングTTSです。音声全体の合成を待つのではなく、残りの音声を同時に生成しながら、約200ms以内に最初の音声チャンクの再生を開始します。

import threading
import queue
import sounddevice as sd
import numpy as np
import io

def stream_tts_and_play(text, voice='nova'):
    audio_queue = queue.Queue()

    def synthesize():
        from openai import OpenAI
        import os
        client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
        with client.audio.speech.with_streaming_response.create(
            model='tts-1', voice=voice, input=text
        ) as response:
            for chunk in response.iter_bytes(chunk_size=4096):
                audio_queue.put(chunk)
        audio_queue.put(None)  # sentinel

    synth_thread = threading.Thread(target=synthesize, daemon=True)
    synth_thread.start()

    # Collect and play (buffering first 2 chunks for smooth start)
    audio_buffer = b''
    min_buffer = 8192
    import pygame
    pygame.mixer.init()

    while True:
        chunk = audio_queue.get()
        if chunk is None:
            break
        audio_buffer += chunk
        if len(audio_buffer) >= min_buffer:
            # play buffer ...
            pass  # simplified — real impl streams to sounddevice

    synth_thread.join()

文単位のストリーミングTTS

最も実用的なストリーミング方法は、LLMの応答を文に分割し、1文ずつ合成して再生することです。最初の文は約300ms以内に再生を開始できます。

import re
import concurrent.futures

def split_sentences(text):
    return [s.strip() for s in re.split(r'(?<=[.!?])\s+', text) if s.strip()]

def tts_and_play_streaming(text, voice='nova'):
    sentences = split_sentences(text)
    if not sentences:
        return

    # Prefetch next sentence while current is playing
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=2)

    # Kick off synthesis of first sentence
    futures = []
    for sentence in sentences:
        futures.append(executor.submit(cached_tts, sentence, voice))

    # Play each sentence as soon as it's ready
    for future in futures:
        audio_path = future.result(timeout=10)
        play_audio_file(audio_path)

    executor.shutdown(wait=False)

# Key insight: while sentence 1 plays (~2 seconds), sentence 2 is being synthesized in parallel

TTFT：最初のトークンまでの時間の最適化

TTFT（最初のトークンまでの時間）とは、LLMリクエストを送信してから、応答の最初のトークンを受信するまでの遅延です。TTFTを最適化すると、ユーザーはより早く回答の冒頭を聞けるようになります。

LLMのストリーミング応答を使用し、文の区切りを検出したらすぐにトークンをTTSへ渡してください。

import openai
import os

client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))

def stream_llm_to_tts(user_text, conversation_history, voice='nova'):
    buffer = ''

    stream = client.chat.completions.create(
        model='gpt-4o-mini',
        messages=conversation_history + [{'role': 'user', 'content': user_text}],
        stream=True  # streaming enabled
    )

    for chunk in stream:
        delta = chunk.choices[0].delta.content or ''
        buffer += delta

        # Check if a sentence is complete
        if buffer.endswith(('.', '!', '?')) and len(buffer) > 20:
            sentence = buffer.strip()
            print(f'Queuing TTS: {sentence[:50]}')
            # Synthesize and play immediately (non-blocking)
            audio_path = cached_tts(sentence, voice)
            play_audio_file(audio_path)
            buffer = ''

    # Flush remaining buffer
    if buffer.strip():
        audio_path = cached_tts(buffer.strip(), voice)
        play_audio_file(audio_path)

定型応答の事前生成

エージェントの応答には、予測しやすいものがあります。挨拶、エラーメッセージ、思考中の表示（「確認しますので、少々お待ちください」）などです。起動時に音声を事前生成しておけば、レイテンシーなしですぐに再生できます。

import os

PRE_GENERATED = {
    'greeting':      'Hello! How can I help you today?',
    'thinking':      'Let me look that up for you.',
    'not_found':     'I could not find information on that. Could you rephrase?',
    'error':         'Sorry, something went wrong. Please try again.',
    'goodbye':       'Goodbye! Have a great day.',
    'clarify':       'Could you give me a bit more detail?',
    'working_on_it': 'Working on it, this may take a moment.'
}

pre_gen_cache = {}

def prewarm_responses(voice='nova'):
    for key, text in PRE_GENERATED.items():
        audio_path = cached_tts(text, voice=voice)
        pre_gen_cache[key] = audio_path
    print(f'Pre-generated {len(pre_gen_cache)} common responses')

def instant_response(key):
    path = pre_gen_cache.get(key)
    if path:
        play_audio_file(path)
    else:
        say(PRE_GENERATED.get(key, ''))

# Usage: while LLM is thinking, play a stall message instantly
instant_response('thinking')

ローカルTTSとクラウドTTSのレイテンシー

クラウドTTSでは、ネットワークの往復時間（約100～300ms）が加わります。短い応答や頻繁に使うフレーズでは、ローカルTTSエンジンの方が高速になる場合があります。ただし、音声品質は低下する可能性があります。

pyttsx3は、Pythonで使える最も簡単なオフラインTTSです。

import pyttsx3
import time

# Local TTS with pyttsx3
def local_tts(text, voice_index=0, rate=175):
    engine = pyttsx3.init()
    voices = engine.getProperty('voices')
    if voice_index < len(voices):
        engine.setProperty('voice', voices[voice_index].id)
    engine.setProperty('rate', rate)  # words per minute
    t0 = time.perf_counter()
    engine.say(text)
    engine.runAndWait()
    print(f'Local TTS latency: {(time.perf_counter()-t0)*1000:.0f}ms')

# Comparison (rough benchmarks):
# pyttsx3 (local):     ~50ms start, robotic quality
# OpenAI TTS-1:        ~200-400ms, good quality
# ElevenLabs turbo:    ~150-250ms, excellent quality
# OpenAI TTS-1-hd:     ~400-800ms, best quality

# Recommendation:
# Real-time voice agent -> OpenAI TTS-1 or ElevenLabs turbo
# Quality recording     -> OpenAI TTS-1-hd or ElevenLabs standard

より高速なLLMモデルの選択

モデルの選択はTTFTに大きく影響します。GPT-4o-miniは、ほとんどの音声エージェントのタスクでGPT-4oより5～10倍高速です。品質要件を満たす範囲で、最も小さいモデルを使用してください。

# Model latency comparison (rough benchmarks for voice agent use case):
MODEL_BENCHMARKS = {
    'gpt-4o-mini':   {'ttft_ms': 300,  'quality': 'good',      'cost': 'very low'},
    'gpt-4o':        {'ttft_ms': 800,  'quality': 'excellent',  'cost': 'medium'},
    'claude-haiku':  {'ttft_ms': 250,  'quality': 'good',       'cost': 'very low'},
    'claude-sonnet': {'ttft_ms': 600,  'quality': 'excellent',  'cost': 'medium'},
    'llama3-8b':     {'ttft_ms': 100,  'quality': 'decent',     'cost': 'free (local)'},
}

# Strategy: use fast model for simple factual queries,
# fall back to powerful model for complex reasoning
def select_model(question):
    # Short, simple questions -> fast model
    if len(question.split()) < 15:
        return 'gpt-4o-mini'
    # Complex, multi-step -> better model
    if any(kw in question.lower() for kw in ['analyze', 'compare', 'explain why', 'write a']):
        return 'gpt-4o'
    return 'gpt-4o-mini'

if __name__ == '__main__':
    for q in ['What time is it in Tokyo?', 'Analyze why sales dropped last quarter and compare to competitors']:
        print(f'{select_model(q)!r} chosen for: "{q}"')

繰り返し質問へのレスポンスキャッシュ

ユーザーは、同じ質問や似た質問を繰り返し尋ねることがよくあります。同一のクエリに対するLLMとTTSのレスポンスをキャッシュし、再度尋ねられたときに即座に提供します。

import hashlib
import json

RESPONSE_CACHE = {}  # in production: use Redis with TTL

def get_cache_key(user_text):
    # Normalize: lowercase, strip punctuation
    import re
    normalized = re.sub(r'[^a-z0-9 ]', '', user_text.lower()).strip()
    return hashlib.md5(normalized.encode()).hexdigest()

def cached_agent_respond(user_text, voice='nova'):
    key = get_cache_key(user_text)

    if key in RESPONSE_CACHE:
        print('Response cache hit!')
        entry = RESPONSE_CACHE[key]
        play_audio_file(entry['audio_path'])
        return entry['text']

    # Cache miss
    text = agent.respond(user_text)
    audio_path = cached_tts(text, voice=voice)

    RESPONSE_CACHE[key] = {'text': text, 'audio_path': audio_path}
    play_audio_file(audio_path)
    return text

エンドツーエンドの遅延目標

すべての最適化を適用すると、適切に構築された音声エージェントは、ユーザーの発話終了からエージェントのレスポンス開始までを1秒未満にできるはずです。

# Target latency budget breakdown (1000ms total):

LATENCY_BUDGET = {
    'silence_detection_end':   0,    # user stops speaking
    'audio_processing':        50,   # RMS check, noise gate
    'transcription_whisper':   400,  # STT API call
    'llm_ttft':                300,  # time to first token (gpt-4o-mini)
    'tts_first_sentence':      200,  # first sentence synthesized
    'playback_start':          1000  # TOTAL: user hears response within 1 second
}

# Optimizations applied:
# - Streaming LLM responses (saves 200-500ms vs waiting for full response)
# - Sentence-by-sentence TTS (play while rest generates)
# - gpt-4o-mini instead of gpt-4o (saves 500ms TTFT)
# - TTS cache for common phrases (saves 200ms on cached phrases)
# - Pre-generated stall messages ('Let me check...' plays instantly)

print('Target: < 1000ms from speech end to first audio byte played')
print('Typical with optimizations: 600-900ms')

非同期文字起こしのオーバーラップ

エージェントがレスポンスを生成して発話している間に、次のユーザー入力の録音をすぐに開始します。パイプラインの各段階を重ねて実行することで、ターン間の待ち時間を短縮できます。

import threading
import time

class PipelineOverlapAgent:
    def __init__(self):
        self.next_audio = None
        self.recording_thread = None

    def start_background_recording(self):
        def record():
            self.next_audio = record_until_silence()
        self.recording_thread = threading.Thread(target=record, daemon=True)
        self.recording_thread.start()

    def get_recorded_audio(self, timeout=30):
        if self.recording_thread:
            self.recording_thread.join(timeout=timeout)
        audio = self.next_audio
        self.next_audio = None
        return audio

    def conversation_turn(self, audio, sr):
        # Transcribe and run agent
        text = transcribe_file(audio_to_file(audio, sr))
        if not text.strip():
            return

        # Start recording NEXT turn BEFORE speaking
        # (user can start speaking as soon as agent starts)
        response = agent.respond(text)

        # Start next recording while speaking
        self.start_background_recording()

        # Speak current response
        speak_with_interrupt(response)

        # Next audio is already being captured in background
        return self.get_recorded_audio()

理解度チェック

音声エージェントで通常、最も大きな遅延削減効果が得られる単一の最適化はどれですか？

まとめ：音声エージェントの遅延最適化

音声の遅延最適化の構成要素は、LLMレスポンスのストリーミング（トークンが到着するたびに文の境界でTTSを開始）、文単位のTTS再生（合成しながら再生）、TTSキャッシュ（繰り返し使われるフレーズを即座に再生）、よく使うレスポンスの事前生成（挨拶やつなぎのフレーズ）、高速なLLMモデルの選択（単純なクエリにはgpt-4o-miniを使用）です。

目標は、ユーザーの発話終了から最初の音声バイトまでを1秒未満にすることです。各コンポーネントを測定してください。文字起こしは、全体の遅延の40%を占めることがよくあります。ローカルTTSは品質と速度のトレードオフになります。多くのエージェントにとっては、文単位のストリーミングに対応したクラウドTTSのほうがバランスに優れています。

よくある質問

「音声エージェントのレイテンシ最適化」レッスンは無料ですか？

はい。「音声エージェントのレイテンシ最適化」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。

「音声エージェントのレイテンシ最適化」で何を学びますか？

ストリーミングTTS、応答の分割、最初の発話までのレイテンシ短縮を学びます。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

AI Agentsを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン4/4です。

「音声エージェントのレイテンシ最適化」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このAI Agentsレッスンでコードを書いて実行できますか？

はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. WhisperとDeepgramによる音声認識
  2. エージェント応答における音声合成
  3. 音声会話ループの構築
  4. 音声エージェントのレイテンシ最適化
← AI Agentsに戻る