From cb49b6a0d678662cdc2483b2497e00fbf0af28bc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 23 Sep 2024 23:39:05 -0700 Subject: [PATCH 1/2] rtvi: add llm-text and tts-text server messages --- src/pipecat/processors/frameworks/rtvi.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 820ea716c..7a35c5c06 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -242,6 +242,22 @@ class RTVILLMFunctionCallResultData(BaseModel): result: dict | str +class RTVITextMessageData(BaseModel): + text: str + + +class RTVILLMTextMessage(BaseModel): + label: Literal["rtvi-ai"] = "rtvi-ai" + type: Literal["llm-text"] = "llm-text" + data: RTVITextMessageData + + +class RTVITTSTextMessage(BaseModel): + label: Literal["rtvi-ai"] = "rtvi-ai" + type: Literal["tts-text"] = "tts-text" + data: RTVITextMessageData + + class RTVITranscriptionMessageData(BaseModel): text: str user_id: str From 08ac311971afb82fff28efa58af81814d9c052c0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 24 Sep 2024 09:35:36 -0700 Subject: [PATCH 2/2] rtvi: use task to process incoming action frames --- src/pipecat/processors/frameworks/rtvi.py | 24 ++++++++++++++++++++--- 1 file changed, 21 insertions(+), 3 deletions(-) diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 7a35c5c06..bb23c9856 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -316,6 +316,11 @@ class RTVIProcessor(FrameProcessor): self._registered_actions: Dict[str, RTVIAction] = {} self._registered_services: Dict[str, RTVIService] = {} + # A task to process incoming action frames. + self._action_task = self.get_event_loop().create_task(self._action_task_handler()) + self._action_queue = asyncio.Queue() + + # A task to process incoming transport messages. self._message_task = self.get_event_loop().create_task(self._message_task_handler()) self._message_queue = asyncio.Queue() @@ -401,7 +406,7 @@ class RTVIProcessor(FrameProcessor): elif isinstance(frame, TransportMessageFrame): await self._message_queue.put(frame) elif isinstance(frame, RTVIActionFrame): - await self._handle_action(frame.message_id, frame.rtvi_action_run) + await self._action_queue.put(frame) # Other frames else: await self.push_frame(frame, direction) @@ -415,12 +420,16 @@ class RTVIProcessor(FrameProcessor): await self._maybe_send_bot_ready() async def _stop(self, frame: EndFrame): - # We need to cancel the message task handler because that one is not - # processing EndFrames. + self._action_task.cancel() + await self._action_task + self._message_task.cancel() await self._message_task async def _cancel(self, frame: CancelFrame): + self._action_task.cancel() + await self._action_task + self._message_task.cancel() await self._message_task @@ -471,6 +480,15 @@ class RTVIProcessor(FrameProcessor): if message: await self._push_transport_message(message) + async def _action_task_handler(self): + while True: + try: + frame = await self._action_queue.get() + await self._handle_action(frame.message_id, frame.rtvi_action_run) + self._action_queue.task_done() + except asyncio.CancelledError: + break + async def _message_task_handler(self): while True: try: