概述

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

语音智能体就是其中一例。你无需依赖键盘和鼠标向智能体输入文本，而是可以用说话的方式与它交互。这可以成为一种更自然、更投入的 AI 交互方式，并且在某些场景下尤其有用。

什么是语音智能体？

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

它们适用于多种使用场景，包括：

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

语音智能体如何工作？

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

1. 倾听（Listen） ——捕获音频并转写
2. 思考（Think） ——解释意图、推理、规划
3. 说话（Speak） ——生成音频并流式传输回用户

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

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

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

优点：

• 完全掌控每个组件（可按需更换 STT/TTS 提供商）
• 可访问现代文本模态模型的最新能力
• 组件之间边界清晰，行为透明

缺点：

• 需要编排多个服务
• 管理流水线时额外增加复杂性
• 语音转文本的转换会丢失信息（例如语调、情感）

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

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

优点：

• 架构更简单，动件更少
• 简单交互的延迟通常更低
• 直接处理音频可捕获语调等语音细微差别

缺点：

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

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

演示应用概述

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

在 voice-sandwich-demo 仓库中提供了一个端到端的参考应用。我们将在这里逐步介绍该应用。

该演示使用 WebSockets 在浏览器和服务器之间进行实时双向通信。同样的架构也可以适用于其他传输方式，例如电话系统（Twilio、Vonage）或 WebRTC 连接。

架构

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

客户端（浏览器）

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

服务器（Python）服务器（Node.js）

• 接受来自客户端的 WebSocket 连接

• 编排三步流水线：

  • 语音转文本（STT）：将音频转发到 STT 提供商（例如 AssemblyAI），接收转写事件
  • 智能体：使用 LangChain 智能体处理转写内容，流式输出响应 token
  • 文本转语音（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]:
    """
    Transform stream: Audio (Bytes) → Voice Events (VoiceAgentEvent)

    Uses a producer-consumer pattern where:
    - Producer: Reads audio chunks and sends them to AssemblyAI
    - Consumer: Receives transcription events from AssemblyAI
    """
    stt = AssemblyAISTT(sample_rate=16000)

    async def send_audio():
        """Background task that pumps audio chunks to 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()

import { AssemblyAISTT } from "./assemblyai";
import type { VoiceAgentEvent } from "./types";

async function* sttStream(
  audioStream: AsyncIterable<Uint8Array>
): AsyncGenerator<VoiceAgentEvent> {
  const stt = new AssemblyAISTT({ sampleRate: 16000 });
  const passthrough = writableIterator<VoiceAgentEvent>();

  // 生产者：将音频块推送到 AssemblyAI
  const producer = (async () => {
    try {
      for await (const audioChunk of audioStream) {
        await stt.sendAudio(audioChunk);
      }
    } finally {
      await stt.close();
    }
  })();

  // 消费者：接收转写事件
  const consumer = (async () => {
    for await (const event of stt.receiveEvents()) {
      passthrough.push(event);
    }
  })();

  try {
    // 在事件到达时逐个产出
    yield* passthrough;
  } finally {
    // 等待生产者和消费者完成
    await Promise.all([producer, consumer]);
  }
}

该应用实现了一个 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:
        """Send PCM audio bytes to AssemblyAI."""
        ws = await self._ensure_connection()
        await ws.send(audio_chunk)

    async def receive_events(self) -> AsyncIterator[STTEvent]:
        """Yield STT events as they arrive from AssemblyAI."""
        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:
        """Establish WebSocket connection if not already connected."""
        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

export class AssemblyAISTT {
  protected _bufferIterator = writableIterator<VoiceAgentEvent.STTEvent>();
  protected _connectionPromise: Promise<WebSocket> | null = null;

  async sendAudio(buffer: Uint8Array): Promise<void> {
    const conn = await this._connection;
    conn.send(buffer);
  }

  async *receiveEvents(): AsyncGenerator<VoiceAgentEvent.STTEvent> {
    yield* this._bufferIterator;
  }

  protected get _connection(): Promise<WebSocket> {
    if (this._connectionPromise) return this._connectionPromise;

    this._connectionPromise = new Promise((resolve, reject) => {
      const params = new URLSearchParams({
        sample_rate: this.sampleRate.toString(),
        format_turns: "true",
      });
      const url = `wss://streaming.assemblyai.com/v3/ws?${params}`;
      const ws = new WebSocket(url, {
        headers: { Authorization: this.apiKey },
      });

      ws.on("open", () => resolve(ws));

      ws.on("message", (data) => {
        const message = JSON.parse(data.toString());
        if (message.type === "Turn") {
          if (message.turn_is_formatted) {
            this._bufferIterator.push({
              type: "stt_output",
              transcript: message.transcript,
              ts: Date.now()
            });
          } else {
            this._bufferIterator.push({
              type: "stt_chunk",
              transcript: message.transcript,
              ts: Date.now()
            });
          }
        }
      });
    });

    return this._connectionPromise;
  }
}

2. LangChain 智能体

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

关键概念

流式输出响应：智能体使用 stream_events(version="v3") 配合 stream.messages，在生成响应 token 时即时发出，而不是等待完整响应。这使得 TTS 阶段可以立即开始合成。

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

实现

from langchain_core.utils.uuid import uuid7
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:
    """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."

# 使用工具和记忆创建智能体
agent = create_agent(
    model="google_genai:gemini-3.6-flash",  # 选择你的模型
    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]:
    """
    Transform stream: Voice Events → Voice Events (with Agent Responses)

    Passes through all upstream events and adds agent_chunk events
    when processing STT transcripts.
    """
    # 为对话记忆生成唯一的线程 ID
    thread_id = str(uuid7())

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

        # 通过智能体处理最终转写内容
        if event.type == "stt_output":
            # 流式输出带对话上下文的智能体响应
            stream = await agent.astream_events(
                {"messages": [HumanMessage(content=event.transcript)]},
                {"configurable": {"thread_id": thread_id}},
                version="v3",
            )

            # 在响应块到达时逐个产出
            async for message in stream.messages:
                async for token in message.text:
                    yield AgentChunkEvent.create(token)

import { createAgent } from "langchain";
import { HumanMessage } from "@langchain/core/messages";
import { MemorySaver } from "@langchain/langgraph";
import { tool } from "@langchain/core/tools";
import { z } from "zod";

// 定义智能体工具
const addToOrder = tool(
  async ({ item, quantity }) => {
    return `Added ${quantity} x ${item} to the order.`;
  },
  {
    name: "add_to_order",
    description: "Add an item to the customer's sandwich order.",
    schema: z.object({
      item: z.string(),
      quantity: z.number(),
    }),
  }
);

const confirmOrder = tool(
  async ({ orderSummary }) => {
    return `Order confirmed: ${orderSummary}. Sending to kitchen.`;
  },
  {
    name: "confirm_order",
    description: "Confirm the final order with the customer.",
    schema: z.object({
      orderSummary: z.string().describe("Summary of the order"),
    }),
  }
);

// 使用工具和记忆创建智能体
const agent = createAgent({
  model: "claude-haiku-4-5",
  tools: [addToOrder, confirmOrder],
  checkpointer: new MemorySaver(),
  systemPrompt: `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.`,
});

async function* agentStream(
  eventStream: AsyncIterable<VoiceAgentEvent>
): AsyncGenerator<VoiceAgentEvent> {
  // 为对话记忆生成唯一的线程 ID
  const threadId = crypto.randomUUID();

  for await (const event of eventStream) {
    // 透传所有上游事件
    yield event;

    // 通过智能体处理最终转写内容
    if (event.type === "stt_output") {
      const stream = await agent.streamEvents(
        { messages: [new HumanMessage(event.transcript)] },
        {
          configurable: { thread_id: threadId },
          version: "v3",
        }
      );

      // 在响应块到达时逐个产出
      for await (const message of stream.messages) {
        for await (const token of message.text) {
          yield { type: "agent_chunk", text: token, ts: Date.now() };
        }
      }
    }
  }
}

3. 文本转语音

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

关键概念

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

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

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

事件透传：所有上游事件都会原样流过，使客户端或其他观察者能够跟踪完整的流水线状态。

实现

from cartesia_tts import CartesiaTTS
from utils import merge_async_iters

async def tts_stream(
    event_stream: AsyncIterator[VoiceAgentEvent],
) -> AsyncIterator[VoiceAgentEvent]:
    """
    Transform stream: Voice Events → Voice Events (with Audio)

    Merges two concurrent streams:
    1. process_upstream(): passes through events and sends text to Cartesia
    2. tts.receive_events(): yields audio chunks from Cartesia
    """
    tts = CartesiaTTS()

    async def process_upstream() -> AsyncIterator[VoiceAgentEvent]:
        """Process upstream events and send agent text to Cartesia."""
        async for event in event_stream:
            # 透传所有事件
            yield event
            # 将智能体文本发送给 Cartesia 进行合成
            if event.type == "agent_chunk":
                await tts.send_text(event.text)

    try:
        # 合并上游事件与 TTS 音频事件
        # 两个流并发运行
        async for event in merge_async_iters(
            process_upstream(),
            tts.receive_events()
        ):
            yield event
    finally:
        await tts.close()

import { CartesiaTTS } from "./cartesia";

async function* ttsStream(
  eventStream: AsyncIterable<VoiceAgentEvent>
): AsyncGenerator<VoiceAgentEvent> {
  const tts = new CartesiaTTS();
  const passthrough = writableIterator<VoiceAgentEvent>();

  // 生产者：读取上游事件并将文本发送给 Cartesia
  const producer = (async () => {
    try {
      for await (const event of eventStream) {
        passthrough.push(event);
        if (event.type === "agent_chunk") {
          await tts.sendText(event.text);
        }
      }
    } finally {
      await tts.close();
    }
  })();

  // 消费者：从 Cartesia 接收音频
  const consumer = (async () => {
    for await (const event of tts.receiveEvents()) {
      passthrough.push(event);
    }
  })();

  try {
    // 产出生产者和消费者双方的事件
    yield* passthrough;
  } finally {
    await Promise.all([producer, consumer]);
  }
}

该应用实现了一个 Cartesia 客户端来管理 WebSocket 连接和音频流。具体实现见下方；可以为其他 TTS 提供商构建类似的适配器。

Cartesia 客户端

import base64
import json
import websockets

class CartesiaTTS:
    def __init__(
        self,
        api_key: Optional[str] = None,
        voice_id: str = "f6ff7c0c-e396-40a9-a70b-f7607edb6937",
        model_id: str = "sonic-3",
        sample_rate: int = 24000,
        encoding: str = "pcm_s16le",
    ):
        self.api_key = api_key or os.getenv("CARTESIA_API_KEY")
        self.voice_id = voice_id
        self.model_id = model_id
        self.sample_rate = sample_rate
        self.encoding = encoding
        self._ws: WebSocketClientProtocol | None = None

    def _generate_context_id(self) -> str:
        """Generate a valid context_id for Cartesia."""
        timestamp = int(time.time() * 1000)
        counter = self._context_counter
        self._context_counter += 1
        return f"ctx_{timestamp}_{counter}"

    async def send_text(self, text: str | None) -> None:
        """Send text to Cartesia for synthesis."""
        if not text or not text.strip():
            return

        ws = await self._ensure_connection()
        payload = {
            "model_id": self.model_id,
            "transcript": text,
            "voice": {
                "mode": "id",
                "id": self.voice_id,
            },
            "output_format": {
                "container": "raw",
                "encoding": self.encoding,
                "sample_rate": self.sample_rate,
            },
            "language": self.language,
            "context_id": self._generate_context_id(),
        }
        await ws.send(json.dumps(payload))

    async def receive_events(self) -> AsyncIterator[TTSChunkEvent]:
        """Yield audio chunks as they arrive from Cartesia."""
        async for raw_message in self._ws:
            message = json.loads(raw_message)

            # 解码并产出音频块
            if "data" in message and message["data"]:
                audio_chunk = base64.b64decode(message["data"])
                if audio_chunk:
                    yield TTSChunkEvent.create(audio_chunk)

    async def _ensure_connection(self) -> WebSocketClientProtocol:
        """Establish WebSocket connection if not already connected."""
        if self._ws is None:
            url = (
                f"wss://api.cartesia.ai/tts/websocket"
                f"?api_key={self.api_key}&cartesia_version={self.cartesia_version}"
            )
            self._ws = await websockets.connect(url)

        return self._ws

export class CartesiaTTS {
  protected _bufferIterator = writableIterator<VoiceAgentEvent.TTSEvent>();
  protected _connectionPromise: Promise<WebSocket> | null = null;

  async sendText(text: string | null): Promise<void> {
    if (!text || !text.trim()) return;

    const conn = await this._connection;
    const payload = { text, try_trigger_generation: false };
    conn.send(JSON.stringify(payload));
  }

  async *receiveEvents(): AsyncGenerator<VoiceAgentEvent.TTSEvent> {
    yield* this._bufferIterator;
  }

  protected _generateContextId(): string {
    const timestamp = Date.now();
    const counter = this._contextCounter++;
    return `ctx_${timestamp}_${counter}`;
  }

  protected get _connection(): Promise<WebSocket> {
    if (this._connectionPromise) return this._connectionPromise;

    this._connectionPromise = new Promise((resolve, reject) => {
      const params = new URLSearchParams({
        api_key: this.apiKey,
        cartesia_version: this.cartesiaVersion,
      });
      const url = `wss://api.cartesia.ai/tts/websocket?${params.toString()}`;
      const ws = new WebSocket(url);

      ws.on("open", () => {
        resolve(ws);
      });

      ws.on("message", (data: WebSocket.RawData) => {
        const message: CartesiaTTSResponse = JSON.parse(data.toString());
        if (message.data) {
          this._bufferIterator.push({
            type: "tts_chunk",
            audio: message.data,
            ts: Date.now(),
          });
        } else if (message.error) {
          throw new Error(`Cartesia error: ${message.error}`);
        }
      });
    });

    return this._connectionPromise;
  }
}

LangSmith

你使用 LangChain 构建的许多应用都包含多个步骤以及多次 LLM 调用。随着这些应用变得越来越复杂，能够检查你的链或智能体内部究竟发生了什么就变得至关重要。做到这一点的最佳方式就是使用 LangSmith。

在上面的链接处注册后，请务必设置你的环境变量以开始记录追踪：

export LANGSMITH_TRACING="true"
export LANGSMITH_API_KEY="..."

或者在 Python 中设置：

import getpass
import os

os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = getpass.getpass()

综合应用

完整的流水线将三个阶段串联在一起：

from langchain_core.runnables import RunnableGenerator

pipeline = (
    RunnableGenerator(stt_stream)      # 音频 → STT 事件
    | RunnableGenerator(agent_stream)  # STT 事件 → 智能体事件
    | RunnableGenerator(tts_stream)    # 智能体事件 → TTS 音频
)

# 在 WebSocket 端点中使用
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

    async def websocket_audio_stream():
        """Yield audio bytes from WebSocket."""
        while True:
            data = await websocket.receive_bytes()
            yield data

    # 通过流水线转换音频
    output_stream = pipeline.atransform(websocket_audio_stream())

    # 将 TTS 音频发送回客户端
    async for event in output_stream:
        if event.type == "tts_chunk":
            await websocket.send_bytes(event.audio)

我们使用 RunnableGenerators 组合流水线的每一步。这是 LangChain 在内部用来管理跨组件流式传输的抽象。

// 使用 https://hono.dev/
app.get("/ws", upgradeWebSocket(async () => {
  const inputStream = writableIterator<Uint8Array>();

  // 串联三个阶段
  const transcriptEventStream = sttStream(inputStream);
  const agentEventStream = agentStream(transcriptEventStream);
  const outputEventStream = ttsStream(agentEventStream);

  // 处理流水线并将 TTS 音频发送到客户端
  const flushPromise = (async () => {
    for await (const event of outputEventStream) {
      if (event.type === "tts_chunk") {
        currentSocket?.send(event.audio);
      }
    }
  })();

  return {
    onMessage(event) {
      // 将传入的音频推入流水线
      const data = event.data;
      if (Buffer.isBuffer(data)) {
        inputStream.push(new Uint8Array(data));
      }
    },
    async onClose() {
      inputStream.cancel();
      await flushPromise;
    },
  };
}));

每个阶段都独立且并发地处理事件：音频一到达就开始转写，转写内容一可用智能体就开始推理，智能体文本一生成就开始语音合成。这种架构可以实现低于 700ms 的延迟，以支持自然对话。

有关使用 LangChain 构建智能体的更多信息，请参阅智能体指南。