用 LiveKit Agents 搭可打断的语音智能体（STT→LLM→TTS 流水线）

用一个干净的虚拟环境，避免和别的项目依赖打架。下面按"可换厂商"的经典流水线装：OpenAI 做 LLM、Deepgram 做 STT、Cartesia 做 TTS、Silero 做 VAD、并带上降噪插件。

python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install "livekit-agents[openai,deepgram,cartesia,silero,noise-cancellation]"

LiveKit Agents 需要一个实时房间（LiveKit Cloud 或自托管 Server），再各选一家 STT / LLM / TTS 供应商。把密钥写进项目根目录的 .env：

LIVEKIT_URL=wss://your-project.livekit.cloud
LIVEKIT_API_KEY=your_livekit_api_key
LIVEKIT_API_SECRET=your_livekit_api_secret
DEEPGRAM_API_KEY=your_deepgram_key
OPENAI_API_KEY=your_openai_key
CARTESIA_API_KEY=your_cartesia_key

新建 agent.py。核心是 AgentSession：把 STT、LLM、TTS 分别挂上去，框架自动处理 VAD、轮次和打断。VoicePipelineAgent 是经典流水线封装，allow_interruptions=True 打开打断。

import asyncio
from dotenv import load_dotenv
from livekit import agents
from livekit.agents import AgentSession, Agent
from livekit.plugins import openai, deepgram, cartesia, silero

load_dotenv()

class Assistant(Agent):
    def __init__(self):
        super().__init__(instructions="你是简洁友好的语音客服，每次回答不超过两句话。")

async def entrypoint(ctx: agents.JobContext):
    await ctx.connect()
    session = AgentSession(
        vad=silero.VAD.load(),
        stt=deepgram.STT(model="nova-3", language="multi"),
        llm=openai.LLM(model="gpt-4.1-mini"),
        tts=cartesia.TTS(),
    )
    await session.start(agent=Assistant(), room=ctx.room)
    await session.generate_reply(instructions="向用户问好，并说明你能帮什么忙。")

if __name__ == "__main__":
    agents.cli.run_app(agents.WorkerOptions(entrypoint_fnc=entrypoint))

打断在 LiveKit 里是"开箱即用"的：VAD 一旦检测到用户又开始说话，会取消正在播放的 TTS、截断 Agent 上句未念完的部分，把新的用户语音路由给 STT。你只需要显式允许，并控制 VAD 灵敏度。

session = AgentSession(
    vad=silero.VAD.load(min_speech_duration=0.2),   # 多短算"开口"
    stt=deepgram.STT(model="nova-3"),
    llm=openai.LLM(model="gpt-4.1-mini"),
    tts=cartesia.TTS(),
    turn_detection="stt",        # 用 STT 做端点检测
)
await session.start(agent=Assistant(), room=ctx.room, allow_interruptions=True)

语音 Agent 也能在对话中途调工具，例如查订单。用 @function_tool 装饰一个异步函数，框架会把函数注册进会话，模型按名字调用，schema 写清楚能显著降低幻觉。

from livekit.agents import function_tool

class Assistant(Agent):
    @function_tool
    async def lookup_order(self, order_id: str) -> str:
        """按订单号查询客户订单状态。"""
        # 这里换成你真实的 CRM 查询
        return f"订单 {order_id}：已发货，预计明天送达。"

不接 LiveKit 房间也能先验证逻辑：console 模式用麦克风本地对话；dev 模式连 LiveKit Cloud 的开发房间，打开 agents-playground 网页即可对话。

python agent.py console     # 本地麦克风，无需 LiveKit 连接
python agent.py dev         # 连 LiveKit Cloud 开发房间
python agent.py start       # 生产 worker 模式