LiveKit Plugin Architecture — Swap STT/LLM/TTS Providers

简介

LiveKit 的 plugin 架构把你的语音 agent 跟具体 STT、LLM、TTS 提供商解耦 —— 每个 plugin 实现同一接口，所以 Deepgram → AssemblyAI → Groq Whisper 切换就一行代码。并行 A/B 测多家、搭 fallback 链、按 room 路由到不同栈。适合优化语音 agent 质量和成本、避免厂商锁定、合规多供应商部署。兼容所有官方 LiveKit plugin（openai / anthropic / deepgram / assemblyai / cartesia / elevenlabs / silero / groq）加社区 plugin。装机时间 5 分钟。

一行换提供商

# 默认栈
assistant = agents.VoicePipelineAgent(
    stt=deepgram.STT(model="nova-3"),
    llm=openai.LLM(model="gpt-4o-mini"),
    tts=cartesia.TTS(voice="alloy"),
)

# STT 换成 AssemblyAI
assistant.stt = assemblyai.STT(model="universal-2")

# LLM 换成 Anthropic
assistant.llm = anthropic.LLM(model="claude-3-5-sonnet-20241022")

# TTS 换成 ElevenLabs
assistant.tts = elevenlabs.TTS(voice="Adam", model="eleven_turbo_v2_5")

Fallback 链

from livekit.agents.tts import FallbackAdapter

primary_tts = cartesia.TTS(voice="alloy")
backup_tts  = elevenlabs.TTS(voice="Adam")
emergency   = openai.TTS(voice="alloy")

tts = FallbackAdapter([primary_tts, backup_tts, emergency], timeout=2.0)
# 主超时/错误就降级到备份；备份错误再降级到应急。

生产 A/B 测

import random

async def entrypoint(ctx: JobContext):
    variant = "a" if random.random() < 0.5 else "b"
    llm = (openai.LLM(model="gpt-4o-mini") if variant == "a"
           else anthropic.LLM(model="claude-3-5-haiku-20241022"))

    assistant = agents.VoicePipelineAgent(stt=..., llm=llm, tts=...)
    assistant.start(ctx.room)

    # 记录变体供离线分析
    ctx.log.info("agent_started", variant=variant, room=ctx.room.name)

提供商优势 cheat sheet

FAQ

Q: 通话中途切换有效吗？ A: 有效 —— start 后可以重新赋值 .stt / .llm / .tts。在飞的音频在旧 provider 上跑完，新发言路由到新 provider。适合把高价值呼叫者路由到更聪明的 LLM。

Q: 怎么写自定义 plugin？ A: 继承 livekit.agents.stt.STT / llm.LLM / tts.TTS 实现流式方法。大多数社区 plugin <300 行。看 livekit-plugins-deepgram 学接口最快。

Q: fallback 触发时延迟如何？ A: FallbackAdapter 在主 provider 上探测 timeout 秒。timeout=2.0 时主失败给备份让出最多 2 秒。要更紧 SLO 用 timeout=0.5 —— 误报上升但尾部延迟降。