wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Sweep agent->task and task->job in docstrings and identifiers

Browse Source 
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.
This commit is contained in:
Aleix Conchillo Flaqué
2026-05-13 22:13:09 -07:00
parent e546471bef
commit 86f8f137a8
15 changed files with 131 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/pipecat/bus/__init__.py
View File

@@ -4,18 +4,18 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Agent bus package -- pub/sub messaging between agents and the runner.
"""Task bus package -- pub/sub messaging between tasks and the runner.
Provides the pub/sub infrastructure that connects agents to each other and to
Provides the pub/sub infrastructure that connects tasks to each other and to
the runner. Key components:
- `TaskBus` -- abstract base class defining the send/receive interface.
- `AsyncQueueBus` -- in-process implementation backed by ``asyncio.Queue``.
- `BusBridgeProcessor` -- bidirectional mid-pipeline bridge for
transport/session agents that exchanges frames with other agents
transport/session tasks that exchanges frames with other tasks
through the bus.
- `BusMessage` and its subclasses -- the typed message hierarchy used for
agent lifecycle events (activation, cancellation, shutdown), task
task lifecycle events (activation, cancellation, shutdown), job
coordination, and frame transport.
"""

32
src/pipecat/bus/bridge_processor.py
View File

@@ -41,8 +41,8 @@ _PASSTHROUGH_FRAMES = (OutputTransportMessageUrgentFrame,)
class BusBridgeProcessor(FrameProcessor, BusSubscriber):
"""Bidirectional mid-pipeline bridge between a Pipecat pipeline and the bus.
Placed in a transport or session agent's pipeline to exchange frames
with other agents via the `TaskBus`. Lifecycle and excluded frames
Placed in a transport or session task's pipeline to exchange frames
with other tasks via the `TaskBus`. Lifecycle and excluded frames
pass through locally without crossing the bus.
"""
@@ -50,8 +50,8 @@ class BusBridgeProcessor(FrameProcessor, BusSubscriber):
self,
*,
bus: TaskBus,
agent_name: str,
target_agent: str | None = None,
task_name: str,
target_task: str | None = None,
bridge: str | None = None,
exclude_frames: tuple[type[Frame], ...] | None = None,
**kwargs,
@@ -59,20 +59,20 @@ class BusBridgeProcessor(FrameProcessor, BusSubscriber):
"""Initialize the BusBridgeProcessor.
Args:
bus: The ``TaskBus`` to exchange frames with.
agent_name: Name of this agent, used as message source.
target_agent: When set, only exchange frames with this agent.
bus: The `TaskBus` to exchange frames with.
task_name: Name of the owning task, used as message source.
target_task: When set, only exchange frames with this task.
bridge: Optional bridge name for routing. When set, outgoing
frames are tagged with this name and only incoming frames
with the same bridge name are accepted.
exclude_frames: Extra frame types that should never cross the bus
(on top of lifecycle frames which are always excluded).
**kwargs: Additional arguments passed to ``FrameProcessor``.
**kwargs: Additional arguments passed to `FrameProcessor`.
"""
super().__init__(**kwargs)
self._bus = bus
self._agent_name = agent_name
self._target_agent = target_agent
self._task_name = task_name
self._target_task = target_task
self._bridge = bridge
self._exclude_frames = exclude_frames or ()
@@ -101,7 +101,7 @@ class BusBridgeProcessor(FrameProcessor, BusSubscriber):
return
# Urgent transport frames pass through directly. They need to
# reach the transport even when no child agent is active yet.
# reach the transport even when no child task is active yet.
if isinstance(frame, _PASSTHROUGH_FRAMES):
await self.push_frame(frame, direction)
return
@@ -113,7 +113,7 @@ class BusBridgeProcessor(FrameProcessor, BusSubscriber):
# Send to bus
msg = BusFrameMessage(
source=self._agent_name,
source=self._task_name,
frame=frame,
direction=direction,
bridge=self._bridge,
@@ -130,19 +130,19 @@ class BusBridgeProcessor(FrameProcessor, BusSubscriber):
return
# Skip own frames
if message.source == self._agent_name:
if message.source == self._task_name:
return
# Filter by bridge name
if self._bridge and message.bridge != self._bridge:
return
# If target_agent set, only accept from that agent
if self._target_agent and message.source != self._target_agent:
# If target_task set, only accept from that task
if self._target_task and message.source != self._target_task:
return
# If message targeted at someone else, skip
if message.target and message.target != self._agent_name:
if message.target and message.target != self._task_name:
return
await self.push_frame(message.frame, message.direction)

4
src/pipecat/bus/bus.py
View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Abstract agent bus for inter-agent pub/sub messaging.
"""Abstract task bus for inter-task pub/sub messaging.
Provides the abstract `TaskBus` base class. Concrete implementations
(e.g. `AsyncQueueBus`) live in separate modules.
@@ -45,7 +45,7 @@ class BusSubscription:
class TaskBus(BaseObject):
"""Abstract base for inter-agent and runner-agent communication.
"""Abstract base for inter-task and runner-task communication.
Provides pub/sub messaging where each subscriber receives messages
independently through its own priority queue. System messages

2
src/pipecat/bus/local/async_queue.py
View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""In-process agent bus backed by asyncio queues."""
"""In-process task bus backed by asyncio queues."""
from loguru import logger

116
src/pipecat/bus/messages.py
View File

@@ -4,10 +4,10 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Bus message types for inter-agent communication.
"""Bus message types for inter-task communication.
Defines the message hierarchy used by the `TaskBus` for pub/sub messaging
between agents, the session, and the runner.
between tasks, the session, and the runner.
"""
from __future__ import annotations
@@ -31,7 +31,7 @@ if TYPE_CHECKING:
class BusMessage:
"""Mixin carrying source/target metadata for bus messages.
Not a frame itself. Combined with ``DataFrame`` or ``SystemFrame``
Not a frame itself. Combined with `DataFrame` or `SystemFrame`
to create concrete message types with appropriate priority.
"""
@@ -53,8 +53,8 @@ class BusDataMessage(BusMessage, DataFrame):
"""Normal-priority bus message.
Parameters:
source: Name of the agent or component that sent this message.
target: Name of the intended recipient agent, or None for broadcast.
source: Name of the task or component that sent this message.
target: Name of the intended recipient task, or None for broadcast.
"""
source: str
@@ -66,8 +66,8 @@ class BusSystemMessage(BusMessage, SystemFrame):
"""High-priority bus message that preempts normal messages in subscriber queues.
Parameters:
source: Name of the agent or component that sent this message.
target: Name of the intended recipient agent, or None for broadcast.
source: Name of the task or component that sent this message.
target: Name of the intended recipient task, or None for broadcast.
"""
source: str
@@ -95,16 +95,16 @@ class BusFrameMessage(BusDataMessage):
# ---------------------------------------------------------------------------
# Agent lifecycle
# Task lifecycle
# ---------------------------------------------------------------------------
@dataclass
class BusActivateTaskMessage(BusDataMessage):
"""Tells a targeted agent to become active and start processing.
"""Tells a targeted task to become active and start processing.
Parameters:
args: Optional activation arguments forwarded to ``on_activated``.
args: Optional activation arguments forwarded to `on_activated`.
"""
args: dict | None = None
@@ -112,7 +112,7 @@ class BusActivateTaskMessage(BusDataMessage):
@dataclass
class BusDeactivateTaskMessage(BusDataMessage):
"""Tells a targeted agent to become inactive and stop processing."""
"""Tells a targeted task to become inactive and stop processing."""
pass
@@ -121,8 +121,8 @@ class BusDeactivateTaskMessage(BusDataMessage):
class BusEndMessage(BusDataMessage):
"""Request a graceful end of the session.
Sent by an agent to the runner, which responds by sending
`BusEndTaskMessage` to each agent.
Sent by a task to the runner, which responds by sending
`BusEndTaskMessage` to each task.
Parameters:
reason: Optional human-readable reason for ending.
@@ -133,9 +133,9 @@ class BusEndMessage(BusDataMessage):
@dataclass
class BusEndTaskMessage(BusDataMessage):
"""Tells a targeted agent to end its pipeline gracefully.
"""Tells a targeted task to end its pipeline gracefully.
Sent by the runner to individual agents during shutdown.
Sent by the runner to individual tasks during shutdown.
Parameters:
reason: Optional human-readable reason for ending.
@@ -148,8 +148,8 @@ class BusEndTaskMessage(BusDataMessage):
class BusCancelMessage(BusSystemMessage):
"""Request a hard cancel of the session.
Sent by an agent to the runner, which responds by sending
`BusCancelTaskMessage` to each agent.
Sent by a task to the runner, which responds by sending
`BusCancelTaskMessage` to each task.
Parameters:
reason: Optional human-readable reason for the cancellation.
@@ -160,9 +160,9 @@ class BusCancelMessage(BusSystemMessage):
@dataclass
class BusCancelTaskMessage(BusSystemMessage):
"""Tells a targeted agent to cancel its pipeline task.
"""Tells a targeted task to cancel its pipeline.
Sent by the runner to individual agents during cancellation.
Sent by the runner to individual tasks during cancellation.
Parameters:
reason: Optional human-readable reason for the cancellation.
@@ -172,7 +172,7 @@ class BusCancelTaskMessage(BusSystemMessage):
# ---------------------------------------------------------------------------
# Agent registry and errors
# Task registry and errors
# ---------------------------------------------------------------------------
@@ -208,18 +208,18 @@ class BusTaskRegistryMessage(BusSystemMessage):
@dataclass
class BusTaskReadyMessage(BusDataMessage):
"""Announces that an agent is ready.
"""Announces that a task is ready.
Sent when any agent (root or child) becomes ready. Carries the
agent's parent name so observers can reconstruct the full hierarchy.
Sent when any task (root or child) becomes ready. Carries the
task's parent name so observers can reconstruct the full hierarchy.
Parameters:
runner: Name of the runner managing this agent.
parent: Name of the parent agent, or None for root agents.
active: Whether the agent started active.
bridged: Whether the agent is bridged (receives pipeline frames
runner: Name of the runner managing this task.
parent: Name of the parent task, or None for root tasks.
active: Whether the task started active.
bridged: Whether the task is bridged (receives pipeline frames
from the bus).
started_at: Unix timestamp when the agent became ready.
started_at: Unix timestamp when the task became ready.
"""
runner: str
@@ -231,10 +231,10 @@ class BusTaskReadyMessage(BusDataMessage):
@dataclass
class BusTaskErrorMessage(BusSystemMessage):
"""Reports an error from a root agent.
"""Reports an error from a root task.
Sent over the network so remote agents can react. For child agent
errors, see ``BusTaskLocalErrorMessage``.
Sent over the network so remote tasks can react. For child task
errors, see `BusTaskLocalErrorMessage`.
Parameters:
error: Description of the error.
@@ -245,10 +245,10 @@ class BusTaskErrorMessage(BusSystemMessage):
@dataclass
class BusTaskLocalErrorMessage(BusSystemMessage, BusLocalMessage):
"""Reports an error from a child agent to its parent.
"""Reports an error from a child task to its parent.
Local-only: never crosses the network. The parent receives it
via ``on_task_failed()``.
via `on_task_failed()`.
Parameters:
error: Description of the error.
@@ -258,17 +258,17 @@ class BusTaskLocalErrorMessage(BusSystemMessage, BusLocalMessage):
# ---------------------------------------------------------------------------
# Tasks
# Jobs
# ---------------------------------------------------------------------------
@dataclass
class BusJobRequestMessage(BusDataMessage):
"""Requests a task agent to start work.
"""Requests a worker task to start work.
Parameters:
job_id: Unique identifier for this task.
job_name: Optional task name for routing to named handlers.
job_id: Unique identifier for this job.
job_name: Optional job name for routing to named `@job` handlers.
payload: Optional structured data describing the work.
"""
@@ -279,10 +279,10 @@ class BusJobRequestMessage(BusDataMessage):
@dataclass
class BusJobResponseMessage(BusDataMessage):
"""Response from a task agent when it completes.
"""Response from a worker task when its job completes.
Parameters:
job_id: The task identifier.
job_id: The job identifier.
status: Completion status.
response: Optional result data.
"""
@@ -294,13 +294,13 @@ class BusJobResponseMessage(BusDataMessage):
@dataclass
class BusJobResponseUrgentMessage(BusSystemMessage):
"""High-priority response from a task agent.
"""High-priority job response.
Same semantics as ``BusJobResponseMessage`` but delivered with
Same semantics as `BusJobResponseMessage` but delivered with
system priority, preempting queued data messages.
Parameters:
job_id: The task identifier.
job_id: The job identifier.
status: Completion status.
response: Optional result data.
"""
@@ -312,10 +312,10 @@ class BusJobResponseUrgentMessage(BusSystemMessage):
@dataclass
class BusJobUpdateMessage(BusDataMessage):
"""Progress update from a task agent.
"""Progress update from a worker task.
Parameters:
job_id: The task identifier.
job_id: The job identifier.
update: Optional progress data.
"""
@@ -325,13 +325,13 @@ class BusJobUpdateMessage(BusDataMessage):
@dataclass
class BusJobUpdateUrgentMessage(BusSystemMessage):
"""High-priority progress update from a task agent.
"""High-priority job progress update.
Same semantics as ``BusJobUpdateMessage`` but delivered with
Same semantics as `BusJobUpdateMessage` but delivered with
system priority, preempting queued data messages.
Parameters:
job_id: The task identifier.
job_id: The job identifier.
update: Optional progress data.
"""
@@ -341,10 +341,10 @@ class BusJobUpdateUrgentMessage(BusSystemMessage):
@dataclass
class BusJobUpdateRequestMessage(BusDataMessage):
"""Request a progress update from a task agent.
"""Request a progress update from a worker task.
Parameters:
job_id: The task identifier.
job_id: The job identifier.
"""
job_id: str
@@ -352,10 +352,10 @@ class BusJobUpdateRequestMessage(BusDataMessage):
@dataclass
class BusJobCancelMessage(BusSystemMessage):
"""Cancel a running task.
"""Cancel a running job.
Parameters:
job_id: The task identifier.
job_id: The job identifier.
reason: Optional human-readable reason for cancellation.
"""
@@ -364,16 +364,16 @@ class BusJobCancelMessage(BusSystemMessage):
# ---------------------------------------------------------------------------
# Task streaming
# Job streaming
# ---------------------------------------------------------------------------
@dataclass
class BusJobStreamStartMessage(BusDataMessage):
"""Signals the start of a streaming task response.
"""Signals the start of a streaming job response.
Parameters:
job_id: The task identifier.
job_id: The job identifier.
data: Optional metadata (e.g. content type).
"""
@@ -383,10 +383,10 @@ class BusJobStreamStartMessage(BusDataMessage):
@dataclass
class BusJobStreamDataMessage(BusDataMessage):
"""A chunk of streaming task data.
"""A chunk of streaming job data.
Parameters:
job_id: The task identifier.
job_id: The job identifier.
data: The chunk payload.
"""
@@ -396,10 +396,10 @@ class BusJobStreamDataMessage(BusDataMessage):
@dataclass
class BusJobStreamEndMessage(BusDataMessage):
"""Signals the end of a streaming task response.
"""Signals the end of a streaming job response.
Parameters:
job_id: The task identifier.
job_id: The job identifier.
data: Optional final metadata.
"""

2
src/pipecat/bus/network/__init__.py
View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Network bus implementations for distributed agents.
"""Network bus implementations for distributed tasks.
Each adapter has its own optional dependency. Imports are lazy so the
package can be loaded with only the extras you need; importing a specific

8
src/pipecat/bus/network/pgmq.py
View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""PGMQ (PostgreSQL Message Queue) agent bus for distributed agents."""
"""PGMQ (PostgreSQL Message Queue) task bus for distributed tasks."""
import asyncio
import json
@@ -23,7 +23,7 @@ try:
from pgmq.async_queue import PGMQueue
except ModuleNotFoundError as e: # pragma: no cover - exercised only when extra is missing
logger.error(f"Exception: {e}")
logger.error("In order to use PgmqBus, you need to `pip install pipecat-ai-subagents[pgmq]`.")
logger.error("In order to use PgmqBus, you need to `pip install pipecat-ai[pgmq]`.")
raise Exception(f"Missing module: {e}")
@@ -54,7 +54,7 @@ def _sanitize_channel(channel: str) -> str:
class PgmqBus(TaskBus):
"""Distributed agent bus backed by PGMQ (PostgreSQL Message Queue).
"""Distributed task bus backed by PGMQ (PostgreSQL Message Queue).
Implements pub/sub fan-out on top of PGMQ's point-to-point queue semantics
by giving each ``PgmqBus`` instance its own queue and broadcasting on
@@ -65,7 +65,7 @@ class PgmqBus(TaskBus):
local subscribers.
Requires the ``pgmq`` and ``asyncpg`` packages. Install with
``pip install pipecat-ai-subagents[pgmq]``.
``pip install pipecat-ai[pgmq]``.
The provided ``PGMQueue`` must already have its connection pool
initialized via ``await pgmq.init()`` before being passed. The adapter

6
src/pipecat/bus/network/redis.py
View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Redis pub/sub agent bus for distributed agents."""
"""Redis pub/sub task bus for distributed tasks."""
import asyncio
@@ -20,12 +20,12 @@ try:
from redis.asyncio.client import PubSub
except ModuleNotFoundError as e:
logger.error(f"Exception: {e}")
logger.error("In order to use RedisBus, you need to `pip install pipecat-ai-subagents[redis]`.")
logger.error("In order to use RedisBus, you need to `pip install pipecat-ai[redis]`.")
raise Exception(f"Missing module: {e}")
class RedisBus(TaskBus):
"""Distributed agent bus backed by Redis pub/sub.
"""Distributed task bus backed by Redis pub/sub.
Publishes serialized messages to a Redis channel for cross-process
communication. ``BusLocalMessage`` messages bypass Redis and are

36
src/pipecat/pipeline/job_context.py
View File

@@ -50,10 +50,10 @@ class JobGroupError(Exception):
@dataclass
class JobGroupResponse:
"""Collected results from a completed task group.
"""Collected results from a completed job group.
Parameters:
job_id: The shared task identifier.
job_id: The shared job identifier.
responses: Collected responses keyed by worker name.
"""
@@ -81,7 +81,7 @@ class JobEvent:
@dataclass
class JobGroupEvent:
"""An event received from a worker during task group execution.
"""An event received from a worker during job group execution.
Parameters:
type: The event type.
@@ -156,16 +156,16 @@ class JobGroup:
class JobGroupContext:
"""Async context manager and iterator for structured task group execution.
"""Async context manager and iterator for structured job group execution.
Sends task requests on enter, waits for all responses on exit.
Sends job requests on enter, waits for all responses on exit.
Supports ``async for`` to receive intermediate events (updates
and streaming data) from workers while waiting for completion.
On normal completion, results are available via ``responses``.
On worker error (with ``cancel_on_error=True``) or timeout, raises
``JobGroupError``. If the ``async with`` block raises, remaining
tasks are cancelled.
jobs are cancelled.
Example::
@@ -192,10 +192,10 @@ class JobGroupContext:
Args:
task: The parent `BaseTask` that owns this job group.
task_names: Names of the workers to send the job to.
name: Optional task name for routing to named handlers.
name: Optional job name for routing to named ``@job`` handlers.
payload: Optional structured data describing the work.
timeout: Optional timeout in seconds covering both the
ready-wait and task execution.
ready-wait and job execution.
cancel_on_error: Whether to cancel the group if a worker
errors. Defaults to True.
"""
@@ -209,16 +209,16 @@ class JobGroupContext:
@property
def job_id(self) -> str:
"""The shared task identifier for this group."""
"""The shared job identifier for this group."""
if not self._group:
raise RuntimeError("Task group has not been started")
raise RuntimeError("Job group has not been started")
return self._group.job_id
@property
def responses(self) -> dict[str, dict]:
"""Collected responses keyed by worker name."""
if not self._group:
raise RuntimeError("Task group has not been started")
raise RuntimeError("Job group has not been started")
return self._group.responses
def __aiter__(self):
@@ -263,13 +263,13 @@ class JobGroupContext:
class JobContext:
"""Async context manager and iterator for a single-worker job.
Sends a task request on enter, waits for the response on exit.
Sends a job request on enter, waits for the response on exit.
Supports ``async for`` to receive intermediate events (updates
and streaming data) from the worker while waiting for completion.
On normal completion, the result is available via ``response``.
On worker error or timeout, raises ``JobError``. If the
``async with`` block raises, the task is cancelled.
``async with`` block raises, the job is cancelled.
Example::
@@ -294,10 +294,10 @@ class JobContext:
Args:
task: The parent `BaseTask` that owns this job.
task_name: Name of the worker to send the job to.
name: Optional task name for routing to a named handler.
name: Optional job name for routing to a named ``@job`` handler.
payload: Optional structured data describing the work.
timeout: Optional timeout in seconds covering both the
ready-wait and task execution.
ready-wait and job execution.
"""
self._task = task
self._task_name = task_name
@@ -308,16 +308,16 @@ class JobContext:
@property
def job_id(self) -> str:
"""The task identifier."""
"""The job identifier."""
if not self._group:
raise RuntimeError("Task has not been started")
raise RuntimeError("Job has not been started")
return self._group.job_id
@property
def response(self) -> dict:
"""The worker's response payload."""
if not self._group:
raise RuntimeError("Task has not been started")
raise RuntimeError("Job has not been started")
return self._group.responses.get(self._task_name, {})
def __aiter__(self):

6
src/pipecat/pipeline/runner.py
View File

@@ -328,14 +328,12 @@ class PipelineRunner(BaseObject, BusSubscriber):
await asyncio.gather(*remaining, return_exceptions=True)
async def _load_setup_files(self) -> None:
"""Load setup files from ``PIPECAT_SUBAGENTS_SETUP_FILES``.
"""Load setup files from ``PIPECAT_RUNNER_SETUP_FILES``.
Each file should contain an async ``setup_runner(runner)`` function
that receives the runner instance.
"""
setup_files = [
f for f in os.environ.get("PIPECAT_SUBAGENTS_SETUP_FILES", "").split(":") if f
]
setup_files = [f for f in os.environ.get("PIPECAT_RUNNER_SETUP_FILES", "").split(":") if f]
for f in setup_files:
try:
path = Path(f).resolve()

2
src/pipecat/pipeline/task.py
View File

@@ -266,7 +266,7 @@ class PipelineTask(BaseTask):
idle_timeout_secs: Timeout (in seconds) to consider pipeline idle or
None. If a pipeline is idle the pipeline task will be cancelled
automatically.
name: Optional task name (used for agent-style addressing on the bus).
name: Optional task name (used for task-style addressing on the bus).
observers: List of observers for monitoring pipeline execution.
params: Configuration parameters for the pipeline.
rtvi_observer_params: The RTVI observer parameter to use if RTVI is enabled.

32
src/pipecat/pipeline/task_ready_decorator.py
View File

@@ -4,16 +4,16 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Decorator for marking agent methods as agent-ready handlers."""
"""Decorator for marking methods as task-ready handlers."""
def task_ready(*, name: str):
"""Mark a method as a handler for a specific agent becoming ready.
"""Mark a method as a handler for a specific task becoming ready.
Decorated methods are automatically collected by ``BaseTask`` at
initialization. When ``on_ready`` fires, the agent calls
``watch_task`` for each decorated handler. When the watched agent
registers, the decorated method is called with the ready data.
Decorated methods are automatically collected by `BaseTask` at
initialization. When the task starts, it calls `watch_task` for
each decorated handler. When the watched task registers, the
decorated method is called with the ready data.
Example::
@@ -22,11 +22,11 @@ def task_ready(*, name: str):
await self.activate_task("greeter", args=...)
Args:
name: The name of the agent to watch.
name: The name of the task to watch.
"""
def decorator(fn):
fn.agent_ready_name = name
fn.task_ready_name = name
return fn
return decorator
@@ -35,10 +35,10 @@ def task_ready(*, name: str):
def _collect_task_ready_handlers(obj) -> dict:
"""Collect all ``@task_ready`` decorated bound methods from an object.
Returns a dict mapping agent name to the bound method.
Returns a dict mapping task name to the bound method.
Raises:
ValueError: If two handlers watch the same agent name.
ValueError: If two handlers watch the same task name.
"""
seen: set[str] = set()
handlers: dict[str, object] = {}
@@ -47,13 +47,13 @@ def _collect_task_ready_handlers(obj) -> dict:
if attr_name in seen:
continue
seen.add(attr_name)
if callable(val) and hasattr(val, "agent_ready_name"):
agent_name = val.agent_ready_name
if agent_name in handlers:
existing = handlers[agent_name].__name__
if callable(val) and hasattr(val, "task_ready_name"):
task_name = val.task_ready_name
if task_name in handlers:
existing = handlers[task_name].__name__
raise ValueError(
f"Duplicate @task_ready handler for '{agent_name}': "
f"Duplicate @task_ready handler for '{task_name}': "
f"'{attr_name}' conflicts with '{existing}'"
)
handlers[agent_name] = getattr(obj, attr_name)
handlers[task_name] = getattr(obj, attr_name)
return handlers

2
src/pipecat/registry/__init__.py
View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Agent registry for tracking discovered agents across runners."""
"""Task registry for tracking known tasks across runners."""
from pipecat.registry.registry import TaskRegistry

2
src/pipecat/tasks/llm/__init__.py
View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""LLM agent and tool decorator."""
"""LLM task package -- `LLMTask`, `LLMContextTask`, and the `@tool` decorator."""
from pipecat.tasks.llm.llm_context_task import LLMContextTask
from pipecat.tasks.llm.llm_task import LLMTask, LLMTaskActivationArgs

6
src/pipecat/tasks/llm/tool_decorator.py
View File

@@ -4,7 +4,7 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Decorator for marking agent methods as tools."""
"""Decorator for marking methods as LLM tools."""
def tool(fn=None, *, cancel_on_interruption=True, timeout=None):
@@ -34,7 +34,7 @@ def tool(fn=None, *, cancel_on_interruption=True, timeout=None):
"""
def decorator(fn):
fn.is_agent_tool = True
fn.is_llm_tool = True
fn.cancel_on_interruption = cancel_on_interruption
fn.timeout = timeout
return fn
@@ -57,6 +57,6 @@ def _collect_tools(obj) -> list:
if name in seen:
continue
seen.add(name)
if callable(val) and getattr(val, "is_agent_tool", False):
if callable(val) and getattr(val, "is_llm_tool", False):
tools.append(getattr(obj, name))
return tools