openai-agents-python 语音管道实战：用 VoicePipeline 将智能体工作流接入语音应用
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
VoicePipeline是 openai-agents-python 中用于把"文本型智能体工作流"包装成"语音应用"的核心类：你只需要传入一个工作流，管道会自动完成输入音频的语音转文字（STT）、说话结束检测、在合适的时机调用你的工作流，并把工作流输出的文本重新合成为音频（TTS）。读完本文，你将掌握如何配置与运行VoicePipeline、如何区分静态与流式两种音频输入、如何消费流式事件结果，以及如何基于生命周期事件自行实现打断（interruption）处理等关键实战能力。
VoicePipeline 是什么：三步式语音处理流程
从 VoicePipeline 源码 的类注释可以清晰看到它的工作方式，整个处理过程只有三步：
- 把输入的音频转写成文本（speech-to-text，STT）；
- 运行你提供的
workflow，产出连续的文本响应； - 把文本响应转换为流式音频输出（text-to-speech，TTS）。
从源码看，VoicePipeline.run()会根据输入类型分发到两条内部路径（pipeline.py）：传入AudioInput走_run_single_turn（单轮静态音频），传入StreamedAudioInput走_run_multi_turn（流式多轮）。无论哪条路径，都会返回一个StreamedAudioResult，供上层以异步流的方式消费事件。
配置一个语音管道
创建VoicePipeline时可以设置三类内容（对应 pipeline.py 的构造参数）：
workflow：每次有新的音频被转写成文本时都会执行的代码，类型为VoiceWorkflowBase（见 workflow.py）。stt_model与tts_model：使用的语音转文字模型与文字转语音模型，类型分别为STTModel与TTSModel（见 model.py）。两者都可以直接传模型实例，也可以传模型名字符串；不传时由配置中的模型提供者（model provider）解析出默认模型。config：VoicePipelineConfig配置对象（见 pipeline_config.py），聚合了模型提供者、追踪（tracing）设置以及 STT/TTS 模型参数。它既支持直接传VoicePipelineConfig实例，也支持传dict，SDK 会通过coerce_dataclass_config自动完成字典到数据类的转换。
VoicePipelineConfig 关键字段
| 字段 | 默认值 | 说明 |
|---|---|---|
model_provider | OpenAIVoiceModelProvider | 语音模型提供者，负责把模型名映射为具体的 STT/TTS 模型实例 |
tracing_disabled | False | 是否关闭管道自身的追踪 |
tracing | None | 管道的TracingConfig追踪配置 |
trace_include_sensitive_data | True | 追踪中是否包含敏感数据（仅作用于语音管道本身，不影响你 workflow 内部的内容） |
trace_include_sensitive_audio_data | True | 追踪中是否包含音频数据 |
workflow_name | "Voice Agent" | 追踪中使用的工作流名称 |
group_id | 自动生成 | 用于把同一次对话/进程产生的多条 trace 关联成组的标识 |
trace_metadata | None | 附加到 trace 上的元数据字典 |
stt_settings | STTModelSettings() | STT 模型设置 |
tts_settings | TTSModelSettings() | TTS 模型设置 |
TTS 与 STT 模型设置
TTSModelSettings（model.py）常用字段包括：
voice：使用的音色。OpenAI 内置音色包括alloy、ash、ballad、coral、echo、fable、onyx、nova、sage、shimmer、verse、marin、cedar，也支持传入自定义音色 ID（TTSCustomVoice）。buffer_size：默认120，指流式输出的音频数据块的最小字节规模，决定 TTS 输出被切分为多小的块推给下游。dtype：默认np.int16，返回音频数据的 numpy 数据类型，也可配置为np.float32。transform_data：一个可选的转换函数，用于把 TTS 产出的音频数组转成你需要的形状。instructions：默认指令为"你会收到不完整的句子，不要补全句子，只需朗读文本"，用于控制 TTS 的朗读口吻。text_splitter：用于把文本按句切分后再送入 TTS 的拆分函数，默认是基于句子的拆分器，这样无需等待整段文本生成完毕即可开始合成。speed：朗读速度，取值范围 0.25 到 4.0。
STTModelSettings（model.py）常用字段包括：
prompt：给模型的提示/指令。language：输入音频的语言。temperature：采样温度。turn_detection：使用流式音频输入时的轮次检测（turn detection）设置；OpenAI 默认值为{"type": "semantic_vad"}（见 openai_stt.py），即基于语义的语音活动检测。languages：流式输入时输入音频可能的语言列表（API 语言码），gpt-transcribe与gpt-live-transcribe支持，优先级高于language。keywords：流式输入时用于引导转写的词或短语，同样由gpt-transcribe与gpt-live-transcribe支持。
模型提供者（Model Provider）
VoicePipelineConfig.model_provider默认使用 OpenAIVoiceModelProvider，其默认 STT 模型为gpt-4o-transcribe、默认 TTS 模型为gpt-4o-mini-tts（见 openai_model_provider.py）。它支持通过api_key、base_url、organization、project自定义 OpenAI 客户端，也可以直接传入一个现成的AsyncOpenAI客户端（此时不能再同时传上述参数）。SDK 会共享同一个 HTTP 客户端以复用连接池，且客户端采用懒加载，避免在未设置 API Key 的环境下构造即报错。
from agents.voice import VoicePipeline, VoicePipelineConfig from agents.voice.models.openai_model_provider import OpenAIVoiceModelProvider config = VoicePipelineConfig( model_provider=OpenAIVoiceModelProvider( api_key="...", # 不传则使用默认 API Key ), workflow_name="My Weather Agent", trace_include_sensitive_data=False, tts_settings={"voice": "nova", "speed": 1.0}, stt_settings={"language": "en"}, ) pipeline = VoicePipeline(workflow=my_workflow, config=config)运行管道：两种音频输入方式
VoicePipeline.run()接受两种形式的音频输入（pipeline.py），你需要根据自己的场景选择：
1.AudioInput：静态完整音频
当你已经拥有完整的音频输入、只想为它产生一个结果时使用AudioInput（见 input.py）。它适合不需要检测说话者何时说完的场景，例如：
- 你有预先录制好的音频；
- 你是 push-to-talk（按住说话）类应用，用户说完的时机是明确的。
AudioInput是一个 dataclass，字段包括：
buffer：音频数据，必须是np.int16或np.float32的 numpy 数组；frame_rate：采样率，默认24000；sample_width：采样位宽（字节数），默认2；channels：声道数，默认1。
它还提供了to_audio_file()（返回(filename, bytes, content_type)三元组）和to_base64()两个便捷方法，便于把音频转成文件或 base64 形式交给模型。
import numpy as np from agents.voice import AudioInput # 3 秒的静音，实际项目中应替换为真实的麦克风数据 buffer = np.zeros(24000 * 3, dtype=np.int16) audio_input = AudioInput(buffer=buffer) result = await pipeline.run(audio_input)2.StreamedAudioInput：流式音频与活动检测
当你可能需要检测用户何时说完话时使用StreamedAudioInput（见 input.py）。它内部维护一个asyncio.Queue，你可以随时通过add_audio()把检测到的音频块推入队列；传入None表示音频流结束。语音管道会通过一种称为"活动检测（activity detection）"的机制，在合适的时机自动运行你的智能体工作流。
from agents.voice import StreamedAudioInput audio_input = StreamedAudioInput() # 在音频采集循环中不断推入音频块 await audio_input.add_audio(audio_chunk) # numpy 数组，int16 或 float32 # ... 更多音频块 ... await audio_input.add_audio(None) # 标记流结束在_run_multi_turn的实现中（pipeline.py），管道会先尝试执行工作流可选的on_start()（例如播报开场白/问候语），然后为 STT 创建转录会话，并持续消费transcribe_turns()产出的文本；每检测到一个新的轮次文本，就触发一次工作流运行。这正是文档中所说的"每个检测到的轮次都会触发工作流的单独一次执行"。
一个真实的流式示例可以参考 examples/voice/streamed/main.py：它以 50ms 为读取粒度从麦克风采集int16音频，通过await self._audio_input.add_audio(data)持续推入管道，同时用sd.OutputStream实时播放返回的音频事件。
结果与事件流：StreamedAudioResult
pipeline.run()的返回结果是StreamedAudioResult（见 result.py）。它是一个允许你"事件发生即消费"的异步流对象，通过result.stream()逐条产出VoiceStreamEvent。
VoiceStreamEvent有三种类型（见 events.py）：
VoiceStreamEventAudio：包含一段音频数据（numpy 数组），对应type == "voice_stream_event_audio"；VoiceStreamEventLifecycle：生命周期事件，对应type == "voice_stream_event_lifecycle"，其event字段取值包括turn_started（新轮次开始处理）、turn_ended（该轮次所有音频已派发完毕）、session_ended（会话结束）；VoiceStreamEventError：错误事件，对应type == "voice_stream_event_error"，携带error异常对象。
错误语义：什么在stream()时抛出
关于错误处理，原文档明确了两点，且与 result.py 中stream()的实现一致：
- 终止性（terminal）的管道错误会在应用消费
StreamedAudioResult.stream()时被抛出； - 如果一次运行本身是干净的，但语音转文字的转录会话未能成功关闭，流不会无限期等待，而是抛出那个关闭错误；
- 如果轮次本身已经失败、且随后转录会话关闭也失败，流会保留原始的轮次错误作为主错误，而不是用关闭错误覆盖它。
消费事件的典型代码：
result = await pipeline.run(input) async for event in result.stream(): if event.type == "voice_stream_event_audio": # 播放音频 pass elif event.type == "voice_stream_event_lifecycle": # 处理生命周期事件（turn_started / turn_ended / session_ended） pass elif event.type == "voice_stream_event_error": # 处理错误 pass从实现上看，StreamedAudioResult内部会为每一段文本创建独立的音频合成任务，并用一个有序派发器（_dispatch_audio）保证音频块按文本产生顺序输出；文本会先经过text_splitter切句，再按buffer_size分块推入队列。若整个会话没有产生任何音频，_done()也会确保派发器启动并最终发出session_ended终止事件，避免stream()永久等待。
最佳实践：中断（Interruption）处理
当前 Agents SDK没有为StreamedAudioInput提供内置的中断处理能力。每个检测到的轮次都会触发工作流的单独一次运行。如果要在自己的应用里实现打断逻辑，官方推荐的做法是监听VoiceStreamEventLifecycle生命周期事件：
turn_started：表示新的轮次已被转写、处理即将开始；turn_ended：表示该轮次的所有音频都已派发完毕。
利用这两个事件，你可以实现"模型开始说话时静音麦克风、应用播放完该轮次全部音频后再取消静音"的经典打断处理：当turn_started到达时停止采集用户输入（避免用户的声音被转录进正在进行的回复轮次），当turn_ended到达时恢复采集。
async for event in result.stream(): if event.type == "voice_stream_event_lifecycle": if event.event == "turn_started": mute_microphone() # 模型开始回复，静音麦克风 elif event.event == "turn_ended": unmute_microphone() # 本轮音频播完，恢复收音深入：如何编写 workflow
VoiceWorkflowBase（workflow.py）是一个抽象基类，你需要实现run(transcription) -> AsyncIterator[str]：它接收一次转写文本，然后不断yield出将要被 TTS 朗读的文本。绝大多数情况下，你会创建Agent并用Runner.run_streamed()运行它们，再从流中提取文本事件——VoiceWorkflowHelper.stream_text_from()正好封装了这个过程（过滤raw_response_event中的response.output_text.delta事件）。
如果工作流足够简单（只有一个起始 Agent、没有自定义逻辑），可以直接使用内置的SingleAgentVoiceWorkflow。它会在内部维护输入历史（_input_history），把每次转写追加为用户消息，用Runner.run_streamed()运行 Agent，并把流式文本透传出来，同时更新历史与当前 Agent（支持 handoff 后继续对话），见 workflow.py。
from agents import Agent from agents.voice import SingleAgentVoiceWorkflow, VoicePipeline agent = Agent( name="Assistant", instructions="You're speaking to a human, so be polite and concise.", model="gpt-5.6-sol", ) pipeline = VoicePipeline(workflow=SingleAgentVoiceWorkflow(agent))对于更复杂的场景——多次Runner调用、自定义消息历史、自定义逻辑、自定义运行配置——则继承VoiceWorkflowBase自己实现：
from collections.abc import AsyncIterator from agents.voice import VoiceWorkflowBase class MyWorkflow(VoiceWorkflowBase): async def run(self, transcription: str) -> AsyncIterator[str]: # 在这里运行任意逻辑（可多次调用 Runner、调用工具等） yield f"你说的是：{transcription}" async def on_start(self) -> AsyncIterator[str]: # 可选：在收到任何用户输入前先播报问候语 yield "你好，请问有什么可以帮你？"注意on_start()默认不做任何事（workflow.py），需要开场白时再覆写；管道会在多轮会话中先消费on_start()的文本并结束该轮次（pipeline.py），避免开场白被合并进第一个用户轮次。
完整可运行示例
结合 docs/voice/quickstart.md，一个完整的静态音频示例是这样的（运行前需先按 docs/quickstart.md 配置好环境，并安装可选依赖pip install 'openai-agents[voice]'与pip install sounddevice）：
import asyncio import numpy as np import sounddevice as sd from agents import Agent from agents.decorators import tool from agents.voice import AudioInput, SingleAgentVoiceWorkflow, VoicePipeline @tool def get_weather(city: str) -> str: """Get the weather for a given city.""" choices = ["sunny", "cloudy", "rainy", "snowy"] return f"The weather in {city} is {choices[0]}." agent = Agent( name="Assistant", instructions="You're speaking to a human, so be polite and concise.", model="gpt-5.6-sol", tools=[get_weather], ) async def main(): pipeline = VoicePipeline(workflow=SingleAgentVoiceWorkflow(agent)) # 3 秒静音占位，实际应使用麦克风数据 audio_input = AudioInput(buffer=np.zeros(24000 * 3, dtype=np.int16)) result = await pipeline.run(audio_input) # 用 sounddevice 实时播放返回的音频 player = sd.OutputStream(samplerate=24000, channels=1, dtype=np.int16) player.start() async for event in result.stream(): if event.type == "voice_stream_event_audio": player.write(event.data) elif event.type == "voice_stream_event_lifecycle": print(f"[lifecycle] {event.event}") elif event.type == "voice_stream_event_error": print(f"[error] {event.error}") player.close() if __name__ == "__main__": asyncio.run(main())更完整的流式对话演示（真实麦克风采集 + 按键控制 + 生命周期事件处理）可以进一步研究 examples/voice/streamed/main.py 及其配套的 workflow 实现，把VoicePipeline与StreamedAudioInput、turn_started/turn_ended生命周期事件组合成可打断的双向语音对话应用。
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
