OpenAI Agents SDK #18：拆掉黑盒，「三段流水线」里究竟跑着什么？

一、三步流水线全景

AudioInput  ──→  [1] STT（转录）  ──→  [2] Workflow（你的代码）  ──→  [3] TTS  ──→  StreamedAudioResult

pipeline = VoicePipeline(
    workflow=my_workflow,          # 必选：你的业务逻辑
    stt_model=None,                # 可选：默认用 OpenAI
    tts_model=None,                # 可选：默认用 OpenAI
    config=None,                   # 可选：默认 VoicePipelineConfig()
)

result: StreamedAudioResult = await pipeline.run(audio_input)

二、STT 替换：两条路径，选哪条？

class MySTTModel(STTModel):
    @property
    def model_name(self) -> str:
        return "deepgram-nova-2"

async def transcribe(
        self, input: AudioInput, settings: STTModelSettings,
        trace_include_sensitive_data: bool,
        trace_include_sensitive_audio_data: bool,
    ) -> str:
        # 调用你的 STT API，返回转录文本
        ...

async def create_session(
        self, input: StreamedAudioInput, settings: STTModelSettings,
        trace_include_sensitive_data: bool,
        trace_include_sensitive_audio_data: bool,
    ) -> StreamedTranscriptionSession:
        # 返回支持 transcribe_turns() 的会话对象
        ...

pipeline = VoicePipeline(
    workflow=workflow,
    stt_model=MySTTModel(),   # 直接传实例，绕过 provider
)

class MyProvider(VoiceModelProvider):
    def get_stt_model(self, name: str | None) -> STTModel:
        if name == "deepgram":
            return DeepgramSTTModel()
        return OpenAISTTModel(name)

def get_tts_model(self, name: str | None) -> TTSModel:
        ...

config = VoicePipelineConfig(model_provider=MyProvider())
pipeline = VoicePipeline(workflow=workflow, config=config)

三、TTS 的 7 个配置项：哪些真的值得调？

四、Workflow：Base 还是 Single？

class MyWorkflow(VoiceWorkflowBase):
    async def run(self, transcription: str) -> AsyncIterator[str]:
        # 接收转录文本，产出待朗读的文本流
        result = await Runner.run_streamed(my_agent, transcription)
        async for chunk in VoiceWorkflowHelper.stream_text_from(result):
            yield chunk

async def on_start(self) -> AsyncIterator[str]:
        # 可选：多轮模式的欢迎语，通过 TTS 播放
        yield "你好，有什么可以帮你的？"

from agents.voice import SingleAgentVoiceWorkflow

pipeline = VoicePipeline(
    workflow=SingleAgentVoiceWorkflow(my_agent),
)

• 需要在转录之前做预处理（比如语言检测）
• 需要多个 Agent 并行处理同一段语音
• 需要条件分支逻辑（见官方示例 my_workflow.py 中的 secret_word 分支5）

五、事件流消费：三种 VoiceStreamEvent

# VoiceStreamEvent 是 TypeAlias：
# VoiceStreamEventAudio | VoiceStreamEventLifecycle | VoiceStreamEventError

async for event in result.stream():
    if event.type == "voice_stream_event_audio":
        # event.data: NDArray[np.int16 | np.float32] | None
        audio_player.write(event.data)

elif event.type == "voice_stream_event_lifecycle":
        # event.event: "turn_started" | "turn_ended" | "session_ended"
        if event.event == "turn_started":
            print("[新一轮 TTS 开始]")
        elif event.event == "turn_ended":
            print("[当前轮播放完毕]")
        elif event.event == "session_ended":
            print("[整个会话结束]")
            break

elif event.type == "voice_stream_event_error":
        # event.error: Exception
        print(f"[错误] {event.error}")

• turn_started：StreamedAudioResult._start_turn() 调用时，每轮 TTS 开始前触发
• turn_ended：当前轮 TTS 完成（finish_turn=True）后触发
• session_ended：_dispatch_audio() 完成所有分派，整个会话结束时触发

1. voice_stream_event_audio 的 data 字段可以是 None——处理前要判空
2. 多轮对话时必须监听 session_ended 才能知道什么时候关闭播放器；只监听音频事件会漏掉会话结束信号

六、VoicePipelineConfig：10 个字段的分类记法

config = VoicePipelineConfig(
    trace_include_sensitive_audio_data=False,   # 别把用户声音存进追踪
    trace_include_sensitive_data=False,          # 别把转录文本存进追踪
    workflow_name="客服语音 v2",
    tts_settings=TTSModelSettings(
        voice="nova",
        speed=1.1,
        buffer_size=80,                          # 降低首帧延迟
    ),
    stt_settings=STTModelSettings(
        language="zh",                           # 指定中文，提升转录准确率
    ),
)

七、理清两个容易混淆的类名

• StreamedTranscriptionSession：STT 层的流式会话，由 STTModel.create_session() 返回，暴露 transcribe_turns() 和 close() 方法——这是多轮对话持续接收音频并逐轮输出转录文本的对象3
• StreamedAudioResult：Pipeline 输出层的事件流，由 pipeline.run() 返回，暴露 stream() 方法产出三种 VoiceStreamEvent7

八、三条实践建议

下期预告 #19：进入语音子系统的另一半——Realtime Agents 内部的 RealtimeAgent、RealtimeSession、事件处理器完整解析，以及与 Voice Pipeline 在实际项目中如何做最终选型。