From 2d0f3341c3d3676388154ccf746a6647a019012e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 14 Jan 2025 14:44:06 -0800 Subject: [PATCH 1/7] frames: add LLMTextFrame and TTSTextFrame This is to distinguish what type of service has generated the TextFrames. --- src/pipecat/frames/frames.py | 14 ++++++++++++++ src/pipecat/processors/frameworks/rtvi.py | 9 +++++---- src/pipecat/services/ai_services.py | 5 +++-- src/pipecat/services/anthropic.py | 4 ++-- .../services/gemini_multimodal_live/gemini.py | 6 +++--- src/pipecat/services/google.py | 4 ++-- src/pipecat/services/openai.py | 4 ++-- .../services/openai_realtime_beta/openai.py | 4 ++-- 8 files changed, 33 insertions(+), 17 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 321f8514b..6e5fa744f 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -177,6 +177,20 @@ class TextFrame(DataFrame): return f"{self.name}(pts: {pts}, text: [{self.text}])" +@dataclass +class LLMTextFrame(TextFrame): + """A text frame generated by LLM services.""" + + pass + + +@dataclass +class TTSTextFrame(TextFrame): + """A text frame generated by TTS services.""" + + pass + + @dataclass class TranscriptionFrame(TextFrame): """A text frame with transcription-specific data. Will be placed in the diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 96d51cb5d..36b782f80 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -34,14 +34,15 @@ from pipecat.frames.frames import ( InterimTranscriptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, + LLMTextFrame, MetricsFrame, StartFrame, SystemFrame, - TextFrame, TranscriptionFrame, TransportMessageUrgentFrame, TTSStartedFrame, TTSStoppedFrame, + TTSTextFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, ) @@ -479,7 +480,7 @@ class RTVIBotTranscriptionProcessor(RTVIFrameProcessor): if isinstance(frame, UserStartedSpeakingFrame): await self._push_aggregation() - elif isinstance(frame, TextFrame): + elif isinstance(frame, LLMTextFrame): self._aggregation += frame.text if match_endofsentence(self._aggregation): await self._push_aggregation() @@ -504,7 +505,7 @@ class RTVIBotLLMProcessor(RTVIFrameProcessor): await self._push_transport_message_urgent(RTVIBotLLMStartedMessage()) elif isinstance(frame, LLMFullResponseEndFrame): await self._push_transport_message_urgent(RTVIBotLLMStoppedMessage()) - elif type(frame) is TextFrame: + elif isinstance(frame, LLMTextFrame): message = RTVIBotLLMTextMessage(data=RTVITextMessageData(text=frame.text)) await self._push_transport_message_urgent(message) @@ -522,7 +523,7 @@ class RTVIBotTTSProcessor(RTVIFrameProcessor): await self._push_transport_message_urgent(RTVIBotTTSStartedMessage()) elif isinstance(frame, TTSStoppedFrame): await self._push_transport_message_urgent(RTVIBotTTSStoppedMessage()) - elif type(frame) is TextFrame: + elif isinstance(frame, TTSTextFrame): message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=frame.text)) await self._push_transport_message_urgent(message) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 973332623..0a03b2eaa 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -30,6 +30,7 @@ from pipecat.frames.frames import ( TTSSpeakFrame, TTSStartedFrame, TTSStoppedFrame, + TTSTextFrame, TTSUpdateSettingsFrame, UserImageRequestFrame, VisionImageRawFrame, @@ -358,7 +359,7 @@ class TTSService(AIService): if self._push_text_frames: # We send the original text after the audio. This way, if we are # interrupted, the text is not added to the assistant context. - await self.push_frame(TextFrame(text)) + await self.push_frame(TTSTextFrame(text)) async def _stop_frame_handler(self): try: @@ -437,7 +438,7 @@ class WordTTSService(TTSService): frame = TTSStoppedFrame() frame.pts = last_pts else: - frame = TextFrame(word) + frame = TTSTextFrame(word) frame.pts = self._initial_word_timestamp + timestamp if frame: last_pts = frame.pts diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 08bde31b7..1590296af 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -26,10 +26,10 @@ from pipecat.frames.frames import ( LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesFrame, + LLMTextFrame, LLMUpdateSettingsFrame, OpenAILLMContextAssistantTimestampFrame, StartInterruptionFrame, - TextFrame, UserImageRawFrame, UserImageRequestFrame, VisionImageRawFrame, @@ -191,7 +191,7 @@ class AnthropicLLMService(LLMService): if event.type == "content_block_delta": if hasattr(event.delta, "text"): - await self.push_frame(TextFrame(event.delta.text)) + await self.push_frame(LLMTextFrame(event.delta.text)) completion_tokens_estimate += self._estimate_tokens(event.delta.text) elif hasattr(event.delta, "partial_json") and tool_use_block: json_accumulator += event.delta.partial_json diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 6ce09e721..57753e9cd 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -28,10 +28,10 @@ from pipecat.frames.frames import ( LLMFullResponseStartFrame, LLMMessagesAppendFrame, LLMSetToolsFrame, + LLMTextFrame, LLMUpdateSettingsFrame, StartFrame, StartInterruptionFrame, - TextFrame, TranscriptionFrame, TTSAudioRawFrame, TTSStartedFrame, @@ -295,7 +295,7 @@ class GeminiMultimodalLiveLLMService(LLMService): # definitely feels like a hack. Need to revisit when the API evolves. # context.add_message({"role": "assistant", "content": [{"type": "text", "text": text}]}) await self.push_frame(LLMFullResponseStartFrame()) - await self.push_frame(TextFrame(text=text)) + await self.push_frame(LLMTextFrame(text=text)) await self.push_frame(LLMFullResponseEndFrame()) async def _transcribe_audio(self, audio, context): @@ -628,7 +628,7 @@ class GeminiMultimodalLiveLLMService(LLMService): await self.push_frame(LLMFullResponseStartFrame()) self._bot_text_buffer += text - await self.push_frame(TextFrame(text=text)) + await self.push_frame(LLMTextFrame(text=text)) inline_data = part.inlineData if not inline_data: diff --git a/src/pipecat/services/google.py b/src/pipecat/services/google.py index 341c51636..4e4db5990 100644 --- a/src/pipecat/services/google.py +++ b/src/pipecat/services/google.py @@ -23,9 +23,9 @@ from pipecat.frames.frames import ( LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesFrame, + LLMTextFrame, LLMUpdateSettingsFrame, OpenAILLMContextAssistantTimestampFrame, - TextFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -698,7 +698,7 @@ class GoogleLLMService(LLMService): try: for c in chunk.parts: if c.text: - await self.push_frame(TextFrame(c.text)) + await self.push_frame(LLMTextFrame(c.text)) elif c.function_call: logger.debug(f"!!! Function call: {c.function_call}") args = type(c.function_call).to_dict(c.function_call).get("args", {}) diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index 4a70838f4..19c8f6d99 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -25,10 +25,10 @@ from pipecat.frames.frames import ( LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesFrame, + LLMTextFrame, LLMUpdateSettingsFrame, OpenAILLMContextAssistantTimestampFrame, StartInterruptionFrame, - TextFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -258,7 +258,7 @@ class BaseOpenAILLMService(LLMService): # Keep iterating through the response to collect all the argument fragments arguments += tool_call.function.arguments elif chunk.choices[0].delta.content: - await self.push_frame(TextFrame(chunk.choices[0].delta.content)) + await self.push_frame(LLMTextFrame(chunk.choices[0].delta.content)) # if we got a function name and arguments, check to see if it's a function with # a registered handler. If so, run the registered callback, save the result to diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py index 44057606f..c1d7c4a4d 100644 --- a/src/pipecat/services/openai_realtime_beta/openai.py +++ b/src/pipecat/services/openai_realtime_beta/openai.py @@ -24,11 +24,11 @@ from pipecat.frames.frames import ( LLMFullResponseStartFrame, LLMMessagesAppendFrame, LLMSetToolsFrame, + LLMTextFrame, LLMUpdateSettingsFrame, StartFrame, StartInterruptionFrame, StopInterruptionFrame, - TextFrame, TranscriptionFrame, TTSAudioRawFrame, TTSStartedFrame, @@ -458,7 +458,7 @@ class OpenAIRealtimeBetaLLMService(LLMService): async def _handle_evt_audio_transcript_delta(self, evt): if evt.delta: - await self.push_frame(TextFrame(evt.delta)) + await self.push_frame(LLMTextFrame(evt.delta)) async def _handle_evt_speech_started(self, evt): await self._truncate_current_audio_response() From 25bcaf5c7c9572afffa32a01092b25960e580b3d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 15 Jan 2025 09:39:05 -0800 Subject: [PATCH 2/7] observers: introduce pipeline observers --- src/pipecat/observers/__init__.py | 0 src/pipecat/observers/base_observer.py | 38 ++++++++++++++++++++++++++ 2 files changed, 38 insertions(+) create mode 100644 src/pipecat/observers/__init__.py create mode 100644 src/pipecat/observers/base_observer.py diff --git a/src/pipecat/observers/__init__.py b/src/pipecat/observers/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/observers/base_observer.py b/src/pipecat/observers/base_observer.py new file mode 100644 index 000000000..2a17b5668 --- /dev/null +++ b/src/pipecat/observers/base_observer.py @@ -0,0 +1,38 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from abc import ABC, abstractmethod + +from pipecat.frames.frames import Frame +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + + +class BaseObserver(ABC): + """This is the base class for pipeline frame observers. Observers can view + all the frames that go through the pipeline without the need to inject + processors in the pipeline. This can be useful, for example, to implement + frame loggers or debuggers among other things. + + """ + + @abstractmethod + async def on_push_frame( + self, src: FrameProcessor, dst: FrameProcessor, frame: Frame, direction: FrameDirection + ): + """Abstract method to handle the event when a frame is pushed from one + processor to another. + + Args: + src (FrameProcessor): The source frame processor that is sending the frame. + dst (FrameProcessor): The destination frame processor that will receive the frame. + frame (Frame): The frame being transferred between processors. + direction (FrameDirection): The direction of the frame transfer. + + This method should be implemented by subclasses to define specific behavior + when a frame is pushed. + + """ + pass From c8da5314020bd525df1727f348a14121efa66728 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 15 Jan 2025 09:48:10 -0800 Subject: [PATCH 3/7] pipeline(task): add support for pipeline frame observers --- src/pipecat/frames/frames.py | 6 +++- src/pipecat/pipeline/task.py | 42 +++++++++++++++++++++-- src/pipecat/processors/frame_processor.py | 6 ++++ 3 files changed, 51 insertions(+), 3 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 6e5fa744f..e2ac25e7a 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -5,7 +5,7 @@ # from dataclasses import dataclass, field -from typing import Any, Awaitable, Callable, List, Literal, Mapping, Optional, Tuple +from typing import TYPE_CHECKING, Any, Awaitable, Callable, List, Literal, Mapping, Optional, Tuple from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.clocks.base_clock import BaseClock @@ -14,6 +14,9 @@ from pipecat.transcriptions.language import Language from pipecat.utils.time import nanoseconds_to_str from pipecat.utils.utils import obj_count, obj_id +if TYPE_CHECKING: + from pipecat.observers.base_observer import BaseObserver + def format_pts(pts: int | None): return nanoseconds_to_str(pts) if pts else None @@ -386,6 +389,7 @@ class StartFrame(SystemFrame): enable_metrics: bool = False enable_usage_metrics: bool = False report_only_initial_ttfb: bool = False + observer: Optional["BaseObserver"] = None @dataclass diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 79a708c74..db9a00049 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -5,10 +5,10 @@ # import asyncio -from typing import AsyncIterable, Iterable +from typing import AsyncIterable, Iterable, List from loguru import logger -from pydantic import BaseModel +from pydantic import BaseModel, ConfigDict from pipecat.clocks.base_clock import BaseClock from pipecat.clocks.system_clock import SystemClock @@ -24,20 +24,31 @@ from pipecat.frames.frames import ( StopTaskFrame, ) from pipecat.metrics.metrics import ProcessingMetricsData, TTFBMetricsData +from pipecat.observers.base_observer import BaseObserver from pipecat.pipeline.base_pipeline import BasePipeline from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.utils.utils import obj_count, obj_id class PipelineParams(BaseModel): + model_config = ConfigDict(arbitrary_types_allowed=True) + allow_interruptions: bool = False enable_metrics: bool = False enable_usage_metrics: bool = False send_initial_empty_metrics: bool = True report_only_initial_ttfb: bool = False + observers: List[BaseObserver] = [] class Source(FrameProcessor): + """This is the source processor that is linked at the beginning of the + pipeline given to the pipeline task. It allows us to easily push frames + downstream to the pipeline and also receive upstream frames coming from the + pipeline. + + """ + def __init__(self, up_queue: asyncio.Queue): super().__init__() self._up_queue = up_queue @@ -68,6 +79,12 @@ class Source(FrameProcessor): class Sink(FrameProcessor): + """This is the sink processor that is linked at the end of the pipeline + given to the pipeline task. It allows us to receive downstream frames and + act on them, for example, waiting to receive an EndFrame. + + """ + def __init__(self, down_queue: asyncio.Queue): super().__init__() self._down_queue = down_queue @@ -80,6 +97,24 @@ class Sink(FrameProcessor): await self._down_queue.put(frame) +class Observer(BaseObserver): + """This is a pipeline frame observer that is used as a proxy to the user + provided observers. That is, this is the only observer passed to the frame + processors. Then, every time a frame is pushed this observer will call all + the observers registered to the pipeline task. + + """ + + def __init__(self, observers: List[BaseObserver] = []): + self._observers = observers + + async def on_push_frame( + self, src: FrameProcessor, dst: FrameProcessor, frame: Frame, direction: FrameDirection + ): + for observer in self._observers: + await observer.on_push_frame(src, dst, frame, direction) + + class PipelineTask: def __init__( self, @@ -105,6 +140,8 @@ class PipelineTask: self._sink = Sink(self._down_queue) pipeline.link(self._sink) + self._observer = Observer(params.observers) + def has_finished(self): return self._finished @@ -156,6 +193,7 @@ class PipelineTask: enable_metrics=self._params.enable_metrics, enable_usage_metrics=self._params.enable_usage_metrics, report_only_initial_ttfb=self._params.report_only_initial_ttfb, + observer=self._observer, clock=self._clock, ) await self._source.queue_frame(start_frame, FrameDirection.DOWNSTREAM) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 653b313b8..ed6f8a4cb 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -58,6 +58,7 @@ class FrameProcessor: self._enable_metrics = False self._enable_usage_metrics = False self._report_only_initial_ttfb = False + self._observer = None # Cancellation is done through CancelFrame (a system frame). This could # cause other events being triggered (e.g. closing a transport) which @@ -194,6 +195,7 @@ class FrameProcessor: self._enable_metrics = frame.enable_metrics self._enable_usage_metrics = frame.enable_usage_metrics self._report_only_initial_ttfb = frame.report_only_initial_ttfb + self._observer = frame.observer elif isinstance(frame, StartInterruptionFrame): await self._start_interruption() await self.stop_all_metrics() @@ -258,9 +260,13 @@ class FrameProcessor: try: if direction == FrameDirection.DOWNSTREAM and self._next: logger.trace(f"Pushing {frame} from {self} to {self._next}") + if self._observer: + await self._observer.on_push_frame(self, self._next, frame, direction) await self._next.queue_frame(frame, direction) elif direction == FrameDirection.UPSTREAM and self._prev: logger.trace(f"Pushing {frame} upstream from {self} to {self._prev}") + if self._observer: + await self._observer.on_push_frame(self, self._prev, frame, direction) await self._prev.queue_frame(frame, direction) except Exception as e: logger.exception(f"Uncaught exception in {self}: {e}") From dd9f9179cc948853fdbbf5e5094e24540b9542ae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 15 Jan 2025 09:49:08 -0800 Subject: [PATCH 4/7] rtvi(RTVIObserver): use observers for RTVI server->client messages --- src/pipecat/processors/frameworks/rtvi.py | 146 ++++++++++++++++++++++ 1 file changed, 146 insertions(+) diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 36b782f80..fa72d5499 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -52,6 +52,7 @@ from pipecat.metrics.metrics import ( TTFBMetricsData, TTSUsageMetricsData, ) +from pipecat.observers.base_observer import BaseObserver from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContext, OpenAILLMContextFrame, @@ -564,6 +565,148 @@ class RTVIMetricsProcessor(RTVIFrameProcessor): await self._push_transport_message_urgent(message) +class RTVIObserver(BaseObserver): + """This is a pipeline frame observer that is used to send RTVI server + messages to clients. The observer does not handle incoming RTVI client + messages, which is done by the RTVIProcessor. + + """ + + def __init__(self, rtvi: FrameProcessor): + super().__init__() + self._rtvi = rtvi + self._bot_transcription = "" + self._frames_seen = set() + + async def on_push_frame( + self, src: FrameProcessor, dst: FrameProcessor, frame: Frame, direction: FrameDirection + ): + # If we have already seen this frame, let's skip it. + if frame.id in self._frames_seen: + return + self._frames_seen.add(frame.id) + + if isinstance(frame, (UserStartedSpeakingFrame, UserStoppedSpeakingFrame)): + await self._handle_interruptions(frame) + elif isinstance(frame, (BotStartedSpeakingFrame, BotStoppedSpeakingFrame)): + await self._handle_bot_speaking(frame) + elif isinstance(frame, (TranscriptionFrame, InterimTranscriptionFrame)): + await self._handle_user_transcriptions(frame) + elif isinstance(frame, OpenAILLMContextFrame): + await self._handle_context(frame) + elif isinstance(frame, UserStartedSpeakingFrame): + await self._push_bot_transcription() + elif isinstance(frame, LLMFullResponseStartFrame): + await self._push_transport_message_urgent(RTVIBotLLMStartedMessage()) + elif isinstance(frame, LLMFullResponseEndFrame): + await self._push_transport_message_urgent(RTVIBotLLMStoppedMessage()) + elif isinstance(frame, LLMTextFrame): + await self._handle_llm_text_frame(frame) + elif isinstance(frame, TTSStartedFrame): + await self._push_transport_message_urgent(RTVIBotTTSStartedMessage()) + elif isinstance(frame, TTSStoppedFrame): + await self._push_transport_message_urgent(RTVIBotTTSStoppedMessage()) + elif isinstance(frame, TTSTextFrame): + message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=frame.text)) + await self._push_transport_message_urgent(message) + elif isinstance(frame, MetricsFrame): + await self._handle_metrics(frame) + + async def _push_transport_message_urgent(self, model: BaseModel, exclude_none: bool = True): + frame = TransportMessageUrgentFrame(message=model.model_dump(exclude_none=exclude_none)) + await self._rtvi.push_frame(frame) + + async def _push_bot_transcription(self): + if len(self._bot_transcription) > 0: + message = RTVIBotTranscriptionMessage( + data=RTVITextMessageData(text=self._bot_transcription) + ) + await self._push_transport_message_urgent(message) + self._bot_transcription = "" + + async def _handle_interruptions(self, frame: Frame): + message = None + if isinstance(frame, UserStartedSpeakingFrame): + message = RTVIUserStartedSpeakingMessage() + elif isinstance(frame, UserStoppedSpeakingFrame): + message = RTVIUserStoppedSpeakingMessage() + + if message: + await self._push_transport_message_urgent(message) + + async def _handle_bot_speaking(self, frame: Frame): + message = None + if isinstance(frame, BotStartedSpeakingFrame): + message = RTVIBotStartedSpeakingMessage() + elif isinstance(frame, BotStoppedSpeakingFrame): + message = RTVIBotStoppedSpeakingMessage() + + if message: + await self._push_transport_message_urgent(message) + + async def _handle_llm_text_frame(self, frame: LLMTextFrame): + message = RTVIBotLLMTextMessage(data=RTVITextMessageData(text=frame.text)) + await self._push_transport_message_urgent(message) + + self._bot_transcription += frame.text + if match_endofsentence(self._bot_transcription): + await self._push_bot_transcription() + + async def _handle_user_transcriptions(self, frame: Frame): + message = None + if isinstance(frame, TranscriptionFrame): + message = RTVIUserTranscriptionMessage( + data=RTVIUserTranscriptionMessageData( + text=frame.text, user_id=frame.user_id, timestamp=frame.timestamp, final=True + ) + ) + elif isinstance(frame, InterimTranscriptionFrame): + message = RTVIUserTranscriptionMessage( + data=RTVIUserTranscriptionMessageData( + text=frame.text, user_id=frame.user_id, timestamp=frame.timestamp, final=False + ) + ) + + if message: + await self._push_transport_message_urgent(message) + + async def _handle_context(self, frame: OpenAILLMContextFrame): + messages = frame.context.messages + if len(messages) > 0: + message = messages[-1] + if message["role"] == "user": + content = message["content"] + if isinstance(content, list): + text = " ".join(item["text"] for item in content if "text" in item) + else: + text = content + rtvi_message = RTVIUserLLMTextMessage(data=RTVITextMessageData(text=text)) + await self._push_transport_message_urgent(rtvi_message) + + async def _handle_metrics(self, frame: MetricsFrame): + metrics = {} + for d in frame.data: + if isinstance(d, TTFBMetricsData): + if "ttfb" not in metrics: + metrics["ttfb"] = [] + metrics["ttfb"].append(d.model_dump(exclude_none=True)) + elif isinstance(d, ProcessingMetricsData): + if "processing" not in metrics: + metrics["processing"] = [] + metrics["processing"].append(d.model_dump(exclude_none=True)) + elif isinstance(d, LLMUsageMetricsData): + if "tokens" not in metrics: + metrics["tokens"] = [] + metrics["tokens"].append(d.value.model_dump(exclude_none=True)) + elif isinstance(d, TTSUsageMetricsData): + if "characters" not in metrics: + metrics["characters"] = [] + metrics["characters"].append(d.model_dump(exclude_none=True)) + + message = RTVIMetricsMessage(data=metrics) + await self._push_transport_message_urgent(message) + + class RTVIProcessor(FrameProcessor): def __init__( self, @@ -594,6 +737,9 @@ class RTVIProcessor(FrameProcessor): self._register_event_handler("on_bot_started") self._register_event_handler("on_client_ready") + def observer(self) -> RTVIObserver: + return RTVIObserver(self) + def register_action(self, action: RTVIAction): id = self._action_id(action.service, action.action) self._registered_actions[id] = action From e50c76d07535042da904a87e6d434936cb02bd2d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 15 Jan 2025 09:50:04 -0800 Subject: [PATCH 5/7] examples(simple-chatbot): use RTVIObserver for server-client messages --- examples/simple-chatbot/server/bot-gemini.py | 30 ++------------------ examples/simple-chatbot/server/bot-openai.py | 28 ++---------------- 2 files changed, 4 insertions(+), 54 deletions(-) diff --git a/examples/simple-chatbot/server/bot-gemini.py b/examples/simple-chatbot/server/bot-gemini.py index 19e8137cb..16e3f8fcd 100644 --- a/examples/simple-chatbot/server/bot-gemini.py +++ b/examples/simple-chatbot/server/bot-gemini.py @@ -41,17 +41,8 @@ from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.processors.frameworks.rtvi import ( - RTVIBotTranscriptionProcessor, - RTVIConfig, - RTVIMetricsProcessor, - RTVIProcessor, - RTVISpeakingProcessor, - RTVIUserTranscriptionProcessor, -) -from pipecat.services.elevenlabs import ElevenLabsTTSService +from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIProcessor from pipecat.services.gemini_multimodal_live.gemini import GeminiMultimodalLiveLLMService -from pipecat.services.openai import OpenAILLMService from pipecat.transports.services.daily import DailyParams, DailyTransport load_dotenv(override=True) @@ -168,20 +159,6 @@ async def main(): # # RTVI events for Pipecat client UI # - - # This will send `user-*-speaking` and `bot-*-speaking` messages. - rtvi_speaking = RTVISpeakingProcessor() - - # This will emit UserTranscript events. - rtvi_user_transcription = RTVIUserTranscriptionProcessor() - - # This will emit BotTranscript events. - rtvi_bot_transcription = RTVIBotTranscriptionProcessor() - - # This will send `metrics` messages. - rtvi_metrics = RTVIMetricsProcessor() - - # Handles RTVI messages from the client rtvi = RTVIProcessor(config=RTVIConfig(config=[])) pipeline = Pipeline( @@ -190,11 +167,7 @@ async def main(): rtvi, context_aggregator.user(), llm, - rtvi_speaking, - rtvi_user_transcription, - rtvi_bot_transcription, ta, - rtvi_metrics, transport.output(), context_aggregator.assistant(), ] @@ -206,6 +179,7 @@ async def main(): allow_interruptions=True, enable_metrics=True, enable_usage_metrics=True, + observers=[rtvi.observer()], ), ) await task.queue_frame(quiet_frame) diff --git a/examples/simple-chatbot/server/bot-openai.py b/examples/simple-chatbot/server/bot-openai.py index da691d571..d23d254bf 100644 --- a/examples/simple-chatbot/server/bot-openai.py +++ b/examples/simple-chatbot/server/bot-openai.py @@ -41,14 +41,7 @@ from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.processors.frameworks.rtvi import ( - RTVIBotTranscriptionProcessor, - RTVIConfig, - RTVIMetricsProcessor, - RTVIProcessor, - RTVISpeakingProcessor, - RTVIUserTranscriptionProcessor, -) +from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIProcessor from pipecat.services.elevenlabs import ElevenLabsTTSService from pipecat.services.openai import OpenAILLMService from pipecat.transports.services.daily import DailyParams, DailyTransport @@ -189,34 +182,16 @@ async def main(): # # RTVI events for Pipecat client UI # - - # This will send `user-*-speaking` and `bot-*-speaking` messages. - rtvi_speaking = RTVISpeakingProcessor() - - # This will emit UserTranscript events. - rtvi_user_transcription = RTVIUserTranscriptionProcessor() - - # This will emit BotTranscript events. - rtvi_bot_transcription = RTVIBotTranscriptionProcessor() - - # This will send `metrics` messages. - rtvi_metrics = RTVIMetricsProcessor() - - # Handles RTVI messages from the client rtvi = RTVIProcessor(config=RTVIConfig(config=[])) pipeline = Pipeline( [ transport.input(), rtvi, - rtvi_speaking, - rtvi_user_transcription, context_aggregator.user(), llm, - rtvi_bot_transcription, tts, ta, - rtvi_metrics, transport.output(), context_aggregator.assistant(), ] @@ -228,6 +203,7 @@ async def main(): allow_interruptions=True, enable_metrics=True, enable_usage_metrics=True, + observers=[rtvi.observer()], ), ) await task.queue_frame(quiet_frame) From 45039e7cde1921aa0455eb6f67d558b7a9c61d39 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 15 Jan 2025 09:54:24 -0800 Subject: [PATCH 6/7] update CHANGELOG.md --- CHANGELOG.md | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 80c7c0c7d..232bf0f4f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,9 +9,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Introduced pipeline frame observers. Observers can view all the frames that go + through the pipeline without the need to inject processors in the + pipeline. This can be useful, for example, to implement frame loggers or + debuggers among other things. + - Added `OpenRouter` for OpenRouter integration with an OpenAI-compatible interface. Added foundational example `14m-function-calling-openrouter.py`. - + - Added a new `WebsocketService` based class for TTS services, containing base functions and retry logic. From 08f1dda94ebbd657d38a91d0851e359aecd7b753 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 15 Jan 2025 10:15:36 -0800 Subject: [PATCH 7/7] observers: add a timestamp to on_push_frame() --- src/pipecat/observers/base_observer.py | 8 +++++++- src/pipecat/pipeline/task.py | 9 +++++++-- src/pipecat/processors/frame_processor.py | 9 +++++++-- src/pipecat/processors/frameworks/rtvi.py | 7 ++++++- 4 files changed, 27 insertions(+), 6 deletions(-) diff --git a/src/pipecat/observers/base_observer.py b/src/pipecat/observers/base_observer.py index 2a17b5668..46f746946 100644 --- a/src/pipecat/observers/base_observer.py +++ b/src/pipecat/observers/base_observer.py @@ -20,7 +20,12 @@ class BaseObserver(ABC): @abstractmethod async def on_push_frame( - self, src: FrameProcessor, dst: FrameProcessor, frame: Frame, direction: FrameDirection + self, + src: FrameProcessor, + dst: FrameProcessor, + frame: Frame, + direction: FrameDirection, + timestamp: int, ): """Abstract method to handle the event when a frame is pushed from one processor to another. @@ -30,6 +35,7 @@ class BaseObserver(ABC): dst (FrameProcessor): The destination frame processor that will receive the frame. frame (Frame): The frame being transferred between processors. direction (FrameDirection): The direction of the frame transfer. + timestamp (int): The timestamp when the frame was pushed (based on the pipeline clock). This method should be implemented by subclasses to define specific behavior when a frame is pushed. diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index db9a00049..fc404f75b 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -109,10 +109,15 @@ class Observer(BaseObserver): self._observers = observers async def on_push_frame( - self, src: FrameProcessor, dst: FrameProcessor, frame: Frame, direction: FrameDirection + self, + src: FrameProcessor, + dst: FrameProcessor, + frame: Frame, + direction: FrameDirection, + timestamp: int, ): for observer in self._observers: - await observer.on_push_frame(src, dst, frame, direction) + await observer.on_push_frame(src, dst, frame, direction, timestamp) class PipelineTask: diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index ed6f8a4cb..302ec06d1 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -258,15 +258,20 @@ class FrameProcessor: async def __internal_push_frame(self, frame: Frame, direction: FrameDirection): try: + timestamp = self._clock.get_time() if direction == FrameDirection.DOWNSTREAM and self._next: logger.trace(f"Pushing {frame} from {self} to {self._next}") if self._observer: - await self._observer.on_push_frame(self, self._next, frame, direction) + await self._observer.on_push_frame( + self, self._next, frame, direction, timestamp + ) await self._next.queue_frame(frame, direction) elif direction == FrameDirection.UPSTREAM and self._prev: logger.trace(f"Pushing {frame} upstream from {self} to {self._prev}") if self._observer: - await self._observer.on_push_frame(self, self._prev, frame, direction) + await self._observer.on_push_frame( + self, self._prev, frame, direction, timestamp + ) await self._prev.queue_frame(frame, direction) except Exception as e: logger.exception(f"Uncaught exception in {self}: {e}") diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index fa72d5499..a689b93ad 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -579,7 +579,12 @@ class RTVIObserver(BaseObserver): self._frames_seen = set() async def on_push_frame( - self, src: FrameProcessor, dst: FrameProcessor, frame: Frame, direction: FrameDirection + self, + src: FrameProcessor, + dst: FrameProcessor, + frame: Frame, + direction: FrameDirection, + timestamp: int, ): # If we have already seen this frame, let's skip it. if frame.id in self._frames_seen: