AI Agents · 课时

构建语音对话循环

Q: 「构建语音对话循环」这节课中我会学到什么？

录音 → 转写 → 推理 → 语音输出循环，并处理打断。 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

录音 → 转写 → 推理 → 语音输出循环，并处理打断。

第 3 / 4 课13 个步骤

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

语音对话闭环

完整的语音对话闭环会将麦克风输入与代理输出连接成连续循环：录音 → 转写 → 代理处理 → TTS → 播放 → 重复。

本课将介绍每个组件，以及使语音闭环区别于基于文本的代理的各种挑战：静音检测、中断处理和语音结束检测。

从麦克风录音

使用 sounddevice 从默认麦克风捕获音频。您可以录制固定时长，也可以一直录制到检测到静音为止。请始终以 16kHz 单声道录音，这是 Whisper 所需的采样率。

使用 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。

免费开始

用 AI 导师学习 AI Agents — 免费

在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。

课程: 60
课程: 239

免费开始学习查看完整课程

常见问题解答

「构建语音对话循环」课时是免费的吗？

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

「构建语音对话循环」这节课中我会学到什么？

录音 → 转写 → 推理 → 语音输出循环，并处理打断。你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

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

「构建语音对话循环」课时需要多长时间？

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

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

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

此课程中的所有课时

← 返回 AI Agents