pipeline

Pipeline implementation for connecting and managing frame processors.

This module provides the main Pipeline class that connects frame processors in sequence and manages frame flow between them, along with helper classes for pipeline source and sink operations.

class pipecat.pipeline.pipeline.PipelineSource(upstream_push_frame: Callable[[Frame, FrameDirection], Coroutine], **kwargs)[source]

Bases: FrameProcessor

Source processor that forwards frames to an upstream handler.

This processor acts as the entry point for a pipeline, forwarding downstream frames to the next processor and upstream frames to a provided upstream handler function.

__init__(upstream_push_frame: Callable[[Frame, FrameDirection], Coroutine], **kwargs)[source]

Initialize the pipeline source.

Parameters:

  • upstream_push_frame – Coroutine function to handle upstream frames.

  • **kwargs – Additional arguments passed to parent class.

async process_frame(frame: Frame, direction: FrameDirection)[source]

Process frames and route them based on direction.

Parameters:

  • frame – The frame to process.

  • direction – The direction of frame flow.

class pipecat.pipeline.pipeline.PipelineSink(downstream_push_frame: Callable[[Frame, FrameDirection], Coroutine], **kwargs)[source]

Bases: FrameProcessor

Sink processor that forwards frames to a downstream handler.

This processor acts as the exit point for a pipeline, forwarding upstream frames to the previous processor and downstream frames to a provided downstream handler function.

__init__(downstream_push_frame: Callable[[Frame, FrameDirection], Coroutine], **kwargs)[source]

Initialize the pipeline sink.

Parameters:

  • downstream_push_frame – Coroutine function to handle downstream frames.

  • **kwargs – Additional arguments passed to parent class.

async process_frame(frame: Frame, direction: FrameDirection)[source]

Process frames and route them based on direction.

Parameters:

  • frame – The frame to process.

  • direction – The direction of frame flow.

class pipecat.pipeline.pipeline.Pipeline(processors: Sequence[FrameProcessor], *, source: FrameProcessor | None = None, sink: FrameProcessor | None = None)[source]

Bases: BasePipeline

Main pipeline implementation that connects frame processors in sequence.

Creates a linear chain of frame processors with automatic source and sink processors for external frame handling. Manages processor lifecycle and provides metrics collection from contained processors.

__init__(processors: Sequence[FrameProcessor], *, source: FrameProcessor | None = None, sink: FrameProcessor | None = None)[source]

Initialize the pipeline with a list of processors.

Parameters:

  • processors – Sequence of frame processors to connect in sequence.

  • source – An optional pipeline source processor.

  • sink – An optional pipeline sink processor.

property processors

Return the list of sub-processors contained within this processor.

Only compound processors (e.g. pipelines and parallel pipelines) have sub-processors. Non-compound processors will return an empty list.

Returns:

The list of sub-processors if this is a compound processor.

property entry_processors: list[FrameProcessor]

Return the list of entry processors for this processor.

Entry processors are the first processors in a compound processor (e.g. pipelines, parallel pipelines). Note that pipelines can also be an entry processor as pipelines are processors themselves. Non-compound processors will simply return an empty list.

Returns:

The list of entry processors.

processors_with_metrics()[source]

Return processors that can generate metrics.

Recursively collects all processors that support metrics generation, including those from nested pipelines.

Returns:

List of frame processors that can generate metrics.

async setup(setup: FrameProcessorSetup)[source]

Set up the pipeline and all contained processors.

Parameters:

setup – Configuration for frame processor setup.

async cleanup()[source]

Clean up the pipeline and all contained processors.

async process_frame(frame: Frame, direction: FrameDirection)[source]

Process frames by routing them through the pipeline.

Parameters:

  • frame – The frame to process.

  • direction – The direction of frame flow.