0Pricing
AI Agents · 课时

优化语音代理的延迟

流式 TTS、响应分块和最小化首词延迟。

优化语音代理的延迟 是 CoddyKit 上的免费 AI Agents 课时。 这是第 4 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

语音场景为何重视延迟

在文本聊天中，3 秒的延迟通常可以接受。但在语音对话中，超过1.5 秒就会让人感觉不自然，并打断对话流程。

语音延迟主要由三个部分组成：转写时间（STT）、LLM 处理时间（TTFT 加生成时间）以及 TTS 合成时间。分别优化每个部分后，整体效果会叠加，显著改善用户体验。

测量延迟预算

在进行优化之前，请先测量每个组件。为闭环加入计时调用，以便了解时间实际花费在哪里。

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。无需等待整个音频合成完成，而是让第一块音频在约 200 毫秒内开始播放，同时生成其余内容。

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 响应拆分成句子，逐句合成并播放。第一句话可在约 300 毫秒内开始播放。

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～300 毫秒）。对于简短响应或高频短语，本地 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 是更好的平衡选择。

常见问题解答

「优化语音代理的延迟」课时是免费的吗？

是的 — 「优化语音代理的延迟」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「优化语音代理的延迟」这节课中我会学到什么？

流式 TTS、响应分块和最小化首词延迟。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 4 节课，共 4 节。

「优化语音代理的延迟」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 使用 Whisper 和 Deepgram 进行语音转文本
  2. 代理回复中的文本转语音
  3. 构建语音对话循环
  4. 优化语音代理的延迟
← 返回 AI Agents