openai-agents-python 语音工作流Voice Workflow开发指南从SingleAgentVoiceWorkflow到自定义多轮对话流程【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-pythonagents.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, )参数类型说明agentAgent[TContext]每个语音轮次都会运行的这个 Agent。可以带tools、handoffs、instructions等完整配置callbacksSingleAgentWorkflowCallbacks \| None可选回调目前包含on_run在工作流每次运行时被触发contextTContext \| 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, contextself._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 回调接口SingleAgentWorkflowCallbacksclass 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_fromclassmethod 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.md5.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( nameAssistant, instructionsprompt_with_handoff_instructions( Youre speaking to a human, so be polite and concise., ), modelgpt-5.6-sol, # tools[...], handoffs[...] 均可按需配置 ) pipeline VoicePipeline(workflowSingleAgentVoiceWorkflow(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, dtypenp.int16) # 示例3 秒静音实际应使用麦克风数据 audio_input AudioInput(bufferbuffer) result await pipeline.run(audio_input) 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)pipeline.run()接受两种输入docs/voice/pipeline.mdAudioInput一次性提供完整音频适合预录音频或按键对讲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 fThe weather in {city} is {random.choice(choices)}. spanish_agent Agent( nameSpanish, handoff_descriptionA spanish speaking agent., instructionsprompt_with_handoff_instructions( Youre speaking to a human, so be polite and concise. Speak in Spanish., ), modelgpt-5.6-sol, ) agent Agent( nameAssistant, instructionsprompt_with_handoff_instructions( Youre speaking to a human, so be polite and concise. If the user speaks in Spanish, handoff to the spanish agent., ), modelgpt-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(..., contextself._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实现自己的run8.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），仅供参考 openai-agents-python 语音工作流（Voice Workflow）开发指南：从 `SingleAgentVoiceWorkflow` 到自定义多轮对话流程
openai-agents-python 语音工作流Voice Workflow开发指南从SingleAgentVoiceWorkflow到自定义多轮对话流程【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-pythonagents.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, )参数类型说明agentAgent[TContext]每个语音轮次都会运行的这个 Agent。可以带tools、handoffs、instructions等完整配置callbacksSingleAgentWorkflowCallbacks \| None可选回调目前包含on_run在工作流每次运行时被触发contextTContext \| 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, contextself._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 回调接口SingleAgentWorkflowCallbacksclass 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_fromclassmethod 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.md5.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( nameAssistant, instructionsprompt_with_handoff_instructions( Youre speaking to a human, so be polite and concise., ), modelgpt-5.6-sol, # tools[...], handoffs[...] 均可按需配置 ) pipeline VoicePipeline(workflowSingleAgentVoiceWorkflow(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, dtypenp.int16) # 示例3 秒静音实际应使用麦克风数据 audio_input AudioInput(bufferbuffer) result await pipeline.run(audio_input) 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)pipeline.run()接受两种输入docs/voice/pipeline.mdAudioInput一次性提供完整音频适合预录音频或按键对讲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 fThe weather in {city} is {random.choice(choices)}. spanish_agent Agent( nameSpanish, handoff_descriptionA spanish speaking agent., instructionsprompt_with_handoff_instructions( Youre speaking to a human, so be polite and concise. Speak in Spanish., ), modelgpt-5.6-sol, ) agent Agent( nameAssistant, instructionsprompt_with_handoff_instructions( Youre speaking to a human, so be polite and concise. If the user speaks in Spanish, handoff to the spanish agent., ), modelgpt-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(..., contextself._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实现自己的run8.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），仅供参考