base_task

Base pipeline task implementation for managing pipeline execution.

This module provides the abstract base class and configuration for pipeline tasks that manage the lifecycle and execution of frame processing pipelines.

class pipecat.pipeline.base_task.PipelineTaskParams(loop: AbstractEventLoop)

Bases: object

Configuration parameters for pipeline task execution.

Parameters:

loop – The asyncio event loop to use for task execution.

loop: AbstractEventLoop

class pipecat.pipeline.base_task.BasePipelineTask(*, name: str | None = None, **kwargs)

Bases: BaseObject

Abstract base class for pipeline task implementations.

Defines the interface for managing pipeline execution lifecycle, including starting, stopping, and frame queuing operations.

abstractmethod has_finished() → bool

Check if the pipeline task has finished execution.

Returns:

True if all processors have stopped and the task is complete.

abstractmethod async stop_when_done()

Schedule the pipeline to stop after processing all queued frames.

Implementing classes should send an EndFrame or equivalent signal to gracefully terminate the pipeline once all current processing is complete.

abstractmethod async cancel()

Immediately stop the running pipeline.

Implementing classes should cancel all running tasks and stop frame processing without waiting for completion.

abstractmethod async run(params: PipelineTaskParams)

Start and run the pipeline with the given parameters.

Implementing classes should initialize and execute the pipeline using the provided configuration parameters.

Parameters:

params – Configuration parameters for pipeline execution.

abstractmethod async queue_frame(frame: Frame)

Queue a single frame for processing by the pipeline.

Implementing classes should add the frame to their processing queue for downstream handling.

Parameters:

frame – The frame to be processed.

abstractmethod async queue_frames(frames: Iterable[Frame] | AsyncIterable[Frame])

Queue multiple frames for processing by the pipeline.

Implementing classes should process the iterable/async iterable and add all frames to their processing queue.

Parameters:

frames – An iterable or async iterable of frames to be processed.