Pipecat — Pipelines, Frames & Processors¶

transport.input()  → InputAudioRawFrame ...           (microphone)
VAD / aggregator   → UserStartedSpeakingFrame, UserStoppedSpeakingFrame
STT                → TranscriptionFrame("Where is my order?")
user aggregator    → LLMContextFrame                  (context + new user message)
LLM                → LLMFullResponseStartFrame, LLMTextFrame("Your "), LLMTextFrame("order "), ..., LLMFullResponseEndFrame
TTS                → TTSStartedFrame, TTSTextFrame, OutputAudioRawFrame ...
transport.output() → audio to the client

from pipecat.frames.frames import TextFrame
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor


class Redactor(FrameProcessor):
    """Masks digits in text before it reaches TTS."""

    async def process_frame(self, frame, direction: FrameDirection):
        await super().process_frame(frame, direction)          # 1. always first
        if isinstance(frame, TextFrame):
            frame = TextFrame(text="".join("*" if c.isdigit() else c for c in frame.text))
        await self.push_frame(frame, direction)                # 2. always push every frame

from pipecat.pipeline.pipeline import Pipeline

pipeline = Pipeline(
    [
        transport.input(),
        stt,
        context_aggregator.user(),
        llm,
        tts,
        transport.output(),
        context_aggregator.assistant(),
    ]
)

from pipecat.pipeline.runner import WorkerRunner
from pipecat.pipeline.worker import PipelineParams, PipelineWorker

worker = PipelineWorker(
    pipeline,
    name="assistant",
    params=PipelineParams(enable_metrics=True, enable_usage_metrics=True),
    observers=[...],
)
runner = WorkerRunner(handle_sigint=runner_args.handle_sigint)
await runner.add_workers(worker)
await runner.run()                      # blocks until the session ends

@worker.event_handler("on_pipeline_started")
async def on_started(worker, frame): ...

@worker.event_handler("on_pipeline_finished")        # fires for graceful AND cancelled shutdown
async def on_finished(worker, frame):
    save_transcript()                                # do persistence here, not in on_client_disconnected

import asyncio

from pipecat.frames.frames import EndFrame, TTSSpeakFrame


async def hang_up_after(worker, seconds: float):
    await asyncio.sleep(seconds)
    await worker.queue_frames([TTSSpeakFrame("I have to go now. Goodbye!"), EndFrame()])

import asyncio

from pipecat.frames.frames import EndFrame, TextFrame
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import WorkerRunner
from pipecat.pipeline.worker import PipelineWorker
from pipecat.processors.frame_processor import FrameProcessor


class Collect(FrameProcessor):
    def __init__(self):
        super().__init__()
        self.seen = []

    async def process_frame(self, frame, direction):
        await super().process_frame(frame, direction)
        if isinstance(frame, TextFrame):
            self.seen.append(frame.text)
        await self.push_frame(frame, direction)


async def main():
    collector = Collect()
    worker = PipelineWorker(Pipeline([collector]), name="demo", cancel_on_idle_timeout=False)
    runner = WorkerRunner(handle_sigint=False)
    await runner.add_workers(worker)
    await worker.queue_frames([TextFrame("hello"), TextFrame("world"), EndFrame()])
    await runner.run()
    print(collector.seen)        # ['hello', 'world']


asyncio.run(main())