Controlling the UI

<ui_state>
- heading "Shopping list" [level=1] [ref=e3]
- list:
  - checkbox "milk" [checked] [ref=e5]
  - checkbox "eggs" [ref=e6]
</ui_state>

from pipecat.workers.ui import UIWorker, ui_event

class MyUIWorker(UIWorker):
    @ui_event("note_click")
    async def on_note_click(self, message):
        ref = (message.payload or {}).get("ref")
        await self.scroll_to(ref)
        await self.select_text(ref)

from pipecat.workers.ui import UIWorker

UI_PROMPT = """\
You answer questions about the page the user is looking at, in one or \
two plain sentences. Don't tell the user what you can't see; answer, \
or say you don't know."""

ui_worker = UIWorker(
    "ui",
    llm=OpenAILLMService(
        api_key=os.environ["OPENAI_API_KEY"],
        settings=OpenAILLMService.Settings(system_instruction=UI_PROMPT),
    ),
)

from pipecat.adapters.schemas.direct_function import tool_options
from pipecat.pipeline.job_context import JobError, JobParams


@tool_options(cancel_on_interruption=False, timeout_secs=60)
async def ask_page(params: FunctionCallParams, question: str):
    """Ask about the page the user is looking at.

    Call it for any question that could be about the page; nothing else
    can see it. Returns the answer.
    """
    try:
        async with params.pipeline_worker.job(
            "ui", params=JobParams(name="respond", payload={"query": question}, timeout=30)
        ) as t:
            pass
    except JobError as e:
        await params.result_callback({"error": str(e)})
        return
    await params.result_callback(t.response)


context = LLMContext(tools=[ask_page])

await runner.add_workers(ui_worker, worker)

from pipecat.workers.ui import UIWorker, screen_tools

context = LLMContext(tools=screen_tools("ui"))

ui_worker = UIWorker("ui", llm=ui_llm)

- screen(action="list", target="textbox"): the form's inputs with their
  current values. Call it at the start of every turn to see which fields
  are filled and steer toward the next empty one.
- screen(action="fill", target=..., value=...): write one value into the
  field the target names, such as "the email field".
- screen(action="click", target="the submit button"): submit, at the end.

You cannot know what the user has selected; only your tools can. Whenever
the user says "this", "that" or "this paragraph", call screen("selection")
first and answer from the text it returns.

from pipecat.classifiers.jev.classifier import JevClassifier

ui_worker = UIWorker(
    "ui",
    llm=ui_llm,
    classifier=JevClassifier(api_key=os.getenv("TYPESAFE_API_KEY")),
)

from pipecat.bus.messages import BusJobRequestMessage
from pipecat.classifiers.base_classifier import ChoiceQuestion
from pipecat.pipeline.job_decorator import job
from pipecat.workers.ui import UIWorker


class ListWorker(UIWorker):
    @job(name="update")
    async def _update(self, message: BusJobRequestMessage) -> None:
        payload = message.payload or {}

        # Adding needs no classifier: the voice LLM already carries the text.
        for text in payload.get("add", []):
            await self.send_command("add_item", {"text": text})

        # For items named in the user's words, ask the classifier which
        # checkbox on screen each one means, all in one call.
        items = {ref: name for ref, name in self._checkboxes()}
        to_check = payload.get("check", [])
        questions = {
            text: ChoiceQuestion(instructions=f"the list item {text!r} refers to", options=items)
            for text in to_check
        }
        not_found = []
        if items and questions:
            results = await self.classifier.choice({"items": to_check}, questions)
            for text, result in results.items():
                if result.confidence >= 0.5:
                    await self.send_command("set_checked", {"ref": result.choice, "checked": True})
                else:
                    not_found.append(text)

        await self.send_job_response(message.job_id, {"not_found": not_found})

    def _checkboxes(self) -> list[tuple[str, str]]:
        """The (ref, name) of each checkbox in the latest snapshot."""
        ...

@tool_options(cancel_on_interruption=False, timeout_secs=15)
async def update_list(
    params: FunctionCallParams,
    add: list[str] | None = None,
    check: list[str] | None = None,
):
    """Change the shopping list.

    Args:
        add: Items to add, named as they should appear on the list.
        check: Items to check off, as on the list or as the user said them.
    """
    async with params.pipeline_worker.job(
        "ui", params=JobParams(name="update", payload={"add": add or [], "check": check or []})
    ) as t:
        pass
    await params.result_callback(t.response)

class ReviewWorker(UIWorker):
    @job(name="add_note")
    async def _add_note(self, message: BusJobRequestMessage) -> None:
        text = (message.payload or {}).get("text", "")
        textarea = await self.which_element("the notes textarea")
        save = await self.which_element("the Save button")
        if not text or not textarea or not save:
            await self.send_job_response(message.job_id, {"done": False})
            return
        await self.set_input_value(textarea, text)
        await self.click(save)
        await self.send_job_response(message.job_id, {"done": True})

from pipecat.pipeline.job_context import JobGroupError, JobGroupParams, JobStatus


class ResearchWorker(UIWorker):
    @job(name="research")
    async def _research(self, message: BusJobRequestMessage) -> None:
        query = (message.payload or {}).get("query", "")
        try:
            async with self.job_group(
                "wikipedia", "news", "scholar",
                params=JobGroupParams(payload={"query": query}, label=f"Research: {query}"),
            ) as group:
                pass
        except JobGroupError as e:
            await self.send_job_response(message.job_id, {"error": str(e)}, status=JobStatus.ERROR)
            return
        summaries = {name: r.get("summary") for name, r in group.responses.items()}
        await self.send_job_response(message.job_id, {"results": summaries})

@tool_options(cancel_on_interruption=False, timeout_secs=60)
async def research(params: FunctionCallParams, query: str):
    """Research a topic across three sources and return their summaries."""
    await params.llm.push_frame(TTSSpeakFrame(f"Researching {query} now."))
    async with params.pipeline_worker.job(
        "ui", params=JobParams(name="research", payload={"query": query}, timeout=60)
    ) as t:
        pass
    await params.result_callback(t.response)