PipelineTask

Manage pipeline execution and lifecycle with PipelineTask

Overview

PipelineTask is the central class for managing pipeline execution. It handles the lifecycle of the pipeline, processes frames in both directions, manages task cancellation, and provides event handlers for monitoring pipeline activity.

Basic Usage

from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.runner import PipelineRunner
from pipecat.pipeline.task import PipelineParams, PipelineTask

# Create a pipeline
pipeline = Pipeline([...])

# Create a task with the pipeline
task = PipelineTask(pipeline)

# Queue frames for processing
await task.queue_frame(TTSSpeakFrame("Hello, how can I help you today?"))

# Run the pipeline
runner = PipelineRunner()
await runner.run(task)

Constructor Parameters

pipeline · BasePipeline · required

The pipeline to execute.

params · PipelineParams · default: "PipelineParams()"

Configuration parameters for the pipeline. See PipelineParams for details.

observers · List[BaseObserver] · default: "[]"

List of observers for monitoring pipeline execution. See Observers for details.

clock · BaseClock · default: "SystemClock()"

Clock implementation for timing operations.

task_manager · Optional[BaseTaskManager] · default: "None"

Custom task manager for handling asyncio tasks. If None, a default TaskManager is used.

check_dangling_tasks · bool · default: "True"

Whether to check for processors’ tasks finishing properly.

idle_timeout_secs · Optional[float] · default: "300"

Timeout in seconds before considering the pipeline idle. Set to None to disable idle detection. See Pipeline Idle Detection for details.

idle_timeout_frames · Tuple[Type[Frame], ...] · default: "(BotSpeakingFrame, LLMFullResponseEndFrame)"

Frame types that should prevent the pipeline from being considered idle. See Pipeline Idle Detection for details.

cancel_on_idle_timeout · bool · default: "True"

Whether to automatically cancel the pipeline task when idle timeout is reached. See Pipeline Idle Detection for details.

enable_tracing · bool · default: "False"

Whether to enable OpenTelemetry tracing. See The OpenTelemetry guide for details.

enable_turn_tracking · bool · default: "False"

Whether to enable turn tracking. See The OpenTelemetry guide for details.

conversation_id · Optional[str] · default: "None"

Custom ID for the conversation. If not provided, a UUID will be generated. See The OpenTelemetry guide for details.

additional_span_attributes · Optional[dict] · default: "None"

Any additional attributes to add to top-level OpenTelemetry conversation span. See The OpenTelemetry guide for details.

Methods

Task Lifecycle Management

run() · async

Starts and manages the pipeline execution until completion or cancellation.

await task.run()

stop_when_done() · async

Sends an EndFrame to the pipeline to gracefully stop the task after all queued frames have been processed.

await task.stop_when_done()

cancel() · async

Stops the running pipeline immediately by sending a CancelFrame.

  await task.cancel()

has_finished() · bool

Returns whether the task has finished (all processors have stopped).

if task.has_finished(): print("Task is complete")

Frame Management

queue_frame() · async

Queues a single frame to be pushed down the pipeline.

await task.queue_frame(TTSSpeakFrame("Hello!"))

queue_frames() · async

Queues multiple frames to be pushed down the pipeline.

frames = [TTSSpeakFrame("Hello!"), TTSSpeakFrame("How are you?")]

await task.queue_frames(frames)

Event Handlers

PipelineTask provides an event handler that can be registered using the event_handler decorator:

on_idle_timeout

Triggered when no activity frames (as specified by idle_timeout_frames) have been received within the idle timeout period.

@task.event_handler("on_idle_timeout")
async def on_idle_timeout(task):
    print("Pipeline has been idle too long")
    await task.queue_frame(TTSSpeakFrame("Are you still there?"))