Add distributed-handoff examples (redis and pgmq)

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.

examples/multi-task/distributed-handoff/redis-handoff/llm.py

@@ -0,0 +1,153 @@
+#
+# Copyright (c) 2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""LLM worker task — run on Machine B (or locally alongside ``main.py``).
+
+A standalone process that runs one LLM task (greeter or support)
+attached to the same Redis-backed `TaskBus` as the main task.
+Multiple instances can run on different machines.
+
+Usage::
+
+    python llm.py greeter --redis-url redis://localhost:6379
+    python llm.py support --redis-url redis://localhost:6379
+
+Requirements:
+
+- OPENAI_API_KEY
+"""
+
+import argparse
+import asyncio
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+from redis.asyncio import Redis
+
+from pipecat.bus.network.redis import RedisBus
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.services.llm_service import FunctionCallParams
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.tasks.llm import LLMTask, LLMTaskActivationArgs, tool
+
+load_dotenv(override=True)
+
+WORKER_CONFIG = {
+    "greeter": {
+        "system_instruction": (
+            "You are a friendly greeter for Acme Corp. The available products "
+            "are: the Acme Rocket Boots, the Acme Invisible Paint, and the Acme "
+            "Tornado Kit. Ask which one they'd like to learn more about. "
+            "When the user picks a product or asks a question about one, "
+            "immediately call the transfer_to_agent tool with agent 'support'. "
+            "Do not answer product questions yourself. If the user says goodbye, "
+            "call the end_conversation tool. Do not mention transferring — just do it "
+            "seamlessly. Keep responses brief — this is a voice conversation."
+        ),
+        "watch": ["support"],
+    },
+    "support": {
+        "system_instruction": (
+            "You are a support agent for Acme Corp. You know about three "
+            "products: Acme Rocket Boots (jet-powered boots, $299, run up "
+            "to 60 mph), Acme Invisible Paint (makes anything invisible for "
+            "24 hours, $49 per can), and Acme Tornado Kit (portable tornado "
+            "generator, $199, batteries included). Answer the user's questions "
+            "about these products. If the user wants to browse other products "
+            "or start over, call the transfer_to_agent tool with agent "
+            "'greeter'. If the user says goodbye, call the end_conversation "
+            "tool. Do not mention transferring — just do it seamlessly. "
+            "Keep responses brief — this is a voice conversation."
+        ),
+        "watch": ["greeter"],
+    },
+}
+
+
+class AcmeLLMTask(LLMTask):
+    """LLM task for Acme Corp with transfer and end tools."""
+
+    def __init__(self, name: str, *, system_instruction: str, watch: list[str]):
+        """Initialize the AcmeLLMTask.
+
+        Args:
+            name: Unique task name (``"greeter"`` or ``"support"``).
+            system_instruction: System prompt for this LLM role.
+            watch: Sibling task names this task will watch via the
+                registry so it knows when they become available for
+                handoff.
+        """
+        llm = OpenAILLMService(
+            name=f"{name}::OpenAILLMService",
+            api_key=os.environ["OPENAI_API_KEY"],
+            settings=OpenAILLMService.Settings(system_instruction=system_instruction),
+        )
+        super().__init__(name, llm=llm, bridged=())
+        self._watch = watch
+
+    async def start(self) -> None:
+        """Register watches for sibling tasks once ready."""
+        await super().start()
+        for task_name in self._watch:
+            await self.watch_task(task_name)
+
+    @tool(cancel_on_interruption=False)
+    async def transfer_to_agent(self, params: FunctionCallParams, agent: str, reason: str):
+        """Transfer the user to another agent.
+
+        Args:
+            agent (str): The agent to transfer to (e.g. 'greeter', 'support').
+            reason (str): Why the user is being transferred.
+        """
+        logger.info(f"Task '{self.name}': transferring to '{agent}' ({reason})")
+        await self.handoff_to(
+            agent,
+            activation_args=LLMTaskActivationArgs(
+                messages=[{"role": "developer", "content": reason}]
+            ),
+            result_callback=params.result_callback,
+        )
+
+    @tool
+    async def end_conversation(self, params: FunctionCallParams, reason: str):
+        """End the conversation when the user says goodbye.
+
+        Args:
+            reason (str): Why the conversation is ending.
+        """
+        logger.info(f"Task '{self.name}': ending conversation ({reason})")
+        await self.end(
+            reason=reason,
+            messages=[{"role": "developer", "content": reason}],
+            result_callback=params.result_callback,
+        )
+
+
+async def main_async() -> None:
+    parser = argparse.ArgumentParser(description="LLM worker task (greeter or support)")
+    parser.add_argument("worker", choices=list(WORKER_CONFIG), help="Which worker to run")
+    parser.add_argument("--redis-url", default="redis://localhost:6379", help="Redis URL")
+    parser.add_argument("--channel", default="pipecat:acme", help="Redis pub/sub channel")
+    args = parser.parse_args()
+
+    redis = Redis.from_url(args.redis_url)
+    bus = RedisBus(redis=redis, channel=args.channel)
+
+    config = WORKER_CONFIG[args.worker]
+    worker = AcmeLLMTask(
+        args.worker,
+        system_instruction=config["system_instruction"],
+        watch=config["watch"],
+    )
+
+    runner = PipelineRunner(bus=bus, handle_sigint=True)
+    logger.info(f"Starting {args.worker} worker, waiting for activation...")
+    await runner.run(worker)
+
+
+if __name__ == "__main__":
+    asyncio.run(main_async())