尚拙
Search K

Appearance

使用 LangChain 构建语音智能体 (Build a voice agent) ​

概述 (Overview) ​

聊天界面曾长期主导人机交互方式，但多模态 AI 的最新突破正在开启激动人心的崭新可能。高质量生成模型与极具表现力的文本转语音（TTS）系统让开发者能够构建不再像生硬工具、而更像真实对话伙伴的智能体。

语音智能体就是其中的代表。你无需依赖键盘输入文字，而是可以直接用自然语言与智能体交谈。这带来了更自然、更具沉浸感的交互体验，在许多实际业务场景中尤为实用。

什么是语音智能体？ (What are voice agents?) ​

语音智能体是能够与用户开展自然口语交谈的智能体 (Agents)。它们融合了语音识别（ASR/STT）、自然语言理解、生成式大模型以及语音合成（TTS）技术，实现流畅丝滑的交互体验。

它们非常适用于以下典型场景：

  • 智能客户服务
  • 个人生活与工作助理
  • 驾车或工业操作中的免提（Hands-free）交互
  • 口语陪练与教学培训

语音智能体的工作原理 (How do voice agents work?) ​

从架构顶层来看，所有语音智能体都需要处理三大任务：

  1. 倾听 (Listen)：采集音频并转录为文本
  2. 思考 (Think)：解析意图、执行逻辑推理与规划决策
  3. 表达 (Speak)：合成音频并将其流式推送回客户端播放

各架构方案的核心差异在于这三个步骤的串联与耦合方式。在生产实践中，主流实现通常遵循以下两种架构之一：

1. 三明治架构：STT > Agent > TTS (The "Sandwich") ​

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

mermaid
flowchart LR
    A[User Audio] --> B[Speech-to-Text]
    B --> C[LangChain Agent]
    C --> D[Text-to-Speech]
    D --> E[Audio Output]

    classDef trigger fill:#F6FFDB,stroke:#6E8900,stroke-width:2px,color:#2E3900
    classDef process fill:#E5F4FF,stroke:#006DDD,stroke-width:2px,color:#030710

    class A,E trigger
    class B,C,D process

优势 (Pros)：

  • 对各个组件拥有完全掌控权（可根据需要随时平替 STT 或 TTS 供应商）
  • 能够直接使用顶级文本大语言模型的最新前沿能力
  • 组件之间边界清晰透明，便于调试和日志追踪

劣势 (Cons)：

  • 需要协调多个异步服务
  • 流水线状态管理具有一定复杂度
  • 语音转文本阶段会丢失语音语调、情绪等副语言学信息

2. 端到端原生语音架构 (Speech-to-Speech, S2S) ​

端到端语音架构使用原生多模态模型直接接收音频输入，并原生输出音频流。

mermaid
flowchart LR
    A[User Audio] --> B[Multimodal Model]
    B --> C[Audio Output]

    classDef trigger fill:#F6FFDB,stroke:#6E8900,stroke-width:2px,color:#2E3900
    classDef process fill:#E5F4FF,stroke:#006DDD,stroke-width:2px,color:#030710

    class A,C trigger
    class B process

优势 (Pros)：

  • 架构更极简，流转环节更少
  • 简单交互下的端到端延迟通常更低
  • 原生音频处理保留了语气与微妙的情感起伏

劣势 (Cons)：

  • 可选模型极少，存在高度厂商锁定风险
  • 复杂逻辑推理与工具调用能力往往滞后于顶级纯文本模型
  • 音频内部处理过程缺乏透明度
  • 业务受控度与精细化定制能力较弱

本指南重点演示三明治架构（Sandwich architecture），在性能、系统可控性与模型最强推理能力之间取得最佳平衡。在优秀的 STT 与 TTS 供应商配合下，三明治架构同样可以达到低于 700ms 的极低端到端延迟，同时保持各个模块高度可替换。

Demo Application overview ​

We'll walk through building a voice-based agent using the sandwich architecture. The agent will manage orders for a sandwich shop. The application will demonstrate all three components of the sandwich architecture, using AssemblyAI for STT and Cartesia for TTS (although adapters can be built for most providers).

An end-to-end reference application is available in the voice-sandwich-demo repository. We will walk through that application here.

The demo uses WebSockets for real-time bidirectional communication between the browser and server. The same architecture can be adapted for other transports like telephony systems (Twilio, Vonage) or WebRTC connections.

Architecture ​

The demo implements a streaming pipeline where each stage processes data asynchronously:

Client (Browser)

  • Captures microphone audio and encodes it as PCM
  • Establishes WebSocket connection to the backend server
  • Streams audio chunks to the server in real-time
  • Receives and plays back synthesized speech audio

Server (Node.js)

  • Accepts WebSocket connections from clients

  • Orchestrates the three-step pipeline:

    • Speech-to-text (STT): Forwards audio to the STT provider (e.g., AssemblyAI), receives transcript events
    • Agent: Processes transcripts with LangChain agent, streams response tokens
    • Text-to-speech (TTS): Sends agent responses to the TTS provider (e.g., Cartesia), receives audio chunks

  • Returns synthesized audio to the client for playback

The pipeline uses async iterators to enable streaming at each stage. This allows downstream components to begin processing before upstream stages complete, minimizing end-to-end latency.

环境搭建 (Setup) ​

有关详细安装和环境配置，请参阅示例仓库 README。

1. 语音转文本 (Speech-to-text) ​

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

Key concepts ​

Producer-Consumer Pattern: Audio chunks are sent to the STT service concurrently with receiving transcript events. This allows transcription to begin before all audio has arrived.

Event Types:

  • stt_chunk: Partial transcripts provided as the STT service processes audio
  • stt_output: Final, formatted transcripts that trigger agent processing

WebSocket Connection: Maintains a persistent connection to AssemblyAI's real-time STT API, configured for 16kHz PCM audio with automatic turn formatting.

Implementation ​

typescript
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>();

  // Producer: pump audio chunks to AssemblyAI
  const producer = (async () => {
    try {
      for await (const audioChunk of audioStream) {
        await stt.sendAudio(audioChunk);
      }
    } finally {
      await stt.close();
    }
  })();

  // Consumer: receive transcription events
  const consumer = (async () => {
    for await (const event of stt.receiveEvents()) {
      passthrough.push(event);
    }
  })();

  try {
    // Yield events as they arrive
    yield* passthrough;
  } finally {
    // Wait for producer and consumer to complete
    await Promise.all([producer, consumer]);
  }
}

The application implements an AssemblyAI client to manage the WebSocket connection and message parsing. See below for implementations; similar adapters can be constructed for other STT providers.

AssemblyAI Client

typescript
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 agent) ​

智能体阶段通过 LangChain 智能体运行时处理转录文本，并以流式方式输出回答 Token。在此阶段，我们流式提取智能体生成的全部文本内容块。

Key concepts ​

Streaming Responses: The agent uses stream_events(version="v3") with stream.messages to emit response tokens as they are generated, rather than waiting for the complete response. This enables the TTS stage to begin synthesis immediately.

Conversation Memory: A checkpointer maintains conversation state across turns using a unique thread ID. This allows the agent to reference previous exchanges in the conversation.

Implementation ​

typescript
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 { uuid7 } from "langsmith";

// Define agent tools
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"),
    }),
  }
);

// Create agent with tools and memory
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> {
  // Generate unique thread ID for conversation memory
  const threadId = uuid7();

  for await (const event of eventStream) {
    // Pass through all upstream events
    yield event;

    // Process final transcripts through the agent
    if (event.type === "stt_output") {
      const stream = await agent.streamEvents(
        { messages: [new HumanMessage(event.transcript)] },
        {
          configurable: { thread_id: threadId },
          version: "v3",
        }
      );

      // Yield agent response chunks as they arrive
      for await (const message of stream.messages) {
        for await (const token of message.text) {
          yield { type: "agent_chunk", text: token, ts: Date.now() };
        }
      }
    }
  }
}

3. 文本转语音 (Text-to-speech) ​

TTS 阶段将智能体生成的文本实时合成为音频流，并推回给客户端。与 STT 阶段类似，它使用生产者-消费者模式并发处理文本发送与音频块接收。

Key concepts ​

Concurrent Processing: The implementation merges two async streams:

  • Upstream processing: Passes through all events and sends agent text chunks to the TTS provider
  • Audio reception: Receives synthesized audio chunks from the TTS provider

Streaming TTS: Some providers (such as Cartesia) begin synthesizing audio as soon as it receives text, enabling audio playback to start before the agent finishes generating its complete response.

Event Passthrough: All upstream events flow through unchanged, allowing the client or other observers to track the full pipeline state.

Implementation ​

typescript
import { CartesiaTTS } from "./cartesia";

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

  // Producer: read upstream events and send text to 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();
    }
  })();

  // Consumer: receive audio from Cartesia
  const consumer = (async () => {
    for await (const event of tts.receiveEvents()) {
      passthrough.push(event);
    }
  })();

  try {
    // Yield events from both producer and consumer
    yield* passthrough;
  } finally {
    await Promise.all([producer, consumer]);
  }
}

The application implements a Cartesia client to manage the WebSocket connection and audio streaming. See below for implementations; similar adapters can be constructed for other TTS providers.

Cartesia Client

typescript
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 ​

Many of the applications you build with LangChain will contain multiple steps with multiple invocations of LLM calls. As these applications get more complex, it becomes crucial to be able to inspect what exactly is going on inside your chain or agent. The best way to do this is with LangSmith.

After you sign up at the link above, make sure to set your environment variables to start logging traces:

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

完整串联 (Putting it all together) ​

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

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

  // Chain the three stages
  const transcriptEventStream = sttStream(inputStream);
  const agentEventStream = agentStream(transcriptEventStream);
  const outputEventStream = ttsStream(agentEventStream);

  // Process pipeline and send TTS audio to client
  const flushPromise = (async () => {
    for await (const event of outputEventStream) {
      if (event.type === "tts_chunk") {
        currentSocket?.send(event.audio);
      }
    }
  })();

  return {
    onMessage(event) {
      // Push incoming audio into pipeline
      const data = event.data;
      if (Buffer.isBuffer(data)) {
        inputStream.push(new Uint8Array(data));
      }
    },
    async onClose() {
      inputStream.cancel();
      await flushPromise;
    },
  };
}));

Each stage processes events independently and concurrently: audio transcription begins as soon as audio arrives, the agent starts reasoning as soon as a transcript is available, and speech synthesis begins as soon as agent text is generated. This architecture can achieve sub-700ms latency to support natural conversation.

For more on building agents with LangChain, see the Agents guide.