diff --git a/CHANGELOG.md b/CHANGELOG.md index f5d2d40df..9d232ae79 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,10 +5,19 @@ All notable changes to **Pipecat** will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [0.0.43] - 2024-10-03 +## [Unreleased] + +### Added + +- Added new `RTVIUserLLMTextProcessor`. This processor will send an RTVI + `user-llm-text` message with the user content's that was sent to the LLM. ### Changed +- `TransportMessageFrame` doesn't have an `urgent` field anymore, instead + there's now a `TransportMessageUrgentFrame` which is a `SystemFrame` and + therefore skip all internal queuing. + - For TTS services, convert inputted languages to match each service's language format diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index e2ef78df5..174815651 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -269,7 +269,6 @@ class TTSSpeakFrame(DataFrame): @dataclass class TransportMessageFrame(DataFrame): message: Any - urgent: bool = False def __str__(self): return f"{self.name}(message: {self.message})" @@ -405,6 +404,14 @@ class BotInterruptionFrame(SystemFrame): pass +@dataclass +class TransportMessageUrgentFrame(SystemFrame): + message: Any + + def __str__(self): + return f"{self.name}(message: {self.message})" + + @dataclass class MetricsFrame(SystemFrame): """Emitted by processor that can compute metrics like latencies.""" diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 7a6054c3c..7020ea857 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -6,10 +6,11 @@ import asyncio import base64 - -from typing import Any, Awaitable, Callable, Dict, List, Literal, Optional, Union -from pydantic import BaseModel, Field, PrivateAttr, ValidationError from dataclasses import dataclass +from typing import Any, Awaitable, Callable, Dict, List, Literal, Optional, Union + +from loguru import logger +from pydantic import BaseModel, Field, PrivateAttr, ValidationError from pipecat.frames.frames import ( BotInterruptionFrame, @@ -20,27 +21,28 @@ from pipecat.frames.frames import ( EndFrame, ErrorFrame, Frame, + FunctionCallResultFrame, InterimTranscriptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, OutputAudioRawFrame, StartFrame, SystemFrame, - TTSStartedFrame, - TTSStoppedFrame, TextFrame, TranscriptionFrame, TransportMessageFrame, + TransportMessageUrgentFrame, + TTSStartedFrame, + TTSStoppedFrame, UserStartedSpeakingFrame, - FunctionCallResultFrame, UserStoppedSpeakingFrame, ) -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.aggregators.openai_llm_context import ( + OpenAILLMContext, + OpenAILLMContextFrame, +) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from loguru import logger - - RTVI_PROTOCOL_VERSION = "0.2" ActionResult = Union[bool, int, float, str, list, dict] @@ -291,22 +293,12 @@ class RTVIAudioMessageData(BaseModel): num_channels: int -class RTVIBotAudioMessage(BaseModel): +class RTVIBotTTSAudioMessage(BaseModel): label: Literal["rtvi-ai"] = "rtvi-ai" - type: Literal["bot-audio"] = "bot-audio" + type: Literal["bot-tts-audio"] = "bot-tts-audio" data: RTVIAudioMessageData -class RTVIBotTranscriptionMessageData(BaseModel): - text: str - - -class RTVIBotTranscriptionMessage(BaseModel): - label: Literal["rtvi-ai"] = "rtvi-ai" - type: Literal["bot-transcription"] = "bot-transcription" - data: RTVIBotTranscriptionMessageData - - class RTVIUserTranscriptionMessageData(BaseModel): text: str user_id: str @@ -320,6 +312,12 @@ class RTVIUserTranscriptionMessage(BaseModel): data: RTVIUserTranscriptionMessageData +class RTVIUserLLMTextMessage(BaseModel): + label: Literal["rtvi-ai"] = "rtvi-ai" + type: Literal["user-llm-text"] = "user-llm-text" + data: RTVITextMessageData + + class RTVIUserStartedSpeakingMessage(BaseModel): label: Literal["rtvi-ai"] = "rtvi-ai" type: Literal["user-started-speaking"] = "user-started-speaking" @@ -350,9 +348,11 @@ class RTVIFrameProcessor(FrameProcessor): self._direction = direction async def _push_transport_message(self, model: BaseModel, exclude_none: bool = True): - frame = TransportMessageFrame( - message=model.model_dump(exclude_none=exclude_none), urgent=True - ) + frame = TransportMessageFrame(message=model.model_dump(exclude_none=exclude_none)) + await self.push_frame(frame, self._direction) + + 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.push_frame(frame, self._direction) @@ -378,7 +378,7 @@ class RTVISpeakingProcessor(RTVIFrameProcessor): message = RTVIUserStoppedSpeakingMessage() if message: - await self._push_transport_message(message) + await self._push_transport_message_urgent(message) async def _handle_bot_speaking(self, frame: Frame): message = None @@ -388,7 +388,7 @@ class RTVISpeakingProcessor(RTVIFrameProcessor): message = RTVIBotStoppedSpeakingMessage() if message: - await self._push_transport_message(message) + await self._push_transport_message_urgent(message) class RTVIUserTranscriptionProcessor(RTVIFrameProcessor): @@ -419,7 +419,36 @@ class RTVIUserTranscriptionProcessor(RTVIFrameProcessor): ) if message: - await self._push_transport_message(message) + await self._push_transport_message_urgent(message) + + +class RTVIUserLLMTextProcessor(RTVIFrameProcessor): + def __init__(self, **kwargs): + super().__init__(**kwargs) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + await self.push_frame(frame, direction) + + if isinstance(frame, OpenAILLMContextFrame): + await self._handle_context(frame) + + 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): + print("LIST") + text = " ".join(item["text"] for item in content if "text" in item) + else: + print("STRING") + text = content + + rtvi_message = RTVIUserLLMTextMessage(data=RTVITextMessageData(text=text)) + await self._push_transport_message_urgent(rtvi_message) class RTVIBotLLMProcessor(RTVIFrameProcessor): @@ -432,9 +461,9 @@ class RTVIBotLLMProcessor(RTVIFrameProcessor): await self.push_frame(frame, direction) if isinstance(frame, LLMFullResponseStartFrame): - await self._push_transport_message(RTVIBotLLMStartedMessage()) + await self._push_transport_message_urgent(RTVIBotLLMStartedMessage()) elif isinstance(frame, LLMFullResponseEndFrame): - await self._push_transport_message(RTVIBotLLMStoppedMessage()) + await self._push_transport_message_urgent(RTVIBotLLMStoppedMessage()) class RTVIBotTTSProcessor(RTVIFrameProcessor): @@ -447,9 +476,9 @@ class RTVIBotTTSProcessor(RTVIFrameProcessor): await self.push_frame(frame, direction) if isinstance(frame, TTSStartedFrame): - await self._push_transport_message(RTVIBotTTSStartedMessage()) + await self._push_transport_message_urgent(RTVIBotTTSStartedMessage()) elif isinstance(frame, TTSStoppedFrame): - await self._push_transport_message(RTVIBotTTSStoppedMessage()) + await self._push_transport_message_urgent(RTVIBotTTSStoppedMessage()) class RTVIBotLLMTextProcessor(RTVIFrameProcessor): @@ -466,7 +495,7 @@ class RTVIBotLLMTextProcessor(RTVIFrameProcessor): async def _handle_text(self, frame: TextFrame): message = RTVIBotLLMTextMessage(data=RTVITextMessageData(text=frame.text)) - await self._push_transport_message(message) + await self._push_transport_message_urgent(message) class RTVIBotTTSTextProcessor(RTVIFrameProcessor): @@ -486,7 +515,7 @@ class RTVIBotTTSTextProcessor(RTVIFrameProcessor): await self._push_transport_message(message) -class RTVIBotAudioProcessor(RTVIFrameProcessor): +class RTVIBotTTSAudioProcessor(RTVIFrameProcessor): def __init__(self, **kwargs): super().__init__(**kwargs) @@ -500,7 +529,7 @@ class RTVIBotAudioProcessor(RTVIFrameProcessor): async def _handle_audio(self, frame: OutputAudioRawFrame): encoded = base64.b64encode(frame.audio).decode("utf-8") - message = RTVIBotAudioMessage( + message = RTVIBotTTSAudioMessage( data=RTVIAudioMessageData( audio=encoded, sample_rate=frame.sample_rate, num_channels=frame.num_channels ) @@ -647,9 +676,7 @@ class RTVIProcessor(FrameProcessor): self._message_task = None async def _push_transport_message(self, model: BaseModel, exclude_none: bool = True): - frame = TransportMessageFrame( - message=model.model_dump(exclude_none=exclude_none), urgent=True - ) + frame = TransportMessageUrgentFrame(message=model.model_dump(exclude_none=exclude_none)) await self.push_frame(frame) async def _action_task_handler(self): diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 6ee458f37..7ec503d24 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -33,6 +33,7 @@ from pipecat.frames.frames import ( TTSStoppedFrame, TextFrame, TransportMessageFrame, + TransportMessageUrgentFrame, ) from pipecat.transports.base_transport import TransportParams @@ -148,7 +149,7 @@ class BaseOutputTransport(FrameProcessor): await self._audio_out_task self._audio_out_task = None - async def send_message(self, frame: TransportMessageFrame): + async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): pass async def send_metrics(self, frame: MetricsFrame): @@ -186,6 +187,8 @@ class BaseOutputTransport(FrameProcessor): elif isinstance(frame, MetricsFrame): await self.push_frame(frame, direction) await self.send_metrics(frame) + elif isinstance(frame, TransportMessageUrgentFrame): + await self.send_message(frame) elif isinstance(frame, SystemFrame): await self.push_frame(frame, direction) # Control frames. @@ -198,8 +201,6 @@ class BaseOutputTransport(FrameProcessor): await self._handle_audio(frame) elif isinstance(frame, (OutputImageRawFrame, SpriteFrame)): await self._handle_image(frame) - elif isinstance(frame, TransportMessageFrame) and frame.urgent: - await self.send_message(frame) # TODO(aleix): Images and audio should support presentation timestamps. elif frame.pts: await self._sink_clock_queue.put((frame.pts, frame.id, frame)) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 50c2ae085..628fd61e2 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -35,6 +35,7 @@ from pipecat.frames.frames import ( StartFrame, TranscriptionFrame, TransportMessageFrame, + TransportMessageUrgentFrame, UserImageRawFrame, UserImageRequestFrame, ) @@ -70,6 +71,11 @@ class DailyTransportMessageFrame(TransportMessageFrame): participant_id: str | None = None +@dataclass +class DailyTransportMessageUrgentFrame(TransportMessageUrgentFrame): + participant_id: str | None = None + + class WebRTCVADAnalyzer(VADAnalyzer): def __init__(self, *, sample_rate=16000, num_channels=1, params: VADParams = VADParams()): super().__init__(sample_rate=sample_rate, num_channels=num_channels, params=params) @@ -234,12 +240,12 @@ class DailyTransportClient(EventHandler): def set_callbacks(self, callbacks: DailyCallbacks): self._callbacks = callbacks - async def send_message(self, frame: TransportMessageFrame): + async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): if not self._client: return participant_id = None - if isinstance(frame, DailyTransportMessageFrame): + if isinstance(frame, (DailyTransportMessageFrame, DailyTransportMessageUrgentFrame)): participant_id = frame.participant_id future = self._loop.create_future() @@ -736,7 +742,7 @@ class DailyOutputTransport(BaseOutputTransport): await super().cleanup() await self._client.cleanup() - async def send_message(self, frame: TransportMessageFrame): + async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): await self._client.send_message(frame) async def send_metrics(self, frame: MetricsFrame): diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index 6e5e48d0b..d97b9596e 100644 --- a/src/pipecat/transports/services/livekit.py +++ b/src/pipecat/transports/services/livekit.py @@ -22,6 +22,7 @@ from pipecat.frames.frames import ( MetricsFrame, StartFrame, TransportMessageFrame, + TransportMessageUrgentFrame, ) from pipecat.metrics.metrics import ( LLMUsageMetricsData, @@ -51,6 +52,11 @@ class LiveKitTransportMessageFrame(TransportMessageFrame): participant_id: str | None = None +@dataclass +class LiveKitTransportMessageUrgentFrame(TransportMessageUrgentFrame): + participant_id: str | None = None + + class LiveKitParams(TransportParams): audio_out_sample_rate: int = 48000 audio_out_channels: int = 1 @@ -420,8 +426,8 @@ class LiveKitOutputTransport(BaseOutputTransport): await super().cancel(frame) await self._client.disconnect() - async def send_message(self, frame: TransportMessageFrame): - if isinstance(frame, LiveKitTransportMessageFrame): + async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + if isinstance(frame, (LiveKitTransportMessageFrame, LiveKitTransportMessageUrgentFrame)): await self._client.send_data(frame.message.encode(), frame.participant_id) else: await self._client.send_data(frame.message.encode()) @@ -596,6 +602,13 @@ class LiveKitTransport(BaseTransport): frame = LiveKitTransportMessageFrame(message=message, participant_id=participant_id) await self._output.send_message(frame) + async def send_message_urgent(self, message: str, participant_id: str | None = None): + if self._output: + frame = LiveKitTransportMessageUrgentFrame( + message=message, participant_id=participant_id + ) + await self._output.send_message(frame) + async def cleanup(self): if self._input: await self._input.cleanup()