概述

聊天界面主导了我们与 AI 的交互方式，但最近多模态 AI 的突破正在开启令人兴奋的新可能性。高质量的生成模型和富有表现力的文本转语音（TTS）系统，现在使得构建感觉更像对话伙伴而非工具的智能体成为可能。

语音智能体就是其中一个例子。你不再需要依赖键盘和鼠标向智能体输入文字，而是可以使用口语与之交互。这是一种更自然、更具吸引力的 AI 交互方式，在某些场景下尤其有用。

什么是语音智能体？

语音智能体是能够与用户进行自然口语对话的智能体。这些智能体结合了语音识别、自然语言处理、生成式 AI 和文本转语音技术，以创建无缝、自然的对话。

它们适用于多种用例，包括：

• 客户支持
• 个人助理
• 免提界面
• 辅导与培训

语音智能体如何工作？

从高层次来看，每个语音智能体都需要处理三个任务：

1. 聆听 - 捕获音频并将其转录为文本
2. 思考 - 解释意图、推理、规划
3. 说话 - 生成音频并流式传输回用户

区别在于这些步骤如何排序和耦合。在实践中，生产级智能体遵循以下两种主要架构之一：

1. STT > 智能体 > TTS 架构（"三明治"架构）

三明治架构由三个独立的组件组成：语音转文本（STT）、基于文本的 LangChain 智能体和文本转语音（TTS）。

flowchart LR
    A[用户音频] --> B[语音转文本]
    B --> C[LangChain 智能体]
    C --> D[文本转语音]
    D --> E[音频输出]

优点：

• 完全控制每个组件（可根据需要更换 STT/TTS 提供商）
• 能够利用现代文本模态模型的最新能力
• 行为透明，组件间边界清晰

缺点：

• 需要协调多个服务
• 管理流水线增加了复杂性
• 从语音到文本的转换会丢失信息（例如，语调、情感）

2. 语音到语音架构（S2S）

语音到语音架构使用多模态模型，该模型原生处理音频输入并生成音频输出。

flowchart LR
    A[用户音频] --> B[多模态模型]
    B --> C[音频输出]

优点：

• 架构更简单，组件更少
• 对于简单交互，通常延迟更低
• 直接音频处理能捕捉语音的语调和其他细微差别

缺点：

• 模型选择有限，供应商锁定的风险更大
• 功能可能落后于文本模态模型
• 音频处理过程透明度较低
• 可控制性和自定义选项减少

本指南演示了三明治架构，以在性能、可控性和现代模型能力访问之间取得平衡。使用某些 STT 和 TTS 提供商时，三明治架构可以实现低于 700 毫秒的延迟，同时保持对模块化组件的控制。

演示应用概述

我们将逐步构建一个使用三明治架构的基于语音的智能体。该智能体将管理一家三明治店的订单。该应用将演示三明治架构的所有三个组件，使用 AssemblyAI 进行 STT，使用 Cartesia 进行 TTS（尽管可以为大多数提供商构建适配器）。

一个端到端的参考应用可在 voice-sandwich-demo 仓库中找到。我们将在此处逐步讲解该应用。

该演示使用 WebSockets 实现浏览器和服务器之间的实时双向通信。相同的架构可以适配到其他传输方式，如电话系统（Twilio, Vonage）或 WebRTC 连接。

架构

该演示实现了一个流式处理流水线，其中每个阶段异步处理数据：

客户端（浏览器）

• 捕获麦克风音频并将其编码为 PCM
• 建立到后端服务器的 WebSocket 连接
• 实时将音频块流式传输到服务器
• 接收并播放合成的语音音频

服务器（Python）

• 接受来自客户端的 WebSocket 连接
• 协调三步流水线：
  • 语音转文本（STT）：将音频转发给 STT 提供商（例如 AssemblyAI），接收转录事件
  • 智能体：使用 LangChain 智能体处理转录文本，流式传输响应令牌
  • 文本转语音（TTS）：将智能体响应发送给 TTS 提供商（例如 Cartesia），接收音频块
• 将合成的音频返回给客户端进行播放

流水线使用异步生成器来实现每个阶段的流式处理。这使得下游组件可以在上游阶段完成之前开始处理，从而最小化端到端延迟。

设置

有关详细的安装说明和设置，请参阅仓库 README。

1. 语音转文本

STT 阶段将传入的音频流转换为文本转录。该实现使用生产者-消费者模式来并发处理音频流和转录接收。

关键概念

生产者-消费者模式：音频块被并发地发送到 STT 服务，同时接收转录事件。这使得转录可以在所有音频到达之前开始。

事件类型：

• stt_chunk：STT 服务处理音频时提供的部分转录文本
• stt_output：触发智能体处理的最终格式化转录文本

WebSocket 连接：维护与 AssemblyAI 实时 STT API 的持久连接，配置为 16kHz PCM 音频并启用自动轮次格式化。

实现

from typing import AsyncIterator
import asyncio
from assemblyai_stt import AssemblyAISTT
from events import VoiceAgentEvent

async def stt_stream(
    audio_stream: AsyncIterator[bytes],
) -> AsyncIterator[VoiceAgentEvent]:
    """
    转换流：音频（字节）→ 语音事件（VoiceAgentEvent）

    使用生产者-消费者模式，其中：
    - 生产者：读取音频块并将其发送到 AssemblyAI
    - 消费者：从 AssemblyAI 接收转录事件
    """
    stt = AssemblyAISTT(sample_rate=16000)

    async def send_audio():
        """后台任务，将音频块泵送到 AssemblyAI。"""
        try:
            async for audio_chunk in audio_stream:
                await stt.send_audio(audio_chunk)
        finally:
            # 音频流结束时发出完成信号
            await stt.close()

    # 在后台启动音频发送
    send_task = asyncio.create_task(send_audio())

    try:
        # 接收并产出到达的转录事件
        async for event in stt.receive_events():
            yield event
    finally:
        # 清理
        with contextlib.suppress(asyncio.CancelledError):
            send_task.cancel()
            await send_task
        await stt.close()

该应用实现了一个 AssemblyAI 客户端来管理 WebSocket 连接和消息解析。实现见下文；可以为其他 STT 提供商构建类似的适配器。

AssemblyAI 客户端

class AssemblyAISTT:
    def __init__(self, api_key: str | None = None, sample_rate: int = 16000):
        self.api_key = api_key or os.getenv("ASSEMBLYAI_API_KEY")
        self.sample_rate = sample_rate
        self._ws: WebSocketClientProtocol | None = None

    async def send_audio(self, audio_chunk: bytes) -> None:
        """将 PCM 音频字节发送到 AssemblyAI。"""
        ws = await self._ensure_connection()
        await ws.send(audio_chunk)

    async def receive_events(self) -> AsyncIterator[STTEvent]:
        """产出从 AssemblyAI 到达的 STT 事件。"""
        async for raw_message in self._ws:
            message = json.loads(raw_message)

            if message["type"] == "Turn":
                # 最终格式化转录文本
                if message.get("turn_is_formatted"):
                    yield STTOutputEvent.create(message["transcript"])
                # 部分转录文本
                else:
                    yield STTChunkEvent.create(message["transcript"])

    async def _ensure_connection(self) -> WebSocketClientProtocol:
        """如果尚未连接，则建立 WebSocket 连接。"""
        if self._ws is None:
            url = f"wss://streaming.assemblyai.com/v3/ws?sample_rate={self.sample_rate}&format_turns=true"
            self._ws = await websockets.connect(
                url,
                additional_headers={"Authorization": self.api_key}
            )
        return self._ws

2. LangChain 智能体

智能体阶段通过 LangChain 智能体处理文本转录，并流式传输响应令牌。在本例中，我们流式传输智能体生成的所有文本内容块。

关键概念

流式响应：智能体使用 stream_mode="messages" 在生成时发出响应令牌，而不是等待完整响应。这使得 TTS 阶段可以立即开始合成。

对话记忆：检查点器使用唯一的线程 ID 在多个轮次中维护对话状态。这使得智能体能够引用对话中先前的交流。

实现

from uuid import uuid4
from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver

# 定义智能体工具
def add_to_order(item: str, quantity: int) -> str:
    """将商品添加到客户的三明治订单中。"""
    return f"Added {quantity} x {item} to the order."

def confirm_order(order_summary: str) -> str:
    """与客户确认最终订单。"""
    return f"Order confirmed: {order_summary}. Sending to kitchen."

# 创建带有工具和记忆的智能体
agent = create_agent(
    model="anthropic:claude-haiku-4-5",  # 选择你的模型
    tools=[add_to_order, confirm_order],
    system_prompt="""You are a helpful sandwich shop assistant.
    Your goal is to take the user's order. Be concise and friendly.
    Do NOT use emojis, special characters, or markdown.
    Your responses will be read by a text-to-speech engine.""",
    checkpointer=InMemorySaver(),
)

async def agent_stream(
    event_stream: AsyncIterator[VoiceAgentEvent],
) -> AsyncIterator[VoiceAgentEvent]:
    """
    转换流：语音事件 → 语音事件（包含智能体响应）

    传递所有上游事件，并在处理 STT 转录文本时添加 agent_chunk 事件。
    """
    # 为对话记忆生成唯一的线程 ID
    thread_id = str(uuid4())

    async for event in event_stream:
        # 传递所有上游事件
        yield event

        # 通过智能体处理最终转录文本
        if event.type == "stt_output":
            # 流式传输带有对话上下文的智能体响应
            stream = agent.astream(
                {"messages": [HumanMessage(content=event.transcript)]},
                {"configurable": {"thread_id": thread_id}},
                stream_mode="messages",
            )

            # 产出到达的智能体响应块
            async for message, _ in stream:
                if message.text:
                    yield AgentChunkEvent.create(message.text)

3. 文本转语音

TTS 阶段将智能体响应文本合成为音频，并将其流式传输回客户端。与 STT 阶段类似，它使用生产者-消费者模式来处理并发的文本发送和音频接收。

关键概念

并发处理：该实现合并了两个异步流：

• 上游处理：传递所有事件并将智能体文本块发送给 TTS 提供商
• 音频接收：从 TTS 提供商接收合成的音频块

流式 TTS：一些提供商（例如 Cartesia）在收到文本后立即开始合成音频，使得音频播放可以在智能体完成生成完整响应之前开始。

事件透传：所有上游事件都保持不变地流过，允许客户端或其他观察者跟踪整个流水线的状态。

实现

:::python

from cartesia_tts import CartesiaTTS
from utils import merge_async_iters

async def tts_stream(
event_stream: AsyncIterator[VoiceAgentEvent],
) -> AsyncIterator[VoiceAgentEvent]:
"""
转换流：语音事件 → 语音事件（包含音频）

合并两个并发流：
    1. process_upstream(): 传递事件并将文本发送到 Cartesia
    2. tts.receive_events(): 产出来自 Cartesia 的音频块
"""