Pipecat

一句话定位

一个开源的 实时语音与多模态对话 AI 框架（pyproject 自述 “framework for voice (and multimodal) assistants”）：把音频/视频、STT/LLM/TTS 服务、WebRTC/WebSocket 传输统一抽象成在 Pipeline 中单向流动的 Frame，由一串 FrameProcessor 事件驱动地逐级处理，专门解决实时语音对话里的低延迟、打断（barge-in）、轮次（turn）管理问题。

设计理念 / 顶层架构

Pipecat 的核心范式不是”agent 推理循环”，而是 frame-based 流式管道（pipeline of frame processors）——更接近 GStreamer / 数字音频工作站的数据流模型，而非 ReAct。设计取舍：

• 一切皆 Frame：音频、文本、图像、控制信号都是 Frame dataclass（src/pipecat/frames/frames.py 定义 ~236 个 Frame 类型）。Frame 有方向：DOWNSTREAM（输入→输出，承载数据）与 UPSTREAM（确认/错误回传），见 processors/frame_processor.py:56 的 FrameDirection。
• 一切皆 FrameProcessor：FrameProcessor（processors/frame_processor.py:175）是唯一处理单元；它接收 frame、处理、用 push_frame()（:702）推给下游。STT/LLM/TTS service、transport、aggregator、filter 全是它的子类。处理器用 link()（:536）串成链。
• Pipeline 即处理器链：Pipeline（pipeline/pipeline.py:91）把一个 processor 列表首尾相连，并自动在两端包上 PipelineSource/PipelineSink（:21/:55）。ParallelPipeline（pipeline/parallel_pipeline.py:24）让多条子管道并行。
• Worker = 可运行单元（1.3.0 新模型）：管道本身不自跑，要包进 PipelineWorker（pipeline/worker.py:170，BaseWorker 子类），由 WorkerRunner（workers/runner.py:80）驱动并管信号/优雅退出。旧的 PipelineTask（pipeline/worker.py:1273）/PipelineRunner 已弃用为别名。
• 事件驱动：几乎所有对象继承 BaseObject，支持事件回调（如 transport 的 @event_handler("on_client_connected")）；打断、轮次开始/结束都通过特定 Frame 在管道里传播来驱动状态变化，而非集中式调度。
• 多 worker / bus（合并自 pipecat-subagents）：单管道之外，多个 worker 可经 WorkerBus（bus/bus.py，默认 AsyncQueueBus，分布式可换 pgmq/redis）用 job RPC 协作。

最小示例（取自 examples/getting-started/06-voice-agent.py，典型 STT→LLM→TTS 语音回环）：

from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineParams, PipelineWorker
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair, LLMUserAggregatorParams,
)
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.services.deepgram.stt import DeepgramSTTService
from pipecat.services.openai.llm import OpenAILLMService
from pipecat.services.cartesia.tts import CartesiaTTSService
from pipecat.workers.runner import WorkerRunner

stt = DeepgramSTTService(api_key=...)
llm = OpenAILLMService(api_key=..., settings=OpenAILLMService.Settings(
    system_instruction="You are a helpful assistant in a voice conversation."))
tts = CartesiaTTSService(api_key=...)

context = LLMContext()
user_agg, assistant_agg = LLMContextAggregatorPair(
    context, user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()))

pipeline = Pipeline([
    transport.input(),   # 传输层进帧（麦克风/WebRTC）
    stt,                 # 语音转文字
    user_agg,            # 聚合用户轮次
    llm,                 # 大模型
    tts,                 # 文字转语音
    transport.output(),  # 传输层出帧（扬声器）
    assistant_agg,       # 聚合助手轮次
])

worker = PipelineWorker(pipeline, params=PipelineParams(enable_metrics=True))
runner = WorkerRunner()
await runner.add_workers(worker)
await runner.run()

组件实现（横向逐项，无则标 N/A）

设计权衡与特性

• 为”实时语音”而生，不是通用 agent harness：与 ConnectOnion 这类 single-agent ReAct 框架正交——Pipecat 的一等公民是音频流、延迟、打断、轮次，而不是”思考-行动循环”。LLM 在这里只是管道中一个 processor，function-calling 多轮是 service 内部行为。
• frame/processor 抽象的代价与收益：统一成 ~236 个 Frame + 单一 FrameProcessor 接口，换来极强的可组合性（任意 STT/LLM/TTS/transport 自由拼接、并行、加 filter/observer），但学习曲线偏陡——必须理解 frame 方向、push 时机、打断时 frame.complete() 的契约（不向下游传 InterruptionFrame 的处理器必须调 complete()，否则会卡住等待者）。
• 打断处理是核心难点的一等抽象：InterruptionFrame 携带 asyncio.Event、uninterruptible frame（EndFrame/StopFrame）、VAD 触发的用户轮次开始策略，共同构成低延迟 barge-in 机制——这是语音框架区别于文本 agent 框架的关键工程点。
• 极广的 provider 生态：pyproject 列了 60+ 可选 extra（STT/TTS/LLM/传输/电话/avatar），全部走统一 base 类 + adapter，换 provider 基本只改构造行。
• 新 worker/bus 模型（1.3.0）：把”pipeline task”重构为 BaseWorker/WorkerRunner，并内置多 worker job RPC（本地 AsyncQueueBus，分布式 pgmq/redis），向”多协作 worker”演进；旧 PipelineTask/PipelineRunner 保留为弃用别名。
• 待确认/坑：①star 数为约数，本次未联网核实；②PipelineTask/PipelineRunner/tool_resources/pipeline_task 等大量 1.2–1.3 弃用别名仍在，跟教程时注意版本；③不少能力靠 optional extra，缺装会 import 失败。

关联

• component-taxonomy · single-vs-multi-agent · tool-use
• 范式（voice / 流式管道）：本库唯一 · 源码：agents-example/pipecat/