用 LangChain 做语音智能体

这篇适合谁，读完能做到什么

适合已经会用 LangChain 智能体、想给它加上语音对话能力的开发者。读完你应该能分清「三段式」和「语音直出」两种架构，理解一个三段式语音智能体的数据流，知道每一段该怎么做流式处理，并能把三段串成一条可以通过 WebSocket 对外服务的流水线。

语音智能体是什么

语音智能体是能与用户进行自然语音对话的智能体。它把语音识别、自然语言处理、生成式 AI 和语音合成结合在一起^[1]。典型场景包括客服、个人助理、免手操作的界面，以及陪练和培训^[1]。

每个语音智能体都要完成三件事：听（采集音频并转写）、想（理解意图、推理、规划）、说（生成音频并流式返回给用户）^[1]。

两种架构怎么选

第一种是「三明治」架构：语音转文字（STT）、基于文本的 LangChain 智能体、文字转语音（TTS）三段拼在一起^[1]。它的好处是每个部件都能控制，STT 或 TTS 的厂商可以随时换，还能用上文本模型的最新能力，边界也清楚^[1]。代价是要编排多个服务，流水线本身更复杂。把语音转成文字也会丢掉语气、情绪这类信息^[1]。

第二种是语音直出（speech-to-speech）：用一个多模态模型直接处理音频输入、原生生成音频输出^[1]。架构更简单，简单对话的延迟通常更低^[1]。但可选的模型少，厂商锁定的风险更大，音频处理过程的透明度也较低，可控性和定制空间有限^[1]。

本文用的是三明治架构。在部分 STT 和 TTS 厂商组合下，三明治架构可以做到 700 毫秒以内的延迟，同时保持各部件可替换^[1]。

示例：三明治店的点单语音助手

示例是一个管理三明治店订单的语音智能体，语音识别用的是 AssemblyAI，语音合成用的是 Cartesia^[1]。配套的参考应用是 voice-sandwich-demo 仓库，安装和运行步骤见它的 README^[1]。

浏览器端负责采集麦克风音频、编码成 PCM，通过 WebSocket 把音频块持续发给服务端，并播放收到的合成语音^[1]。服务端通过 WebSocket 接收连接，依次调度三个阶段^[1]。WebSocket 是双向实时通信，这套架构换到电话系统（如 Twilio、Vonage）或 WebRTC 上也能用^[1]。

整条流水线用异步生成器（async generator）或异步迭代器（async iterator）串起来，下游在上游还没结束时就能开始处理，从而压缩端到端延迟^[1]。

第一段：语音转文字

STT 阶段采用生产者—消费者模式：音频块在发送给识别服务的同时，接收转写事件，所以在音频还没全部到达时就能开始转写^[1]。

事件分两类。stt_chunk 是识别过程中的部分转写结果；stt_output 是最终的、带格式的转写文本，它会触发智能体处理^[1]。示例连接的是 AssemblyAI 的实时语音识别接口，这是一条持久的 WebSocket 连接，音频配置为 16kHz PCM，并开启自动的轮次格式化^[1]。

第二段：LangChain 智能体

智能体阶段接收转写文本，经过 LangChain 智能体处理，并把它生成的所有文本内容块流式输出^[1]。示例中定义了两个工具，分别用来把菜品加入订单、确认订单：

# Define agent tools
def add_to_order(item: str, quantity: int) -> str:
    """Add an item to the customer's sandwich order."""
    return f"Added {quantity} x {item} to the order."
def confirm_order(order_summary: str) -> str:
    """Confirm the final order with the customer."""
    return f"Order confirmed: {order_summary}. Sending to kitchen."

系统提示词要求回答简洁友好，不使用表情符号、特殊字符和 Markdown，因为回答会被语音合成引擎朗读出来^[1]。

流式输出用的是 stream_events(version="v3") 配合 stream.messages，模型边生成边输出 token，不必等完整回复^[1]。为了让多轮对话能引用前文，智能体还配了检查点（checkpointer），用唯一的 thread ID 保存每个会话的状态^[1]。

第三段：文字转语音

TTS 阶段把智能体的回复文本合成为音频，流式返回给客户端。它同样采用生产者—消费者模式，内部合并两路异步流：一路把上游事件原样传递，并把智能体文本发给 TTS 厂商；另一路接收合成好的音频块^[1]。

Cartesia 这类 TTS 厂商收到文本就会开始合成，所以在智能体还没写完整个回复时，播放就能开始^[1]。上游的所有事件都会原样穿过 TTS 阶段，客户端和其他观察者因此能跟踪整条流水线的状态^[1]。

把三段串成流水线

用 RunnableGenerator 把三个阶段依次连接起来：

pipeline = (
    RunnableGenerator(stt_stream)      # Audio → STT events

接入 WebSocket 时，从客户端接收的音频字节流进入流水线，流水线输出的 TTS 音频再发回客户端：

    # Transform audio through pipeline
    output_stream = pipeline.atransform(websocket_audio_stream())
    # Send TTS audio back to client
    async for event in output_stream:
        if event.type == "tts_chunk":
            await websocket.send_bytes(event.audio)

每个阶段都独立、并发地处理事件：音频一到就开始转写，拿到转写文本就开始推理，智能体生成文本就开始合成语音^[1]。