Cartesia Streaming WebSocket — Full-Duplex Voice Agent TTS

简介

Cartesia 流式 WebSocket TTS 让你把 LLM 流式输出的文本直接管道给 Cartesia，音频边到边播。不等 LLM 结束、不等 Cartesia 结束 —— 两者重叠。这就是生产语音 agent 打到 <1.5 秒往返的方式。适合 LiveKit Agents、Vapi、自定义语音 agent 流水线、任何 TTFA + LLM TTFB 叠加的场景。兼容 cartesia Python/JS SDK + 任何异步 LLM 流源。装机时间 15 分钟。

流水线 LLM 流式 → Cartesia 流式 → 扬声器

import asyncio
from openai import AsyncOpenAI
from cartesia import AsyncCartesia
import sounddevice as sd
import numpy as np

oai = AsyncOpenAI()
cartesia = AsyncCartesia(api_key=os.environ["CARTESIA_API_KEY"])

async def voice_response(user_text: str):
    # 先开 Cartesia WebSocket，LLM 第一块到时它已准备好
    ws = await cartesia.tts.websocket()

    async def feed_llm_to_tts():
        stream = await oai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": user_text}],
            stream=True,
        )
        async for chunk in stream:
            text = chunk.choices[0].delta.content
            if text:
                await ws.send_text(text)
        await ws.flush()   # 告诉 Cartesia：没文本了

    async def play_audio():
        async for chunk in ws.receive():
            audio = np.frombuffer(chunk.audio, dtype=np.int16)
            sd.play(audio, 22_050, blocking=False)

    await asyncio.gather(feed_llm_to_tts(), play_audio())
    await ws.close()

asyncio.run(voice_response("Tell me about state space models in one sentence."))

为啥流水线重要

干净处理打断

async def voice_response_with_barge_in(user_text: str, interrupt_event: asyncio.Event):
    ws = await cartesia.tts.websocket()

    async def stream_audio():
        async for chunk in ws.receive():
            if interrupt_event.is_set():
                await ws.cancel()    # 让 Cartesia 停止生成
                return
            sd.play(np.frombuffer(chunk.audio, dtype=np.int16), 22_050)

    # ...同上把 LLM token 喂给 ws...

输出格式选择

output_format={
    "container": "raw",            # 直接播用 raw 字节；存档用 mp3
    "encoding": "pcm_s16le",       # PCM 16 位小端
    "sample_rate": 22_050,         # 电话 16k、web 22k、高保真 44k
}

FAQ

Q: 为啥不分块 LLM 输出每块调一次 /tts.bytes？ A: 每次 HTTP 调用的连接开销主导 —— 每块约 50ms TCP/TLS 握手即使缓存。WebSocket 让一条连接保活，无开销流式发自然 <1 秒块。

Q: 词/句边界怎么办？ A: Cartesia 优雅处理部分输入 —— 内部等安全边界点（词中 vs 句末）。也可以 flush(continue=True) 强制分段做显式边界。

Q: 怎么检测发言结束？ A: 你 flush() 后 Cartesia 发一个 is_final=True 的最终 WebSocket 消息。用它清音频队列，向 VAD 信号 agent 说完了。