Add TwoLayerLLMService: a roll-your-own two-layer bot with any frontend

changelog/5689.added.md

@@ -0,0 +1 @@
+- Added `TwoLayerLLMService` (`pipecat.pipeline.two_layer_llm_service`), which makes any LLM service, text or speech-to-speech, the conversational frontend of a two-layer bot, with a `BackendLLMWorker` as the backend that does the work it hands off: the split `OpenAILiveLLMService` uses for client delegation, for any frontend. The service drops into a pipeline's LLM slot, installs the `delegate` tool on the frontend and registers a local backend worker itself; a backend given by name may run in another process on a shared bus. A `BackendConnector` sets how delegation works from a `BackendRequestStrategy`, which defines the tool and what it sends (`TranscriptBackendRequestStrategy`: the conversation since the previous delegation; `ExplicitBackendRequestStrategy`: a request the model words itself), and a `BackendReplyStrategy`, which sets what the frontend hears of the backend's progress (`StrictSpeechFlagBackendReplyStrategy`: spoken as the backend's `prefers_spoken` flag says; `AdvisorySpeechFlagBackendReplyStrategy`: the flag passed on for the frontend to weigh, with a silence marker it can answer with; `FinalOnlyBackendReplyStrategy`: the answer only). Defaults are picked by frontend kind. See `examples/multi-worker/two-layer/`.

changelog/5689.changed.md

@@ -0,0 +1 @@
+- `BackendLLMWorker` now appends output-style guidance to its LLM's system instruction (plain spoken text a voice assistant can relay, no Markdown, no claiming an action completed without a tool result), so a backend prompt need only say what the backend does.

changelog/5689.fixed.2.md

@@ -0,0 +1 @@
+- Fixed a `PipelineWorker` whose pipeline ended on its own (an `EndFrame` it queued, an idle timeout, a fatal error) leaving its child workers running, which kept the `WorkerRunner` from exiting. The worker now cancels its children as it finishes.

changelog/5689.fixed.md

@@ -0,0 +1 @@
+- Tweaked the async-tool guidance given to LLMs (`ASYNC_TOOL_INSTRUCTIONS` and the final-result message) so a result that lands after the model has already answered the user is more reliably delivered on its own, without the model repeating its previous reply first.

examples/multi-worker/README.md

@@ -51,6 +51,7 @@ A Pipecat **worker** is a unit of work attached to a shared bus. Workers exchang
 | [`code-assistant/`](code-assistant/)          | Voice access to your codebase via a Claude Agent SDK worker behind `job(...)`.    |
 | [`sensor-controller/`](sensor-controller/)    | Voice agent forwards questions to a sidecar `PipelineWorker` owning a simulated sensor. |
 | [`openclaw-agent/`](openclaw-agent/)          | Voice loop stays responsive while an OpenClaw agent works; steer or stop it mid-task. |
+| [`two-layer/`](two-layer/)                    | A fast conversational frontend (cascade or realtime) delegating to a tool-using backend via `TwoLayerLLMService`. |
 
 ### Distributed (separate processes, network bus)
 

examples/multi-worker/two-layer/README.md

@@ -0,0 +1,78 @@
+# Two-layer LLM
+
+A conversational **frontend** holds the conversation with a fast model and no tools of its own. Anything that needs tools, current information or careful reasoning it hands to a **backend**: a `BackendLLMWorker` running a heavier model with the tools. `TwoLayerLLMService` wraps the frontend so the pair drops into a pipeline where an LLM goes, installs the `delegate` tool that joins them, and runs the backend as a worker of its own.
+
+```python
+llm = TwoLayerLLMService(
+    frontend=OpenAILLMService(...),
+    backend=BackendLLMWorker(
+        llm=AnthropicLLMService(...),
+        context=LLMContext(tools=[get_current_weather, get_restaurant_recommendation]),
+    ),
+)
+```
+
+How a delegation crosses is the `BackendConnector`'s business, built from two strategies the service picks by frontend kind unless told otherwise:
+
+|                    | request (frontend → backend)                                        | reply (backend → frontend)                                   |
+| ------------------ | ------------------------------------------------------------------- | ------------------------------------------------------------ |
+| text frontend      | `TranscriptBackendRequestStrategy`: the conversation since the previous delegation | `StrictSpeechFlagBackendReplyStrategy`: progress relayed, spoken as the backend's flag says |
+| realtime frontend  | `ExplicitBackendRequestStrategy`: a request the model words itself   | `FinalOnlyBackendReplyStrategy`: the answer only             |
+
+## Examples
+
+| Example                                            | What it shows                                                                                          |
+| -------------------------------------------------- | ------------------------------------------------------------------------------------------------------ |
+| [`cascade-frontend.py`](cascade-frontend.py)       | A cascade pipeline (STT + GPT + TTS) as the frontend, all defaults.                                    |
+| [`realtime-frontend.py`](realtime-frontend.py)     | OpenAI Realtime as the frontend, all defaults. Same backend, same prompts as the cascade example.       |
+| [`advisory-speech-flag.py`](advisory-speech-flag.py) | `AdvisorySpeechFlagBackendReplyStrategy`: the frontend hears every piece of progress and decides what to say. |
+
+Run any of them the usual way, then connect a client:
+
+```bash
+python cascade-frontend.py
+```
+
+Or drive one with a behavioral eval:
+
+```bash
+python cascade-frontend.py -t eval --port 7860
+pipecat eval run ../../../scripts/release-evals/scenarios/scripted/weather_function_call_audio.yaml --bot-url ws://localhost:7860 -v
+```
+
+## A backend in another process
+
+The service addresses the backend by name over the bus, so it need not run in the same process. Give `TwoLayerLLMService` the worker's name instead of the worker, and run the backend under its own `WorkerRunner` on a shared network bus. See [`distributed-handoff`](../distributed-handoff/) for the bus setup.
+
+Backend process:
+
+```python
+bus = RedisBus(redis=Redis.from_url(REDIS_URL), channel="pipecat:two-layer")
+
+backend = BackendLLMWorker(
+    name="backend",
+    llm=AnthropicLLMService(...),
+    context=LLMContext(tools=[get_current_weather, get_restaurant_recommendation]),
+)
+
+runner = WorkerRunner(bus=bus, handle_sigint=True)
+await runner.add_workers(backend)
+await runner.run()
+```
+
+Frontend process:
+
+```python
+bus = RedisBus(redis=Redis.from_url(REDIS_URL), channel="pipecat:two-layer")
+
+llm = TwoLayerLLMService(
+    frontend=OpenAILLMService(...),
+    backend="backend",  # registered in the other process
+)
+
+runner = WorkerRunner(bus=bus, handle_sigint=runner_args.handle_sigint)
+await runner.add_workers(worker)
+await runner.run()
+```
+
+The first delegation waits for the registry to report the backend ready, so a backend that starts a little late is fine. A `PgmqBus` works the same way.

examples/multi-worker/two-layer/advisory-speech-flag.py

@@ -0,0 +1,217 @@
+#
+# Copyright (c) 2024-2026, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""A two-layer voice agent whose frontend decides what backend progress to speak.
+
+The same cascade frontend and backend as ``cascade-frontend.py``, with one
+change to the connector: ``AdvisorySpeechFlagBackendReplyStrategy``. By
+default the backend's ``prefers_spoken`` flag decides whether the frontend
+speaks a piece of progress. Here the flag is passed on as advice, the
+frontend is run on every piece of progress, and its model decides: it either
+says something, or answers with the silence marker alone and nothing is
+said.
+
+The backend's weather lookup is slow, so there is progress worth deciding
+about. Watch the frontend's log: a response of ``∅`` is a decision to stay
+quiet.
+
+Requirements:
+
+- OPENAI_API_KEY
+- ANTHROPIC_API_KEY
+- DEEPGRAM_API_KEY
+- CARTESIA_API_KEY
+"""
+
+import asyncio
+import os
+from datetime import datetime
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.evals.transport import EvalTransportParams
+from pipecat.frames.frames import LLMRunFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.two_layer_llm_service import (
+    AdvisorySpeechFlagBackendReplyStrategy,
+    BackendConnector,
+    TwoLayerLLMService,
+)
+from pipecat.pipeline.worker import PipelineParams, PipelineWorker, ProcessorUnusablePolicy
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import (
+    LLMContextAggregatorPair,
+    LLMUserAggregatorParams,
+)
+from pipecat.processors.frameworks.rtvi import (
+    RTVIFunctionCallReportLevel,
+    RTVIObserverParams,
+)
+from pipecat.runner.types import RunnerArguments
+from pipecat.runner.utils import create_transport
+from pipecat.services.anthropic.llm import AnthropicLLMService
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.llm_service import FunctionCallParams
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.daily.transport import DailyParams
+from pipecat.workers.llm import BackendLLMWorker
+from pipecat.workers.runner import WorkerRunner
+
+load_dotenv(override=True)
+
+FRONTEND_INSTRUCTIONS = """You are a friendly, concise voice assistant. Your responses are spoken
+aloud, so keep them to one or two natural sentences without any formatting."""
+
+BACKEND_INSTRUCTIONS = """You are the backend of a voice assistant. Use the available tools to
+answer questions about the weather and restaurants. Before a slow lookup, say in a few
+words what you are about to do."""
+
+BACKEND_DESCRIPTION = "current information such as the weather or a restaurant recommendation"
+
+transport_params = {
+    "eval": lambda: EvalTransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "daily": lambda: DailyParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+    "webrtc": lambda: TransportParams(
+        audio_in_enabled=True,
+        audio_out_enabled=True,
+    ),
+}
+
+
+async def get_current_weather(params: FunctionCallParams, location: str, format: str):
+    """Get the current weather. Slow: several seconds.
+
+    Args:
+        location: The city and state, e.g. "San Francisco, CA".
+        format: The temperature unit to use. Must be either "celsius" or "fahrenheit". Infer this from the user's location.
+    """
+    await asyncio.sleep(6)
+    temperature = 75 if format == "fahrenheit" else 24
+    await params.result_callback(
+        {
+            "conditions": "nice",
+            "temperature": temperature,
+            "format": format,
+            "timestamp": datetime.now().strftime("%Y%m%d_%H%M%S"),
+        }
+    )
+
+
+async def get_restaurant_recommendation(params: FunctionCallParams, location: str):
+    """Get a restaurant recommendation.
+
+    Args:
+        location: The city and state, e.g. "San Francisco, CA".
+    """
+    await params.result_callback({"name": "The Golden Dragon"})
+
+
+async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
+    logger.info("Starting bot")
+
+    stt = DeepgramSTTService(api_key=os.environ["DEEPGRAM_API_KEY"])
+    tts = CartesiaTTSService(
+        api_key=os.environ["CARTESIA_API_KEY"],
+        settings=CartesiaTTSService.Settings(
+            voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc",  # Jacqueline
+        ),
+    )
+
+    llm = TwoLayerLLMService(
+        frontend=OpenAILLMService(
+            api_key=os.environ["OPENAI_API_KEY"],
+            settings=OpenAILLMService.Settings(
+                model="gpt-5.4-mini", system_instruction=FRONTEND_INSTRUCTIONS
+            ),
+        ),
+        backend=BackendLLMWorker(
+            name="backend",
+            llm=AnthropicLLMService(
+                api_key=os.environ["ANTHROPIC_API_KEY"],
+                settings=AnthropicLLMService.Settings(system_instruction=BACKEND_INSTRUCTIONS),
+            ),
+            context=LLMContext(tools=[get_current_weather, get_restaurant_recommendation]),
+        ),
+        # The frontend hears every piece of the backend's progress, with the
+        # backend's speak-or-not flag as advice, and decides for itself.
+        connector=BackendConnector(
+            reply=AdvisorySpeechFlagBackendReplyStrategy(),
+            backend_description=BACKEND_DESCRIPTION,
+        ),
+    )
+
+    context = LLMContext()
+    aggregators = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            stt,
+            aggregators.user(),
+            llm,
+            tts,
+            transport.output(),
+            aggregators.assistant(),
+        ]
+    )
+
+    worker = PipelineWorker(
+        pipeline,
+        name="frontend",
+        params=PipelineParams(
+            enable_metrics=True,
+            enable_usage_metrics=True,
+        ),
+        rtvi_observer_params=RTVIObserverParams(
+            function_call_report_level={"delegate": RTVIFunctionCallReportLevel.DISABLED},
+        ),
+        idle_timeout_secs=runner_args.pipeline_idle_timeout_secs,
+        processor_unusable_policy=ProcessorUnusablePolicy.END,
+    )
+
+    runner = WorkerRunner(handle_sigint=runner_args.handle_sigint)
+
+    await runner.add_workers(worker)
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info("Client connected")
+        context.add_message(
+            {"role": "developer", "content": "Greet the user and ask how you can help."}
+        )
+        await worker.queue_frame(LLMRunFrame())
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info("Client disconnected")
+        await runner.cancel()
+
+    await runner.run()
+
+
+async def bot(runner_args: RunnerArguments):
+    """Main bot entry point compatible with Pipecat Cloud."""
+    transport = await create_transport(runner_args, transport_params)
+    await run_bot(transport, runner_args)
+
+
+if __name__ == "__main__":
+    from pipecat.runner.run import main
+
+    main()