diff --git a/src/pipecat/bus/__init__.py b/src/pipecat/bus/__init__.py index fbd24354a..513f9f2a1 100644 --- a/src/pipecat/bus/__init__.py +++ b/src/pipecat/bus/__init__.py @@ -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. """ diff --git a/src/pipecat/bus/bridge_processor.py b/src/pipecat/bus/bridge_processor.py index bb6af249d..b296cae3c 100644 --- a/src/pipecat/bus/bridge_processor.py +++ b/src/pipecat/bus/bridge_processor.py @@ -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) diff --git a/src/pipecat/bus/bus.py b/src/pipecat/bus/bus.py index e81fd0abd..f8fec267b 100644 --- a/src/pipecat/bus/bus.py +++ b/src/pipecat/bus/bus.py @@ -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 diff --git a/src/pipecat/bus/local/async_queue.py b/src/pipecat/bus/local/async_queue.py index 2ef82cef7..bc3cbd3a8 100644 --- a/src/pipecat/bus/local/async_queue.py +++ b/src/pipecat/bus/local/async_queue.py @@ -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 diff --git a/src/pipecat/bus/messages.py b/src/pipecat/bus/messages.py index 8a6f27827..973819644 100644 --- a/src/pipecat/bus/messages.py +++ b/src/pipecat/bus/messages.py @@ -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. """ diff --git a/src/pipecat/bus/network/__init__.py b/src/pipecat/bus/network/__init__.py index 8f42393bc..2146207e3 100644 --- a/src/pipecat/bus/network/__init__.py +++ b/src/pipecat/bus/network/__init__.py @@ -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 diff --git a/src/pipecat/bus/network/pgmq.py b/src/pipecat/bus/network/pgmq.py index 88df8d5d6..6a7dba637 100644 --- a/src/pipecat/bus/network/pgmq.py +++ b/src/pipecat/bus/network/pgmq.py @@ -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 diff --git a/src/pipecat/bus/network/redis.py b/src/pipecat/bus/network/redis.py index c89a9bee0..784b2b5e0 100644 --- a/src/pipecat/bus/network/redis.py +++ b/src/pipecat/bus/network/redis.py @@ -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 diff --git a/src/pipecat/pipeline/job_context.py b/src/pipecat/pipeline/job_context.py index 0767b127a..4cc42b1e3 100644 --- a/src/pipecat/pipeline/job_context.py +++ b/src/pipecat/pipeline/job_context.py @@ -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): diff --git a/src/pipecat/pipeline/runner.py b/src/pipecat/pipeline/runner.py index 44070f255..2d7c98c98 100644 --- a/src/pipecat/pipeline/runner.py +++ b/src/pipecat/pipeline/runner.py @@ -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() diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index e43f10840..985c5527c 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -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. diff --git a/src/pipecat/pipeline/task_ready_decorator.py b/src/pipecat/pipeline/task_ready_decorator.py index ee773d293..982652a61 100644 --- a/src/pipecat/pipeline/task_ready_decorator.py +++ b/src/pipecat/pipeline/task_ready_decorator.py @@ -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 diff --git a/src/pipecat/registry/__init__.py b/src/pipecat/registry/__init__.py index 118bdf128..083513eab 100644 --- a/src/pipecat/registry/__init__.py +++ b/src/pipecat/registry/__init__.py @@ -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 diff --git a/src/pipecat/tasks/llm/__init__.py b/src/pipecat/tasks/llm/__init__.py index dc6304c09..6383557b5 100644 --- a/src/pipecat/tasks/llm/__init__.py +++ b/src/pipecat/tasks/llm/__init__.py @@ -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 diff --git a/src/pipecat/tasks/llm/tool_decorator.py b/src/pipecat/tasks/llm/tool_decorator.py index 6cd9f7076..69f9827fe 100644 --- a/src/pipecat/tasks/llm/tool_decorator.py +++ b/src/pipecat/tasks/llm/tool_decorator.py @@ -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