Demonstrates the WebSocket proxy tasks: a local `main.py` voice
bot uses `WebSocketProxyClientTask` to forward bus messages
(including `BusFrameMessage`s) to a remote `assistant.py`
FastAPI server. Each incoming connection spawns a
`WebSocketProxyServerTask` plus an `LLMTask` assistant on a
per-session `PipelineRunner`.
Two transports of the same shape: a main task that hosts the
voice pipeline plus a network-backed `TaskBus` (`RedisBus` or
`PgmqBus`), and a standalone `llm.py` worker process for the
greeter / support LLM. Workers connect to the same bus channel,
register on the shared `TaskRegistry`, and the main task waits
on `runner.registry.watch("greeter", ...)` before sending the
welcome activation so it doesn't fire before the worker is up.
A voice moderator that fans out a debate topic to three worker
tasks (advocate, critic, analyst) via `task.job_group(...)`,
then synthesizes their replies. Workers are `LLMContextTask`s
that keep their own conversation context across rounds and use
the assistant-aggregator's `on_assistant_turn_stopped` event
to ship the completed turn back as a job response.
Variant of the local handoff example with per-task TTS voices.
Each child task wraps the LLM with its own `CartesiaTTSService`
in a custom pipeline override, so the main task has no TTS and
audio comes from whichever child is active over the bus.
Voice code assistant that dispatches questions to a Claude Agent
SDK worker. The main task runs the voice pipeline (STT + LLM + TTS)
and an `ask_code` direct function. `CodeWorker` is a bus-only
`BaseTask` spawned on the runner: it accepts `@job`-style
requests through the bus, queues them onto an asyncio queue, and
runs them sequentially through a persistent Claude SDK session so
follow-ups share context. The example shows the job-RPC surface
(`task.job("code_worker", ...)`), bus-only tasks (no pipeline),
and the `pipeline_task` field on `FunctionCallParams`.
Two LLM tasks (greeter and support) handing off to each other over
the local `AsyncQueueBus`. The main task owns the transport
pipeline (STT, TTS, transport I/O) and the child tasks each run
their own LLM behind a `BusBridgeProcessor`. Each child uses
`bridged=()` so `PipelineTask` auto-wraps its pipeline with
the bus edge processors, and `transfer_to_agent` / `end_conversation`
tools demonstrate `handoff_to(...)` and `end(...)`.
- `TaskBus._router_task`: cast the narrowed `SystemFrame` back
to `BusMessage` for the subscriber callback.
- `bus.network.__init__`: expose `PgmqBus` / `RedisBus` to
the type-checker via a TYPE_CHECKING block so `__all__` is
satisfied; runtime path still goes through `__getattr__`.
- `RedisBus`: subscribe through a local before assigning
`self._pubsub`, and `assert self._pubsub is not None` in
the reader loop.
- `BaseTask.on_job_error` accepts
`BusJobResponseMessage | BusJobResponseUrgentMessage` to match
what is dispatched.
- `JobGroupContext.__aexit__` / `JobContext.__aexit__`: assert
`self._group is not None` before `wait()`.
- `@task_ready` collector: type handlers dict as `dict[str, Callable]`
so the `.__name__` read on a duplicate handler typechecks.
- WebSocket proxy client/server: assert the socket is set in
`_receive_loop`, and decode `str` payloads to bytes before
handing them to the serializer.
`WebSocketProxyServerAgent` / `WebSocketProxyClientAgent` are
renamed to `WebSocketProxyServerTask` / `WebSocketProxyClientTask`
and updated for the post-refactor surface:
- Drop `bus=` from the constructor; the bus arrives via
`BaseTask.attach` from the runner.
- Constructor params `agent_name` / `remote_agent_name` /
`local_agent_name` → `task_name` / `remote_task_name` /
`local_task_name` (matching `BusBridgeProcessor`).
- Move setup logic from the now-removed `on_ready` hook into
`start()`; replace `_stop()` overrides with `stop()`.
- Add `_handle_task_end` / `_handle_task_cancel` overrides that
set `_finished_event` so `PipelineRunner._cancel_spawned_tasks`
can drive these bus-only tasks to a clean exit.
- Update the registry-message field reference
(`agents=`/`message.agents` → `tasks=`/`message.tasks`)
and `TaskReadyData.task_name` access.
- Tighten the server's `_send_ws` exception handling to only
catch `WebSocketDisconnect`.
- Update install hints (`pipecat-ai[websockets-base]` for the
client, `starlette` for the server) and refresh docstrings/
examples to use `runner.spawn(...)`.
Cleans up leftover "agent" terminology in module/class/method
docstrings across `pipecat.bus`, `pipecat.registry`,
`pipecat.pipeline`, and `pipecat.tasks.llm`, and renames
job-RPC phrasing ("task request", "task identifier",
"task group execution") to use "job" consistently.
API-visible changes:
- `BusBridgeProcessor(agent_name=, target_agent=)` → `task_name=` /
`target_task=`.
- `@task_ready` decorator's internal marker
`fn.agent_ready_name` → `fn.task_ready_name`.
- `@tool` decorator's internal marker
`fn.is_agent_tool` → `fn.is_llm_tool`.
- `PIPECAT_SUBAGENTS_SETUP_FILES` env var →
`PIPECAT_RUNNER_SETUP_FILES`.
- pgmq/redis bus install hints point at `pipecat-ai\[extra\]`
rather than the old `pipecat-ai-subagents\[extra\]` package.
Fixes a name collision where `_handle_task_cancel` was defined
twice — once for `BusCancelTaskMessage` (task lifecycle) and
once for `BusJobCancelMessage` (job RPC) — the second silently
shadowing the first. Job-side dispatchers are now consistently
named `_handle_job_*` and the internal helpers
`_run_task_handler` / `_send_task_request` become
`_run_job_handler` / `_send_job_request`. Task-lifecycle
handlers (`_handle_task_end`, `_handle_task_cancel`,
`_handle_task_activate`, `_handle_task_deactivate`,
`_handle_task_error`) keep their names.
`PipelineRunner.run(task)` now calls `spawn(task)` first (which
runs `task.attach()`) and lets `_setup_session` start every
registered entry — main and pre-spawned — through the same path,
instead of relying on `spawn`'s post-running fast-path to start
the main task after setup. The two-branch wait stays for the
`task is None` case but reads the runner_task directly off the
freshly-spawned entry.
`BaseTask` no longer takes `bus=` in its constructor. Instead
the runner now hands both the registry and the bus to a task via
`task.attach(registry=..., bus=...)` (called from
`PipelineRunner.spawn()`), and `bus` / `registry` are
properties that raise if accessed before attach. `PipelineTask`,
`LLMTask`, and `LLMContextTask` lose their `bus=` parameters
to match, and `_BusEdgeProcessor` now stores only a task
reference and reads `task.bus` lazily so bridged pipelines work
even though the bus isn't known at construction time.
`FrameProcessorSetup.pipeline_task` is now mandatory and
`FrameProcessor.pipeline_task` raises if accessed before setup
instead of returning `None`. `FunctionCallParams` gains a
required `pipeline_task` field and `LLMService._run_function_call`
populates it (plus reads `app_resources` directly off the
pipeline task). Tests that build a processor or
`FunctionCallParams` outside a real pipeline stub it with a
`SimpleNamespace`.
Adds `pipecat.tasks.llm` with `LLMTask` (LLM pipeline + `@tool`
collection + tool-call deferral via `PipelineFlushFrame`),
`LLMContextTask` (LLM + `LLMContextAggregatorPair`), and the
`@tool` decorator. Also includes `pipecat.tasks.proxy.websocket`
client/server stubs that need a follow-up port to the new
`BaseTask` lifecycle.
`PipelineRunner` now owns the shared `TaskBus` and
`TaskRegistry` and runs all tasks (the main one plus any
spawned ones) through a unified `_start_task` / `_run_task`
background-task path. Adds `spawn(task)` for fire-and-forget
task registration, threads `end()` / `cancel()` through
`BusEndTaskMessage` / `BusCancelTaskMessage` to all root
tasks, and broadcasts/handles `BusTaskRegistryMessage` for
remote-runner discovery. The runner now wires its own
`TaskManager` via `super().setup(...)` so internal
`create_task` calls go through `BaseObject`.
`PipelineTask` now extends `BaseTask` so every pipeline task is
also a bus participant. Adds optional `bus`, `bridged`, and
`exclude_frames` parameters: when `bridged` is set, the user's
pipeline is wrapped with `_BusEdgeProcessor` source/sink edges so
frames are mirrored onto the bus. Bridges pipeline lifecycle
events to `start()`/`stop()`, overrides `_handle_task_end` /
`_handle_task_cancel` to drive the pipeline shutdown, subscribes
to the bus in setup, and exposes the `bridged` property to the
registry. Moves `PipelineTaskParams` here and updates the
matching test import.
Drops the old abstract `BasePipelineTask` and replaces it with
`BaseTask` — the common base for any runtime task. `BaseTask`
subscribes to a `TaskBus`, participates in the shared
`TaskRegistry`, handles activation / deactivation, end / cancel,
and the full `@job` RPC surface (request_job, job, job_group,
send_job_response / update / stream_*, etc.). It ships a default
`run()` for bus-only tasks; subclasses with their own runtime
(e.g. `PipelineTask`) override it.
Adds `JobContext` / `JobGroupContext` async context managers,
the `JobGroup` / `JobGroupEvent` / `JobGroupResponse` /
`JobGroupError` types, the `@job` decorator (with collector),
and the `@task_ready` decorator (with collector). These power
the bus-driven job RPC between tasks.
Adds ``pipecat.bus.network.pgmq.PgmqBus``, a PGMQ-backed
:class:`TaskBus` adapter that implements pub/sub fan-out over
PGMQ's point-to-point queue semantics. Each bus instance owns its
own queue, broadcasts on publish to peers discovered by channel
prefix, and long-polls its queue to dispatch received messages
to local subscribers.
Requires the optional ``pgmq`` extra
(``pip install pipecat-ai[pgmq]``).
Introduces `TaskBus`, the in-process `AsyncQueueBus`, the bus
message hierarchy (lifecycle, jobs, frames, registry), a
priority-aware bus queue, the `BusSubscriber` mixin, and the
`BusBridgeProcessor` / internal `_BusEdgeProcessor` used to
exchange frames between a local pipeline and the bus.
Introduces `TaskRegistry` and the supporting `TaskReadyData`,
`TaskErrorData`, and `TaskRegistryEntry` dataclasses used to track
local and remote tasks discovered through the bus.
Adds InceptionLLMService, an OpenAI-compatible service for Inception's
Mercury-2 diffusion-based reasoning model. Supports reasoning_effort
(instant/low/medium/high) and realtime mode for reduced TTFT.
Adds tests for AggregatedFrameSequencer, WordCompletionTracker, and
word_timestamp_utils (including CJK language scenarios). Updates existing
Cartesia TTS and TTS frame ordering tests to cover the new behaviours.
TTSTextFrame entries were losing their original text structure when word
timestamps were enabled. AggregatedTextFrame now carries a raw_text field with
the original LLM-produced text (including pattern delimiters such as
<card>...</card>). The assistant context receives properly-tagged content
rather than the cleaned words returned by the TTS provider. Also handles words
that straddle two sentence boundaries by splitting and attributing each part
to its correct source frame.
SSML markup (e.g. <spell>, <emotion>, <break>) was leaking into word entries
returned by the Cartesia word-timestamps API. Tags are now stripped before
processing so word-to-text attribution remains accurate when SSML is present
in the TTS input.
Frames sharing the same presentation timestamp were being reordered by the
priority queue. Adds a monotonic counter as a tiebreaker so frames with equal
PTS are always emitted in insertion order, preventing subtle audio/text
sequencing bugs.
Skipped frames (e.g. code blocks filtered via skip_aggregator_types) were
emitted to the assistant context immediately instead of waiting for preceding
spoken frames to finish. Introduces AggregatedFrameSequencer to hold each
frame's slot and flush only after all earlier spoken sentences are complete,
keeping context ordering correct.
