# Pipelines & workers

> How a Pipeline links processors, how a PipelineWorker runs it, and how a WorkerRunner runs workers.

Source: https://polvallverdu.github.io/pipedog/docs/concepts/pipelines-and-workers/



Three classes take a pipeline from a list of processors to a running bot:

| Class            | Module                            | Job                                                                                                         |
| ---------------- | --------------------------------- | ----------------------------------------------------------------------------------------------------------- |
| `Pipeline`       | `@pipedog/core/pipeline/pipeline` | Links frame processors in sequence. It is itself a processor, so pipelines nest.                            |
| `PipelineWorker` | `@pipedog/core/pipeline/worker`   | Runs one pipeline: start, frame queueing, lifecycle events, idle and heartbeat monitoring, observers, RTVI. |
| `WorkerRunner`   | `@pipedog/core/workers/runner`    | Runs workers to completion: signal handling, the shared bus, the worker registry.                           |

<Callout>
  `PipelineTask` and `PipelineRunner` are deprecated aliases upstream (of `PipelineWorker` and
  `WorkerRunner`) and are not ported. Use the `Worker` names.
</Callout>

## Pipeline [#pipeline]

A `Pipeline` takes an array of processors and connects them in order, adding a source before the
first and a sink after the last:

```ts
const pipeline = new Pipeline([a, b, c]);
```

A typical voice pipeline (once the service packages are ported) looks exactly like Pipecat's:

```ts
const pipeline = new Pipeline([
  transport.input(), // receives user audio
  stt, // speech to text
  userAggregator, // collects user turns into the context
  llm, // generates responses
  tts, // text to speech
  transport.output(), // sends audio to the user
  assistantAggregator, // collects assistant turns
]);
```

**Order matters**: each processor must sit after the processors that produce the frames it needs.
And **processors push frames on** rather than consuming them, so several processors can use the
same audio or text without interfering.

### Parallel pipelines [#parallel-pipelines]

`ParallelPipeline` creates branches that each receive every frame; their output is merged back.
Branches are usually paired with filters so each handles only some frames. Like Python, each
branch is a separate array argument:

```ts
// Every branch receives every frame; their output is merged downstream.
const parallel = new Pipeline([a, new ParallelPipeline([b], [c]), d]);
```

## PipelineWorker [#pipelineworker]

`PipelineWorker` runs a pipeline. It takes the pipeline and an options object
(`PipelineWorkerOptions`), including the pipeline-wide `PipelineParams`:

```ts
const worker = new PipelineWorker(pipeline, {
  params: new PipelineParams({
    audioInSampleRate: 16000,
    audioOutSampleRate: 24000,
    enableMetrics: true,
    enableUsageMetrics: true,
  }),
  idleTimeoutSecs: 300, // seconds, as upstream
  cancelOnIdleTimeout: true,
  processorUnusablePolicy: ProcessorUnusablePolicy.CONTINUE,
  enableRtvi: false,
});
```

`PipelineParams` is passed to every processor through `FrameProcessorSetup`. Its fields (see the
[PipelineWorker API](https://polvallverdu.github.io/pipedog/docs/api/pipeline-worker.md)):

| Field                     | Default | Description                                                |
| ------------------------- | ------- | ---------------------------------------------------------- |
| `audioInSampleRate`       | `16000` | Input audio sample rate in Hz.                             |
| `audioOutSampleRate`      | `24000` | Output audio sample rate in Hz.                            |
| `enableMetrics`           | `false` | Collect performance metrics (TTFB, processing time).       |
| `enableUsageMetrics`      | `false` | Collect usage metrics (LLM tokens, TTS characters).        |
| `reportOnlyInitialTtfb`   | `false` | Report only the first TTFB per service.                    |
| `sendInitialEmptyMetrics` | `true`  | Send a zeroed `MetricsFrame` when the pipeline starts.     |
| `enableHeartbeats`        | `false` | Push `HeartbeatFrame`s to detect a stalled pipeline.       |
| `heartbeatsPeriodSecs`    | `1.0`   | Seconds between heartbeats.                                |
| `heartbeatsMonitorSecs`   | `10.0`  | Seconds without a heartbeat before `on_heartbeat_timeout`. |
| `startMetadata`           | `{}`    | Extra metadata for pipeline start.                         |

Other worker behavior is set on the worker options: `idleTimeoutSecs` (default 300) and
`idleTimeoutFrames`, `cancelOnIdleTimeout`, `cancelTimeoutSecs`, `setupTimeoutSecs` and
`startTimeoutSecs` (20 s each), `observers`, `enableRtvi`, `enableTurnTracking`, `enableTracing`,
`conversationId`, `appResources`, `processorUnusablePolicy`, `bridged` and more.

### Event handlers [#event-handlers]

| Event                         | Arguments                     | Fires when                                                                            |
| ----------------------------- | ----------------------------- | ------------------------------------------------------------------------------------- |
| `on_pipeline_started`         | `(worker, frame: StartFrame)` | The `StartFrame` reached the end of the pipeline.                                     |
| `on_pipeline_finished`        | `(worker, frame)`             | The pipeline reached a terminal state: `StopFrame`, `EndFrame` or `CancelFrame`.      |
| `on_pipeline_error`           | `(worker, frame: ErrorFrame)` | An `ErrorFrame` reached the source. Check `frame.processor.isUsable`.                 |
| `on_pipeline_timeout`         | `(worker, frame)`             | A frame the worker waited on (a `StartFrame` or `CancelFrame`) never reached the end. |
| `on_setup_timeout`            | `(worker)`                    | Setting the processors up took longer than `setupTimeoutSecs`.                        |
| `on_idle_timeout`             | `(worker)`                    | None of `idleTimeoutFrames` arrived within `idleTimeoutSecs`.                         |
| `on_heartbeat_timeout`        | `(worker)`                    | No heartbeat within `heartbeatsMonitorSecs` (repeats while stalled).                  |
| `on_frame_reached_upstream`   | `(worker, frame)`             | A frame matching the upstream filter reached the source.                              |
| `on_frame_reached_downstream` | `(worker, frame)`             | A frame matching the downstream filter reached the sink.                              |

```ts
worker.eventHandler("on_pipeline_started", async (_worker: PipelineWorker, _frame: Frame) => {
  console.log("pipeline started");
});

worker.eventHandler("on_pipeline_finished", async (_worker: PipelineWorker, frame: Frame) => {
  // StopFrame, EndFrame or CancelFrame: inspect it to tell the cases apart.
  console.log(`pipeline finished: ${frame}`);
});

worker.eventHandler("on_pipeline_error", async (_worker: PipelineWorker, frame: ErrorFrame) => {
  if (frame.processor && !frame.processor.isUsable) {
    await worker.cancel({ reason: `${frame.processor} is no longer usable` });
  }
});

worker.eventHandler("on_idle_timeout", async (_worker: PipelineWorker) => {
  console.log("nobody said anything for a while");
});
```

Set the frame types `on_frame_reached_*` fire for with `setReachedUpstreamFilter([...])` /
`addReachedDownstreamFilter([...])`.

### Queueing frames [#queueing-frames]

Frames queued on the worker enter at the start of the pipeline (or, upstream, at the end). They
are processed in order within their lane, so a goodbye is spoken before the `EndFrame` ends the
pipeline:

```ts
// Executed in order: speak first, then end the pipeline once that is flushed.
await worker.queueFrames([
  new TTSSpeakFrame({ text: "Goodbye!" }),
  new EndFrame({ reason: "user said goodbye" }),
]);

// Interrupt whatever the bot is doing.
await worker.queueFrame(new InterruptionWorkerFrame());

// Push from the end of the pipeline, towards the start.
await worker.queueFrame(new TextFrame({ text: "upstream" }), FrameDirection.UPSTREAM);
```

## Ending a pipeline [#ending-a-pipeline]

| Frame               | Job                                                    | How to push it                                                                 |
| ------------------- | ------------------------------------------------------ | ------------------------------------------------------------------------------ |
| `EndFrame`          | Graceful shutdown, after pending frames are processed. | From outside: `worker.queueFrame(new EndFrame())`, or `worker.stopWhenDone()`. |
| `CancelFrame`       | Immediate shutdown, discarding pending frames.         | `worker.cancel()`.                                                             |
| `EndWorkerFrame`    | Graceful shutdown requested from inside the pipeline.  | `this.pushFrame(new EndWorkerFrame(), FrameDirection.DOWNSTREAM)`.             |
| `CancelWorkerFrame` | Immediate shutdown requested from inside the pipeline. | `this.pushFrame(new CancelWorkerFrame())`.                                     |

```ts
await worker.stopWhenDone(); // queue an EndFrame behind everything already queued
await worker.end({ reason: "done" }); // request a graceful end, draining first
await worker.cancel({ reason: "shutting down" }); // stop now, dropping queued frames
```

From inside a processor, push the worker frames. Push `EndWorkerFrame` **downstream**, so it
queues behind the work ahead of it instead of jumping past it:

```ts
/** Ends the conversation when it sees a "goodbye" TextFrame. */
export class GoodbyeDetector extends FrameProcessor {
  override async processFrame(frame: Frame, direction: FrameDirection): Promise<void> {
    await super.processFrame(frame, direction);
    await this.pushFrame(frame, direction);

    if (frame instanceof TextFrame && frame.text.toLowerCase().includes("goodbye")) {
      await this.pushFrame(new TTSSpeakFrame({ text: "Have a nice day!" }));
      // Downstream (the default), so the goodbye is flushed before the pipeline ends.
      await this.pushFrame(new EndWorkerFrame(), FrameDirection.DOWNSTREAM);
    } else if (frame instanceof TextFrame && frame.text === "abort") {
      // Ends right away, abandoning queued frames.
      await this.pushFrame(new CancelWorkerFrame());
    }
  }
}
```

## WorkerRunner [#workerrunner]

`WorkerRunner` runs workers until they finish. For a single-pipeline bot, add the worker and run:

```ts
const runner = new WorkerRunner({
  handleSigint: true, // the default: Ctrl-C cancels the runner (Bun, Node, Deno)
});
await runner.addWorkers(worker);
await runner.run(); // returns once every root worker has finished
```

* `run()` resolves once every root worker has finished, so a one-pipeline bot ends when its
  pipeline does. Pass `run({ autoEnd: false })` for a long-lived host (an HTTP server, say) that
  adds and removes workers over many sessions; it then runs until `runner.end()` or
  `runner.cancel()`.
* With `handleSigint` (default `true`) and `handleSigterm` (default `false`) the runner cancels
  itself on those signals, on runtimes with process signals (Bun, Node, Deno). On Cloudflare
  Workers there are none, and it is a no-op.
* `runner.end(reason)` ends every worker gracefully; `runner.cancel(reason)` cancels them.
* Multi-worker bots add several workers with `addWorkers(a, b, ...)`; they talk over the runner's
  `bus` (an in-process `AsyncQueueBus` by default) and are found with `runner.getWorker(name)`.