AGENTS.md · pipecat-ai/pipecat

Sections

1. Project Overview
2. Common Commands
3. Architecture
4. Code Style
5. Writing for Future Readers
6. Service Implementation
7. Testing

First lines (shown under the repo's BSD-2-Clause license)

# AGENTS.md

This file provides guidance to AI coding agents when working with code in this repository.

## Project Overview

Pipecat is an open-source Python framework for building real-time voice and multimodal conversational AI agents. It orchestrates audio/video, AI services, transports, and conversation pipelines using a frame-based architecture.

## Common Commands

```bash
# Setup development environment
uv sync --group dev --all-extras --no-extra gstreamer --no-extra local

# Install pre-commit hooks
uv run pre-commit install

# Run all tests (in parallel, one worker per CPU core; `-n 0` runs them in one process)
uv run pytest

# Run a single test file
uv run pytest tests/test_name.py

# Run a specific test
uv run pytest tests/test_name.py::test_function_name

# Preview changelog
uv run towncrier build --draft --version Unreleased

# Run a behavioral eval scenario against a running bot (started with `python bot.py`)
pipecat eval run scenarios/<name>.yaml --bot-url ws://localhost:7860/ws

# Run the full release-eval suite (spawns bots from a manifest, runs scenarios in parallel)
pipecat eval suite evals/release/manifest.yaml -p <bot-pattern> -s <scenario>

# Judge saved simulation conversations (<stem>.conversation.json) again, without the bot
pipecat eval judge scenarios/<name>.yaml -s <scenario> logs/*.conversation.json -o judged.jsonl

# Lint and format check
uv run ruff check
uv run ruff format --check

# Update dependencies (after editing pyproject.toml)
uv lock && uv sync

# Refresh the Pipecat UI component snapshot vendored into the CLI's React client templates
node scripts/cli/sync-pipecat-ui.mjs
```

## Architecture

### Frame-Based Pipeline Processing

All data flows as **Frame** objects through a pipeline of **FrameProcessors**:

```
[Processor1] → [Processor2] → ... → [ProcessorN]
```

**Key components:**

- **Frames** (`src/pipecat/frames/frames.py`): Data units (audio, text, video) and control signals. Flow DOWNSTREAM (input→output) or UPSTREAM (acknowledgments/errors).

- **FrameProcessor** (`src/pipecat/processors/frame_processor.py`): Base processing unit. Each processor receives frames, processes them, and pushes results downstream.

- **Pipeline** (`src/pipecat/pipeline/pipeline.py`): Chains processors together.

- **ParallelPipeline** (`src/pipecat/pipeline/parallel_pipeline.py`): Runs multiple pipelines in parallel.

- **Transports** (`src/pipecat/transports/`): Transports are frame processors used for external I/O layer (Daily WebRTC, LiveKit WebRTC, WebSocket, Local). Abstract interface via `BaseTransport`, `BaseInputTransport` and `BaseOutputTransport`.

- **Workers (`src/pipecat/workers/`)**: A **worker** is the top-level runnable unit (it replaces the old "pipeline task"). `BaseWorker` (`src/pipecat/workers/base_worker.py`) is the abstract base: it owns activation, end/cancel, bus subscription, job RPC, and a default `run()` for bus-only workers. `WorkerParams` configures it. Specialized workers live under `src/pipecat/workers/llm/` (`LLMWorker` adds `@tool` collection; `LLMContextWorker` adds an `LLMContext` plus aggregator pair) and `src/pipecat/workers/ui/` (`UIWorker`).

- **Pipeline Worker (`src/pipecat/pipeline/worker.py`)**: `PipelineWorker` is the `BaseWorker` subclass that wraps a user-defined pipeline — the common single-bot case. It sends the first frame, `StartFrame`, so processors know they can start processing and pushing frames, and internally wraps the pipeline with a source processor (before the pipeline) and a sink processor (at the end) used for error handling, worker-level events, heartbeat monitoring, etc. Optional `bridged=` adds bus edge processors so the pipeline can exchange frames with other workers. `PipelineTask` is a deprecated (1.3.0) alias — construct `PipelineWorker` instead.

- **Worker Runner (`src/pipecat/workers/runner.py`)**: `WorkerRunner` is the high-level entry point for executing workers. It owns the shared `WorkerBus` and `WorkerRegistry` and handles signal management (SIGINT/SIGTERM) for graceful shutdown. Register workers with `await runner.add_workers(*workers)`, then `await runner.run()`. By default (`auto_end=True`) the runner ends once every root worker finishes — so a single-pipeline bot ends when its pipeline does; pass `auto_end=False` for long-lived hosts (e.g. a FastAPI server) that add/remove workers across sessions. `PipelineRunner`, and passing a worker directly to `run(worker)`, are deprecated (1.3.0).

- **Services** (`src/pipecat/services/`): 60+ AI provider integrations (STT, TTS, LLM, etc.). Extend base classes: `AIService`, `LLMService`, `STTService`, `TTSService`, `VisionService`.

- **Serializers** (`src/pipecat/serializers/`): Convert frames to/from wire formats for WebSocket transports. `FrameSerializer` base class defines `serialize()` and `deserialize()`. Telephony serializers (Twilio, Plivo, Vonage, Telnyx, Exotel, Genesys) handle provider-specific protocols and audio encoding (e.g., μ-law).

- **RTVI** (`src/pipecat/processors/frameworks/rtvi/`): Real-Time Voice Interface protocol bridging clients and the pipeline. `RTVIProcessor` handles incoming client messages (text input, audio, DTMF keypresses, function call results). `RTVIObserver` converts pipeline frames to outgoing messages: user/bot speaking events, transcriptions, LLM/TTS lifecycle, function calls, metrics, and audio levels.

- **Observers** (`src/pipecat/observers/`): Monitor frame flow without modifying the pipeline. Passed to `PipelineWorker` via the `observers` parameter. Implement `on_process_frame()` and `on_push_frame()` callbacks.

### Workers, Bus, and Jobs

Beyond the single-pipeline case, Pipecat supports multiple cooperating **workers** coordinated by a shared bus (folded in from the former `pipecat-subagents` package). Terminology note: a "worker" is a runnable unit, "task" now refers only to asyncio tasks, and cross-worker RPC uses "jobs" and "job groups".

- **Bus** (`src/pipecat/bus/`): `WorkerBus` is the abstract pub/sub message bus; `AsyncQueueBus` (`bus/local/async_queue.py`) is the default in-process implementation, with `PgmqBus` / `RedisBus` (`bus/network/`) for distributed workers. Typed messages live in `bus/messages.py` (a `BusMessage` hierarchy split into normal-priority data messages and high-priority system messages, covering frame transport, worker lifecycle, and job RPC). `BusBridgeProcessor` (`bus/bridge_processor.py`) bridges a pipeline to the bus; `BusSubscriber` is the receive-side mixin.

- **Registry** (`src/pipecat/registry/`): `WorkerRegistry` tracks local and remote workers. The runner manages registration; code uses `watch(name, handler)` (or the `@worker_ready(name=...)` decorator) to be notified when a named worker becomes ready.

- **Jobs** (`src/pipecat/pipeline/job_context.py`, `job_decorator.py`, `worker_ready_decorator.py`): A worker exposes handlers with `@job(name=..., sequential=...)`. A caller opens `async with self.job(worker_name, ...)` (single worker) or `self.job_group(*worker_names, ...)` (fan-out) to send a request and await `JobStatus` results / streamed updates over the bus.

- **LLM tools** (`src/pipecat/workers/llm/tool_decorator.py`): On an `LLMWorker`, methods marked `@tool` are auto-collected and registered with the LLM service (supports `cancel_on_interruption` and `timeout`).

Runnable examples live in `examples/multi-worker/` (local handoff, distributed handoff via pgmq/redis, parallel debate, remote proxy, UI worker).

### Important Patterns

- **Context Aggregation**: `LLMContext` accumulates messages for LLM calls; the aggregators created by `LLMContextAggregatorPair` keep it updated with user and assistant turns

- **Turn Management**: Turn management is done through `LLMUserAggregator` and
  `LLMAssistantAggregator`, created with `LLMContextAggregatorPair`

- **User turn strategies**: Detection of when the user starts and stops speaking is done via user turn start/stop strategies. They push `UserStartedSpeakingFrame` and `UserStoppedSpeakingFrame` respectively.

- **Interruptions**: Interruptions are usually triggered by a user turn start strategy (e.g. `VADUserTurnStartStrategy`), but any processor can trigger one by calling `await self.broadcast_interruption()`, which broadcasts an `InterruptionFrame` both upstream and downstream. The old `push_interruption_task_frame_and_wait()` is deprecated and delegates to `broadcast_interruption()`.

- **Uninterruptible Frames**: These are frames that will not be removed from internal queues even if there's an interruption. Every frame carries an `interruptible` flag, True by default; a frame class such as `EndFrame` or `StopFrame` declares `interruptible: bool = field(default=False, init=False)` to be uninterruptible by default, and setting the flag on a frame before pushing it decides for that frame alone. A frame that must act after the bot speaks, such as a transfer, needs `interruptible = False` on both it and the `TTSSpeakFrame` before it, so the user talking over the bot can't drop either one.

- **Frames are the primary public interface**: Some classes have public methods that do the same work as a frame. When the frame gets the job done, prefer it, and don't add a new method that duplicates it. Use the method only when the work has to happen outside the frame order: a method acts at once, ahead of frames already queued. A method that duplicates a frame says so in its docstring and names the frame.

- **Events**: Most classes in Pipecat have `BaseObject` as the very base class. `BaseObject` has support for events. Events can run in the background in an async task (default) or synchronously (`sync=True`) if we want immediate action. Synchronous event handlers need to execute fast.

- **Async Task Management**: Always use `self.create_task(coroutine, name)` instead of raw `asyncio.create_task()`. The `TaskManager` automatically tracks tasks and cleans them up on processor shutdown. Use `await self.cancel_task(task, timeout)` for cancellation.

- **Keep `process_frame` fast**: a processor handles one frame at a time, so whatever `process_frame` is doing, every other frame arriving at that processor waits, and everything behind the processor in the pipeline waits with them. System frames jump the queue (`StartFrame` first, then `SystemFrame`, then the rest) but cannot interrupt a call in progress. Never do slow work inside `process_frame` (model inference, a blocking call, a long await): hand it to a task created with `self.create_task()` and push the result when it is ready. `SegmentedSTTService` transcribes each speech segment in a background task for this reason; transcribing inline held the input audio frames behind it, so a VAD downstream could not analyze them until it finished.

- **Error Handling**: Use `await self.push_error(msg, exception, fatal)` to push errors upstream. Services should use `fatal=False` (the default) so application code can handle errors and take action (e.g. switch to another service).

- **Accessing the worker**: Reach the running worker via the `pipeline_worker` property on `FrameProcessor` and the `pipeline_worker` field on `FunctionCallParams`. Both are required once the processor is set up (the property raises if accessed before setup). The old `pipeline_task` accessor is deprecated (1.3.0).

### Key Directories

| Directory                  | Purpose                                            |
| -------------------------- | -------------------------------------------------- |
| `src/pipecat/frames/`      | Frame definitions (100+ types)                     |
| `src/pipecat/processors/`  | FrameProcessor base + aggregators, filters, audio  |
| `src/pipecat/pipeline/`    | Pipeline orchestration; PipelineWorker + job RPC   |
| `src/pipecat/workers/`     | Worker model: BaseWorker, runner, LLM/UI workers   |
| `src/pipecat/bus/`         | Inter-worker message bus (local + pgmq/redis)      |
| `src/pipecat/registry/`    | Worker registry (local + remote tracking)          |
| `src/pipecat/services/`    | AI service integrations (60+ providers)            |
| `src/pipecat/transports/`  | Transport layer (Daily, LiveKit, WebSocket, Local) |
| `src/pipecat/serializers/` | Frame serialization for WebSocket protocols        |
| `src/pipecat/observers/`   | Pipeline observers for monitoring frame flow       |
| `src/pipecat/audio/`       | VAD, filters, mixers, turn detection, DTMF         |
| `src/pipecat/turns/`       | User turn management                               |
| `src/pipecat/adapters/`    | LLM adapters (`llm/`): context/tools conversion    |
| `src/pipecat/runner/`      | Development runner (multi-transport bot hosting)   |
| `src/pipecat/cli/`         | `pipecat` CLI (`init`, `eval`)                     |
| `src/pipecat/evals/`       | Behavioral eval framework (run via `pipecat eval`) |
| `src/pipecat/metrics/`     | Metrics data models                                |
| `evals/`                   | Eval suites; `evals/release/` is the release suite |

## Code Style

- **Docstrings**: Google-style. Classes describe purpose; `__init__` has `Args:` section; dataclasses use `Parameters:` section.
- **Deprecations**: Every deprecation needs a `.. deprecated:: <version>` directive in the docstring (never inline `[DEPRECATED]` tags) — it's the registry's source of truth. Its body must **lead with the replacement as the first reference** — `Use :class:`X` instead.` / `Moved to :mod:`X`.` / `Merged into :class:`X`.` — or state `No replacement.` explicitly; **never lead with a contextual reference** (the deprecated thing itself, a `DeprecationWarning`, or a related-but-not-replacement API), and don't rely on incidental words like "no longer" to signal no-replacement. Prefer Sphinx roles (`:class:`/`:meth:`/`:func:`/`:attr:`/`:mod:`) over plain backticks, but use a backtick when a role wouldn't resolve (aliases like `Service.Settings`, usage idioms, parameters). For the runtime warning: **classes, functions, methods, and properties** use the PEP 702 `@deprecated` decorator from `pipecat.utils.deprecation` with a string-literal message matching the canonical template — `` `Subject` is deprecated since X.Y.Z and will be removed in A.B.C. Use `Replacement` instead. `` — where the removal is a concrete version (e.g. `2.0.0`, never "a future release") and the tail is `No replacement.` when nothing replaces it. Parameters, fields, module moves, and behavior/value changes can't use the decorator — call `warn_deprecated(message, stacklevel=...)` from `pipecat.utils.deprecation`, never `warnings.warn` directly, with a literal message in the same template and `stacklevel` pointing at the caller's code where it can. Enforced by `tests/test_deprecation_markers.py`; full conventions in `CONTRIBUTING.md`.
- **Linting**: Ruff (line length 100). Pre-commit hooks enforce formatting.
- **Type hints**: Required for complex async code.
- **Dataclass vs Pydantic**: Use `@dataclass` for frames and internal pipeline data (high-frequency, no validation needed). Use Pydantic `BaseModel` for configuration, parameters, metrics, and external API data (benefits from validation and serialization). Specifically:
  - `@dataclass`: Frame types, context aggregator pairs, internal data containers
  - `BaseModel`: Service `InputParams`, transport/VAD/turn params, metrics data, API request/response models, serializer params

### Docstring Example

```python