wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Inject <ui_state> via the LLM's on_before_process_frame hook

Browse Source 
Move <ui_state> 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 <ui_state> 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.
This commit is contained in:
Mark Backman
2026-05-21 22:42:12 -04:00
parent 950fc10f05
commit 1c94feaaff
3 changed files with 123 additions and 104 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
changelog/xxxx.added.md
View File

@@ -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 `<ui_state>` 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 `<ui_state>` 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.

89
src/pipecat/workers/ui/ui_worker.py
View File

@@ -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: ``<ui_state>`` 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
``<ui_state>`` 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 ``<ui_state>``
@@ -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 ``<ui_state>`` 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 ``<ui_state>`` 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 ``<ui_state>`` 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 ``<ui_state>`` 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")

136
tests/test_ui_worker.py
View File

@@ -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 <ui_state> 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("<ui_state>"))
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("<ui_state>"))
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
# <ui_state> 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("<ui_state>"))
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("<ui_state>"))
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("<ui_state>"))
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")