From 1c94feaaff0448c352afa85e07266daa966daa56 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 21 May 2026 22:42:12 -0400 Subject: [PATCH] Inject via the LLM's on_before_process_frame hook MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Move snapshot injection out of respond_with_llm into a cross-cutting on_before_process_frame handler on the UIWorker's LLM, so it appends the current snapshot to the context the request is built from, just before each inference. Injection is gated to the user-turn-initiating inference so a tool-calling turn never stacks duplicate blocks; respond_with_llm no longer injects manually. Also drop the bridged parameter from UIWorker: there is no viable way to bridge a UIWorker between workers — a shared, teed context would be polluted by the injection, and per-worker turn detection off teed frames isn't supported. Other workers keep their PipelineWorker bridging. --- changelog/xxxx.added.md | 2 +- src/pipecat/workers/ui/ui_worker.py | 89 +++++++++--------- tests/test_ui_worker.py | 136 +++++++++++++++------------- 3 files changed, 123 insertions(+), 104 deletions(-) diff --git a/changelog/xxxx.added.md b/changelog/xxxx.added.md index f0dd7fe15..ff0858208 100644 --- a/changelog/xxxx.added.md +++ b/changelog/xxxx.added.md @@ -1,3 +1,3 @@ -- Added `pipecat.workers.ui.UIWorker`, an `LLMContextWorker` that observes and drives a client GUI over the RTVI UI channel: it stores live accessibility snapshots, auto-injects `` at the start of each `respond` job, dispatches client events to `@on_ui_event` handlers, and sends UI commands (`scroll_to`, `highlight`, `select_text`, `click`, `set_input_value`) back to the client. The optional `ReplyToolMixin` exposes a bundled `reply` tool, and `user_job_group(...)` surfaces fan-out work to the client as cancellable task cards. A native RTVI⇄bus UI bridge is built into `PipelineWorker` (active whenever RTVI is enabled), so no decorator or manual wiring is needed: inbound UI messages are broadcast on the bus as `BusUIEventMessage`, and outbound `BusUICommandMessage` / `BusUITask*` carriers are translated into RTVI frames for the client. +- Added `pipecat.workers.ui.UIWorker`, an `LLMContextWorker` that observes and drives a client GUI over the RTVI UI channel: it stores live accessibility snapshots, auto-injects `` into the LLM context before every inference (via the LLM's `on_before_process_frame` hook), dispatches client events to `@on_ui_event` handlers, and sends UI commands (`scroll_to`, `highlight`, `select_text`, `click`, `set_input_value`) back to the client. The optional `ReplyToolMixin` exposes a bundled `reply` tool, and `user_job_group(...)` surfaces fan-out work to the client as cancellable task cards. A native RTVI⇄bus UI bridge is built into `PipelineWorker` (active whenever RTVI is enabled), so no decorator or manual wiring is needed: inbound UI messages are broadcast on the bus as `BusUIEventMessage`, and outbound `BusUICommandMessage` / `BusUITask*` carriers are translated into RTVI frames for the client. - `UIWorker` auto-injects the UI wire-format guide (`UI_STATE_PROMPT_GUIDE`) into its LLM's system instruction by default, via a `prompt_guide` parameter — pass your own string to override the guide, or `None` to disable. Apps no longer need to concatenate `UI_STATE_PROMPT_GUIDE` into the LLM's `system_instruction` by hand. diff --git a/src/pipecat/workers/ui/ui_worker.py b/src/pipecat/workers/ui/ui_worker.py index 3f45e533d..ea099baa2 100644 --- a/src/pipecat/workers/ui/ui_worker.py +++ b/src/pipecat/workers/ui/ui_worker.py @@ -31,7 +31,7 @@ from pipecat.bus.ui_messages import ( BusUITaskCompletedMessage, BusUITaskUpdateMessage, ) -from pipecat.frames.frames import LLMMessagesAppendFrame, LLMMessagesUpdateFrame +from pipecat.frames.frames import LLMContextFrame, LLMMessagesAppendFrame, LLMMessagesUpdateFrame from pipecat.pipeline.job_context import JobGroupError, JobStatus from pipecat.pipeline.job_decorator import job from pipecat.processors.aggregators.llm_context import LLMContext @@ -79,10 +79,11 @@ class UIWorker(LLMContextWorker): ## Canonical pattern - A ``UIWorker`` is the delegate side of a voice ↔ UI split: a - bridged ``LLMWorker`` (the voice layer) receives the user's - transcript and delegates UI-relevant work to this worker via - ``self.job("ui_worker_name", name="respond", payload={"query": text})``. + A ``UIWorker`` is the delegate side of a voice ↔ UI split: the + voice layer (the main pipeline's LLM, or a separate ``LLMWorker``) + receives the user's transcript and delegates UI-relevant work to + this worker via + ``job("ui_worker_name", name="respond", payload={"query": text})``. The built-in ``respond`` job runs: ```` is auto-injected, the LLM picks a tool, and the job completes with a spoken reply the voice worker hands to TTS. @@ -176,7 +177,6 @@ class UIWorker(LLMContextWorker): *, llm: LLMService[Any], active: bool = True, - bridged: tuple[str, ...] | None = None, defer_tool_frames: bool = True, context: LLMContext | None = None, user_params: LLMUserAggregatorParams | None = None, @@ -198,8 +198,6 @@ class UIWorker(LLMContextWorker): should self-activate as soon as its pipeline starts. Pass ``active=False`` only if you have a handoff use case. - bridged: Bridge configuration. See ``PipelineWorker`` for - details. defer_tool_frames: Forwarded to ``LLMContextWorker``. See ``LLMWorker`` for details. context: Optional pre-built ``LLMContext``. Forwarded to @@ -230,9 +228,10 @@ class UIWorker(LLMContextWorker): the injected content, or set this to False to disable. auto_inject_ui_state: When True (the default), the latest ```` snapshot is appended to the LLM context - at the start of every job request, so the worker always - reasons over the current screen. Set to False if you - want to call ``inject_ui_state()`` yourself. + just before every inference (via the LLM's + ``on_before_process_frame`` hook), so the worker always + reasons over the current screen. Set to False to manage + injection yourself with ``inject_ui_state()``. keep_history: When False (the default), the LLM context is cleared at the start of every job. Each job starts from an empty messages list, the current ```` @@ -264,37 +263,11 @@ class UIWorker(LLMContextWorker): ``system_instruction`` too. Use ``keep_history=True`` if the seeded messages genuinely need to live in the conversation history. - - Raises: - ValueError: If ``bridged`` is set together with the default - ``auto_inject_ui_state=True``. The two are - incompatible: auto-injection fires at the start of the - ``respond`` job, but a bridged ``UIWorker`` receives - user voice frames through the bridge instead of job - messages, so the snapshot would never reach the LLM - context. The canonical pattern is a non-bridged - ``UIWorker`` that receives delegated jobs from a - separate voice ``LLMWorker``. If you really want a - bridged ``UIWorker`` (advanced cases), pass - ``auto_inject_ui_state=False`` explicitly and call - ``inject_ui_state()`` yourself. """ - if bridged is not None and auto_inject_ui_state: - raise ValueError( - f"UIWorker '{name}': bridged + auto_inject_ui_state=True is " - "incompatible. Auto-injection fires at the start of the respond " - "job, but a bridged UIWorker receives frames through the bridge — the " - "snapshot would never land in the LLM context and the worker " - "would silently hallucinate. Use the canonical pattern " - "(non-bridged UIWorker receiving jobs from a separate " - "LLMWorker) or pass auto_inject_ui_state=False if you really " - "want a bridged UIWorker and will manage injection manually." - ) super().__init__( name, llm=llm, active=active, - bridged=bridged, defer_tool_frames=defer_tool_frames, context=context, user_params=user_params, @@ -331,6 +304,25 @@ class UIWorker(LLMContextWorker): # client as ``ui-task`` envelopes. self._user_job_groups: dict[str, _UserJobGroupRegistration] = {} + # Auto-inject the current ```` snapshot into the context just + # before each inference. Driven by the LLM's ``on_before_process_frame`` + # so it fires whenever the worker runs its LLM (e.g. a ``respond`` job), + # appending the snapshot to the same context the request is built from. + # The snapshot is a normal, persistent developer message; growth is + # managed by ``keep_history`` + context summarization. + @self.llm.event_handler("on_before_process_frame") + async def _inject_ui_state(_llm, frame): + if not (self._auto_inject_ui_state and isinstance(frame, LLMContextFrame)): + return + # Only inject on a user-turn-initiating inference, not the follow-up + # inference the LLM runs after a tool result (which would stack a + # duplicate ```` within the same turn). + if not _is_user_turn(frame.context): + return + content = self.render_ui_state() + if content: + frame.context.add_message({"role": "developer", "content": content}) + async def send_command(self, name: str, payload: Any = None) -> None: """Send a named UI command to the client. @@ -558,11 +550,11 @@ class UIWorker(LLMContextWorker): """Run one LLM turn for a job and respond when a tool completes it. Records the in-flight job for ``respond_to_job`` to close out, - clears the LLM context when ``keep_history=False``, injects the - current snapshot when ``auto_inject_ui_state=True``, appends the + clears the LLM context when ``keep_history=False``, appends the rendered query, and runs the LLM. Then blocks until a ``@tool`` calls ``respond_to_job``, and sends that result as the job - response. + response. The current ```` snapshot is injected by the + shared ``on_before_process_frame`` hook just before the inference. This is the body of the built-in ``@job(name="respond", sequential=True)`` handler. Because it spans the full LLM @@ -577,8 +569,6 @@ class UIWorker(LLMContextWorker): try: if not self._keep_history: await self.reset_context() - if self._auto_inject_ui_state: - await self.inject_ui_state() await self.queue_frame( LLMMessagesAppendFrame( messages=[{"role": "user", "content": self.render_query(message)}], @@ -1064,6 +1054,21 @@ class UIWorker(LLMContextWorker): ) +def _is_user_turn(context: LLMContext) -> bool: + """Whether the context's last message is the user's turn. + + Distinguishes a fresh user-turn inference (tail is the user message) from + the follow-up inference the LLM runs after a tool result (tail is the tool + result / assistant output), so the ```` snapshot is injected once + per turn rather than again on each tool round. + """ + messages = context.messages + if not messages: + return False + last = messages[-1] + return isinstance(last, dict) and last.get("role") == "user" + + def _collect_visible(node: dict[str, Any], out: list[dict[str, Any]]) -> None: """Depth-first collect nodes whose state does not include offscreen.""" state = node.get("state") diff --git a/tests/test_ui_worker.py b/tests/test_ui_worker.py index eb9e99c1d..37c8b9024 100644 --- a/tests/test_ui_worker.py +++ b/tests/test_ui_worker.py @@ -13,8 +13,14 @@ from unittest.mock import AsyncMock, MagicMock from pipecat.bus.messages import BusJobCancelMessage, BusJobRequestMessage from pipecat.bus.ui_messages import _UI_SNAPSHOT_BUS_EVENT_NAME, BusUIEventMessage -from pipecat.frames.frames import LLMMessagesAppendFrame, LLMMessagesUpdateFrame +from pipecat.frames.frames import ( + LLMContextFrame, + LLMMessagesAppendFrame, + LLMMessagesUpdateFrame, +) +from pipecat.processors.aggregators.llm_context import LLMContext from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.openai.llm import OpenAILLMService from pipecat.utils.asyncio.task_manager import TaskManager, TaskManagerParams from pipecat.workers.ui import UI_STATE_PROMPT_GUIDE, UIWorker, on_ui_event @@ -180,16 +186,6 @@ class TestUIWorkerDispatch(unittest.IsolatedAsyncioTestCase): _Bad("ui", llm=MagicMock()) - async def test_bridged_with_default_auto_inject_raises(self): - with self.assertRaises(ValueError) as ctx: - _PlainWorker("ui", llm=MagicMock(), bridged=()) - self.assertIn("bridged", str(ctx.exception)) - self.assertIn("auto_inject_ui_state", str(ctx.exception)) - - async def test_bridged_with_explicit_auto_inject_disabled_is_allowed(self): - worker = _PlainWorker("ui", llm=MagicMock(), bridged=(), auto_inject_ui_state=False) - self.assertFalse(worker._auto_inject_ui_state) - async def test_default_construction_unaffected(self): worker = _PlainWorker("ui", llm=MagicMock()) self.assertTrue(worker._auto_inject_ui_state) @@ -497,57 +493,71 @@ class TestUIWorkerSnapshot(unittest.IsolatedAsyncioTestCase): self.assertIn("e5", refs) -class TestUIWorkerAutoInject(unittest.IsolatedAsyncioTestCase): - async def test_respond_auto_injects_latest_snapshot(self): - worker = await _make_worker() +class TestUIWorkerSnapshotInjection(unittest.IsolatedAsyncioTestCase): + """The snapshot is injected just before inference via the LLM's + on_before_process_frame hook (e.g. during a respond job). + """ + + def _worker(self, **kwargs) -> UIWorker: + # A real LLM service so the on_before_process_frame event actually fires. + llm = OpenAILLMService(api_key="sk-test") + return _PlainWorker("ui", llm=llm, active=False, **kwargs) + + async def _fire(self, worker: UIWorker, context: LLMContext) -> None: + await worker.llm._call_event_handler( + "on_before_process_frame", LLMContextFrame(context=context) + ) + + def _developer_messages(self, context: LLMContext) -> list: + return [m for m in context.messages if isinstance(m, dict) and m.get("role") == "developer"] + + async def test_injects_ui_state_on_user_turn(self): + worker = self._worker() worker._latest_snapshot = _SAMPLE_SNAPSHOT + ctx = LLMContext([{"role": "user", "content": "hi"}]) + await self._fire(worker, ctx) + devs = self._developer_messages(ctx) + self.assertEqual(len(devs), 1) + self.assertTrue(devs[0]["content"].startswith("")) - t = await _start( - worker, - BusJobRequestMessage(source="voice", target="ui", job_id="t1", payload={"query": "hi"}), - ) - - frames = _append_frames(worker) - self.assertEqual(len(frames), 2) - self.assertEqual(frames[0].messages[0]["role"], "developer") - self.assertTrue(frames[0].messages[0]["content"].startswith("")) - self.assertFalse(frames[0].run_llm) - self.assertEqual(frames[1].messages[0]["content"], "hi") - self.assertTrue(frames[1].run_llm) - - await worker.respond_to_job() - await t - - async def test_auto_inject_ui_state_false_suppresses_injection(self): - worker = await _make_worker(auto_inject_ui_state=False) + async def test_skips_tool_result_continuation(self): + # The follow-up inference after a tool result must not stack a second + # within the same turn. + worker = self._worker() worker._latest_snapshot = _SAMPLE_SNAPSHOT - - t = await _start( - worker, - BusJobRequestMessage(source="voice", target="ui", job_id="t1", payload={"query": "hi"}), + ctx = LLMContext( + [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": "ok"}, + {"role": "tool", "content": "result"}, + ] ) + before = len(ctx.messages) + await self._fire(worker, ctx) + self.assertEqual(len(ctx.messages), before) - frames = _append_frames(worker) - self.assertEqual(len(frames), 1) - self.assertFalse(frames[0].messages[0]["content"].startswith("")) + async def test_no_injection_when_disabled(self): + worker = self._worker(auto_inject_ui_state=False) + worker._latest_snapshot = _SAMPLE_SNAPSHOT + ctx = LLMContext([{"role": "user", "content": "hi"}]) + await self._fire(worker, ctx) + self.assertEqual(len(ctx.messages), 1) - await worker.respond_to_job() - await t + async def test_no_injection_without_snapshot(self): + worker = self._worker() + ctx = LLMContext([{"role": "user", "content": "hi"}]) + await self._fire(worker, ctx) + self.assertEqual(len(ctx.messages), 1) - async def test_auto_inject_no_op_without_snapshot(self): - worker = await _make_worker() - - t = await _start( - worker, - BusJobRequestMessage(source="voice", target="ui", job_id="t1", payload={"query": "hi"}), - ) - - frames = _append_frames(worker) - self.assertEqual(len(frames), 1) - self.assertFalse(frames[0].messages[0]["content"].startswith("")) - - await worker.respond_to_job() - await t + async def test_injects_once_per_turn(self): + # After injecting, the tail is the developer message, so a re-fire on the + # same context is a no-op (no accumulation within a turn). + worker = self._worker() + worker._latest_snapshot = _SAMPLE_SNAPSHOT + ctx = LLMContext([{"role": "user", "content": "hi"}]) + await self._fire(worker, ctx) + await self._fire(worker, ctx) + self.assertEqual(len(self._developer_messages(ctx)), 1) class TestUIWorkerKeepHistory(unittest.IsolatedAsyncioTestCase): @@ -594,8 +604,12 @@ class TestUIWorkerKeepHistory(unittest.IsolatedAsyncioTestCase): BusJobRequestMessage(source="voice", target="ui", job_id="t1", payload={"query": "hi"}), ) + # Injection moved to the on_before_process_frame hook, so respond_with_llm + # itself queues only the query append. self.assertEqual(_update_frames(worker), []) - self.assertEqual(len(_append_frames(worker)), 2) + appends = _append_frames(worker) + self.assertEqual(len(appends), 1) + self.assertEqual(appends[0].messages[0]["content"], "hi") await worker.respond_to_job() await t @@ -739,12 +753,12 @@ class TestUIWorkerRespondJob(unittest.IsolatedAsyncioTestCase): ), ) + # The snapshot is injected by the hook at inference time, so the handler + # itself queues only the rendered query. appends = _append_frames(worker) - self.assertEqual(len(appends), 2) - self.assertTrue(appends[0].messages[0]["content"].startswith("")) - self.assertFalse(appends[0].run_llm) - self.assertEqual(appends[1].messages[0]["content"], "hello") - self.assertTrue(appends[1].run_llm) + self.assertEqual(len(appends), 1) + self.assertEqual(appends[0].messages[0]["content"], "hello") + self.assertTrue(appends[0].run_llm) self.assertEqual(worker.current_job.job_id, "t1") await worker.respond_to_job(speak="done")