openai-agents-python 语音 Agent 快速入门：用 VoicePipeline 实现「语音输入 → 智能体推理 → 语音输出」全流程

本篇指南基于 openai-agents-python 官方快速入门文档展开，讲解如何用 VoicePipeline 把智能体工作流包装成一套完整的语音应用：麦克风音频经语音转文本（STT）后进入 Agent 工作流，推理结果再经文本转语音（TTS）流式播报出来。读完本文你将掌握语音依赖的安装方式、VoicePipeline 三步架构、SingleAgentVoiceWorkflow 的使用，以及如何用 sounddevice 播放流式音频结果，并能直接跑通一个支持工具调用与智能体转移（handoff）的语音天气助手。

前提条件

开始之前，请先按照 Agents SDK 的基础快速入门说明完成环境准备，并设置好虚拟环境。随后从 SDK 安装可选的语音依赖项：

pip install 'openai-agents[voice]'

下面的演示代码还会用到 sounddevice 来处理麦克风和扬声器 I/O，但它不属于 voice extra，需要单独安装：

pip install sounddevice

此外，代码中直接操作 numpy 数组，因此也需要 numpy（通常随 openai-agents 的依赖链一同安装，若缺失可执行 pip install numpy）。

核心概念：VoicePipeline 的三步流程

语音功能的核心是 VoicePipeline（源码位于 src/agents/voice/pipeline.py）。它是一个「有主见」（opinionated）的语音管线，固定按三步工作：

1. 语音转文本（STT）：运行语音转文本模型，把音频转成文本。
2. 运行你的代码：通常是智能体工作流（agentic workflow），基于转录文本产生结果文本。
3. 文本转语音（TTS）：运行文本转语音模型，把结果文本转换回音频，并以流式方式输出。

整个过程可用下面的流程图概括：

graph LR
    A["🎤 Audio Input"]
    subgraph Voice_Pipeline [Voice Pipeline]
        direction TB
        B["Transcribe (speech-to-text)"]
        C["Your Code"]
        D["Text-to-speech"]
        B --> C --> D
    end
    E["🎧 Audio Output"]
    A --> Voice_Pipeline
    Voice_Pipeline --> E

从源码看，VoicePipeline 在 pipeline.py 的类文档中明确给出了上述三步定义。它的构造函数接收三个可选组件：

• workflow：要运行的工作流（继承自 VoiceWorkflowBase），必填；
• stt_model：语音转文本模型，不传则使用默认 OpenAI 模型；
• tts_model：文本转语音模型，不传则使用默认 OpenAI 模型；
• config：VoicePipelineConfig 或其字典形式，不传则使用默认配置。

其中默认模型在 openai_model_provider.py 中定义：STT 默认使用 gpt-4o-transcribe，TTS 默认使用 gpt-4o-mini-tts。

pipeline.run() 根据输入类型分派两条路径（见 pipeline.py）：

• 传入 AudioInput（静态音频缓冲区）时执行单轮流程：STT 转录 → workflow.run(文本) → 逐段送 TTS；
• 传入 StreamedAudioInput（可追加的音频流）时执行多轮流程：先创建流式转录会话（create_session），逐轮转录、逐轮推理、逐轮合成，并支持 workflow.on_start() 先播报开场白。

第一步：定义智能体

如果你曾用本 SDK 构建过智能体，这一步会非常熟悉。我们将设置两个智能体、一项已配置的任务转移（handoff）和一个工具。

import random

from agents import Agent
from agents.decorators import tool
from agents.extensions.handoff_prompt import prompt_with_handoff_instructions


@tool
def get_weather(city: str) -> str:
    """Get the weather for a given city."""
    print(f"[debug] get_weather called with city: {city}")
    choices = ["sunny", "cloudy", "rainy", "snowy"]
    return f"The weather in {city} is {random.choice(choices)}."


spanish_agent = Agent(
    name="Spanish",
    handoff_description="A Spanish-speaking agent.",
    instructions=prompt_with_handoff_instructions(
        "You're speaking to a human, so be polite and concise. Speak in Spanish.",
    ),
    model="gpt-5.6-sol",
)

agent = Agent(
    name="Assistant",
    instructions=prompt_with_handoff_instructions(
        "You're speaking to a human, so be polite and concise. If the user speaks in Spanish, hand off to the Spanish agent.",
    ),
    model="gpt-5.6-sol",
    handoffs=[spanish_agent],
    tools=[get_weather],
)

这里有两个值得注意的语音场景细节：

• prompt_with_handoff_instructions：来自 agents.extensions.handoff_prompt，它会把「用户可能随时切换语言/话题」的提示注入系统指令，让 Agent 在听到西班牙语时主动把手头对话转移给 Spanish 智能体。在语音对话中，这种主动转移能力是让多语言助手自然运作的关键。
• handoff_description：为转移目标提供一段描述，帮助主智能体判断何时转移。

完整示例可在 examples/voice/static/main.py 中看到几乎相同的 Agent 定义。

第二步：构建语音管线

接下来设置一个简单的语音管线，并使用 SingleAgentVoiceWorkflow 作为工作流：

from agents.voice import SingleAgentVoiceWorkflow, VoicePipeline

pipeline = VoicePipeline(workflow=SingleAgentVoiceWorkflow(agent))

SingleAgentVoiceWorkflow 是内置的「单智能体」工作流实现，其核心行为（见 workflow.py）：

1. 把本次转录文本追加进内部输入历史 _input_history；
2. 调用 Runner.run_streamed(agent, input_history) 流式运行 Agent；
3. 借助 VoiceWorkflowHelper.stream_text_from 从结果流中逐段提取 response.output_text.delta 文本增量并 yield 出去；
4. 运行结束后把 result.to_input_list() 写回输入历史，并用 result.last_agent 更新当前 Agent —— 这意味着多轮对话中即使发生了 handoff，下一轮也会从正确的智能体继续。

如果你需要更复杂的逻辑（多次 Runner 调用、自定义消息历史、自定义上下文等），可以继承 VoiceWorkflowBase 实现自己的 run(transcription) 方法（见 workflow.py）。SingleAgentVoiceWorkflow 还支持传入 callbacks（如 SingleAgentWorkflowCallbacks.on_run）和 context 参数；VoiceWorkflowBase.on_start() 则可在用户说话前先由 TTS 播报问候语。

第三步：运行管线并播放音频

import numpy as np
import sounddevice as sd
from agents.voice import AudioInput

# For simplicity, we'll just create 3 seconds of silence
# In reality, you'd get microphone data
buffer = np.zeros(24000 * 3, dtype=np.int16)
audio_input = AudioInput(buffer=buffer)

result = await pipeline.run(audio_input)

# Create an audio player using `sounddevice`
player = sd.OutputStream(samplerate=24000, channels=1, dtype=np.int16)
player.start()

# Play the audio stream as it comes in
async for event in result.stream():
    if event.type == "voice_stream_event_audio":
        player.write(event.data)

几个关键点：

• 采样率约定：示例用 24000 Hz 生成 3 秒静音，这与 AudioInput 的默认 frame_rate=24000（见 input.py）以及播放器的 samplerate=24000 保持一致。AudioInput 的完整字段为：buffer（必须是 int16 或 float32 的 numpy 数组）、frame_rate（默认 24000）、sample_width（默认 2，即 16-bit PCM）、channels（默认 1）。

• 流式消费：pipeline.run() 返回 StreamedAudioResult，其 stream() 方法是异步迭代器（见 result.py），会持续产出三类事件（定义在 events.py）：

  • voice_stream_event_audio：携带 numpy 音频数组的 data，写入播放器即可发声；
  • voice_stream_event_lifecycle：生命周期事件，取值为 turn_started / turn_ended / session_ended；
  • voice_stream_event_error：管线运行中发生的错误。

  音频是边生成边播放的：TTS 内部按 TTSModelSettings.buffer_size（默认 120 字节的最小分块）缓冲 PCM 数据，并保证各文本段的音频按顺序派发（见 result.py），所以首字节延迟很低。

完整整合：可直接运行的语音 Agent 示例

把以上步骤拼起来，就是文档中的完整代码：

import asyncio
import random

import numpy as np
import sounddevice as sd

from agents import Agent
from agents.decorators import tool
from agents.voice import (
    AudioInput,
    SingleAgentVoiceWorkflow,
    VoicePipeline,
)
from agents.extensions.handoff_prompt import prompt_with_handoff_instructions


@tool
def get_weather(city: str) -> str:
    """Get the weather for a given city."""
    print(f"[debug] get_weather called with city: {city}")
    choices = ["sunny", "cloudy", "rainy", "snowy"]
    return f"The weather in {city} is {random.choice(choices)}."


spanish_agent = Agent(
    name="Spanish",
    handoff_description="A Spanish-speaking agent.",
    instructions=prompt_with_handoff_instructions(
        "You're speaking to a human, so be polite and concise. Speak in Spanish.",
    ),
    model="gpt-5.6-sol",
)

agent = Agent(
    name="Assistant",
    instructions=prompt_with_handoff_instructions(
        "You're speaking to a human, so be polite and concise. If the user speaks in Spanish, hand off to the Spanish agent.",
    ),
    model="gpt-5.6-sol",
    handoffs=[spanish_agent],
    tools=[get_weather],
)


async def main():
    pipeline = VoicePipeline(workflow=SingleAgentVoiceWorkflow(agent))
    buffer = np.zeros(24000 * 3, dtype=np.int16)
    audio_input = AudioInput(buffer=buffer)

    result = await pipeline.run(audio_input)

    # Create an audio player using `sounddevice`
    player = sd.OutputStream(samplerate=24000, channels=1, dtype=np.int16)
    player.start()

    # Play the audio stream as it comes in
    async for event in result.stream():
        if event.type == "voice_stream_event_audio":
            player.write(event.data)


if __name__ == "__main__":
    asyncio.run(main())

运行此代码示例后，智能体将生成可供你收听的语音音频！

深入源码：管线配置与模型调优

默认模型与模型提供者

不显式传 stt_model / tts_model 时，VoicePipeline 会通过 config.model_provider（默认为 OpenAIVoiceModelProvider，见 pipeline_config.py）按名称创建模型：STT 默认 gpt-4o-transcribe，TTS 默认 gpt-4o-mini-tts（openai_model_provider.py）。你也可以在构造 VoicePipeline 时用字符串或模型实例覆盖：

from agents.voice import VoicePipeline, SingleAgentVoiceWorkflow, OpenAIVoiceModelProvider

pipeline = VoicePipeline(
    workflow=SingleAgentVoiceWorkflow(agent),
    stt_model="gpt-4o-transcribe",
    tts_model="gpt-4o-mini-tts",
)

TTS 设置：音色、语速、缓冲与文本切分

VoicePipelineConfig.tts_settings（类型 TTSModelSettings，见 model.py）支持以下常用参数：

参数	默认值	说明
voice	None	使用的音色。内置可选 alloy、ash、ballad、coral、echo、fable、onyx、nova、sage、shimmer、verse、marin、cedar，也支持传自定义音色 ID 字典 {"id": "..."}
buffer_size	120	流式输出的最小音频分块大小（字节），越小首包越快、开销越大
dtype	np.int16	输出音频数组的数据类型，可选 int16 或 float32
instructions	提示模型「只朗读收到的片段，不要补全句子」	控制 TTS 的朗读风格与语气
text_splitter	基于句子的分割器	把文本切成若干片段分别送 TTS，避免等待整段文本才发声
speed	None	朗读速度，取值范围 0.25 ~ 4.0
transform_data	None	对输出音频数组的自定义变换函数

STT 设置：转录提示、语言与流式轮次检测

VoicePipelineConfig.stt_settings（STTModelSettings）支持 prompt（转录指令）、language（音频语言）、temperature（模型温度）；用于流式音频输入时还支持 turn_detection（轮次检测配置）、languages（候选语言代码列表，gpt-transcribe / gpt-live-transcribe 支持，优先级高于 language）、keywords（热词提示，引导转录结果）。

追踪与敏感数据

VoicePipelineConfig 还提供了面向可观测性的配置：workflow_name（默认 "Voice Agent"，用于追踪命名）、group_id（把同一次会话的多条 trace 归组）、trace_metadata（附加元数据）、tracing_disabled、trace_include_sensitive_data 与 trace_include_sensitive_audio_data（控制 trace 中是否包含转录文本与音频数据，默认均为 True，生产环境可按合规要求关闭）。

从静音到真对话：参考官方示例

本文的示例为了演示方便使用了 3 秒静音作为输入。想要真正「和智能体说话」，仓库提供了两个可直接运行的示例：

• examples/voice/static:python -m examples.voice.static.main。它使用 util.py 中的 record_audio() 先在终端录制一段音频，再送入 VoicePipeline 处理并把结果流式播放出来。示例还演示了 SingleAgentWorkflowCallbacks.on_run 回调，可在每次工作流运行时打印转录文本（见 main.py）。你可以依次尝试：让智能体讲个笑话、询问 Tokyo 的天气（触发 get_weather 工具）、用西班牙语问好（触发 handoff 到 Spanish 智能体）。
• examples/voice/streamed:基于 StreamedAudioInput 的双向流式示例，通过队列持续追加麦克风音频，实现更接近真实通话体验的多轮语音交互。

更深入的内容（如自定义工作流、自定义 STT/TTS 模型、管线配置细节）可继续阅读 voice 文档总览 以及 voice 快速入门（英文原版） 对照学习。