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

[kompfner]

Overview

Lets developers roll their own OpenAI-Live-like two-layer architecture: a conversational frontend that holds the conversation with a fast model, plus a smarter-but-slower backend it delegates to for anything needing tools, current information or careful reasoning. OpenAILiveLLMService is this shape with the frontend fixed; this PR makes the shape available with any LLM service, text or speech-to-speech, as the frontend.

llm = TwoLayerLLMService(
    frontend=OpenAILLMService(...),
    backend=BackendLLMWorker(
        llm=AnthropicLLMService(...),
        context=LLMContext(tools=[get_current_weather, get_restaurant_recommendation]),
    ),
)
pipeline = Pipeline([transport.input(), stt, user_agg, llm, tts, transport.output(), assistant_agg])

What's added

• TwoLayerLLMService (pipecat.pipeline.two_layer_llm_service) wraps the frontend so the pair drops into a pipeline's LLM slot, the way LLMSwitcher does. It installs the delegate tool on the frontend (adding it to the context's tools as they pass), appends the connector's guidance to the frontend's system instruction, and registers a local backend worker with the pipeline worker, so the app never wires the backend up. backend may also be a name, for a worker registered elsewhere: the app's own registration, or another process on a RedisBus/PgmqBus.

• BackendConnector sets how a delegation crosses, from two strategy objects with defaults picked by frontend kind:

  	request (frontend → backend)	reply (backend → frontend)
  text frontend	TranscriptBackendRequestStrategy: the conversation since the previous delegation; the tool takes no arguments	StrictSpeechFlagBackendReplyStrategy: progress relayed as intermediate tool results, the frontend run on those the backend's prefers_spoken flag asks for
  realtime frontend	ExplicitBackendRequestStrategy: the model words the request itself (its context can lag the audio); the tool takes request	FinalOnlyBackendReplyStrategy: the answer only, since a realtime function call takes one result

  A third reply strategy, AdvisorySpeechFlagBackendReplyStrategy, passes the flag on as advice: the frontend is run on every piece of progress and decides for itself, answering with a silence marker (∅) alone when it chooses to say nothing, which the service drops so nothing is said. Text frontends only.

• The request strategy owns the tool's interface. It supplies the delegate tool's description and parameters and reads them back in compose_request(), so a custom strategy can declare whatever it needs.

• Prompts shrink to persona. The mechanism prose lives in the strategies (frontend guidance, appended next to the existing async-tool instructions) and in BackendLLMWorker, which appends output-style guidance to its own model. The cascade and realtime examples share their prompts word for word.

• The job update payload names its type, so the stream can carry other kinds later (tool-call metadata for RTVI) without changing what a reader of outputs sees. The worker-side helpers that put a request to the backend and render the transcript stay module-private: a custom strategy works with FunctionCallParams and BackendOutput and never needs them.

• PipelineWorker cancels its children when its pipeline ends on its own. A parent told to end over the bus already passed that on; one that ended by itself (an EndFrame it queued, an idle timeout, a fatal error) never told its children, so they ran on and the runner waited forever. The same gap applied to OpenAILiveLLMService's client-delegation backend.

• Async-tool guidance (pipecat-wide): ASYNC_TOOL_INSTRUCTIONS and the final-result message distinguish a result arriving while the user has an unanswered request from one arriving on a run of its own, so the model delivers just the result in the latter case instead of re-answering first. A minimal edit of the wording from fix(llm): state async tool handling in the system instruction #5278; replaying the motivating context against gpt-5.4-mini: 8/8 repeats with the old wording, 0/16 with the new; fix(llm): state async tool handling in the system instruction #5278's own release-eval sweep passes 159/160.

Usage

All defaults. A text frontend hands the backend the conversation. Every piece of the backend's progress is recorded in the frontend's context as an intermediate tool result, but the frontend is only run on the ones the backend flags prefers_spoken, which by default is just the final answer: the frontend knows what the backend is doing, and speaks when the backend says to. A realtime frontend words the request and takes the answer only. Both examples, cascade-frontend.py and realtime-frontend.py, are this with different frontends:

llm = TwoLayerLLMService(
    frontend=OpenAILLMService(api_key=..., settings=OpenAILLMService.Settings(
        model="gpt-5.4-mini", system_instruction=FRONTEND_INSTRUCTIONS,   # persona only
    )),
    backend=BackendLLMWorker(
        name="backend",
        llm=AnthropicLLMService(api_key=..., settings=AnthropicLLMService.Settings(
            system_instruction=BACKEND_INSTRUCTIONS,                      # role and tools only
        )),
        context=LLMContext(tools=[get_current_weather, get_restaurant_recommendation]),
    ),
    connector=BackendConnector(backend_description="current information such as the weather"),
)

Which outputs ask to be spoken is the backend worker's transform_output to change (from #5688). To have the frontend voice what the backend says on its way to the answer, not only the answer:

async def speak_progress(output: BackendOutput) -> BackendOutput:
    return replace(output, prefers_spoken=not output.is_thought)

backend=BackendLLMWorker(..., transform_output=speak_progress)

The frontend decides what progress to speak (advisory-speech-flag.py). Progress is recorded the same way, but the frontend is run on every piece of it, with the backend's prefers_spoken flag passed along as advice; its model speaks or answers with the silence marker:

connector=BackendConnector(
    reply=AdvisorySpeechFlagBackendReplyStrategy(),
    backend_description="current information such as the weather",
)

A custom request strategy, declaring its own tool parameters:

class UrgentRequest(BackendRequestStrategy):
    tool_parameters = {
        "request": {"type": "string", "description": "The request, self-contained."},
        "urgency": {"type": "string", "enum": ["now", "whenever"]},
    }
    tool_required = ["request"]

    def tool_description(self, backend_description: str) -> str:
        return f"Hand a request to the backend, for {backend_description}."

    async def compose_request(self, params: FunctionCallParams) -> str:
        urgency = params.arguments.get("urgency", "whenever")
        return f"[{urgency}] {params.arguments['request']}"

connector=BackendConnector(request=UrgentRequest())

A backend in another process: give the service the worker's name and run the backend under its own WorkerRunner on a shared bus (see the examples' README).

llm = TwoLayerLLMService(frontend=..., backend="backend")

Notes

Realtime frontends take the answer only. A realtime function call accepts one result, so nothing mid-delegation (progress, or a future speak_to_user line) reaches a realtime frontend. BackendConnector refuses a reply strategy that streams with a realtime frontend at construction.

The handoff is kept out of RTVI tool-call events. The examples pass function_call_report_level={"delegate": DISABLED}. Reporting it stays the app's decision. The backend's own calls run in the backend worker's pipeline and are not reported; see #5688's compatibility analysis, to which this adds the roll-your-own shape.

Not in this PR: refactoring OpenAILiveLLMService's own client delegation onto BackendConnector. Live's delegation is driven by the model's session.delegation.created event, not a tool call; it shares the worker, the job contract and the transcript rendering with this.

Validation

• uv run pytest tests/test_two_layer_llm_service.py tests/test_backend_llm_worker.py tests/test_openai_live_service.py tests/test_base_worker.py tests/test_runner.py tests/test_ui_worker.py tests/test_pipeline.py — 252 pass. The new file covers the connector defaults by frontend kind, both request strategies, all three reply strategies, the silence filter, tool advertisement on context and tool-change frames with the frontend's own direct functions still registered, and the whole path end to end: a scripted backend worker added by the service itself, called through the delegate tool, answering as tool results.
• uv run pytest tests/test_async_tool_messages.py — 24 pass.
• ruff and pyright clean, examples included.
• Behavioral evals through the eval harness (Kokoro/Moonshine, gemma4:12b judge), one session each:
  • cascade example, live_joke_while_waiting: 3/3 turns. The frontend tells the joke itself while the backend works, then relays the weather.
  • advisory example, live_joke_while_waiting: 3/3 turns; the frontend used the run on the backend's first progress to tell the joke.
  • realtime example, a weather scenario with no function-call expectation: 2/2 turns; the model worded the request, the backend answered, the answer was relayed. The joke scenario's joke turn fails on this frontend: OpenAI Realtime acknowledges and waits while its function call is pending. That is the realtime model's behavior with a pending call, not the delegation path.
  • The release-eval weather_function_call_audio scenario cannot pass for a two-layer bot as written: it expects to observe get_current_weather, which runs in the backend worker's pipeline. The bot's answer in that run was correct ("75 degrees Fahrenheit with nice conditions").

Follow-ups

• Report the backend's tool calls over RTVI — see Add OpenAILiveLLMService (OpenAI Live API, gpt-live-1) with both delegation modes #5688. The type field on job updates and the service, which lives in the main pipeline where RTVIObserver looks, are the hooks.
• A speak_to_user-style tool on the backend — see Add OpenAILiveLLMService (OpenAI Live API, gpt-live-1) with both delegation modes #5688. Any reply strategy already handles one (it is a BackendOutput(prefers_spoken=True) on the same channel); the follow-up folds it into the worker so it passes transform_output.
• An eval scenario shape for two-layer bots in scripts/release-evals, judging the relayed answer rather than expecting the frontend's function call.

[codecov]

Codecov Report

❌ Patch coverage is 96.18644% with 9 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
src/pipecat/pipeline/two_layer_llm_service.py	96.34%	8 Missing ⚠️
src/pipecat/workers/llm/backend_llm_worker.py	87.50%	1 Missing ⚠️

Files with missing lines	Coverage Δ
src/pipecat/pipeline/worker.py	93.69% <100.00%> (+0.04%)	⬆️
...ecat/processors/aggregators/async_tool_messages.py	97.05% <100.00%> (ø)
src/pipecat/services/openai/live/llm.py	88.28% <100.00%> (-0.02%)	⬇️
src/pipecat/workers/llm/backend_llm_worker.py	95.80% <87.50%> (-0.63%)	⬇️
src/pipecat/pipeline/two_layer_llm_service.py	96.34% <96.34%> (ø)

... and 65 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.