Pipecat 实战：用 Python 打造低延迟语音智能体

原文：https://dev.to/stalwartcoder/build-a-low-latency-voice-agent-in-python-with-pipecat-3plb（作者 @stalwartcoder）

📌 封面图（图，点击查看）

这篇文章就是一份可以照着跑的完整演示。读完之后，你的笔记本上会运行起一个语音智能体：在浏览器里直接和它对话、在它说到一半时打断它，并且看着它在每一轮结束后打印出自己的延迟。

整套技术栈完全可替换，下面是本文实际使用的组件以及选择它们的理由。

[Pipecat](https://docs.pipecat.ai)* 作为编排框架。开源、Python 编写，整个智能体读起来就像一份清单。

Smallest AI Pulse 负责语音转文本，Lightning v3.1 Pro* 负责文本转语音。它们是构建语音智能体最好的模型之一，两者都是 Pipecat 官方插件，任何一个都能用一行代码换掉。

Claude Haiku 4.5* 作为 LLM，因为在当时从本地网络实测的首 token 延迟里，它是最快的。后文会详细展开。

进入正题。

语音智能体不是会说话的 LLM

我们大多数人一开始的心智模型是这样的。

📌 文本智能体就是一个方框：token 进、token 出（图，点击查看）

一个方框，文本进、文本出。你所知道的关于构建 LLM 应用的一切，都在这个框里。

而语音智能体是这样的。

📌 语音智能体是一条由麦克风、VAD、轮次检测、STT、LLM、TTS 和扬声器组成的流水线（图，点击查看）

LLM 依然居中，但现在它的周围环绕着语音活动检测（VAD）、轮次检测、语音转文本（STT）、文本转语音（TTS）以及实时传输层。而且这里还有一只在走的时钟。覆盖十种语言的研究显示，人类对话平均每轮之间只留约 200 毫秒的间隙（Stivers 等人，PNAS 2009）。只要网络跳数还在链路里，我们就压不进 200 毫秒，但每超出的一毫秒都会被用户感觉到。

这一节最想让你带走的一个观点是：语音 AI 的所有难点都藏在箭头里，而不是方框里。挑到好模型固然重要，但模型之间的时序衔接才是真正的工程。

为什么一切都要流式处理

如果每一级都等上一级完全跑完再开始，总耗时就是各级时长的全额相加。

record → transcribe → think → synthesize → play      （慢，每一级都在等）
stream → stream     → stream → stream     → stream   （快，没有环节在等"完成"）

在流式流水线里，语音转文本在你话还没说完时就不断吐出部分转写结果，LLM 逐个流出 token，文本转语音则在拿到回复第一个分块时就开始发声。正因为最后这一点，对 TTS 真正重要的指标是首字节时间（time to first byte），而不是总合成时长。第一段音频一到，用户就已经在听了，剩下的部分边生成边播。

这些 Pipecat 都替你处理好了。你的任务只是把合适的方框按正确的顺序摆好。

环境搭建

你需要 Python 3.11 或更高版本、一个 Smallest AI API key 和一个 Anthropic API key。

mkdir voice-agent && cd voice-agent
python3 -m venv venv
source venv/bin/activate
pip install "pipecat-ai[smallest,anthropic,silero,webrtc,runner]>=1.6.0" python-dotenv

把密钥放进 .env 文件。

SMALLEST_API_KEY=your_smallest_key
ANTHROPIC_API_KEY=your_anthropic_key

extras 不是随手写的：smallest 会拉入 Pulse 和 Lightning 服务，anthropic 拉入 LLM，silero 拉入语音活动检测器，webrtc 加 runner 则提供本地浏览器客户端和开发服务器——前端完全不用自己写。

完整的 Agent

下面是完整的 bot.py，大约 100 行，紧接着我会带你过一遍其中有意思的部分。

import os

from dotenv import load_dotenv
from loguru import logger
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.frames.frames import Frame, LLMRunFrame, MetricsFrame
from pipecat.metrics.metrics import TTFBMetricsData
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.runner.types import RunnerArguments
from pipecat.runner.utils import create_transport
from pipecat.services.anthropic.llm import AnthropicLLMService
from pipecat.services.smallest.stt import SmallestSTTService
from pipecat.services.smallest.tts import SmallestTTSService, SmallestTTSSettings
from pipecat.transports.base_transport import BaseTransport, TransportParams

load_dotenv(override=True)

SYSTEM_PROMPT = """You are Pixel, a friendly voice assistant.
You are speaking out loud, so follow these rules.
- Never use bullet points, markdown, emoji or headings.
- Keep answers to one or two short sentences.
- Say numbers and dates the way a person would say them.
- If you get interrupted, answer the new question naturally."""

class LatencyMeter(FrameProcessor):
    """为每个服务、每一轮打印首字节延迟（time-to-first-byte）。"""

    async def process_frame(self, frame: Frame, direction: FrameDirection):
        await super().process_frame(frame, direction)
        if isinstance(frame, MetricsFrame):
            for d in frame.data:
                if isinstance(d, TTFBMetricsData) and d.value > 0:
                    name = d.processor.split("#")[0].replace("Service", "")
                    logger.info(f"⏱  {name:<14} time-to-first-byte {d.value * 1000:6.0f} ms")
        await self.push_frame(frame, direction)

async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
    stt = SmallestSTTService(api_key=os.getenv("SMALLEST_API_KEY"))
    llm = AnthropicLLMService(
        api_key=os.getenv("ANTHROPIC_API_KEY"),
        settings=AnthropicLLMService.Settings(model="claude-haiku-4-5-20251001"),
    )
    tts = SmallestTTSService(
        api_key=os.getenv("SMALLEST_API_KEY"),
        settings=SmallestTTSSettings(voice="meher"),
    )

    context = LLMContext(
        [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": "Say hello in one short sentence."},
        ]
    )
    ctx = LLMContextAggregatorPair(context)

    pipeline = Pipeline(
        [
            transport.input(),   # 麦克风音频输入，VAD 在这里运行
            stt,                 # 音频转文字，流式处理
            ctx.user(),          # 把用户说的话加入对话历史
            llm,                 # 历史转为流式 token
            tts,                 # token 转为流式音频
            LatencyMeter(),      # 打印延迟预算
            transport.output(),  # 音频输出到扬声器
            ctx.assistant(),     # 记录实际播出的内容
        ]
    )

    task = PipelineTask(pipeline, params=PipelineParams(enable_metrics=True))

    @transport.event_handler("on_client_connected")
    async def on_client_connected(transport, client):
        await task.queue_frames([LLMRunFrame()])  # 先打招呼

    @transport.event_handler("on_client_disconnected")
    async def on_client_disconnected(transport, client):
        await task.cancel()

    await PipelineRunner(handle_sigint=runner_args.handle_sigint).run(task)

async def bot(runner_args: RunnerArguments):
    transport = await create_transport(
        runner_args,
        {
            "webrtc": lambda: TransportParams(
                audio_in_enabled=True,
                audio_out_enabled=True,
                vad_analyzer=SileroVADAnalyzer(),
            ),
        },
    )
    await run_bot(transport, runner_args)

if __name__ == "__main__":
    from pipecat.runner.run import main

    main()

Pipeline 就是那张架构图

从上往下读 Pipeline([...]) 这个列表，它就是前文那张图的字面实现。

音频从 transport 进来，语音识别（STT）把它转成文字，用户聚合器把这些文字加进对话，LLM 流式生成回复，TTS 把回复转成音频，最后由 transport 播放出去。Frame 沿着列表向下流动，每个处理器只关心自己认识的那种 Frame。

我想请你注意的是最后一行 ctx.assistant()。它位于扬声器之后，而不是 LLM 之后。这是刻意为之，也是打断（interruption）能正常工作的原因，后面我会再回到这一点。

系统提示词是为“听”而写的

一个好的文字提示词会产出列表、markdown 和详尽的长回答。可一旦念出声，这就是一段 45 秒的独白，用户在第 4 秒就会打断你。所以语音提示词禁止一切格式化、要求回答保持简短，并要求数字按人说话的方式来念："$1,234.56" 应该念成 "twelve hundred thirty four dollars and fifty six cents"（一千二百三十四美元五十六美分），而不是 "dollar sign one comma two three four"（美元符号、一、逗号、二三四）。

VAD 运行在 transport 上

SileroVADAnalyzer() 是直接传给 transport 的，而不是作为一个单独的步骤添加。VAD 是一个小模型，只回答一个问题：“现在有人在说话吗？”。它像一道闸门，控制着下游的所有处理，也正是它让 agent 能察觉到你在它说话时插话。

延迟测量器

LatencyMeter 是一个很小的自定义处理器。当你设置 enable_metrics=True 时，Pipecat 本来就会为每个服务测量首字节时间（time to first byte）。这个处理器只是在这些指标流经时把它们捕获并打印出来。它是整个文件里最有用的 15 行代码，因为它把“感觉好慢”变成了每个阶段的具体数字。

运行

python bot.py

打开 http://localhost:7860，点击 Connect 并允许麦克风权限，Pixel 就会向你打招呼。

请戴上耳机。 这一条我是认真的。不戴耳机时，智能体会通过笔记本麦克风听到自己的声音，以为有人开口说话，于是停下来倾听——听它自己讲话。

📌 two voice agents pointing at each other（图，点击查看）

生产环境中，浏览器会通过 WebRTC 帮你做好回声消除；而在开发时的笔记本上，解决办法就是戴耳机。

打断它

让 Pixel 给你讲一个机器人学唱歌的故事。等它讲上几秒，直接开口盖过它，问点别的。

它会立刻停下，转而回答新问题。这就是 barge-in（抢话打断），刚刚发生了两件事：

1. VAD 听到了你的声音，Pipecat 取消了正在执行中的 LLM 和 TTS 任务，并停止播放。
2. 对话历史里只保留实际播放给你听的那部分故事。

第二点正是语音智能体里一个隐蔽的 bug。如果历史记录写着智能体说完了整句话，智能体就会相信一些用户从未听到的内容。设想它说“我可以在周二下午 3 点帮你预订并发送确认”，而你在“周二”处就打断了它，此时历史记录却宣称下午 3 点已经敲定。

📌 they don't know the agent already told them the price（图，点击查看）

这就是 ctx.assistant() 放在 transport.output() 之后的原因：它记录的是实际说出口的内容，而不是生成的内容。

读懂延迟

每轮对话结束后，终端会打印出类似下面的内容。这是我在班加罗尔的笔记本上、通过 localhost 的 WebRTC 实测得到的真实数字：

AnthropicLLM   time-to-first-byte    874 ms
SmallestTTS    time-to-first-byte    180 ms

当天跑了若干轮，Claude Haiku 的首个 token 落在 0.6 到 1.0 秒之间，Lightning 的首个音频字节在 150 到 250 毫秒之间。这只是一台笔记本、一个网络环境下的结果，把它当作“该测什么”的示例即可，别当成基准测试。当天早些时候我开着 VPN，同样的代码明显更慢——你的网络是你延迟预算的一部分。

盯着这些输出看了一阵之后，我总结出几点：

LLM 通常是最大的一块。* 在我的运行里，它每一轮都是 TTS 数字的好几倍。

模型怎么选要实测，不要想当然。* 我的 OpenAI key 在演讲当天早上失效了，于是换成了 Claude。反正两个都有，我干脆都测了一遍。当天在我的网络下，同一个无头测试中 gpt-4o-mini 的首个 token 要 2.3 到 3.0 秒，而 Claude Haiku 4.5 是 1.2 到 1.7 秒。你所在的地区结果可能反过来，而这正是这套计量代码存在的意义。

这套配置里，语音转文字不会以 TTFB 行的形式出现*，因为流式 STT 的上报方式不同。计量代码会把零值过滤掉，免得你读到一个有误导性的"0 ms"。

如果你只想要一个目标值：Voice AI & Voice Agents primer 在其 2026 年 6 月版中把 1.5 秒的语音到语音延迟称为“一个值得追求的重要目标”。接下来测你的 p95，而不是平均值——平均值会掩盖那些慢的调用，而慢的那几次才是用户真正记得住的。

一行代码换模型

这个智能体里的每个模型都只占一行。想换一个 Smallest 的音色，改 TTS 那一行即可：

tts = SmallestTTSService(
    api_key=os.getenv("SMALLEST_API_KEY"),
    settings=SmallestTTSSettings(voice="nolan"),
)

想换别的 LLM？把 llm = ... 那一行换成任何其他 Pipecat 的 LLM 服务。想换语音转文字厂商？同理。重启、重新连接，你会发现整条 pipeline 一行都没动。

这正是编排器（orchestrator）的全部意义所在。模型是可随时替换的零件，而这个循环才是真正属于你的工程。

Pipecat 还是 LiveKit？

总有人问这个问题。两者都是不错的开源编排器，选一个就行，你不会两个一起用。Pipecat 是用 Python 组合起来的一串处理器 pipeline，这也是上面那个智能体读起来像一份清单的原因。LiveKit Agents 则围绕 LiveKit 的媒体服务器构建，需要你定义一个带 stt、llm 和 tts 的 AgentSession。Smallest 同样有 LiveKit 的官方插件（livekit-plugins-smallestai），所以这些模型选择可以直接沿用过去。

下一步

到这里，你已经用大约 100 行代码得到了一个支持 barge-in、带延迟读数的流式语音智能体。下一篇我们会把这个原封不动的智能体接到一个真实手机号码上，pipeline 保持不变，只换传输层（transport）。之后，我会讲语音智能体的那些难点（端点检测、回声、工具调用延迟、8 kHz 电话音频），以及如何把这一切推向生产环境。

参考资料

* Pipecat 文档，以及其中的 Smallest TTS 与 STT 服务页面

* Smallest AI 文档

* Voice AI & Voice Agents 图解入门

* Stivers 等，Universals and cultural variation in turn-taking in conversation（会话轮换的普适性与文化差异），PNAS 2009

* 本文完整代码见 voice-agent 仓库

原文：https://dev.to/stalwartcoder/build-a-low-latency-voice-agent-in-python-with-pipecat-3plb（作者 @stalwartcoder）