openai-agents-python 语音管道实战用 VoicePipeline 将智能体工作流接入语音应用【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-pythonVoicePipeline是 openai-agents-python 中用于把文本型智能体工作流包装成语音应用的核心类你只需要传入一个工作流管道会自动完成输入音频的语音转文字STT、说话结束检测、在合适的时机调用你的工作流并把工作流输出的文本重新合成为音频TTS。读完本文你将掌握如何配置与运行VoicePipeline、如何区分静态与流式两种音频输入、如何消费流式事件结果以及如何基于生命周期事件自行实现打断interruption处理等关键实战能力。VoicePipeline 是什么三步式语音处理流程从 VoicePipeline 源码 的类注释可以清晰看到它的工作方式整个处理过程只有三步把输入的音频转写成文本speech-to-textSTT运行你提供的workflow产出连续的文本响应把文本响应转换为流式音频输出text-to-speechTTS。从源码看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解析出默认模型。configVoicePipelineConfig配置对象见 pipeline_config.py聚合了模型提供者、追踪tracing设置以及 STT/TTS 模型参数。它既支持直接传VoicePipelineConfig实例也支持传dictSDK 会通过coerce_dataclass_config自动完成字典到数据类的转换。VoicePipelineConfig 关键字段字段默认值说明model_providerOpenAIVoiceModelProvider语音模型提供者负责把模型名映射为具体的 STT/TTS 模型实例tracing_disabledFalse是否关闭管道自身的追踪tracingNone管道的TracingConfig追踪配置trace_include_sensitive_dataTrue追踪中是否包含敏感数据仅作用于语音管道本身不影响你 workflow 内部的内容trace_include_sensitive_audio_dataTrue追踪中是否包含音频数据workflow_nameVoice Agent追踪中使用的工作流名称group_id自动生成用于把同一次对话/进程产生的多条 trace 关联成组的标识trace_metadataNone附加到 trace 上的元数据字典stt_settingsSTTModelSettings()STT 模型设置tts_settingsTTSModelSettings()TTS 模型设置TTS 与 STT 模型设置TTSModelSettingsmodel.py常用字段包括voice使用的音色。OpenAI 内置音色包括alloy、ash、ballad、coral、echo、fable、onyx、nova、sage、shimmer、verse、marin、cedar也支持传入自定义音色 IDTTSCustomVoice。buffer_size默认120指流式输出的音频数据块的最小字节规模决定 TTS 输出被切分为多小的块推给下游。dtype默认np.int16返回音频数据的 numpy 数据类型也可配置为np.float32。transform_data一个可选的转换函数用于把 TTS 产出的音频数组转成你需要的形状。instructions默认指令为你会收到不完整的句子不要补全句子只需朗读文本用于控制 TTS 的朗读口吻。text_splitter用于把文本按句切分后再送入 TTS 的拆分函数默认是基于句子的拆分器这样无需等待整段文本生成完毕即可开始合成。speed朗读速度取值范围 0.25 到 4.0。STTModelSettingsmodel.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 ProviderVoicePipelineConfig.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_providerOpenAIVoiceModelProvider( api_key..., # 不传则使用默认 API Key ), workflow_nameMy Weather Agent, trace_include_sensitive_dataFalse, tts_settings{voice: nova, speed: 1.0}, stt_settings{language: en}, ) pipeline VoicePipeline(workflowmy_workflow, configconfig)运行管道两种音频输入方式VoicePipeline.run()接受两种形式的音频输入pipeline.py你需要根据自己的场景选择1.AudioInput静态完整音频当你已经拥有完整的音频输入、只想为它产生一个结果时使用AudioInput见 input.py。它适合不需要检测说话者何时说完的场景例如你有预先录制好的音频你是 push-to-talk按住说话类应用用户说完的时机是明确的。AudioInput是一个 dataclass字段包括buffer音频数据必须是np.int16或np.float32的 numpy 数组frame_rate采样率默认24000sample_width采样位宽字节数默认2channels声道数默认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, dtypenp.int16) audio_input AudioInput(bufferbuffer) 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实时播放返回的音频事件。结果与事件流StreamedAudioResultpipeline.run()的返回结果是StreamedAudioResult见 result.py。它是一个允许你事件发生即消费的异步流对象通过result.stream()逐条产出VoiceStreamEvent。VoiceStreamEvent有三种类型见 events.pyVoiceStreamEventAudio包含一段音频数据numpy 数组对应type voice_stream_event_audioVoiceStreamEventLifecycle生命周期事件对应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() # 本轮音频播完恢复收音深入如何编写 workflowVoiceWorkflowBaseworkflow.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( nameAssistant, instructionsYoure speaking to a human, so be polite and concise., modelgpt-5.6-sol, ) pipeline VoicePipeline(workflowSingleAgentVoiceWorkflow(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 sounddeviceimport 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 fThe weather in {city} is {choices[0]}. agent Agent( nameAssistant, instructionsYoure speaking to a human, so be polite and concise., modelgpt-5.6-sol, tools[get_weather], ) async def main(): pipeline VoicePipeline(workflowSingleAgentVoiceWorkflow(agent)) # 3 秒静音占位实际应使用麦克风数据 audio_input AudioInput(buffernp.zeros(24000 * 3, dtypenp.int16)) result await pipeline.run(audio_input) # 用 sounddevice 实时播放返回的音频 player sd.OutputStream(samplerate24000, channels1, dtypenp.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），仅供参考