Add UIWorker by markbackman · Pull Request #4540 · pipecat-ai/pipecat

markbackman · 2026-05-22T00:33:33Z

Please describe the changes in your PR. If it is addressing an issue, please reference that as well.

codecov · 2026-05-22T00:36:51Z

Codecov Report

❌ Patch coverage is 93.13929% with 33 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
src/pipecat/workers/ui/ui_worker.py	90.22%	26 Missing ⚠️
...rc/pipecat/processors/frameworks/rtvi/processor.py	0.00%	2 Missing ⚠️
src/pipecat/workers/ui/ui_job_context.py	93.33%	2 Missing ⚠️
src/pipecat/processors/frameworks/rtvi/frames.py	83.33%	1 Missing ⚠️
src/pipecat/processors/frameworks/rtvi/observer.py	50.00%	1 Missing ⚠️
src/pipecat/workers/ui/ui_tools.py	97.14%	1 Missing ⚠️

Files with missing lines	Coverage Δ
src/pipecat/bus/ui/__init__.py	`100.00% <100.00%> (ø)`
src/pipecat/bus/ui/messages.py	`100.00% <100.00%> (ø)`
src/pipecat/pipeline/worker.py	`88.61% <100.00%> (ø)`
src/pipecat/processors/frameworks/rtvi/__init__.py	`100.00% <ø> (ø)`
src/pipecat/processors/frameworks/rtvi/models.py	`100.00% <100.00%> (ø)`
src/pipecat/services/llm_service.py	`64.31% <100.00%> (ø)`
src/pipecat/workers/ui/__init__.py	`100.00% <100.00%> (ø)`
src/pipecat/workers/ui/ui_event_decorator.py	`100.00% <100.00%> (ø)`
src/pipecat/workers/ui/ui_prompts.py	`100.00% <100.00%> (ø)`
src/pipecat/processors/frameworks/rtvi/frames.py	`86.27% <83.33%> (ø)`
... and 5 more

🚀 New features to boost your workflow:

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

aconchillo · 2026-05-22T19:57:08Z

@@ -0,0 +1,178 @@
+#


Should this go into bus/ui/messages.py? following the workers pattern

Good idea. Moving.

aconchillo · 2026-05-22T19:58:06Z

+
+
+@dataclass
+class BusUITaskGroupStartedMessage(BusDataMessage):


This should be called XXXJobGroupXXX I think, same for the ones below.

Updated all terms to align to Worker and JobGroup

aconchillo · 2026-05-22T19:58:40Z

@@ -52,6 +63,16 @@
 from pipecat.pipeline.worker_observer import WorkerObserver
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
 from pipecat.processors.frameworks.rtvi import RTVIObserver, RTVIObserverParams, RTVIProcessor
+from pipecat.processors.frameworks.rtvi.frames import RTVIUICommandFrame, RTVIUITaskFrame
+from pipecat.processors.frameworks.rtvi.models import (
+    UICancelTaskMessage,


UICancelWorkerMessage I think

Updated to UICancelJobGroupMessage.

aconchillo · 2026-05-22T20:09:34Z

+                        event_name=event_name,
+                        payload=payload,
+                    )
+                )


Maybe a small private function:

@self.rtvi.event_handler("on_ui_message") async def on_ui_message(rtvi: RTVIProcessor, message): await self._send_rtvi_bus_message(message)

aconchillo · 2026-05-22T20:12:47Z

+"""Decorator for marking worker methods as UI event handlers."""
+
+
+def on_ui_event(name: str):


should this be @ui_event? we usually only use on_XXX for event handler. for decorators we have @job instead of @on_job_request

Good catch. Done.

Addresses PR #4540 feedback: the inbound RTVI-UI -> bus translation was a ~24-line closure inside PipelineWorker.__init__. Move it to a typed private method, leaving the event handler a one-line delegate. Reads as the inbound counterpart to on_bus_message (the outbound bus -> RTVI translation). Behavior unchanged. Named _republish_ui_message_on_bus rather than the suggested _send_rtvi_bus_message: it matches the comment's verb ('republish ... onto the bus') and is less ambiguous (it sends a BusUIEventMessage derived from the RTVI message, not an 'RTVI bus message').

@job

Addresses PR #4540 feedback. Pipecat reserves the on_XXX form for event-handler names (the string passed to event_handler(...)); decorators that mark methods are named after the thing they handle -- @job (not @on_job_request), @tool, etc. @on_ui_event mixed the two, so it becomes @ui_event("nav_click"), paralleling @job(name="respond"). Updated the decorator, its export, docstrings, the duplicate-handler error message, tests, the document-review example, and the changelog fragment. Internal markers (is_ui_event_handler / ui_event_name) are unchanged.

@job

Addresses PR #4540 feedback. Pipecat reserves the on_XXX form for event-handler names (the string passed to event_handler(...)); decorators that mark methods are named after the thing they handle -- @job (not @on_job_request), @tool, etc. @on_ui_event mixed the two, so it becomes @ui_event("nav_click"), paralleling @job(name="respond"). Updated the decorator, its export, docstrings, the duplicate-handler error message, tests, the document-review example, and the changelog fragment. Internal markers (is_ui_event_handler / ui_event_name) are unchanged.

Addresses PR #4540 feedback: the inbound RTVI-UI -> bus translation was a ~24-line closure inside PipelineWorker.__init__. Move it to a typed private method, leaving the event handler a one-line delegate. Reads as the inbound counterpart to on_bus_message (the outbound bus -> RTVI translation). Behavior unchanged. Named _republish_ui_message_on_bus rather than the suggested _send_rtvi_bus_message: it matches the comment's verb ('republish ... onto the bus') and is less ambiguous (it sends a BusUIEventMessage derived from the RTVI message, not an 'RTVI bus message').

@job

Addresses PR #4540 feedback. Pipecat reserves the on_XXX form for event-handler names (the string passed to event_handler(...)); decorators that mark methods are named after the thing they handle -- @job (not @on_job_request), @tool, etc. @on_ui_event mixed the two, so it becomes @ui_event("nav_click"), paralleling @job(name="respond"). Updated the decorator, its export, docstrings, the duplicate-handler error message, tests, the document-review example, and the changelog fragment. Internal markers (is_ui_event_handler / ui_event_name) are unchanged.

Addresses PR #4540 feedback: the inbound RTVI-UI -> bus translation was a ~24-line closure inside PipelineWorker.__init__. Move it to a typed private method, leaving the event handler a one-line delegate. Reads as the inbound counterpart to on_bus_message (the outbound bus -> RTVI translation). Behavior unchanged. Named _republish_ui_message_on_bus rather than the suggested _send_rtvi_bus_message: it matches the comment's verb ('republish ... onto the bus') and is less ambiguous (it sends a BusUIEventMessage derived from the RTVI message, not an 'RTVI bus message').

@job

Addresses PR #4540 feedback. Pipecat reserves the on_XXX form for event-handler names (the string passed to event_handler(...)); decorators that mark methods are named after the thing they handle -- @job (not @on_job_request), @tool, etc. @on_ui_event mixed the two, so it becomes @ui_event("nav_click"), paralleling @job(name="respond"). Updated the decorator, its export, docstrings, the duplicate-handler error message, tests, the document-review example, and the changelog fragment. Internal markers (is_ui_event_handler / ui_event_name) are unchanged.

Addresses PR #4540 feedback: the inbound RTVI-UI -> bus translation was a ~24-line closure inside PipelineWorker.__init__. Move it to a typed private method, leaving the event handler a one-line delegate. Reads as the inbound counterpart to on_bus_message (the outbound bus -> RTVI translation). Behavior unchanged. Named _republish_ui_message_on_bus rather than the suggested _send_rtvi_bus_message: it matches the comment's verb ('republish ... onto the bus') and is less ambiguous (it sends a BusUIEventMessage derived from the RTVI message, not an 'RTVI bus message').

@job

Addresses PR #4540 feedback. Pipecat reserves the on_XXX form for event-handler names (the string passed to event_handler(...)); decorators that mark methods are named after the thing they handle -- @job (not @on_job_request), @tool, etc. @on_ui_event mixed the two, so it becomes @ui_event("nav_click"), paralleling @job(name="respond"). Updated the decorator, its export, docstrings, the duplicate-handler error message, tests, the document-review example, and the changelog fragment. Internal markers (is_ui_event_handler / ui_event_name) are unchanged.

Addresses PR #4540 feedback: the inbound RTVI-UI -> bus translation was a ~24-line closure inside PipelineWorker.__init__. Move it to a typed private method, leaving the event handler a one-line delegate. Reads as the inbound counterpart to on_bus_message (the outbound bus -> RTVI translation). Behavior unchanged. Named _republish_ui_message_on_bus rather than the suggested _send_rtvi_bus_message: it matches the comment's verb ('republish ... onto the bus') and is less ambiguous (it sends a BusUIEventMessage derived from the RTVI message, not an 'RTVI bus message').

@job

Addresses PR #4540 feedback. Pipecat reserves the on_XXX form for event-handler names (the string passed to event_handler(...)); decorators that mark methods are named after the thing they handle -- @job (not @on_job_request), @tool, etc. @on_ui_event mixed the two, so it becomes @ui_event("nav_click"), paralleling @job(name="respond"). Updated the decorator, its export, docstrings, the duplicate-handler error message, tests, the document-review example, and the changelog fragment. Internal markers (is_ui_event_handler / ui_event_name) are unchanged.

aconchillo · 2026-05-27T03:09:06Z

+
+
+@dataclass
+class BusUIEventMessage(BusDataMessage):


It might be a good idea to have a BusUIDataMessage(BusDataMessage) and all the UI messages inherit from it. There's a small reason in another comment.

aconchillo · 2026-05-27T03:10:30Z

        if message.target and message.target != self.name:
            return

        if isinstance(message, BusTTSSpeakMessage):
            await self.queue_frame(
                TTSSpeakFrame(text=message.text, append_to_context=message.append_to_context)
            )
+            return
+
+        if self._rtvi is None:


if we add BusUIDataMessage we could do:

if self._rtvi and isinstance(message, BusUIDataMessage): await self._handle_ui_bus_message(message)

aconchillo · 2026-05-27T03:11:25Z

-
-        Args:
-            message: The bus message to process.
+        """Handle bus messages for outbound RTVI UI messages.


This function can do more than just RTVI. I'd say the previous line was more generic.

aconchillo · 2026-05-27T03:30:21Z

+            active: Whether the worker starts active. Defaults to ``True``
+                (vs. ``False`` on ``LLMWorker`` / ``LLMContextWorker``) since a
+                UIWorker is typically an always-on delegate. Pass ``False`` for
+                handoff use cases.


I'm wondering if we really need active in UIWorker. This will always be a completely separate LLM that handles UI, plus the user can't enable the bridge. Active usually only makes sense when you have a bridge right now. So I'd suggest to remove it and just set it to True.

aconchillo · 2026-05-27T03:33:18Z

+            response = {"answer": answer} if answer else None
+        pending.set_result({"response": response, "status": status})
+
+    def user_job_group(


Should this be called ui_job_group? Also, UIJobGroupContext. All the new stuff is called UI.

aconchillo · 2026-05-27T03:33:40Z

+            cancellable=cancellable,
+        )
+
+    async def start_user_job_group(


start_ui_job_group()

aconchillo · 2026-05-27T03:34:05Z

+        )
+        return job_id
+
+    def _register_user_job_group(


_register_ui_job_group()

aconchillo · 2026-05-27T03:34:24Z

+            cancellable=cancellable,
+        )
+
+    def _unregister_user_job_group(self, job_id: str) -> None:


_unregister_ui_job_group

Align the job-group API with the rest of the UI surface (UIWorker, BusUI*, UICommand) per review feedback -- user -> ui: - ui_job_group / start_ui_job_group / _register_ui_job_group / _unregister_ui_job_group - UIJobGroupContext (was UserJobGroupContext) - internals _UIJobGroupRegistration and the _ui_job_groups dict Propagated to the async-tasks and document-review examples, the job-group lifecycle tests, and the changelog.

A UIWorker is always a standalone, job-dispatched delegate -- it is never bridged into the voice pipeline where activate/deactivate (handoff) applies -- so active is vestigial. Hardcode active=True in the super() call and drop the param. Updates the __init__ Args + class docstring, the hello-snapshot worker docstring (referenced active=True), and the test call sites that passed active=False purely for isolation (inert -- all UI tests still pass).

…iew) Introduce BusUIDataMessage(BusDataMessage) and reparent all six UI carriers (BusUIEventMessage, BusUICommandMessage, the four BusUIJob*) to it. PipelineWorker.on_bus_message now dispatches on the base with a single isinstance check and delegates the RTVI translation to a new _handle_ui_bus_message; the method's docstring is generalized since it also handles BusTTSSpeakMessage (not just RTVI). Inbound carriers match no translation branch and remain a no-op.

Helper to append text to an LLM service's system instruction. Used by UIWorker to append its wire-format prompt guide, but generally reusable.

UIWorker is an LLMContextWorker that observes and drives a client GUI over the RTVI UI channel: it stores live accessibility snapshots, auto-injects <ui_state> into the LLM context before each inference, dispatches client events to @ui_event handlers, and sends UI commands (scroll_to, highlight, select_text, click, set_input_value) back to the client. - Bus carriers (bus/ui/): a BusUIDataMessage base plus the UI event, command, and job-group lifecycle messages that UIWorker and PipelineWorker exchange. They live in the bus layer because both pipeline and workers reference them. - RTVI UI protocol: PipelineWorker connects a UIWorker to the client whenever RTVI is enabled -- its on_ui_message handler republishes inbound client messages onto the bus, and on_bus_message translates outbound UI carriers into RTVI frames (rtvi models/frames/observer/processor). - Delegate API: a single "respond" job; the worker chooses delivery when it completes the turn with respond_to_job -- data for the requester's voice LLM to phrase, or tts_speak=True for verbatim TTS via the requester's TTS. ReplyToolMixin bundles the standard reply tool; ui_job_group / start_ui_job_group fan work out to peer workers as cancellable job-group cards. Includes UI_STATE_PROMPT_GUIDE, the @ui_event decorator, tests, and changelog.

Seven multi-worker examples demonstrating UIWorker over the RTVI UI channel: pointing, deixis, document-review, form-fill, hello-snapshot, async-tasks, and shopping-list. Each pairs a voice pipeline with a UIWorker on its own worker.

markbackman force-pushed the mb/ui-worker branch from 2d93948 to eeb2a49 Compare May 22, 2026 02:43

aconchillo force-pushed the aleix/port-subagents-to-pipecat branch from 36f3b72 to e8ec7c5 Compare May 22, 2026 02:47

markbackman force-pushed the mb/ui-worker branch 2 times, most recently from e7ce598 to 9054912 Compare May 22, 2026 03:20

markbackman requested a review from aconchillo May 22, 2026 03:21

markbackman marked this pull request as ready for review May 22, 2026 03:21

markbackman force-pushed the mb/ui-worker branch 6 times, most recently from 6c98f7e to 1e5df7c Compare May 22, 2026 16:53

aconchillo reviewed May 22, 2026

View reviewed changes

Comment thread examples/multi-worker/ui-worker/async-tasks/bot.py

aconchillo reviewed May 22, 2026

View reviewed changes

Comment thread examples/multi-worker/ui-worker/deixis/bot.py

aconchillo reviewed May 22, 2026

View reviewed changes

Comment thread examples/multi-worker/ui-worker/document-review/bot.py

aconchillo reviewed May 22, 2026

View reviewed changes

Comment thread examples/multi-worker/ui-worker/form-fill/bot.py

markbackman force-pushed the mb/ui-worker branch from 1e5df7c to 18840cf Compare May 22, 2026 21:38

markbackman force-pushed the mb/ui-worker branch from 0d12a74 to fe6e5ac Compare May 23, 2026 03:09

markbackman force-pushed the mb/ui-worker branch from fe6e5ac to 0dc6cf8 Compare May 23, 2026 03:12

markbackman force-pushed the mb/ui-worker branch from c9fdd04 to 11f3c4a Compare May 23, 2026 13:05

markbackman commented May 23, 2026

View reviewed changes

Comment thread src/pipecat/workers/ui/ui_worker.py Outdated

markbackman mentioned this pull request May 26, 2026

refactor: rename UI protocol vocabulary to job-group and worker pipecat-ai/pipecat-client-web#214

Open

markbackman force-pushed the mb/ui-worker branch from f1220e7 to c069704 Compare May 26, 2026 18:14

markbackman force-pushed the mb/ui-worker branch from c069704 to 59c1b50 Compare May 26, 2026 18:15

markbackman force-pushed the mb/ui-worker branch from 59c1b50 to b5bdce7 Compare May 26, 2026 23:24

aconchillo reviewed May 27, 2026

View reviewed changes

aconchillo approved these changes May 27, 2026

View reviewed changes

markbackman force-pushed the mb/ui-worker branch 2 times, most recently from 138ef6b to ed385dc Compare May 27, 2026 12:15

markbackman added 3 commits May 27, 2026 08:40

Add LLMService.append_system_instruction()

f8488bb

Helper to append text to an LLM service's system instruction. Used by UIWorker to append its wire-format prompt guide, but generally reusable.

Add UIWorker examples

e81a439

Seven multi-worker examples demonstrating UIWorker over the RTVI UI channel: pointing, deixis, document-review, form-fill, hello-snapshot, async-tasks, and shopping-list. Each pairs a voice pipeline with a UIWorker on its own worker.

markbackman force-pushed the mb/ui-worker branch from ed385dc to e81a439 Compare May 27, 2026 12:41



		@dataclass
		class BusUITaskGroupStartedMessage(BusDataMessage):

		"""Decorator for marking worker methods as UI event handlers."""


		def on_ui_event(name: str):

Conversation

markbackman commented May 22, 2026

Please describe the changes in your PR. If it is addressing an issue, please reference that as well.

Uh oh!

codecov Bot commented May 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

codecov Bot commented May 22, 2026 •

edited

Loading