diff --git a/examples/multi-task/sensor-controller/sensor-controller.py b/examples/multi-task/sensor-controller/sensor-controller.py new file mode 100644 index 000000000..61310d3a5 --- /dev/null +++ b/examples/multi-task/sensor-controller/sensor-controller.py @@ -0,0 +1,334 @@ +# +# Copyright (c) 2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Voice agent + sensor-controller worker, both as plain PipelineTasks. + +Two ``PipelineTask`` instances run side by side: + +- The **voice agent** is built inline in ``run_bot`` — a standard + transport + STT + LLM + TTS pipeline. Its LLM has a single tool, + ``ask_controller(question)``, which forwards the user's request to + the controller over the bus and speaks back the response. +- The **sensor controller** (``build_sensor_controller``) is a + ``PipelineTask`` whose pipeline runs a simulated temperature sensor + (see ``sensor.py``) alongside its own LLM. The worker LLM has tool + access to read the current reading, inspect rolling stats, and + mutate the simulated sensor's target temperature and response rate. + +The worker does **not** subclass ``LLMTask`` and is **not** bridged. +The voice agent and the controller communicate exclusively through +``BusJobRequestMessage`` / ``BusJobResponseMessage``. The controller +collects responses by listening to the assistant aggregator's +``on_assistant_turn_stopped`` event and pairing each LLM completion +with the in-flight job id. + +Requirements: + +- OPENAI_API_KEY +- DEEPGRAM_API_KEY +- CARTESIA_API_KEY +- DAILY_API_KEY (for Daily transport) + +Example voice exchange:: + + User: What's the temperature? + Controller: 22.1°C, holding steady. + + User: Make it warmer. + Controller: I set the target to 26°C. Give it about 20 seconds. + + User: Is it stable yet? + Controller: It's at 25.4°C and still climbing — almost there. + + User: Why is it slow? + Controller: The response rate is 5%. I sped it up to 20%; it'll settle faster now. +""" + +import os + +from dotenv import load_dotenv +from loguru import logger +from sensor import SensorReader, SensorStats + +from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.bus import BusJobRequestMessage +from pipecat.frames.frames import LLMMessagesAppendFrame, LLMRunFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import ( + AssistantTurnStoppedMessage, + LLMContextAggregatorPair, + LLMUserAggregatorParams, +) +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.cartesia.tts import CartesiaTTSService +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.llm_service import FunctionCallParams +from pipecat.services.openai.llm import OpenAILLMService +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.daily.transport import DailyParams + +load_dotenv(override=True) + + +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + ), +} + + +def build_sensor_controller() -> PipelineTask: + """Build the controller worker as a plain :class:`PipelineTask`. + + The pipeline shape is:: + + SensorReader -> SensorStats -> user_agg -> llm -> assistant_agg + + ``SensorReader`` runs an autonomous tick loop that emits a + :class:`SensorReadingFrame` every second; ``SensorStats`` consumes + those readings and exposes rolling statistics. The LLM has four + direct tools that read or mutate the sensor. + + Jobs arrive via the ``on_job_request`` event handler. The handler + stores the active ``job_id``, then queues an + :class:`LLMMessagesAppendFrame` with the user's question and runs + the LLM. When the assistant turn finishes (signalled by the + assistant aggregator's ``on_assistant_turn_stopped`` event), the + handler sends a :class:`BusJobResponseMessage` carrying the LLM's + answer back to the voice agent. + """ + sensor = SensorReader() + stats = SensorStats() + + async def get_current_reading(params: FunctionCallParams): + """Read the sensor's current temperature in degrees Celsius.""" + await params.result_callback({"temperature": round(sensor.current, 2)}) + + async def get_stats(params: FunctionCallParams): + """Rolling minimum, maximum, average, and trend of the temperature.""" + await params.result_callback( + { + "min": round(stats.min, 2), + "max": round(stats.max, 2), + "avg": round(stats.avg, 2), + "trend": stats.trend, + } + ) + + async def set_target_temperature(params: FunctionCallParams, target_celsius: float): + """Adjust the target temperature; the sensor will drift toward it. + + Args: + target_celsius (float): The new target temperature in degrees Celsius. + """ + sensor.set_target(target_celsius) + await params.result_callback({"ok": True, "new_target": target_celsius}) + + async def set_response_rate(params: FunctionCallParams, rate: float): + """Set how aggressively the sensor approaches the target. + + Args: + rate (float): Response rate between 0.01 (slow) and 0.3 (fast). + """ + sensor.set_response_rate(rate) + await params.result_callback({"ok": True, "new_rate": rate}) + + llm = OpenAILLMService( + api_key=os.environ["OPENAI_API_KEY"], + settings=OpenAILLMService.Settings( + system_instruction=( + "You are a temperature sensor controller. You manage a single " + "thermometer and answer the user's questions about it. Use the " + "provided tools to read the current temperature, inspect rolling " + "statistics, change the target temperature, or change how fast " + "the sensor responds. When the user asks for a vague change " + "('make it warmer', 'cooler'), pick a sensible target and call " + "set_target_temperature. Always answer in one or two short " + "sentences — your reply is spoken aloud." + ), + ), + ) + llm.register_direct_function(get_current_reading) + llm.register_direct_function(get_stats) + llm.register_direct_function(set_target_temperature) + llm.register_direct_function(set_response_rate) + + context = LLMContext( + tools=ToolsSchema( + standard_tools=[ + get_current_reading, + get_stats, + set_target_temperature, + set_response_rate, + ] + ) + ) + aggregators = LLMContextAggregatorPair(context) + + pipeline = Pipeline( + [ + sensor, + stats, + aggregators.user(), + llm, + aggregators.assistant(), + ] + ) + + worker = PipelineTask(pipeline, name="controller") + + # The controller handles one job at a time (the LLM pipeline can only + # run one turn at a time). ``state["job_id"]`` pairs the in-flight + # job with the next ``on_assistant_turn_stopped`` event. + state: dict[str, str | None] = {"job_id": None} + + @worker.event_handler("on_job_request") + async def on_request(_task, message: BusJobRequestMessage): + question = message.payload["question"] + logger.info(f"Controller: received question '{question}'") + state["job_id"] = message.job_id + await worker.queue_frame( + LLMMessagesAppendFrame( + messages=[{"role": "user", "content": question}], + run_llm=True, + ) + ) + + @aggregators.assistant().event_handler("on_assistant_turn_stopped") + async def on_assistant_turn_stopped(_aggregator, message: AssistantTurnStoppedMessage): + # The aggregator fires this event on every ``LLMFullResponseEndFrame``, + # including the tool-call round that precedes the tool result and has + # no spoken text. Skip those so we only forward the LLM's final + # response to the voice agent. + if not message.content: + return + if state["job_id"] is None: + return + job_id, state["job_id"] = state["job_id"], None + logger.info(f"Controller: answering job {job_id[:8]}") + await worker.send_job_response(job_id, response={"answer": message.content}) + + return worker + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info("Starting sensor-controller bot") + + # Voice agent: standard transport + STT + LLM + TTS pipeline. The + # only tool the voice LLM has is ``ask_controller`` — it does not + # know anything about temperatures, trends, or response rates. + stt = DeepgramSTTService(api_key=os.environ["DEEPGRAM_API_KEY"]) + tts = CartesiaTTSService( + api_key=os.environ["CARTESIA_API_KEY"], + settings=CartesiaTTSService.Settings( + voice="9626c31c-bec5-4cca-baa8-f8ba9e84c8bc", # Jacqueline + ), + ) + + async def ask_controller(params: FunctionCallParams, question: str): + """Ask the temperature sensor controller anything about the sensor. + + Forward the user's request verbatim and speak back the answer. + + Args: + question (str): The user's question or instruction to forward to the controller. + """ + logger.info(f"Voice agent: forwarding to controller: '{question}'") + async with params.pipeline_task.job( + "controller", payload={"question": question}, timeout=30 + ) as t: + pass + await params.result_callback(t.response["answer"]) + + llm = OpenAILLMService( + api_key=os.environ["OPENAI_API_KEY"], + settings=OpenAILLMService.Settings( + system_instruction=( + "You are a friendly voice assistant with access to a temperature " + "sensor controller. For ANY request about the temperature — " + "reading it, adjusting it, checking trends, changing how fast it " + "responds — call the ask_controller tool. Forward the user's " + "request verbatim. Then speak the controller's answer back. " + "Keep responses brief; do not add extra commentary." + ), + ), + ) + llm.register_direct_function(ask_controller, timeout_secs=60) + + context = LLMContext(tools=ToolsSchema(standard_tools=[ask_controller])) + aggregators = LLMContextAggregatorPair( + context, + user_params=LLMUserAggregatorParams(vad_analyzer=SileroVADAnalyzer()), + ) + + pipeline = Pipeline( + [ + transport.input(), + stt, + aggregators.user(), + llm, + tts, + transport.output(), + aggregators.assistant(), + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info("Client connected") + context.add_message( + { + "role": "developer", + "content": ( + "Greet the user and let them know you can read or adjust a " + "temperature sensor on their behalf." + ), + } + ) + await task.queue_frame(LLMRunFrame()) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info("Client disconnected") + await runner.cancel() + + await runner.spawn(build_sensor_controller()) + await runner.spawn(task) + + await runner.run() + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud.""" + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +if __name__ == "__main__": + from pipecat.runner.run import main + + main() diff --git a/examples/multi-task/sensor-controller/sensor.py b/examples/multi-task/sensor-controller/sensor.py new file mode 100644 index 000000000..54843910c --- /dev/null +++ b/examples/multi-task/sensor-controller/sensor.py @@ -0,0 +1,186 @@ +# +# Copyright (c) 2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Temperature sensor processors for the sensor-controller example. + +Two custom :class:`FrameProcessor` subclasses that give the worker +pipeline real autonomous frame flow: + +- :class:`SensorReader` simulates a thermometer. It runs an async tick + loop that advances ``current`` toward ``target`` with a first-order + lag plus Gaussian noise, and pushes a :class:`SensorReadingFrame` on + every tick. ``target`` and ``response_rate`` are mutable so the + worker's LLM can adjust them via tool calls. +- :class:`SensorStats` consumes the readings, maintains a rolling + window, and exposes ``current`` / ``min`` / ``max`` / ``avg`` / + ``trend`` as properties. The worker LLM reads these directly when + answering the user. +""" + +import random +import time +from collections import deque +from dataclasses import dataclass + +from pipecat.frames.frames import DataFrame, Frame, StartFrame +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + + +@dataclass +class SensorReadingFrame(DataFrame): + """A single temperature reading emitted by :class:`SensorReader`. + + Parameters: + temperature: The reading in degrees Celsius. + timestamp: Unix timestamp when the reading was taken. + """ + + temperature: float = 0.0 + timestamp: float = 0.0 + + +class SensorReader(FrameProcessor): + """Simulated temperature sensor with adjustable target and response rate. + + Each tick, ``current`` is updated as:: + + current += (target - current) * response_rate + gauss(0, noise_sigma) + + This is a first-order lag toward ``target``. With ``response_rate=0.05`` + and a 1s tick, the current reading reaches ~halfway to target in ~14s; + with ``response_rate=0.2`` it converges in ~5–10s. + """ + + def __init__( + self, + *, + start_temp: float = 22.0, + sample_period_s: float = 1.0, + response_rate: float = 0.05, + noise_sigma: float = 0.1, + ): + """Initialize the sensor. + + Args: + start_temp: Initial temperature and initial target (°C). + sample_period_s: Seconds between successive readings. + response_rate: Fraction of the gap toward target closed each tick + (clamped to ``[0.0, 1.0]``). + noise_sigma: Standard deviation of the Gaussian noise added to + each reading. + """ + super().__init__() + self._current = start_temp + self._target = start_temp + self._response_rate = max(0.0, min(1.0, response_rate)) + self._noise_sigma = noise_sigma + self._sample_period_s = sample_period_s + self._tick_task = None + + @property + def current(self) -> float: + """The most recent temperature reading (°C).""" + return self._current + + @property + def target(self) -> float: + """The temperature the sensor is drifting toward (°C).""" + return self._target + + @property + def response_rate(self) -> float: + """Fraction of the target-current gap closed per tick.""" + return self._response_rate + + def set_target(self, value: float) -> None: + """Set a new target temperature (°C).""" + self._target = value + + def set_response_rate(self, rate: float) -> None: + """Set how aggressively the sensor approaches the target. + + Args: + rate: Fraction in ``[0.0, 1.0]``. Clamped to that range. + """ + self._response_rate = max(0.0, min(1.0, rate)) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, StartFrame) and self._tick_task is None: + self._tick_task = self.create_task(self._tick_loop(), "ticker") + await self.push_frame(frame, direction) + + async def cleanup(self) -> None: + if self._tick_task is not None: + await self.cancel_task(self._tick_task) + self._tick_task = None + await super().cleanup() + + async def _tick_loop(self) -> None: + import asyncio + + while True: + await asyncio.sleep(self._sample_period_s) + gap = self._target - self._current + self._current += gap * self._response_rate + random.gauss(0, self._noise_sigma) + await self.push_frame( + SensorReadingFrame(temperature=self._current, timestamp=time.time()), + FrameDirection.DOWNSTREAM, + ) + + +class SensorStats(FrameProcessor): + """Rolling-window statistics over :class:`SensorReadingFrame`s. + + Consumes readings as they flow downstream and exposes rolling + ``min`` / ``max`` / ``avg`` / ``trend`` as properties — the worker + LLM reads them directly when responding to the user. + """ + + def __init__(self, window: int = 30): + """Initialize the stats aggregator. + + Args: + window: Number of recent readings to retain. + """ + super().__init__() + self._readings: deque[float] = deque(maxlen=window) + + @property + def current(self) -> float: + """The most recent reading, or 0.0 if none have been seen.""" + return self._readings[-1] if self._readings else 0.0 + + @property + def min(self) -> float: + return min(self._readings) if self._readings else 0.0 + + @property + def max(self) -> float: + return max(self._readings) if self._readings else 0.0 + + @property + def avg(self) -> float: + return sum(self._readings) / len(self._readings) if self._readings else 0.0 + + @property + def trend(self) -> str: + """``"rising"`` / ``"falling"`` / ``"stable"`` based on first vs. last half of the window.""" + if len(self._readings) < 4: + return "stable" + half = len(self._readings) // 2 + old_avg = sum(list(self._readings)[:half]) / half + new_avg = sum(list(self._readings)[half:]) / (len(self._readings) - half) + diff = new_avg - old_avg + if abs(diff) < 0.25: + return "stable" + return "rising" if diff > 0 else "falling" + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, SensorReadingFrame): + self._readings.append(frame.temperature) + await self.push_frame(frame, direction)