Pipecat 小白入门教程

Pipecat 小白入门教程

pipecat作为专注实时交互的Python框架，其核心优势在于将复杂的多模态处理拆解为模块化组。

• 接入层：支持WebRTC、WebSocket等多种传输协议
• 处理层：提供语音编解码、文本处理、图像渲染等基础能力
• 服务层：集成Anthropic、Deepgram等50+ AI服务

1. 核心概念（先理解整体结构）

Pipeline（处理管道）

数据从输入到输出依次经过多个处理器（Processor），你可以理解为一条“处理流水线”。

pipeline = Pipeline(
    [
        transport.input(),          # 输入（音频/文本）
        rtvi,                       # RTVI 协议处理器
        stt,                        # 语音识别
        user_input_store_processor, # 存储用户输入
        context_aggregator.user(),  # 用户上下文聚合
        llm,                        # 大语言模型
        tts,                        # 语音合成
        transport.output(),         # 输出
        context_aggregator.assistant(),
    ]
)

PipelineTask（运行任务）

负责启动 Pipeline 并发出 StartFrame / EndFrame。

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

FlowManager（流程控制器）

负责节点流转、工具注册、上下文更新与跳转逻辑。

flow_manager = FlowManager(
    task=task,
    llm=llm,
    context_aggregator=context_aggregator,
    transport=transport,
)

PipelineRunner（运行器）

启动任务并维持会话运行。

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

2. Transport 重要原理

Transport 是客户端与服务端通信的核心层，负责：

• 设备管理（麦克风/摄像头）
• 音视频推拉流
• 会话生命周期管理

Transport 生命周期（官方状态机）

Transport 会经历以下状态：

1. Disconnected
2. Initializing
3. Initialized
4. Authenticating
5. Authenticated
6. Connecting
7. Connected
8. Ready
9. Disconnecting
10. Error

在 SmallWebRTC 测试页面中，只有当状态进入 Ready，才能正常进行语音/文本交互。
官方说明参考：docs.pipecat.ai/client/js/t…

3. 前端参数传递路径（/start vs /api/offer）

SmallWebRTC 测试页面的 默认参数最终进入 runner_args.body 的路径是 /api/offer 的 request_data。

关键结论

• /start 用于 session 初始化（创建 sessionId、ICE 配置）
• /api/offer 才会把 request_data 写入 runner_args.body

所以如果你需要把默认参数传给后端，请确保注入到 /api/offer。

4. 捕捉用户输入（文本 + 语音）

文本输入

SmallWebRTC UI 发送 send-text，会生成 InputTransportMessageFrame。

if isinstance(frame, InputTransportMessageFrame):
    message = frame.message
    if message.get("type") == "send-text":
        text = message["data"]["content"]

语音输入

语音经过 STT 后输出 TranscriptionFrame。

if isinstance(frame, TranscriptionFrame):
    text = frame.text

5. 将用户输入写入共享变量（flow_manager.state）

使用 flow_manager.state 存储用户每次输入：

flow_manager.state.setdefault("user_inputs", [])
flow_manager.state["user_inputs"].append({"text": text})
flow_manager.state["last_user_input"] = text

在函数 handler 中读取：

last_input = flow_manager.state.get("last_user_input", "")
history = flow_manager.state.get("user_inputs", [])

6. 会话开始传入默认参数

前端（api/offer）参数结构

{
  "body": {
    "init_string": "患者ID=12345",
    "speech_meta": {"name": "lyq", "age": 31}
  }
}

后端读取方式

init_body = runner_args.body or {}
init_string = init_body.get("init_string")
speech_meta = init_body.get("speech_meta", {})

写入上下文或共享变量

flow_manager.state["init_string"] = init_string
context.add_message({"role": "system", "content": f"初始化参数：{init_string}"})

7. Flow 分支教程（条件跳转）

Flow 的分支逻辑靠 handler 返回 (result, next_node) 。

async def handle_choose_pizza(args, flow_manager):
    result = pizza_info()
    if result == 1:
        return result, create_confirm_node()
    else:
        return result, create_pizza_task_node()

原理解释

• FlowManager 读取 handler 返回的 next_node
• 自动调用 _set_node() 完成上下文更新 + 工具注册
• 然后触发下一轮 LLM

8. 自定义组件扩展原理

Pipecat 的核心抽象是 FrameProcessor。
所有组件都遵循：

组件分为三类：

9. 自定义 RTC Transport 实战（简化示例）

如果你要接入第三方 RTC（Agora / 腾讯 / 网易），只需实现 input() / output()：

class CustomRTCTransport(BaseTransport):
    def __init__(self, params: TransportParams, rtc_client):
        super().__init__()
        self._input = CustomRTCInput(params, rtc_client)
        self._output = CustomRTCOutput(params, rtc_client)

    def input(self):
        return self._input

    def output(self):
        return self._output

原理

Transport 只负责：

• 接收外部音视频 → 转换为 Frame
• 将 Frame 输出回外部推流
  不会直接处理 LLM 或 Flow 逻辑。

11. 核心流程总结（一句话）

官方 Transport 文档参考：
docs.pipecat.ai/client/js/t…