使用 LangChain 构建语音智能体

概述

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

语音智能体就是其中之一。你可以用语音与智能体互动，而不是依靠键盘和鼠标输入。这是一种更自然和有吸引力的 AI 交互方式，在某些场景下特别有用。

什么是语音智能体？

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

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

• 客户支持
• 个人助理
• 免提界面
• 教练和培训

语音智能体如何工作？

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

1. 听 - 捕获音频并转录
2. 想 - 理解意图、推理、规划
3. 说 - 生成音频并流式传输回用户

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

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

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

优点：

• 完全控制每个组件（按需切换 STT/TTS 提供商）
• 可以使用现代文本模态模型的最新功能
• 组件之间有清晰边界的透明行为

缺点：

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

2. 语音到语音架构 (S2S)

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

优点：

• 更简单的架构，移动部件更少
• 对于简单交互通常延迟更低
• 直接音频处理捕获语气和其他语音细微差别

缺点：

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

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

演示应用概述

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

完整的端到端参考应用可在 voice-sandwich-demo 仓库中获取。我们将在此处介绍该应用。

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

架构

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

客户端（浏览器）

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

服务器（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 音频并自动进行话轮格式化。

实现

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 客户端

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_mode="messages" 在生成时发出响应 Token，而不是等待完整响应。这使 TTS 阶段能够立即开始合成。

对话记忆：检查点器使用唯一的线程 ID 跨对话轮次维护对话状态。这允许智能体在对话中引用之前的交流。

实现

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";
import { v7 as uuid7 } from "uuid";

// 定义智能体工具
const addToOrder = tool(
  async ({ item, quantity }) => {
    return `Added ${quantity} x ${item} to the order.`;
  },
  {
    name: "add_to_order",
    description: "将商品添加到客户的三明治订单中。",
    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: "与客户确认最终订单。",
    schema: z.object({
      orderSummary: z.string().describe("订单摘要"),
    }),
  }
);

// 创建带有工具和记忆的智能体
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 = uuidv4();

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

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

      // 智能体响应块到达时逐个输出
      for await (const [message] of stream) {
        yield { type: "agent_chunk", text: message.text, ts: Date.now() };
      }
    }
  }
}

3. 文本转语音

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

关键概念

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

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

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

事件透传：所有上游事件不变地流过，允许客户端或其他观察者跟踪完整的管道状态。

实现

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 客户端

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="..."

完整整合

完整的管道将三个阶段链接在一起：

// 使用 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 构建智能体的更多信息，请参阅智能体指南。

---

将这些文档连接到 Claude、VSCode 等工具，通过 MCP 获取实时解答。

在 GitHub 上编辑此页面 或 提交问题。