openai-agents-python 语音工作流（Voice Workflow）开发指南：从SingleAgentVoiceWorkflow到自定义多轮对话流程
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
agents.voice.workflow是 openai-agents-python 语音能力（Voice）的核心抽象层，它定义了"语音工作流"这一概念：一段接收用户语音转写文本、产出将被合成语音回复文本的代码。本文以 workflow 模块 为主线，结合 语音管道文档、语音快速上手、单元测试 tests/voice/test_workflow.py 与官方示例 examples/voice/streamed/my_workflow.py，系统讲解VoiceWorkflowBase、SingleAgentVoiceWorkflow、VoiceWorkflowHelper三大构件，并给出从"单 Agent 开箱即用"到"自定义多轮对话逻辑"的完整落地路径。读完本文，你将能够独立编写、接入并测试自己的语音工作流。
一、Voice Workflow 在语音管道中的位置
在 openai-agents-python 中，语音应用由VoicePipeline承载，它是一个标准的三段式流水线：
- Transcribe（语音转文本）：由 STT 模型把输入音频变成转写文本；
- Your Workflow Code（即 Voice Workflow）：这是整个管道中唯一由开发者完全掌控的环节。它接收转写文本，执行任意业务逻辑（通常是运行 Agent），并产出文本；
- Text-to-speech（文本转语音）：由 TTS 模型把工作流产出的文本合成为音频返回给用户。
也就是说，工作流是"语音管道的大脑"：语音输入与语音输出都被框架接管，而"听懂之后该做什么"完全由你的 workflow 代码决定。这正是 docs/voice/pipeline.md 中workflow配置项所指向的对象——每次检测到新的转写文本，管道都会调用一次你的工作流。
二、核心抽象：VoiceWorkflowBase
所有语音工作流都必须继承VoiceWorkflowBase，它是一个抽象基类（abc.ABC），定义了工作流的契约：
class VoiceWorkflowBase(abc.ABC): @abc.abstractmethod def run(self, transcription: str) -> AsyncIterator[str]: ...2.1run()：唯一必须实现的方法
run(transcription)是工作流的主入口，其语义为：
- 输入：一段用户语音的转写文本（字符串）；
- 输出：一个异步迭代器（
AsyncIterator[str]），逐段产出将被 TTS 合成语音的文本； - 自由度：方法体内可以运行任何逻辑。文档注释明确指出，最常见的做法是创建
Agent，调用Runner.run_streamed()运行它，然后从结果流中把文本事件逐步yield出去。
这里的关键设计是流式输出：工作流不必等 Agent 全部跑完才说话，而是可以一边运行一边把已生成的文本片段吐给 TTS，实现低延迟的"边说边想"体验。
2.2on_start()：可选的主动开口钩子
async def on_start(self) -> AsyncIterator[str]: return yieldon_start()是一个可选方法，会在收到任何用户输入之前被调用，默认实现为空操作。它典型的用途是：通过 TTS 播报一句问候语或操作指引（例如"您好，我是语音助手，请问需要什么帮助？"）。如果你的产品希望"助手先开口"，就重写此方法并yield出开场文本；否则保持默认即可。
三、开箱即用：SingleAgentVoiceWorkflow
对于"单个起始 Agent、无自定义逻辑"的简单场景，无需自己实现抽象类——框架提供了SingleAgentVoiceWorkflow这个现成实现。
3.1 构造参数
SingleAgentVoiceWorkflow( agent: Agent[TContext], callbacks: SingleAgentWorkflowCallbacks | None = None, *, context: TContext | None = None, )| 参数 | 类型 | 说明 |
|---|---|---|
agent | Agent[TContext] | 每个语音轮次都会运行的这个 Agent。可以带tools、handoffs、instructions等完整配置 |
callbacks | SingleAgentWorkflowCallbacks \| None | 可选回调，目前包含on_run，在工作流每次运行时被触发 |
context | TContext \| None | 可选的应用上下文，会被转发到每一次Agent 运行中（关键字参数） |
3.2 运行机制：输入历史的自动管理
SingleAgentVoiceWorkflow.run()的内部实现（src/agents/voice/workflow.py）包含四个步骤：
- 触发回调：若提供了
callbacks，先调用self._callbacks.on_run(self, transcription)； - 追加转写：把当前转写作为
{"role": "user", "content": transcription}追加进self._input_history（类型为list[TResponseInputItem]）； - 流式运行 Agent：调用
Runner.run_streamed(self._current_agent, self._input_history, context=self._context)运行 Agent； - 流式转发文本并同步状态：通过
VoiceWorkflowHelper.stream_text_from(result)逐段yield文本；运行结束后用result.to_input_list()更新输入历史、用result.last_agent更新当前 Agent。
其中第 4 步的"历史回填"是精髓：to_input_list()会把这一轮产生的完整对话（包括 tool call、tool output、assistant 消息）写回_input_history，因此多轮语音对话天然具备记忆——下一轮run()时，Agent 能看到此前所有轮次的上下文。last_agent的更新则保证：如果 Agent 在对话中发生 handoff（交接给别的 Agent），后续轮次会自动继续使用交接后的 Agent。
3.3 回调接口SingleAgentWorkflowCallbacks
class SingleAgentWorkflowCallbacks: def on_run(self, workflow: SingleAgentVoiceWorkflow, transcription: str) -> None: """Called when the workflow is run."""目前该接口只有一个方法on_run，在工作流每次被调用时触发，参数为工作流实例与本次转写文本。可以用于埋点、日志、统计等旁路逻辑，而无需侵入工作流主流程。
四、流式文本提取：VoiceWorkflowHelper.stream_text_from
无论是内置的SingleAgentVoiceWorkflow还是自定义工作流，从 Agent 的流式运行结果中提取"该说的文本"都是一项高频操作。框架为此提供了VoiceWorkflowHelper.stream_text_from：
@classmethod async def stream_text_from(cls, result: RunResultStreaming) -> AsyncIterator[str]: async for event in result.stream_events(): if ( event.type == "raw_response_event" and event.data.type == "response.output_text.delta" ): yield event.data.delta它包装一个RunResultStreaming对象，遍历stream_events()，只筛选出类型为raw_response_event且data.type == "response.output_text.delta"的事件，并把event.data.delta（增量文本片段）逐个产出。这样上层只需关心"文本流"，无需接触底层事件结构的细节。
五、把工作流接入VoicePipeline
工作流本身不直接处理音频；它由VoicePipeline驱动。完整的最小可运行示例来自 docs/voice/quickstart.md：
5.1 安装依赖
pip install 'openai-agents[voice]' pip install sounddevice # 麦克风/扬声器 I/O，不属于 voice extra5.2 定义 Agent 并组装管道
from agents import Agent from agents.decorators import tool from agents.extensions.handoff_prompt import prompt_with_handoff_instructions from agents.voice import SingleAgentVoiceWorkflow, VoicePipeline agent = Agent( name="Assistant", instructions=prompt_with_handoff_instructions( "You're speaking to a human, so be polite and concise.", ), model="gpt-5.6-sol", # tools=[...], handoffs=[...] 均可按需配置 ) pipeline = VoicePipeline(workflow=SingleAgentVoiceWorkflow(agent))管道构造时还可配置 STT/TTS 模型、模型提供者、Tracing、工作流名称、trace ID 等，详见 docs/voice/pipeline.md 中关于VoicePipelineConfig的说明。
5.3 运行管道与消费结果
import numpy as np import sounddevice as sd from agents.voice import AudioInput buffer = np.zeros(24000 * 3, dtype=np.int16) # 示例：3 秒静音，实际应使用麦克风数据 audio_input = AudioInput(buffer=buffer) result = await pipeline.run(audio_input) 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)pipeline.run()接受两种输入（docs/voice/pipeline.md）：
AudioInput：一次性提供完整音频，适合预录音频或按键对讲（push-to-talk）场景；StreamedAudioInput：支持边说话边推送音频分片，由管道通过 "activity detection"（活动检测）自动判断说话结束时机并触发工作流。
result.stream()产出的VoiceStreamEvent有三种类型：VoiceStreamEventAudio（音频分片）、VoiceStreamEventLifecycle（轮次开始/结束等生命周期事件）、VoiceStreamEventError（错误事件）。管道级终态错误会在消费stream()时抛出。
六、自定义工作流实战：完整示例
当业务需要"多个 Runner 调用、自定义消息历史、自定义逻辑或自定义配置"时，官方推荐直接继承VoiceWorkflowBase实现自己的逻辑。examples/voice/streamed/my_workflow.py 给出了一个带"密语猜测"分支的完整示例，可直接作为模板：
import random from collections.abc import AsyncIterator, Callable from agents import Agent, Runner, TResponseInputItem from agents.decorators import tool from agents.extensions.handoff_prompt import prompt_with_handoff_instructions from agents.voice import VoiceWorkflowBase, VoiceWorkflowHelper @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 {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, handoff to the spanish agent.", ), model="gpt-5.6-sol", handoffs=[spanish_agent], tools=[get_weather], ) class MyWorkflow(VoiceWorkflowBase): def __init__(self, secret_word: str, on_start: Callable[[str], None]): self._input_history: list[TResponseInputItem] = [] self._current_agent = agent self._secret_word = secret_word.lower() self._on_start = on_start async def run(self, transcription: str) -> AsyncIterator[str]: self._on_start(transcription) # 把转写加入输入历史，维持多轮记忆 self._input_history.append( {"role": "user", "content": transcription} ) # 命中密语：绕过 Agent，直接回复固定文本 if self._secret_word in transcription.lower(): yield "You guessed the secret word!" self._input_history.append( {"role": "assistant", "content": "You guessed the secret word!"} ) return # 常规路径：运行 Agent 并流式转发文本 result = Runner.run_streamed(self._current_agent, self._input_history) async for chunk in VoiceWorkflowHelper.stream_text_from(result): yield chunk # 更新输入历史与当前 Agent（支持 handoff 后的持续对话） self._input_history = result.to_input_list() self._current_agent = result.last_agent这个例子展示了自定义工作流的三个典型能力：
- 业务分支：转写文本命中特定关键词时，直接
yield固定回复并return，完全绕过 Agent，实现"规则优先、模型兜底"； - 手动历史管理：自己维护
_input_history，也可以决定把哪些内容（比如固定回复）写入历史，从而影响后续轮次的上下文； - 复用流式提取：常规路径下依然借助
VoiceWorkflowHelper.stream_text_from转发 Agent 文本流，不必重复实现事件过滤逻辑。
七、源码级验证：测试用例如何佐证工作流行为
tests/voice/test_workflow.py 使用ScriptedModel对工作流行为做了确定性验证，是理解实现细节的最佳参考。
7.1 多轮输入历史与工具调用
test_single_agent_workflow验证了两轮对话下工作流的状态机行为：
- 第一轮：Agent 产出一个函数调用（
some_function）与一条文本消息，工作流只yield文本（"a_message"），但_input_history会被更新为包含function_call、function_call_output、assistant 消息的完整序列； - 第二轮：由于历史已回填，Agent 能感知第一轮的工具结果，产出
"done"； - 测试断言
workflow._input_history与workflow._current_agent在每轮结束后都被正确更新——这正是多轮记忆与 handoff 连续性的实现证据。
7.2 上下文逐轮转发
test_single_agent_workflow_forwards_context_on_every_turn验证了context参数的语义：工作流以context={"user_id": "user-123"}构造后，每轮运行都通过Runner.run_streamed(..., context=self._context)把同一上下文传给 Agent。测试中的工具read_user_id从RunContextWrapper读取user_id，两轮均返回"user-123"，证实上下文会在每一轮被透传，可用于携带用户身份、会话状态等应用级数据。
八、最佳实践与注意事项
8.1 中断（Interruptions）处理
依据 docs/voice/pipeline.md 的 Best practices 一节：SDK目前不提供内置的中断处理。使用StreamedAudioInput时，每次检测到的语音轮次都会触发一次独立的工作流运行。若应用需要支持"用户打断助手"，可监听VoiceStreamEventLifecycle事件：turn_started表示新轮次转写完成、处理开始；turn_ended表示该轮次所有音频已派发完毕。典型做法是——turn_started时静音用户麦克风，播放完该轮全部音频后再取消静音。
8.2 何时选用哪种工作流
| 场景 | 推荐方案 |
|---|---|
| 单个起始 Agent、无需自定义逻辑 | 直接用SingleAgentVoiceWorkflow |
| 需要问候语/开场白 | 继承VoiceWorkflowBase重写on_start |
| 多 Runner 调用、自定义历史、关键词路由、自定义配置 | 继承VoiceWorkflowBase实现自己的run |
8.3 保持低延迟
工作流应优先使用Runner.run_streamed()加VoiceWorkflowHelper.stream_text_from()的流式链路，让文本片段边生成边交给 TTS，避免等 Agent 完整结束才开始合成语音。
8.4 参考更多示例
- examples/voice/static/main.py：可实际对话的语音演示应用；
- examples/voice/streamed/my_workflow.py：本文剖析的自定义工作流示例；
- docs/voice/pipeline.md 与 docs/voice/quickstart.md：管道配置、结果事件与完整运行示例的权威说明；
- 基础 SDK 上手流程见 docs/quickstart.md。
结语
agents.voice.workflow用不到 120 行代码定义了一个小而美的抽象：VoiceWorkflowBase划定"转写进、文本出"的契约，SingleAgentVoiceWorkflow提供带自动记忆的单 Agent 默认实现，VoiceWorkflowHelper抹平流式事件提取的样板代码，而测试与示例则完整展示了如何在多轮对话、工具调用、handoff、上下文透传等真实场景中驾驭它。掌握了这一层抽象，你就掌握了为 openai-agents-python 语音管道注入任意业务逻辑的钥匙。
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
