From 194790183a08153ef4248ae76fa7598f255deb3c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 17 Jul 2024 14:53:25 -0700 Subject: [PATCH 01/37] processor: add support for setting a processor parent --- src/pipecat/pipeline/pipeline.py | 2 ++ src/pipecat/processors/frame_processor.py | 9 ++++++++- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/src/pipecat/pipeline/pipeline.py b/src/pipecat/pipeline/pipeline.py index c13cc61a1..165717ad3 100644 --- a/src/pipecat/pipeline/pipeline.py +++ b/src/pipecat/pipeline/pipeline.py @@ -91,5 +91,7 @@ class Pipeline(BasePipeline): def _link_processors(self): prev = self._processors[0] for curr in self._processors[1:]: + prev.set_parent(self) prev.link(curr) prev = curr + prev.set_parent(self) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 18c03f169..405936e06 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -72,6 +72,7 @@ class FrameProcessor: **kwargs): self.id: int = obj_id() self.name = name or f"{self.__class__.__name__}#{obj_count(self)}" + self._parent: "FrameProcessor" | None = None self._prev: "FrameProcessor" | None = None self._next: "FrameProcessor" | None = None self._loop: asyncio.AbstractEventLoop = loop or asyncio.get_running_loop() @@ -126,7 +127,7 @@ class FrameProcessor: async def cleanup(self): pass - def link(self, processor: 'FrameProcessor'): + def link(self, processor: "FrameProcessor"): self._next = processor processor._prev = self logger.debug(f"Linking {self} -> {self._next}") @@ -134,6 +135,12 @@ class FrameProcessor: def get_event_loop(self) -> asyncio.AbstractEventLoop: return self._loop + def set_parent(self, parent: "FrameProcessor"): + self._parent = parent + + def get_parent(self) -> "FrameProcessor": + return self._parent + async def process_frame(self, frame: Frame, direction: FrameDirection): if isinstance(frame, StartFrame): self._allow_interruptions = frame.allow_interruptions From 0a69a9e5eff681a5bb75a94fc976cc1553bef0af Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 17 Jul 2024 14:53:53 -0700 Subject: [PATCH 02/37] transport(daily): also accept TransportMessageFrame --- src/pipecat/transports/services/daily.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index baacda50e..b1939269b 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -198,14 +198,18 @@ class DailyTransportClient(EventHandler): def set_callbacks(self, callbacks: DailyCallbacks): self._callbacks = callbacks - async def send_message(self, frame: DailyTransportMessageFrame): + async def send_message(self, frame: TransportMessageFrame): if not self._client: return + participant_id = None + if isinstance(frame, DailyTransportMessageFrame): + participant_id = frame.participant_id + future = self._loop.create_future() self._client.send_app_message( frame.message, - frame.participant_id, + participant_id, completion=completion_callback(future)) await future @@ -655,7 +659,7 @@ class DailyOutputTransport(BaseOutputTransport): await super().cleanup() await self._client.cleanup() - async def send_message(self, frame: DailyTransportMessageFrame): + async def send_message(self, frame: TransportMessageFrame): await self._client.send_message(frame) async def send_metrics(self, frame: MetricsFrame): From 9f012c8002d381767dbba2a502ac74dcaea4076e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 17 Jul 2024 14:54:29 -0700 Subject: [PATCH 03/37] processors: add new RealtimeAIProcessor --- .../processors/frameworks/realtimeai.py | 131 ++++++++++++++++++ 1 file changed, 131 insertions(+) create mode 100644 src/pipecat/processors/frameworks/realtimeai.py diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py new file mode 100644 index 000000000..b68ce2ee5 --- /dev/null +++ b/src/pipecat/processors/frameworks/realtimeai.py @@ -0,0 +1,131 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import dataclasses + +from typing import List, Literal, Optional, Type +from pydantic import BaseModel, ValidationError + +from pipecat.frames.frames import Frame, StartFrame, TransportMessageFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.processors.aggregators.llm_response import LLMAssistantResponseAggregator, LLMUserResponseAggregator +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.services.ai_services import AIService +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.ollama import OLLamaLLMService +from pipecat.transports.base_transport import BaseTransport +from pipecat.vad.silero import SileroVAD + + +class RealtimeAILLMConfig(BaseModel): + model: str + messages: List[dict] + + +class RealtimeAITTSConfig(BaseModel): + voice: str + + +class RealtimeAIConfig(BaseModel): + llm: RealtimeAILLMConfig + tts: RealtimeAITTSConfig + + +class RealtimeAIMessageData(BaseModel): + config: Optional[RealtimeAIConfig] = None + + +class RealtimeAIMessage(BaseModel): + tag: Literal["realtime-ai"] = "realtime-ai" + type: str + data: RealtimeAIMessageData + + +class RealtimeAIResponseMessage(BaseModel): + tag: Literal["realtime-ai"] = "realtime-ai" + type: str + success: bool + error: Optional[str] = None + + +class RealtimeAIProcessor(FrameProcessor): + + def __init__( + self, + *, + transport: BaseTransport, + config: RealtimeAIConfig | None = None, + llm_api_key: str = "", + tts_api_key: str = "", + llm_cls: Type[AIService] = OLLamaLLMService, + tts_cls: Type[AIService] = CartesiaTTSService): + super().__init__() + self._transport = transport + self._config = config + self._llm_api_key = llm_api_key + self._tts_api_key = tts_api_key + self._llm_cls = llm_cls + self._tts_cls = tts_cls + self._start_frame: Frame | None = None + self._llm: FrameProcessor | None = None + self._tts: FrameProcessor | None = None + self._pipeline: FrameProcessor | None = None + + async def process_frame(self, frame: Frame, direction: FrameDirection): + if isinstance(frame, TransportMessageFrame): + await self._handle_message(frame) + else: + await self.push_frame(frame, direction) + + if isinstance(frame, StartFrame): + self._start_frame = frame + if self._config: + await self._handle_config(self._config) + + async def _handle_message(self, frame: TransportMessageFrame): + try: + message = RealtimeAIMessage.model_validate(frame.message) + + match message.type: + case "config": + await self._handle_config(RealtimeAIConfig.model_validate(message.data.config)) + except ValidationError as e: + await self._send_response("config", False, f"invalid configuration: {e}") + + async def _handle_config(self, config: RealtimeAIConfig): + try: + tma_in = LLMUserResponseAggregator(config.llm.messages) + tma_out = LLMAssistantResponseAggregator(config.llm.messages) + + vad = SileroVAD() + + self._llm = self._llm_cls(model=config.llm.model) + + self._tts = self._tts_cls(api_key=self._tts_api_key, voice_id=config.tts.voice) + + pipeline = Pipeline([vad, tma_in, self._llm, self._tts, + self._transport.output(), tma_out]) + self._pipeline = pipeline + + parent = self.get_parent() + if parent and self._start_frame: + parent.link(pipeline) + + # We need to initialize the new pipeline with the same settings + # as the initial one. + start_frame = dataclasses.replace(self._start_frame) + await self.push_frame(start_frame) + + # We now send a message to indicate we successfully initialized + # the pipelines. + await self._send_response("config", True) + except Exception as e: + await self._send_response("config", False, f"unable to create pipeline: {e}") + + async def _send_response(self, type: str, success: bool, error: str | None = None): + response = RealtimeAIResponseMessage(type=type, success=success) + message = TransportMessageFrame(message=response.model_dump(exclude_none=True)) + await self.push_frame(message) From f551f55f03ca20afffd92033142f8235245a04af Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 17 Jul 2024 14:54:44 -0700 Subject: [PATCH 04/37] examples: add new foundational/18-realtime-ai.py --- examples/foundational/18-realtime-ai.py | 60 +++++++++++++++++++++++++ 1 file changed, 60 insertions(+) create mode 100644 examples/foundational/18-realtime-ai.py diff --git a/examples/foundational/18-realtime-ai.py b/examples/foundational/18-realtime-ai.py new file mode 100644 index 000000000..948330749 --- /dev/null +++ b/examples/foundational/18-realtime-ai.py @@ -0,0 +1,60 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import sys +import os + +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.pipeline.runner import PipelineRunner +from pipecat.processors.frameworks.realtimeai import RealtimeAIProcessor +from pipecat.transports.services.daily import DailyParams, DailyTransport + +from runner import configure + +from loguru import logger + +from dotenv import load_dotenv +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def main(room_url, token): + transport = DailyTransport( + room_url, + token, + "Realtime AI", + DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + transcription_enabled=True, + )) + + rtai = RealtimeAIProcessor(transport=transport, tts_api_key=os.getenv("CARTESIA_API_KEY")) + + runner = PipelineRunner() + + pipeline = Pipeline([transport.input(), rtai]) + + task = PipelineTask( + pipeline, + params=PipelineParams( + allow_interruptions=True, + enable_metrics=True, + report_only_initial_ttfb=True)) + + @transport.event_handler("on_participant_joined") + async def on_participant_joined(transport, participant): + transport.capture_participant_transcription(participant["id"]) + + await runner.run(task) + +if __name__ == "__main__": + (url, token) = configure() + asyncio.run(main(url, token)) From 3e738642a7ccd655ac5c48c7c90b5474230ec79b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 18 Jul 2024 14:52:48 -0700 Subject: [PATCH 05/37] processors(realtime-ai): add support for getting/updating LLM context --- src/pipecat/frames/frames.py | 10 ++ .../processors/aggregators/llm_response.py | 10 ++ .../processors/frameworks/realtimeai.py | 109 +++++++++++++----- 3 files changed, 101 insertions(+), 28 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index b8a2a6d06..515098a6e 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -158,6 +158,16 @@ class LLMMessagesFrame(DataFrame): messages: List[dict] +@dataclass +class LLMMessagesUpdateFrame(DataFrame): + """A frame containing a list of new LLM messages. These messages will + replace the current context LLM messages and should generate a new + LLMMessagesFrame. + + """ + messages: List[dict] + + @dataclass class TransportMessageFrame(DataFrame): message: Any diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index aa85c5ade..efc8af179 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -15,6 +15,7 @@ from pipecat.frames.frames import ( LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesFrame, + LLMMessagesUpdateFrame, StartInterruptionFrame, TranscriptionFrame, TextFrame, @@ -120,6 +121,15 @@ class LLMResponseAggregator(FrameProcessor): # Reset anyways self._reset() await self.push_frame(frame, direction) + elif isinstance(frame, LLMMessagesUpdateFrame): + # We push the frame downstream so the assistant aggregator gets + # updated as well. + await self.push_frame(frame) + # We can now reset this one. + self._reset() + self._messages = frame.messages + messages_frame = LLMMessagesFrame(self._messages) + await self.push_frame(messages_frame) else: await self.push_frame(frame, direction) diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py index b68ce2ee5..688538924 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/src/pipecat/processors/frameworks/realtimeai.py @@ -9,7 +9,7 @@ import dataclasses from typing import List, Literal, Optional, Type from pydantic import BaseModel, ValidationError -from pipecat.frames.frames import Frame, StartFrame, TransportMessageFrame +from pipecat.frames.frames import Frame, LLMMessagesFrame, LLMMessagesUpdateFrame, StartFrame, TransportMessageFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.processors.aggregators.llm_response import LLMAssistantResponseAggregator, LLMUserResponseAggregator from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -21,50 +21,61 @@ from pipecat.vad.silero import SileroVAD class RealtimeAILLMConfig(BaseModel): - model: str - messages: List[dict] + model: Optional[str] = None + messages: Optional[List[dict]] = None class RealtimeAITTSConfig(BaseModel): - voice: str + voice: Optional[str] = None class RealtimeAIConfig(BaseModel): - llm: RealtimeAILLMConfig - tts: RealtimeAITTSConfig + llm: Optional[RealtimeAILLMConfig] = None + tts: Optional[RealtimeAITTSConfig] = None + + +class RealtimeAISetup(BaseModel): + config: RealtimeAIConfig class RealtimeAIMessageData(BaseModel): + setup: Optional[RealtimeAISetup] = None config: Optional[RealtimeAIConfig] = None class RealtimeAIMessage(BaseModel): tag: Literal["realtime-ai"] = "realtime-ai" type: str - data: RealtimeAIMessageData + data: Optional[RealtimeAIMessageData] = None -class RealtimeAIResponseMessage(BaseModel): +class RealtimeAIBasicResponse(BaseModel): tag: Literal["realtime-ai"] = "realtime-ai" type: str success: bool error: Optional[str] = None +class RealtimeAILLMContextResponse(BaseModel): + tag: Literal["realtime-ai"] = "realtime-ai" + type: Literal["llm-context"] = "llm-context" + messages: List[dict] + + class RealtimeAIProcessor(FrameProcessor): def __init__( self, *, transport: BaseTransport, - config: RealtimeAIConfig | None = None, + setup: RealtimeAISetup | None = None, llm_api_key: str = "", tts_api_key: str = "", llm_cls: Type[AIService] = OLLamaLLMService, tts_cls: Type[AIService] = CartesiaTTSService): super().__init__() self._transport = transport - self._config = config + self._setup = setup self._llm_api_key = llm_api_key self._tts_api_key = tts_api_key self._llm_cls = llm_cls @@ -82,32 +93,48 @@ class RealtimeAIProcessor(FrameProcessor): if isinstance(frame, StartFrame): self._start_frame = frame - if self._config: - await self._handle_config(self._config) + if self._setup and self._setup.config: + await self._handle_setup(self._setup) async def _handle_message(self, frame: TransportMessageFrame): try: message = RealtimeAIMessage.model_validate(frame.message) - - match message.type: - case "config": - await self._handle_config(RealtimeAIConfig.model_validate(message.data.config)) except ValidationError as e: - await self._send_response("config", False, f"invalid configuration: {e}") + await self._send_response("setup", False, f"invalid message: {e}") + return + + print(message) - async def _handle_config(self, config: RealtimeAIConfig): try: - tma_in = LLMUserResponseAggregator(config.llm.messages) - tma_out = LLMAssistantResponseAggregator(config.llm.messages) + match message.type: + case "setup": + await self._handle_setup(RealtimeAISetup.model_validate(message.data.setup)) + case "llm-get-context": + await self._handle_llm_get_context() + case "llm-update-context": + await self._handle_llm_update_context(RealtimeAIConfig.model_validate(message.data.config)) + except ValidationError as e: + await self._send_response(message.type, False, f"invalid message: {e}") + async def _handle_setup(self, setup: RealtimeAISetup): + try: vad = SileroVAD() - self._llm = self._llm_cls(model=config.llm.model) + self._tma_in = LLMUserResponseAggregator(setup.config.llm.messages) + self._tma_out = LLMAssistantResponseAggregator(setup.config.llm.messages) - self._tts = self._tts_cls(api_key=self._tts_api_key, voice_id=config.tts.voice) + self._llm = self._llm_cls(model=setup.config.llm.model) - pipeline = Pipeline([vad, tma_in, self._llm, self._tts, - self._transport.output(), tma_out]) + self._tts = self._tts_cls(api_key=self._tts_api_key, voice_id=setup.config.tts.voice) + + pipeline = Pipeline([ + vad, + self._tma_in, + self._llm, + self._tts, + self._transport.output(), + self._tma_out + ]) self._pipeline = pipeline parent = self.get_parent() @@ -121,11 +148,37 @@ class RealtimeAIProcessor(FrameProcessor): # We now send a message to indicate we successfully initialized # the pipelines. - await self._send_response("config", True) + await self._send_response("setup", True) except Exception as e: - await self._send_response("config", False, f"unable to create pipeline: {e}") + await self._send_response("setup", False, f"unable to create pipeline: {e}") - async def _send_response(self, type: str, success: bool, error: str | None = None): - response = RealtimeAIResponseMessage(type=type, success=success) + async def _handle_llm_get_context(self): + messages = self._tma_in.messages + response = RealtimeAILLMContextResponse(messages=messages) + message = TransportMessageFrame(message=response.model_dump(exclude_none=True)) + await self.push_frame(message) + + async def _handle_llm_update_context(self, config: RealtimeAIConfig): + if config.llm and config.llm.messages: + frame = LLMMessagesUpdateFrame(config.llm.messages) + await self.push_frame(frame) + + async def _send_response(self, type: str, success: bool, error: str | None = None): + # TODO(aleix): This is a bit hacky, but we might get invalid + # configuration or something might going wrong during setup and we would + # like to send the error to the client. However, if the pipeline is not + # setup yet we don't have an output transport and therefore we can't + # send any messages. So, we setup a super basic pipeline with just the + # output transport so we can send messages. + if not self._pipeline: + # We add the SilerVAD() so the audio doesn't go through. + pipeline = Pipeline([SileroVAD(), self._transport.output()]) + self._pipeline = pipeline + + parent = self.get_parent() + if parent and self._start_frame: + parent.link(pipeline) + + response = RealtimeAIBasicResponse(type=type, success=success, error=error) message = TransportMessageFrame(message=response.model_dump(exclude_none=True)) await self.push_frame(message) From 29a85302215e3558911df3338b101e22f49e45eb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 18 Jul 2024 17:17:03 -0700 Subject: [PATCH 06/37] processors(realtime-ai): add support for updating config (model, voice...) --- src/pipecat/frames/frames.py | 14 ++++++++++ .../processors/frameworks/realtimeai.py | 26 ++++++++++++++++--- src/pipecat/services/ai_services.py | 7 +++++ src/pipecat/services/anthropic.py | 4 +++ src/pipecat/services/azure.py | 4 +++ src/pipecat/services/cartesia.py | 4 +++ src/pipecat/services/deepgram.py | 4 +++ src/pipecat/services/elevenlabs.py | 4 +++ src/pipecat/services/google.py | 9 ++++++- src/pipecat/services/openai.py | 8 ++++++ src/pipecat/services/xtts.py | 4 +++ 11 files changed, 83 insertions(+), 5 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 515098a6e..f28f2b85e 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -345,3 +345,17 @@ class UserImageRequestFrame(ControlFrame): def __str__(self): return f"{self.name}, user: {self.user_id}" + + +@dataclass +class LLMModelUpdateFrame(ControlFrame): + """A control frame containing a request to update to a new LLM model. + """ + model: str + + +@dataclass +class TTSVoiceUpdateFrame(ControlFrame): + """A control frame containing a request to update to a new TTS voice. + """ + voice: str diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py index 688538924..b32a77366 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/src/pipecat/processors/frameworks/realtimeai.py @@ -9,9 +9,16 @@ import dataclasses from typing import List, Literal, Optional, Type from pydantic import BaseModel, ValidationError -from pipecat.frames.frames import Frame, LLMMessagesFrame, LLMMessagesUpdateFrame, StartFrame, TransportMessageFrame +from pipecat.frames.frames import ( + Frame, + LLMMessagesUpdateFrame, + LLMModelUpdateFrame, + StartFrame, + TTSVoiceUpdateFrame, + TransportMessageFrame) from pipecat.pipeline.pipeline import Pipeline -from pipecat.processors.aggregators.llm_response import LLMAssistantResponseAggregator, LLMUserResponseAggregator +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantResponseAggregator, LLMUserResponseAggregator) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.ai_services import AIService from pipecat.services.cartesia import CartesiaTTSService @@ -103,12 +110,12 @@ class RealtimeAIProcessor(FrameProcessor): await self._send_response("setup", False, f"invalid message: {e}") return - print(message) - try: match message.type: case "setup": await self._handle_setup(RealtimeAISetup.model_validate(message.data.setup)) + case "config-update": + await self._handle_config_update(RealtimeAIConfig.model_validate(message.data.config)) case "llm-get-context": await self._handle_llm_get_context() case "llm-update-context": @@ -152,6 +159,17 @@ class RealtimeAIProcessor(FrameProcessor): except Exception as e: await self._send_response("setup", False, f"unable to create pipeline: {e}") + async def _handle_config_update(self, config: RealtimeAIConfig): + if config.llm and config.llm.model: + frame = LLMModelUpdateFrame(config.llm.model) + await self.push_frame(frame) + if config.llm and config.llm.messages: + frame = LLMMessagesUpdateFrame(config.llm.messages) + await self.push_frame(frame) + if config.tts and config.tts.voice: + frame = TTSVoiceUpdateFrame(config.tts.voice) + await self.push_frame(frame) + async def _handle_llm_get_context(self): messages = self._tma_in.messages response = RealtimeAILLMContextResponse(messages=messages) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 8647d3f77..e32419965 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -21,6 +21,7 @@ from pipecat.frames.frames import ( StartInterruptionFrame, TTSStartedFrame, TTSStoppedFrame, + TTSVoiceUpdateFrame, TextFrame, VisionImageRawFrame, ) @@ -148,6 +149,10 @@ class TTSService(AIService): self._push_text_frames: bool = push_text_frames self._current_sentence: str = "" + @abstractmethod + async def set_voice(self, voice: str): + pass + # Converts the text to audio. @abstractmethod async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: @@ -203,6 +208,8 @@ class TTSService(AIService): await self.push_frame(frame, direction) else: await self.push_frame(frame, direction) + elif isinstance(frame, TTSVoiceUpdateFrame): + await self.set_voice(frame.voice) else: await self.push_frame(frame, direction) diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 8c165a750..7854bb792 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -8,6 +8,7 @@ import base64 from pipecat.frames.frames import ( Frame, + LLMModelUpdateFrame, TextFrame, VisionImageRawFrame, LLMMessagesFrame, @@ -134,6 +135,9 @@ class AnthropicLLMService(LLMService): context = OpenAILLMContext.from_messages(frame.messages) elif isinstance(frame, VisionImageRawFrame): context = OpenAILLMContext.from_image_frame(frame) + elif isinstance(frame, LLMModelUpdateFrame): + logger.debug(f"Switching LLM model to: [{frame.model}]") + self._model = frame.model else: await self.push_frame(frame, direction) diff --git a/src/pipecat/services/azure.py b/src/pipecat/services/azure.py index 013129d1d..8991f154a 100644 --- a/src/pipecat/services/azure.py +++ b/src/pipecat/services/azure.py @@ -81,6 +81,10 @@ class AzureTTSService(TTSService): def can_generate_metrics(self) -> bool: return True + async def set_voice(self, voice: str): + logger.debug(f"Switching TTS voice to: [{voice}]") + self._voice = voice + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index be0d80f0b..c672121be 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -87,6 +87,10 @@ class CartesiaTTSService(TTSService): def can_generate_metrics(self) -> bool: return True + async def set_voice(self, voice: str): + logger.debug(f"Switching TTS voice to: [{voice}]") + self._voice_id = voice + async def start(self, frame: StartFrame): await super().start(frame) await self._connect() diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index 59d323c33..e6ac09991 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -59,6 +59,10 @@ class DeepgramTTSService(TTSService): def can_generate_metrics(self) -> bool: return True + async def set_voice(self, voice: str): + logger.debug(f"Switching TTS voice to: [{voice}]") + self._voice = voice + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index c24736672..1bf0fe6ca 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -34,6 +34,10 @@ class ElevenLabsTTSService(TTSService): def can_generate_metrics(self) -> bool: return True + async def set_voice(self, voice: str): + logger.debug(f"Switching TTS voice to: [{voice}]") + self._voice_id = voice + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") diff --git a/src/pipecat/services/google.py b/src/pipecat/services/google.py index 6e719201b..7f20f1b8f 100644 --- a/src/pipecat/services/google.py +++ b/src/pipecat/services/google.py @@ -10,6 +10,7 @@ from typing import List from pipecat.frames.frames import ( Frame, + LLMModelUpdateFrame, TextFrame, VisionImageRawFrame, LLMMessagesFrame, @@ -43,11 +44,14 @@ class GoogleLLMService(LLMService): def __init__(self, *, api_key: str, model: str = "gemini-1.5-flash-latest", **kwargs): super().__init__(**kwargs) gai.configure(api_key=api_key) - self._client = gai.GenerativeModel(model) + self._create_client(model) def can_generate_metrics(self) -> bool: return True + def _create_client(self, model: str): + self._client = gai.GenerativeModel(model) + def _get_messages_from_openai_context( self, context: OpenAILLMContext) -> List[glm.Content]: openai_messages = context.get_messages() @@ -118,6 +122,9 @@ class GoogleLLMService(LLMService): context = OpenAILLMContext.from_messages(frame.messages) elif isinstance(frame, VisionImageRawFrame): context = OpenAILLMContext.from_image_frame(frame) + elif isinstance(frame, LLMModelUpdateFrame): + logger.debug(f"Switching LLM model to: [{frame.model}]") + self._create_client(frame.model) else: await self.push_frame(frame, direction) diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index d2d7ae175..3e1a6effc 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -22,6 +22,7 @@ from pipecat.frames.frames import ( LLMFullResponseEndFrame, LLMFullResponseStartFrame, LLMMessagesFrame, + LLMModelUpdateFrame, TextFrame, URLImageRawFrame, VisionImageRawFrame @@ -227,6 +228,9 @@ class BaseOpenAILLMService(LLMService): context = OpenAILLMContext.from_messages(frame.messages) elif isinstance(frame, VisionImageRawFrame): context = OpenAILLMContext.from_image_frame(frame) + elif isinstance(frame, LLMModelUpdateFrame): + logger.debug(f"Switching LLM model to: [{frame.model}]") + self._model = frame.model else: await self.push_frame(frame, direction) @@ -313,6 +317,10 @@ class OpenAITTSService(TTSService): def can_generate_metrics(self) -> bool: return True + async def set_voice(self, voice: str): + logger.debug(f"Switching TTS voice to: [{voice}]") + self._voice = voice + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") diff --git a/src/pipecat/services/xtts.py b/src/pipecat/services/xtts.py index 590a9dd3d..a17277b88 100644 --- a/src/pipecat/services/xtts.py +++ b/src/pipecat/services/xtts.py @@ -54,6 +54,10 @@ class XTTSService(TTSService): def can_generate_metrics(self) -> bool: return True + async def set_voice(self, voice: str): + logger.debug(f"Switching TTS voice to: [{voice}]") + self._voice_id = voice + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") embeddings = self._studio_speakers[self._voice_id] From 0a672e61e23c0b5d2c31c6abbc43a1c9bebdfdd7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 18 Jul 2024 22:33:01 -0700 Subject: [PATCH 07/37] processors(realtime-ai): update it to use groq by default --- examples/foundational/18-realtime-ai.py | 5 ++++- src/pipecat/processors/frameworks/realtimeai.py | 11 ++++++++--- 2 files changed, 12 insertions(+), 4 deletions(-) diff --git a/examples/foundational/18-realtime-ai.py b/examples/foundational/18-realtime-ai.py index 948330749..d00225a61 100644 --- a/examples/foundational/18-realtime-ai.py +++ b/examples/foundational/18-realtime-ai.py @@ -36,7 +36,10 @@ async def main(room_url, token): transcription_enabled=True, )) - rtai = RealtimeAIProcessor(transport=transport, tts_api_key=os.getenv("CARTESIA_API_KEY")) + rtai = RealtimeAIProcessor( + transport=transport, + llm_api_key=os.getenv("OPENAI_API_KEY"), + tts_api_key=os.getenv("CARTESIA_API_KEY")) runner = PipelineRunner() diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py index b32a77366..a0c9a1f45 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/src/pipecat/processors/frameworks/realtimeai.py @@ -22,7 +22,7 @@ from pipecat.processors.aggregators.llm_response import ( from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.ai_services import AIService from pipecat.services.cartesia import CartesiaTTSService -from pipecat.services.ollama import OLLamaLLMService +from pipecat.services.openai import OpenAILLMService from pipecat.transports.base_transport import BaseTransport from pipecat.vad.silero import SileroVAD @@ -77,13 +77,15 @@ class RealtimeAIProcessor(FrameProcessor): transport: BaseTransport, setup: RealtimeAISetup | None = None, llm_api_key: str = "", + llm_base_url: str = "https://api.groq.com/openai/v1", tts_api_key: str = "", - llm_cls: Type[AIService] = OLLamaLLMService, + llm_cls: Type[AIService] = OpenAILLMService, tts_cls: Type[AIService] = CartesiaTTSService): super().__init__() self._transport = transport self._setup = setup self._llm_api_key = llm_api_key + self._llm_base_url = llm_base_url self._tts_api_key = tts_api_key self._llm_cls = llm_cls self._tts_cls = tts_cls @@ -130,7 +132,10 @@ class RealtimeAIProcessor(FrameProcessor): self._tma_in = LLMUserResponseAggregator(setup.config.llm.messages) self._tma_out = LLMAssistantResponseAggregator(setup.config.llm.messages) - self._llm = self._llm_cls(model=setup.config.llm.model) + self._llm = self._llm_cls( + base_url=self._llm_base_url, + api_key=self._llm_api_key, + model=setup.config.llm.model) self._tts = self._tts_cls(api_key=self._tts_api_key, voice_id=setup.config.tts.voice) From bd9f4eea060b580088f0021e7a234352125b870a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 18 Jul 2024 22:54:49 -0700 Subject: [PATCH 08/37] processors(realtime-ai): provide default values --- .../processors/frameworks/realtimeai.py | 45 +++++++++++++++---- 1 file changed, 36 insertions(+), 9 deletions(-) diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py index a0c9a1f45..1a2b06b76 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/src/pipecat/processors/frameworks/realtimeai.py @@ -4,6 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # +from ctypes import DEFAULT_MODE import dataclasses from typing import List, Literal, Optional, Type @@ -26,6 +27,17 @@ from pipecat.services.openai import OpenAILLMService from pipecat.transports.base_transport import BaseTransport from pipecat.vad.silero import SileroVAD +DEFAULT_MESSAGES = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + } +] + +DEFAULT_MODEL = "llama3-70b-8192" + +DEFAULT_VOICE = "79a125e8-cd45-4c13-8a67-188112f4dd22" + class RealtimeAILLMConfig(BaseModel): model: Optional[str] = None @@ -42,7 +54,7 @@ class RealtimeAIConfig(BaseModel): class RealtimeAISetup(BaseModel): - config: RealtimeAIConfig + config: Optional[RealtimeAIConfig] = None class RealtimeAIMessageData(BaseModel): @@ -115,29 +127,44 @@ class RealtimeAIProcessor(FrameProcessor): try: match message.type: case "setup": - await self._handle_setup(RealtimeAISetup.model_validate(message.data.setup)) + setup = None + if message.data: + setup = message.data.setup + await self._handle_setup(setup) case "config-update": - await self._handle_config_update(RealtimeAIConfig.model_validate(message.data.config)) + await self._handle_config_update(message.data.config) case "llm-get-context": await self._handle_llm_get_context() case "llm-update-context": - await self._handle_llm_update_context(RealtimeAIConfig.model_validate(message.data.config)) + await self._handle_llm_update_context(message.data.config) except ValidationError as e: await self._send_response(message.type, False, f"invalid message: {e}") - async def _handle_setup(self, setup: RealtimeAISetup): + async def _handle_setup(self, setup: RealtimeAISetup | None): try: vad = SileroVAD() - self._tma_in = LLMUserResponseAggregator(setup.config.llm.messages) - self._tma_out = LLMAssistantResponseAggregator(setup.config.llm.messages) + model = DEFAULT_MODEL + if setup and setup.config and setup.config.llm and setup.config.llm.model: + model = setup.config.llm.model + + messages = DEFAULT_MESSAGES + if setup and setup.config and setup.config.llm and setup.config.llm.messages: + messages = setup.config.llm.messages + + voice = DEFAULT_VOICE + if setup and setup.config and setup.config.tts and setup.config.tts.voice: + messages = setup.config.tts.voice + + self._tma_in = LLMUserResponseAggregator(messages) + self._tma_out = LLMAssistantResponseAggregator(messages) self._llm = self._llm_cls( base_url=self._llm_base_url, api_key=self._llm_api_key, - model=setup.config.llm.model) + model=model) - self._tts = self._tts_cls(api_key=self._tts_api_key, voice_id=setup.config.tts.voice) + self._tts = self._tts_cls(api_key=self._tts_api_key, voice_id=voice) pipeline = Pipeline([ vad, From 46a048d7f60fadb503db39252d9eb7475460910d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 19 Jul 2024 08:37:04 -0700 Subject: [PATCH 09/37] processors(realtime-ai): allow default setup to be None --- src/pipecat/processors/frameworks/realtimeai.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py index 1a2b06b76..5aaf0e204 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/src/pipecat/processors/frameworks/realtimeai.py @@ -114,8 +114,7 @@ class RealtimeAIProcessor(FrameProcessor): if isinstance(frame, StartFrame): self._start_frame = frame - if self._setup and self._setup.config: - await self._handle_setup(self._setup) + await self._handle_setup(self._setup) async def _handle_message(self, frame: TransportMessageFrame): try: From f6e22bb3b9f13d2bd17b23adc9655b49645665d7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 19 Jul 2024 11:42:13 -0700 Subject: [PATCH 10/37] processors(realtime-ai): add silero vad to the transport --- examples/foundational/18-realtime-ai.py | 4 +++- src/pipecat/processors/frameworks/realtimeai.py | 7 +------ 2 files changed, 4 insertions(+), 7 deletions(-) diff --git a/examples/foundational/18-realtime-ai.py b/examples/foundational/18-realtime-ai.py index d00225a61..8a0a7683c 100644 --- a/examples/foundational/18-realtime-ai.py +++ b/examples/foundational/18-realtime-ai.py @@ -13,6 +13,7 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.pipeline.runner import PipelineRunner from pipecat.processors.frameworks.realtimeai import RealtimeAIProcessor from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.vad.silero import SileroVADAnalyzer from runner import configure @@ -31,9 +32,10 @@ async def main(room_url, token): token, "Realtime AI", DailyParams( - audio_in_enabled=True, audio_out_enabled=True, transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer() )) rtai = RealtimeAIProcessor( diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py index 5aaf0e204..d296416fa 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/src/pipecat/processors/frameworks/realtimeai.py @@ -25,7 +25,6 @@ from pipecat.services.ai_services import AIService from pipecat.services.cartesia import CartesiaTTSService from pipecat.services.openai import OpenAILLMService from pipecat.transports.base_transport import BaseTransport -from pipecat.vad.silero import SileroVAD DEFAULT_MESSAGES = [ { @@ -141,8 +140,6 @@ class RealtimeAIProcessor(FrameProcessor): async def _handle_setup(self, setup: RealtimeAISetup | None): try: - vad = SileroVAD() - model = DEFAULT_MODEL if setup and setup.config and setup.config.llm and setup.config.llm.model: model = setup.config.llm.model @@ -166,7 +163,6 @@ class RealtimeAIProcessor(FrameProcessor): self._tts = self._tts_cls(api_key=self._tts_api_key, voice_id=voice) pipeline = Pipeline([ - vad, self._tma_in, self._llm, self._tts, @@ -220,8 +216,7 @@ class RealtimeAIProcessor(FrameProcessor): # send any messages. So, we setup a super basic pipeline with just the # output transport so we can send messages. if not self._pipeline: - # We add the SilerVAD() so the audio doesn't go through. - pipeline = Pipeline([SileroVAD(), self._transport.output()]) + pipeline = Pipeline([self._transport.output()]) self._pipeline = pipeline parent = self.get_parent() From 4c629e538e75111bf9310e1fcb2cb2023310374f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 19 Jul 2024 11:42:36 -0700 Subject: [PATCH 11/37] processors(realtime-ai): add assistant before output transport Cartesia can do word-to-word output instead of full sentences. This means that for properly adding things into the context we need to add it before the transport, otherwise some words might be lost. --- src/pipecat/processors/frameworks/realtimeai.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py index d296416fa..df2e7650a 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/src/pipecat/processors/frameworks/realtimeai.py @@ -166,8 +166,8 @@ class RealtimeAIProcessor(FrameProcessor): self._tma_in, self._llm, self._tts, + self._tma_out, self._transport.output(), - self._tma_out ]) self._pipeline = pipeline From 846ae765e5dccf175738e76b298a0a6e1fe2dcb0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 19 Jul 2024 11:48:59 -0700 Subject: [PATCH 12/37] services(TTSService): fix sentence cleanup --- CHANGELOG.md | 3 +++ src/pipecat/services/ai_services.py | 3 ++- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 286325679..c28f5ede1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `TTSService` end of sentence detection has been improved. It now works with acronyms, numbers, hours and others. +- Fixed an issue in `TTSService` that would not properly flush the current + aggregated sentence if an `LLMFullResponseEndFrame` was found. + ### Performance - `CartesiaTTSService` now uses websockets which improves speed. It also diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index e32419965..ec89b9eba 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -201,8 +201,9 @@ class TTSService(AIService): elif isinstance(frame, StartInterruptionFrame): await self._handle_interruption(frame, direction) elif isinstance(frame, LLMFullResponseEndFrame) or isinstance(frame, EndFrame): + sentence = self._current_sentence self._current_sentence = "" - await self._push_tts_frames(self._current_sentence) + await self._push_tts_frames(sentence) if isinstance(frame, LLMFullResponseEndFrame): if self._push_text_frames: await self.push_frame(frame, direction) From b85dd7283af762cdfeb7e30003ee8da7a268ab89 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 19 Jul 2024 13:44:16 -0700 Subject: [PATCH 13/37] processors(realtime-ai): add support for appending to the LLM context --- src/pipecat/frames/frames.py | 9 +++++++++ src/pipecat/processors/aggregators/llm_response.py | 5 +++++ src/pipecat/processors/frameworks/realtimeai.py | 8 ++++++++ 3 files changed, 22 insertions(+) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index f28f2b85e..79fd24322 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -158,6 +158,15 @@ class LLMMessagesFrame(DataFrame): messages: List[dict] +@dataclass +class LLMMessagesAppendFrame(DataFrame): + """A frame containing a list of LLM messages that neeed to be added to the + current context. + + """ + messages: List[dict] + + @dataclass class LLMMessagesUpdateFrame(DataFrame): """A frame containing a list of new LLM messages. These messages will diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index efc8af179..6939a70c4 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -14,6 +14,7 @@ from pipecat.frames.frames import ( InterimTranscriptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, + LLMMessagesAppendFrame, LLMMessagesFrame, LLMMessagesUpdateFrame, StartInterruptionFrame, @@ -121,6 +122,10 @@ class LLMResponseAggregator(FrameProcessor): # Reset anyways self._reset() await self.push_frame(frame, direction) + elif isinstance(frame, LLMMessagesAppendFrame): + self._messages.extend(frame.messages) + messages_frame = LLMMessagesFrame(self._messages) + await self.push_frame(messages_frame) elif isinstance(frame, LLMMessagesUpdateFrame): # We push the frame downstream so the assistant aggregator gets # updated as well. diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py index df2e7650a..742e85198 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/src/pipecat/processors/frameworks/realtimeai.py @@ -12,6 +12,7 @@ from pydantic import BaseModel, ValidationError from pipecat.frames.frames import ( Frame, + LLMMessagesAppendFrame, LLMMessagesUpdateFrame, LLMModelUpdateFrame, StartFrame, @@ -133,6 +134,8 @@ class RealtimeAIProcessor(FrameProcessor): await self._handle_config_update(message.data.config) case "llm-get-context": await self._handle_llm_get_context() + case "llm-append-context": + await self._handle_llm_append_context(message.data.config) case "llm-update-context": await self._handle_llm_update_context(message.data.config) except ValidationError as e: @@ -203,6 +206,11 @@ class RealtimeAIProcessor(FrameProcessor): message = TransportMessageFrame(message=response.model_dump(exclude_none=True)) await self.push_frame(message) + async def _handle_llm_append_context(self, config: RealtimeAIConfig): + if config.llm and config.llm.messages: + frame = LLMMessagesAppendFrame(config.llm.messages) + await self.push_frame(frame) + async def _handle_llm_update_context(self, config: RealtimeAIConfig): if config.llm and config.llm.messages: frame = LLMMessagesUpdateFrame(config.llm.messages) From b0b1475563a0bf4ae3c1ec1d6b84b9108a2878ab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 19 Jul 2024 14:11:58 -0700 Subject: [PATCH 14/37] processors(realtime-ai): add support making TTS to speak --- src/pipecat/frames/frames.py | 9 +++++ .../processors/frameworks/realtimeai.py | 34 ++++++++++++++----- src/pipecat/services/ai_services.py | 5 ++- 3 files changed, 39 insertions(+), 9 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 79fd24322..09c475702 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -177,6 +177,15 @@ class LLMMessagesUpdateFrame(DataFrame): messages: List[dict] +@dataclass +class TTSSpeakFrame(DataFrame): + """A frame that contains a text that should be spoken by the TTS in the + pipeline (if any). + + """ + text: str + + @dataclass class TransportMessageFrame(DataFrame): message: Any diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py index 742e85198..3f032e4e7 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/src/pipecat/processors/frameworks/realtimeai.py @@ -16,6 +16,7 @@ from pipecat.frames.frames import ( LLMMessagesUpdateFrame, LLMModelUpdateFrame, StartFrame, + TTSSpeakFrame, TTSVoiceUpdateFrame, TransportMessageFrame) from pipecat.pipeline.pipeline import Pipeline @@ -57,9 +58,19 @@ class RealtimeAISetup(BaseModel): config: Optional[RealtimeAIConfig] = None +class RealtimeAILLMMessageData(BaseModel): + messages: List[dict] + + +class RealtimeAITTSMessageData(BaseModel): + text: str + + class RealtimeAIMessageData(BaseModel): setup: Optional[RealtimeAISetup] = None config: Optional[RealtimeAIConfig] = None + llm: Optional[RealtimeAILLMMessageData] = None + tts: Optional[RealtimeAITTSMessageData] = None class RealtimeAIMessage(BaseModel): @@ -135,9 +146,11 @@ class RealtimeAIProcessor(FrameProcessor): case "llm-get-context": await self._handle_llm_get_context() case "llm-append-context": - await self._handle_llm_append_context(message.data.config) + await self._handle_llm_append_context(message.data.llm) case "llm-update-context": - await self._handle_llm_update_context(message.data.config) + await self._handle_llm_update_context(message.data.llm) + case "tts-speak": + await self._handle_tts_speak(message.data.tts) except ValidationError as e: await self._send_response(message.type, False, f"invalid message: {e}") @@ -206,14 +219,19 @@ class RealtimeAIProcessor(FrameProcessor): message = TransportMessageFrame(message=response.model_dump(exclude_none=True)) await self.push_frame(message) - async def _handle_llm_append_context(self, config: RealtimeAIConfig): - if config.llm and config.llm.messages: - frame = LLMMessagesAppendFrame(config.llm.messages) + async def _handle_llm_append_context(self, data: RealtimeAILLMMessageData): + if data and data.messages: + frame = LLMMessagesAppendFrame(data.messages) await self.push_frame(frame) - async def _handle_llm_update_context(self, config: RealtimeAIConfig): - if config.llm and config.llm.messages: - frame = LLMMessagesUpdateFrame(config.llm.messages) + async def _handle_llm_update_context(self, data: RealtimeAILLMMessageData): + if data and data.messages: + frame = LLMMessagesUpdateFrame(data.messages) + await self.push_frame(frame) + + async def _handle_tts_speak(self, data: RealtimeAITTSMessageData): + if data and data.text: + frame = TTSSpeakFrame(text=data.text) await self.push_frame(frame) async def _send_response(self, type: str, success: bool, error: str | None = None): diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index ec89b9eba..bc00accdf 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -19,6 +19,7 @@ from pipecat.frames.frames import ( LLMFullResponseEndFrame, StartFrame, StartInterruptionFrame, + TTSSpeakFrame, TTSStartedFrame, TTSStoppedFrame, TTSVoiceUpdateFrame, @@ -178,7 +179,7 @@ class TTSService(AIService): if text: await self._push_tts_frames(text) - async def _push_tts_frames(self, text: str): + async def _push_tts_frames(self, text: str, text_passthrough: bool = True): text = text.strip() if not text: return @@ -209,6 +210,8 @@ class TTSService(AIService): await self.push_frame(frame, direction) else: await self.push_frame(frame, direction) + elif isinstance(frame, TTSSpeakFrame): + await self._push_tts_frames(frame.text, False) elif isinstance(frame, TTSVoiceUpdateFrame): await self.set_voice(frame.voice) else: From 09c05354c212592e1c093f78c5c654796a37b122 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 19 Jul 2024 14:34:30 -0700 Subject: [PATCH 15/37] processors(realtime-ai): fix voice initialization --- src/pipecat/processors/frameworks/realtimeai.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py index 3f032e4e7..a7a89913c 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/src/pipecat/processors/frameworks/realtimeai.py @@ -166,7 +166,7 @@ class RealtimeAIProcessor(FrameProcessor): voice = DEFAULT_VOICE if setup and setup.config and setup.config.tts and setup.config.tts.voice: - messages = setup.config.tts.voice + voice = setup.config.tts.voice self._tma_in = LLMUserResponseAggregator(messages) self._tma_out = LLMAssistantResponseAggregator(messages) From 32170b47d90d81b84418e0508481b967c4f255cf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 19 Jul 2024 15:15:36 -0700 Subject: [PATCH 16/37] processors(realtime-ai): add user-[start|stopped]-speaking messages --- .../processors/frameworks/realtimeai.py | 28 +++++++++++++++++-- 1 file changed, 26 insertions(+), 2 deletions(-) diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py index a7a89913c..fdfbd5b47 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/src/pipecat/processors/frameworks/realtimeai.py @@ -4,7 +4,6 @@ # SPDX-License-Identifier: BSD 2-Clause License # -from ctypes import DEFAULT_MODE import dataclasses from typing import List, Literal, Optional, Type @@ -18,7 +17,9 @@ from pipecat.frames.frames import ( StartFrame, TTSSpeakFrame, TTSVoiceUpdateFrame, - TransportMessageFrame) + TransportMessageFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame) from pipecat.pipeline.pipeline import Pipeline from pipecat.processors.aggregators.llm_response import ( LLMAssistantResponseAggregator, LLMUserResponseAggregator) @@ -92,6 +93,16 @@ class RealtimeAILLMContextResponse(BaseModel): messages: List[dict] +class RealtimeAIUserStartedSpeakingMessage(BaseModel): + tag: Literal["realtime-ai"] = "realtime-ai" + type: Literal["user-started-speaking"] = "user-started-speaking" + + +class RealtimeAIUserStoppedSpeakingMessage(BaseModel): + tag: Literal["realtime-ai"] = "realtime-ai" + type: Literal["user-stopped-speaking"] = "user-stopped-speaking" + + class RealtimeAIProcessor(FrameProcessor): def __init__( @@ -126,6 +137,19 @@ class RealtimeAIProcessor(FrameProcessor): if isinstance(frame, StartFrame): self._start_frame = frame await self._handle_setup(self._setup) + elif isinstance(frame, UserStartedSpeakingFrame) or isinstance(frame, UserStoppedSpeakingFrame): + await self._handle_interruptions(frame) + + async def _handle_interruptions(self, frame: Frame): + message = None + if isinstance(frame, UserStartedSpeakingFrame): + message = RealtimeAIUserStartedSpeakingMessage() + elif isinstance(frame, UserStoppedSpeakingFrame): + message = RealtimeAIUserStoppedSpeakingMessage() + + if message: + frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) + await self.push_frame(frame) async def _handle_message(self, frame: TransportMessageFrame): try: From 13827e1282b4e0d58c5797c847b8698e6a0f3462 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 19 Jul 2024 16:57:00 -0700 Subject: [PATCH 17/37] processors(realtime-ai): send a successful response for every command --- src/pipecat/processors/frameworks/realtimeai.py | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py index fdfbd5b47..de87ab136 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/src/pipecat/processors/frameworks/realtimeai.py @@ -175,6 +175,9 @@ class RealtimeAIProcessor(FrameProcessor): await self._handle_llm_update_context(message.data.llm) case "tts-speak": await self._handle_tts_speak(message.data.tts) + + # Send a message to indicate we successfully executed the command. + await self._send_response(message.type, True) except ValidationError as e: await self._send_response(message.type, False, f"invalid message: {e}") @@ -219,10 +222,6 @@ class RealtimeAIProcessor(FrameProcessor): # as the initial one. start_frame = dataclasses.replace(self._start_frame) await self.push_frame(start_frame) - - # We now send a message to indicate we successfully initialized - # the pipelines. - await self._send_response("setup", True) except Exception as e: await self._send_response("setup", False, f"unable to create pipeline: {e}") From f094c427285d294cdca25cdb44250491de8a99a7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 19 Jul 2024 17:33:49 -0700 Subject: [PATCH 18/37] processors(realtime-ai): add transcription messages --- .../processors/frameworks/realtimeai.py | 42 +++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py index de87ab136..c76d7e74d 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/src/pipecat/processors/frameworks/realtimeai.py @@ -11,12 +11,14 @@ from pydantic import BaseModel, ValidationError from pipecat.frames.frames import ( Frame, + InterimTranscriptionFrame, LLMMessagesAppendFrame, LLMMessagesUpdateFrame, LLMModelUpdateFrame, StartFrame, TTSSpeakFrame, TTSVoiceUpdateFrame, + TranscriptionFrame, TransportMessageFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame) @@ -93,6 +95,24 @@ class RealtimeAILLMContextResponse(BaseModel): messages: List[dict] +class RealtimeAITranscriptionMessageData(BaseModel): + text: str + user_id: str + timestamp: str + + +class RealtimeAITranscriptionMessage(BaseModel): + tag: Literal["realtime-ai"] = "realtime-ai" + type: Literal["user-transcription"] = "user-transcription" + data: RealtimeAITranscriptionMessageData + + +class RealtimeAIInterimTranscriptionMessage(BaseModel): + tag: Literal["realtime-ai"] = "realtime-ai" + type: Literal["user-interim-transcription"] = "user-interim-transcription" + data: RealtimeAITranscriptionMessageData + + class RealtimeAIUserStartedSpeakingMessage(BaseModel): tag: Literal["realtime-ai"] = "realtime-ai" type: Literal["user-started-speaking"] = "user-started-speaking" @@ -137,9 +157,31 @@ class RealtimeAIProcessor(FrameProcessor): if isinstance(frame, StartFrame): self._start_frame = frame await self._handle_setup(self._setup) + elif isinstance(frame, TranscriptionFrame) or isinstance(frame, InterimTranscriptionFrame): + await self._handle_transcriptions(frame) elif isinstance(frame, UserStartedSpeakingFrame) or isinstance(frame, UserStoppedSpeakingFrame): await self._handle_interruptions(frame) + # TODO(aleix): Once we add support for using custom piplines, the STTs will + # be in the pipeline after this processor. This means the STT will have to + # push transcriptions upstream as well. + async def _handle_transcriptions(self, frame: Frame): + message = None + if isinstance(frame, TranscriptionFrame): + message = RealtimeAITranscriptionMessage( + data=RealtimeAITranscriptionMessageData( + text=frame.text, + user_id=frame.user_id, + timestamp=frame.timestamp)) + elif isinstance(frame, InterimTranscriptionFrame): + message = RealtimeAIInterimTranscriptionMessage( + data=RealtimeAITranscriptionMessageData( + text=frame.text, user_id=frame.user_id, timestamp=frame.timestamp)) + + if message: + frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) + await self.push_frame(frame) + async def _handle_interruptions(self, frame: Frame): message = None if isinstance(frame, UserStartedSpeakingFrame): From 6fbf98d8e24adfbcd1231c840e5747a61476e941 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 19 Jul 2024 17:40:20 -0700 Subject: [PATCH 19/37] processors(realtime-ai): llm-context now uses a data field --- src/pipecat/processors/frameworks/realtimeai.py | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py index c76d7e74d..4eed08154 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/src/pipecat/processors/frameworks/realtimeai.py @@ -89,10 +89,14 @@ class RealtimeAIBasicResponse(BaseModel): error: Optional[str] = None -class RealtimeAILLMContextResponse(BaseModel): +class RealtimeAILLMContextMessageData(BaseModel): + messages: List[dict] + + +class RealtimeAILLMContextMessage(BaseModel): tag: Literal["realtime-ai"] = "realtime-ai" type: Literal["llm-context"] = "llm-context" - messages: List[dict] + data: RealtimeAILLMContextMessageData class RealtimeAITranscriptionMessageData(BaseModel): @@ -279,10 +283,10 @@ class RealtimeAIProcessor(FrameProcessor): await self.push_frame(frame) async def _handle_llm_get_context(self): - messages = self._tma_in.messages - response = RealtimeAILLMContextResponse(messages=messages) - message = TransportMessageFrame(message=response.model_dump(exclude_none=True)) - await self.push_frame(message) + data = RealtimeAILLMContextMessageData(messages=self._tma_in.messages) + message = RealtimeAILLMContextMessage(data=data) + frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) + await self.push_frame(frame) async def _handle_llm_append_context(self, data: RealtimeAILLMMessageData): if data and data.messages: From a46ac3cc9285a8c40615dfb9c1286317110ed403 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 19 Jul 2024 18:01:27 -0700 Subject: [PATCH 20/37] examples: moved 18-realtime-ai.py to examples/realtime-ai --- .../18-realtime-ai.py => realtime-ai/bot.py} | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) rename examples/{foundational/18-realtime-ai.py => realtime-ai/bot.py} (76%) diff --git a/examples/foundational/18-realtime-ai.py b/examples/realtime-ai/bot.py similarity index 76% rename from examples/foundational/18-realtime-ai.py rename to examples/realtime-ai/bot.py index 8a0a7683c..29fb020ea 100644 --- a/examples/foundational/18-realtime-ai.py +++ b/examples/realtime-ai/bot.py @@ -11,7 +11,7 @@ import os from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.pipeline.runner import PipelineRunner -from pipecat.processors.frameworks.realtimeai import RealtimeAIProcessor +from pipecat.processors.frameworks.realtimeai import RealtimeAIConfig, RealtimeAILLMConfig, RealtimeAIProcessor, RealtimeAISetup, RealtimeAITTSConfig from pipecat.transports.services.daily import DailyParams, DailyTransport from pipecat.vad.silero import SileroVADAnalyzer @@ -38,8 +38,16 @@ async def main(room_url, token): vad_analyzer=SileroVADAnalyzer() )) + llm = RealtimeAILLMConfig( + model="llama3-70b-8192", + messages=[{"role": "system", "content": "You are a helpful assistant named Gary. Briefly say hello!"}] + ) + tts = RealtimeAITTSConfig(voice="79a125e8-cd45-4c13-8a67-188112f4dd22") + setup = RealtimeAISetup(config=RealtimeAIConfig(llm=llm, tts=tts)) + rtai = RealtimeAIProcessor( transport=transport, + setup=setup, llm_api_key=os.getenv("OPENAI_API_KEY"), tts_api_key=os.getenv("CARTESIA_API_KEY")) From 6e00f31014002955a9be534a316dd9fa3a3a45f1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 19 Jul 2024 18:01:50 -0700 Subject: [PATCH 21/37] updated CHANGELOG with new frames and realtime-ai changes --- CHANGELOG.md | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index c28f5ede1..3f6fcbc1e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Added `RealtimeAIProcessor`... + +- Added `LLMMessagesAppendFrame` which allows appending messages to the current + LLM context. + +- Added `LLMMessagesUpdateFrame` which allows changing the LLM context for the + one provided in this new frame. + +- Added `LLMModelUpdateFrame` which allows updating the LLM model. + +- Added `TTSSpeakFrame` which causes the bot say some text. This text will not + be part of the LLM context. + +- Added `TTSVoiceUpdateFrame` which allows updating the TTS voice. + ### Removed - We remove the `LLMResponseStartFrame` and `LLMResponseEndFrame` frames. These @@ -33,6 +50,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 leverages the new Cartesia contexts which maintains generated audio prosody when multiple inputs are sent, therefore improving audio quality a lot. +### Other + +- Added `examples/realtime-ai/bot.py`... + ## [0.0.36] - 2024-07-02 ### Added From 82d539d1740998d20d0bc5a343faaa2491497d7b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 19 Jul 2024 22:33:46 -0700 Subject: [PATCH 22/37] processors(realtime-ai): add support for interrupting the bot --- CHANGELOG.md | 2 + examples/realtime-ai/bot.py | 7 ++- src/pipecat/frames/frames.py | 10 ++++ .../processors/frameworks/realtimeai.py | 50 ++++++++++++++++--- src/pipecat/transports/base_input.py | 42 ++++++++++------ 5 files changed, 89 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3f6fcbc1e..a2512d378 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added `RealtimeAIProcessor`... +- Added `BotInterruptionFrame` which allows interrupting the bot while talking. + - Added `LLMMessagesAppendFrame` which allows appending messages to the current LLM context. diff --git a/examples/realtime-ai/bot.py b/examples/realtime-ai/bot.py index 29fb020ea..a2f7320c0 100644 --- a/examples/realtime-ai/bot.py +++ b/examples/realtime-ai/bot.py @@ -11,7 +11,12 @@ import os from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.pipeline.runner import PipelineRunner -from pipecat.processors.frameworks.realtimeai import RealtimeAIConfig, RealtimeAILLMConfig, RealtimeAIProcessor, RealtimeAISetup, RealtimeAITTSConfig +from pipecat.processors.frameworks.realtimeai import ( + RealtimeAIConfig, + RealtimeAILLMConfig, + RealtimeAIProcessor, + RealtimeAISetup, + RealtimeAITTSConfig) from pipecat.transports.services.daily import DailyParams, DailyTransport from pipecat.vad.silero import SileroVADAnalyzer diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 09c475702..46e2408bc 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -268,6 +268,16 @@ class StopInterruptionFrame(SystemFrame): pass +@dataclass +class BotInterruptionFrame(SystemFrame): + """Emitted by when the bot should be interrupted. This will mainly cause the + same actions as if the user interrupted except that the + UserStartedSpeakingFrame and UserStoppedSpeakingFrame won't be generated. + + """ + pass + + @dataclass class BotSpeakingFrame(SystemFrame): """Emitted by transport outputs while the bot is still speaking. This can be diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py index 4eed08154..3332a0ef7 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/src/pipecat/processors/frameworks/realtimeai.py @@ -4,18 +4,21 @@ # SPDX-License-Identifier: BSD 2-Clause License # +import asyncio import dataclasses from typing import List, Literal, Optional, Type from pydantic import BaseModel, ValidationError from pipecat.frames.frames import ( + BotInterruptionFrame, Frame, InterimTranscriptionFrame, LLMMessagesAppendFrame, LLMMessagesUpdateFrame, LLMModelUpdateFrame, StartFrame, + SystemFrame, TTSSpeakFrame, TTSVoiceUpdateFrame, TranscriptionFrame, @@ -67,6 +70,7 @@ class RealtimeAILLMMessageData(BaseModel): class RealtimeAITTSMessageData(BaseModel): text: str + interrupt: Optional[bool] = False class RealtimeAIMessageData(BaseModel): @@ -152,24 +156,49 @@ class RealtimeAIProcessor(FrameProcessor): self._tts: FrameProcessor | None = None self._pipeline: FrameProcessor | None = None + self._frame_handler_task = self.get_event_loop().create_task(self._frame_handler()) + self._frame_queue = asyncio.Queue() + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, SystemFrame): + await self.push_frame(frame, direction) + else: + await self._frame_queue.put((frame, direction)) + + if isinstance(frame, StartFrame): + self._start_frame = frame + await self._handle_setup(self._setup) + + async def cleanup(self): + self._frame_handler_task.cancel() + await self._frame_handler_task + + async def _frame_handler(self): + while True: + try: + (frame, direction) = await self._frame_queue.get() + await self._handle_frame(frame, direction) + except asyncio.CancelledError: + break + + async def _handle_frame(self, frame: Frame, direction: FrameDirection): if isinstance(frame, TransportMessageFrame): await self._handle_message(frame) else: await self.push_frame(frame, direction) - if isinstance(frame, StartFrame): - self._start_frame = frame - await self._handle_setup(self._setup) - elif isinstance(frame, TranscriptionFrame) or isinstance(frame, InterimTranscriptionFrame): + if isinstance(frame, TranscriptionFrame) or isinstance(frame, InterimTranscriptionFrame): await self._handle_transcriptions(frame) elif isinstance(frame, UserStartedSpeakingFrame) or isinstance(frame, UserStoppedSpeakingFrame): await self._handle_interruptions(frame) - # TODO(aleix): Once we add support for using custom piplines, the STTs will - # be in the pipeline after this processor. This means the STT will have to - # push transcriptions upstream as well. async def _handle_transcriptions(self, frame: Frame): + # TODO(aleix): Once we add support for using custom piplines, the STTs will + # be in the pipeline after this processor. This means the STT will have to + # push transcriptions upstream as well. + message = None if isinstance(frame, TranscriptionFrame): message = RealtimeAITranscriptionMessage( @@ -221,6 +250,8 @@ class RealtimeAIProcessor(FrameProcessor): await self._handle_llm_update_context(message.data.llm) case "tts-speak": await self._handle_tts_speak(message.data.tts) + case "tts-interrupt": + await self._handle_tts_interrupt() # Send a message to indicate we successfully executed the command. await self._send_response(message.type, True) @@ -300,9 +331,14 @@ class RealtimeAIProcessor(FrameProcessor): async def _handle_tts_speak(self, data: RealtimeAITTSMessageData): if data and data.text: + if data.interrupt: + await self._handle_tts_interrupt() frame = TTSSpeakFrame(text=data.text) await self.push_frame(frame) + async def _handle_tts_interrupt(self): + await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) + async def _send_response(self, type: str, success: bool, error: str | None = None): # TODO(aleix): This is a bit hacky, but we might get invalid # configuration or something might going wrong during setup and we would diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 828dc3e40..f87772c01 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -11,6 +11,7 @@ from concurrent.futures import ThreadPoolExecutor from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.frames.frames import ( AudioRawFrame, + BotInterruptionFrame, CancelFrame, StartFrame, EndFrame, @@ -78,6 +79,8 @@ class BaseInputTransport(FrameProcessor): elif isinstance(frame, EndFrame): await self._internal_push_frame(frame, direction) await self.stop() + elif isinstance(frame, BotInterruptionFrame): + await self._handle_interruptions(frame, False) else: await self._internal_push_frame(frame, direction) @@ -108,24 +111,35 @@ class BaseInputTransport(FrameProcessor): # Handle interruptions # - async def _handle_interruptions(self, frame: Frame): + async def _start_interruption(self): + # Cancel the task. This will stop pushing frames downstream. + self._push_frame_task.cancel() + await self._push_frame_task + # Push an out-of-band frame (i.e. not using the ordered push + # frame task) to stop everything, specially at the output + # transport. + await self.push_frame(StartInterruptionFrame()) + # Create a new queue and task. + self._create_push_task() + + async def _stop_interruption(self): + await self.push_frame(StopInterruptionFrame()) + + async def _handle_interruptions(self, frame: Frame, push_frame: bool): if self.interruptions_allowed: # Make sure we notify about interruptions quickly out-of-band - if isinstance(frame, UserStartedSpeakingFrame): + if isinstance(frame, BotInterruptionFrame): + logger.debug("Bot interruption") + await self._start_interruption() + elif isinstance(frame, UserStartedSpeakingFrame): logger.debug("User started speaking") - # Cancel the task. This will stop pushing frames downstream. - self._push_frame_task.cancel() - await self._push_frame_task - # Push an out-of-band frame (i.e. not using the ordered push - # frame task) to stop everything, specially at the output - # transport. - await self.push_frame(StartInterruptionFrame()) - # Create a new queue and task. - self._create_push_task() + await self._start_interruption() elif isinstance(frame, UserStoppedSpeakingFrame): logger.debug("User stopped speaking") - await self.push_frame(StopInterruptionFrame()) - await self._internal_push_frame(frame) + await self._stop_interruption() + + if push_frame: + await self._internal_push_frame(frame) # # Audio input @@ -149,7 +163,7 @@ class BaseInputTransport(FrameProcessor): frame = UserStoppedSpeakingFrame() if frame: - await self._handle_interruptions(frame) + await self._handle_interruptions(frame, True) vad_state = new_vad_state return vad_state From dfbc11300c65622104142e8d9c540f7ebb3bd279 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sun, 21 Jul 2024 08:26:12 -0700 Subject: [PATCH 23/37] processors(realtime-ai): use label instead of tag --- .../processors/frameworks/realtimeai.py | 29 ++++++++++++------- 1 file changed, 19 insertions(+), 10 deletions(-) diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/realtimeai.py index 3332a0ef7..25e394e42 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/src/pipecat/processors/frameworks/realtimeai.py @@ -81,13 +81,13 @@ class RealtimeAIMessageData(BaseModel): class RealtimeAIMessage(BaseModel): - tag: Literal["realtime-ai"] = "realtime-ai" + label: Literal["realtime-ai"] = "realtime-ai" type: str data: Optional[RealtimeAIMessageData] = None class RealtimeAIBasicResponse(BaseModel): - tag: Literal["realtime-ai"] = "realtime-ai" + label: Literal["realtime-ai"] = "realtime-ai" type: str success: bool error: Optional[str] = None @@ -97,8 +97,13 @@ class RealtimeAILLMContextMessageData(BaseModel): messages: List[dict] +class RealtimeAIBotReady(BaseModel): + label: Literal["realtime-ai"] = "realtime-ai" + type: Literal["bot-ready"] = "bot-ready" + + class RealtimeAILLMContextMessage(BaseModel): - tag: Literal["realtime-ai"] = "realtime-ai" + label: Literal["realtime-ai"] = "realtime-ai" type: Literal["llm-context"] = "llm-context" data: RealtimeAILLMContextMessageData @@ -110,24 +115,24 @@ class RealtimeAITranscriptionMessageData(BaseModel): class RealtimeAITranscriptionMessage(BaseModel): - tag: Literal["realtime-ai"] = "realtime-ai" + label: Literal["realtime-ai"] = "realtime-ai" type: Literal["user-transcription"] = "user-transcription" data: RealtimeAITranscriptionMessageData class RealtimeAIInterimTranscriptionMessage(BaseModel): - tag: Literal["realtime-ai"] = "realtime-ai" + label: Literal["realtime-ai"] = "realtime-ai" type: Literal["user-interim-transcription"] = "user-interim-transcription" data: RealtimeAITranscriptionMessageData class RealtimeAIUserStartedSpeakingMessage(BaseModel): - tag: Literal["realtime-ai"] = "realtime-ai" + label: Literal["realtime-ai"] = "realtime-ai" type: Literal["user-started-speaking"] = "user-started-speaking" class RealtimeAIUserStoppedSpeakingMessage(BaseModel): - tag: Literal["realtime-ai"] = "realtime-ai" + label: Literal["realtime-ai"] = "realtime-ai" type: Literal["user-stopped-speaking"] = "user-stopped-speaking" @@ -299,6 +304,10 @@ class RealtimeAIProcessor(FrameProcessor): # as the initial one. start_frame = dataclasses.replace(self._start_frame) await self.push_frame(start_frame) + + message = RealtimeAIBotReady() + frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) + await self.push_frame(frame) except Exception as e: await self._send_response("setup", False, f"unable to create pipeline: {e}") @@ -354,6 +363,6 @@ class RealtimeAIProcessor(FrameProcessor): if parent and self._start_frame: parent.link(pipeline) - response = RealtimeAIBasicResponse(type=type, success=success, error=error) - message = TransportMessageFrame(message=response.model_dump(exclude_none=True)) - await self.push_frame(message) + message = RealtimeAIBasicResponse(type=type, success=success, error=error) + frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) + await self.push_frame(frame) From be6995cfdfea9d7d72452a310fff64f3067f8b7e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sun, 21 Jul 2024 14:53:49 -0700 Subject: [PATCH 24/37] processors(rtvi): renamed realtime-ai to rtvi --- examples/{realtime-ai => rtvi}/bot.py | 20 ++-- examples/rtvi/runner.py | 58 ++++++++++++ .../frameworks/{realtimeai.py => rtvi.py} | 92 +++++++++---------- 3 files changed, 114 insertions(+), 56 deletions(-) rename examples/{realtime-ai => rtvi}/bot.py (82%) create mode 100644 examples/rtvi/runner.py rename src/pipecat/processors/frameworks/{realtimeai.py => rtvi.py} (82%) diff --git a/examples/realtime-ai/bot.py b/examples/rtvi/bot.py similarity index 82% rename from examples/realtime-ai/bot.py rename to examples/rtvi/bot.py index a2f7320c0..6214a6c72 100644 --- a/examples/realtime-ai/bot.py +++ b/examples/rtvi/bot.py @@ -11,12 +11,12 @@ import os from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.pipeline.runner import PipelineRunner -from pipecat.processors.frameworks.realtimeai import ( - RealtimeAIConfig, - RealtimeAILLMConfig, - RealtimeAIProcessor, - RealtimeAISetup, - RealtimeAITTSConfig) +from pipecat.processors.frameworks.rtvi import ( + RTVIConfig, + RTVILLMConfig, + RTVIProcessor, + RTVISetup, + RTVITTSConfig) from pipecat.transports.services.daily import DailyParams, DailyTransport from pipecat.vad.silero import SileroVADAnalyzer @@ -43,14 +43,14 @@ async def main(room_url, token): vad_analyzer=SileroVADAnalyzer() )) - llm = RealtimeAILLMConfig( + llm = RTVILLMConfig( model="llama3-70b-8192", messages=[{"role": "system", "content": "You are a helpful assistant named Gary. Briefly say hello!"}] ) - tts = RealtimeAITTSConfig(voice="79a125e8-cd45-4c13-8a67-188112f4dd22") - setup = RealtimeAISetup(config=RealtimeAIConfig(llm=llm, tts=tts)) + tts = RTVITTSConfig(voice="79a125e8-cd45-4c13-8a67-188112f4dd22") + setup = RTVISetup(config=RTVIConfig(llm=llm, tts=tts)) - rtai = RealtimeAIProcessor( + rtai = RTVIProcessor( transport=transport, setup=setup, llm_api_key=os.getenv("OPENAI_API_KEY"), diff --git a/examples/rtvi/runner.py b/examples/rtvi/runner.py new file mode 100644 index 000000000..6d1a8113d --- /dev/null +++ b/examples/rtvi/runner.py @@ -0,0 +1,58 @@ +import argparse +import os +import time +import urllib +import requests + + +def configure(): + parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample") + parser.add_argument( + "-u", + "--url", + type=str, + required=False, + help="URL of the Daily room to join") + parser.add_argument( + "-k", + "--apikey", + type=str, + required=False, + help="Daily API Key (needed to create an owner token for the room)", + ) + + args, unknown = parser.parse_known_args() + + url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL") + key = args.apikey or os.getenv("DAILY_API_KEY") + + if not url: + raise Exception( + "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL.") + + if not key: + raise Exception("No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.") + + # Create a meeting token for the given room with an expiration 1 hour in + # the future. + room_name: str = urllib.parse.urlparse(url).path[1:] + expiration: float = time.time() + 60 * 60 + + res: requests.Response = requests.post( + f"https://api.daily.co/v1/meeting-tokens", + headers={ + "Authorization": f"Bearer {key}"}, + json={ + "properties": { + "room_name": room_name, + "is_owner": True, + "exp": expiration}}, + ) + + if res.status_code != 200: + raise Exception( + f"Failed to create meeting token: {res.status_code} {res.text}") + + token: str = res.json()["token"] + + return (url, token) diff --git a/src/pipecat/processors/frameworks/realtimeai.py b/src/pipecat/processors/frameworks/rtvi.py similarity index 82% rename from src/pipecat/processors/frameworks/realtimeai.py rename to src/pipecat/processors/frameworks/rtvi.py index 25e394e42..8a7de1e96 100644 --- a/src/pipecat/processors/frameworks/realtimeai.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -46,103 +46,103 @@ DEFAULT_MODEL = "llama3-70b-8192" DEFAULT_VOICE = "79a125e8-cd45-4c13-8a67-188112f4dd22" -class RealtimeAILLMConfig(BaseModel): +class RTVILLMConfig(BaseModel): model: Optional[str] = None messages: Optional[List[dict]] = None -class RealtimeAITTSConfig(BaseModel): +class RTVITTSConfig(BaseModel): voice: Optional[str] = None -class RealtimeAIConfig(BaseModel): - llm: Optional[RealtimeAILLMConfig] = None - tts: Optional[RealtimeAITTSConfig] = None +class RTVIConfig(BaseModel): + llm: Optional[RTVILLMConfig] = None + tts: Optional[RTVITTSConfig] = None -class RealtimeAISetup(BaseModel): - config: Optional[RealtimeAIConfig] = None +class RTVISetup(BaseModel): + config: Optional[RTVIConfig] = None -class RealtimeAILLMMessageData(BaseModel): +class RTVILLMMessageData(BaseModel): messages: List[dict] -class RealtimeAITTSMessageData(BaseModel): +class RTVITTSMessageData(BaseModel): text: str interrupt: Optional[bool] = False -class RealtimeAIMessageData(BaseModel): - setup: Optional[RealtimeAISetup] = None - config: Optional[RealtimeAIConfig] = None - llm: Optional[RealtimeAILLMMessageData] = None - tts: Optional[RealtimeAITTSMessageData] = None +class RTVIMessageData(BaseModel): + setup: Optional[RTVISetup] = None + config: Optional[RTVIConfig] = None + llm: Optional[RTVILLMMessageData] = None + tts: Optional[RTVITTSMessageData] = None -class RealtimeAIMessage(BaseModel): +class RTVIMessage(BaseModel): label: Literal["realtime-ai"] = "realtime-ai" type: str - data: Optional[RealtimeAIMessageData] = None + data: Optional[RTVIMessageData] = None -class RealtimeAIBasicResponse(BaseModel): +class RTVIBasicResponse(BaseModel): label: Literal["realtime-ai"] = "realtime-ai" type: str success: bool error: Optional[str] = None -class RealtimeAILLMContextMessageData(BaseModel): +class RTVILLMContextMessageData(BaseModel): messages: List[dict] -class RealtimeAIBotReady(BaseModel): +class RTVIBotReady(BaseModel): label: Literal["realtime-ai"] = "realtime-ai" type: Literal["bot-ready"] = "bot-ready" -class RealtimeAILLMContextMessage(BaseModel): +class RTVILLMContextMessage(BaseModel): label: Literal["realtime-ai"] = "realtime-ai" type: Literal["llm-context"] = "llm-context" - data: RealtimeAILLMContextMessageData + data: RTVILLMContextMessageData -class RealtimeAITranscriptionMessageData(BaseModel): +class RTVITranscriptionMessageData(BaseModel): text: str user_id: str timestamp: str -class RealtimeAITranscriptionMessage(BaseModel): +class RTVITranscriptionMessage(BaseModel): label: Literal["realtime-ai"] = "realtime-ai" type: Literal["user-transcription"] = "user-transcription" - data: RealtimeAITranscriptionMessageData + data: RTVITranscriptionMessageData -class RealtimeAIInterimTranscriptionMessage(BaseModel): +class RTVIInterimTranscriptionMessage(BaseModel): label: Literal["realtime-ai"] = "realtime-ai" type: Literal["user-interim-transcription"] = "user-interim-transcription" - data: RealtimeAITranscriptionMessageData + data: RTVITranscriptionMessageData -class RealtimeAIUserStartedSpeakingMessage(BaseModel): +class RTVIUserStartedSpeakingMessage(BaseModel): label: Literal["realtime-ai"] = "realtime-ai" type: Literal["user-started-speaking"] = "user-started-speaking" -class RealtimeAIUserStoppedSpeakingMessage(BaseModel): +class RTVIUserStoppedSpeakingMessage(BaseModel): label: Literal["realtime-ai"] = "realtime-ai" type: Literal["user-stopped-speaking"] = "user-stopped-speaking" -class RealtimeAIProcessor(FrameProcessor): +class RTVIProcessor(FrameProcessor): def __init__( self, *, transport: BaseTransport, - setup: RealtimeAISetup | None = None, + setup: RTVISetup | None = None, llm_api_key: str = "", llm_base_url: str = "https://api.groq.com/openai/v1", tts_api_key: str = "", @@ -206,14 +206,14 @@ class RealtimeAIProcessor(FrameProcessor): message = None if isinstance(frame, TranscriptionFrame): - message = RealtimeAITranscriptionMessage( - data=RealtimeAITranscriptionMessageData( + message = RTVITranscriptionMessage( + data=RTVITranscriptionMessageData( text=frame.text, user_id=frame.user_id, timestamp=frame.timestamp)) elif isinstance(frame, InterimTranscriptionFrame): - message = RealtimeAIInterimTranscriptionMessage( - data=RealtimeAITranscriptionMessageData( + message = RTVIInterimTranscriptionMessage( + data=RTVITranscriptionMessageData( text=frame.text, user_id=frame.user_id, timestamp=frame.timestamp)) if message: @@ -223,9 +223,9 @@ class RealtimeAIProcessor(FrameProcessor): async def _handle_interruptions(self, frame: Frame): message = None if isinstance(frame, UserStartedSpeakingFrame): - message = RealtimeAIUserStartedSpeakingMessage() + message = RTVIUserStartedSpeakingMessage() elif isinstance(frame, UserStoppedSpeakingFrame): - message = RealtimeAIUserStoppedSpeakingMessage() + message = RTVIUserStoppedSpeakingMessage() if message: frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) @@ -233,7 +233,7 @@ class RealtimeAIProcessor(FrameProcessor): async def _handle_message(self, frame: TransportMessageFrame): try: - message = RealtimeAIMessage.model_validate(frame.message) + message = RTVIMessage.model_validate(frame.message) except ValidationError as e: await self._send_response("setup", False, f"invalid message: {e}") return @@ -263,7 +263,7 @@ class RealtimeAIProcessor(FrameProcessor): except ValidationError as e: await self._send_response(message.type, False, f"invalid message: {e}") - async def _handle_setup(self, setup: RealtimeAISetup | None): + async def _handle_setup(self, setup: RTVISetup | None): try: model = DEFAULT_MODEL if setup and setup.config and setup.config.llm and setup.config.llm.model: @@ -305,13 +305,13 @@ class RealtimeAIProcessor(FrameProcessor): start_frame = dataclasses.replace(self._start_frame) await self.push_frame(start_frame) - message = RealtimeAIBotReady() + message = RTVIBotReady() frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) await self.push_frame(frame) except Exception as e: await self._send_response("setup", False, f"unable to create pipeline: {e}") - async def _handle_config_update(self, config: RealtimeAIConfig): + async def _handle_config_update(self, config: RTVIConfig): if config.llm and config.llm.model: frame = LLMModelUpdateFrame(config.llm.model) await self.push_frame(frame) @@ -323,22 +323,22 @@ class RealtimeAIProcessor(FrameProcessor): await self.push_frame(frame) async def _handle_llm_get_context(self): - data = RealtimeAILLMContextMessageData(messages=self._tma_in.messages) - message = RealtimeAILLMContextMessage(data=data) + data = RTVILLMContextMessageData(messages=self._tma_in.messages) + message = RTVILLMContextMessage(data=data) frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) await self.push_frame(frame) - async def _handle_llm_append_context(self, data: RealtimeAILLMMessageData): + async def _handle_llm_append_context(self, data: RTVILLMMessageData): if data and data.messages: frame = LLMMessagesAppendFrame(data.messages) await self.push_frame(frame) - async def _handle_llm_update_context(self, data: RealtimeAILLMMessageData): + async def _handle_llm_update_context(self, data: RTVILLMMessageData): if data and data.messages: frame = LLMMessagesUpdateFrame(data.messages) await self.push_frame(frame) - async def _handle_tts_speak(self, data: RealtimeAITTSMessageData): + async def _handle_tts_speak(self, data: RTVITTSMessageData): if data and data.text: if data.interrupt: await self._handle_tts_interrupt() @@ -363,6 +363,6 @@ class RealtimeAIProcessor(FrameProcessor): if parent and self._start_frame: parent.link(pipeline) - message = RealtimeAIBasicResponse(type=type, success=success, error=error) + message = RTVIBasicResponse(type=type, success=success, error=error) frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) await self.push_frame(frame) From 37b04ed2830a3fe4362863777d26b6ef8ee9915e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sun, 21 Jul 2024 16:34:38 -0700 Subject: [PATCH 25/37] processors(rtvi): use send a type=response as command responses --- src/pipecat/processors/frameworks/rtvi.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 8a7de1e96..062dc63a0 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -86,12 +86,16 @@ class RTVIMessage(BaseModel): data: Optional[RTVIMessageData] = None -class RTVIBasicResponse(BaseModel): - label: Literal["realtime-ai"] = "realtime-ai" +class RTVIResponseData(BaseModel): type: str success: bool error: Optional[str] = None +class RTVIResponse(BaseModel): + label: Literal["realtime-ai"] = "realtime-ai" + type: Literal["response"] = "response" + data: RTVIResponseData + class RTVILLMContextMessageData(BaseModel): messages: List[dict] @@ -363,6 +367,6 @@ class RTVIProcessor(FrameProcessor): if parent and self._start_frame: parent.link(pipeline) - message = RTVIBasicResponse(type=type, success=success, error=error) + message = RTVIResponse(data=RTVIResponseData(type=type, success=success, error=error)) frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) await self.push_frame(frame) From 302ea90dce9fa725baf32973ba0429eb9c558d6c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sun, 21 Jul 2024 16:54:40 -0700 Subject: [PATCH 26/37] processors(rtvi): messages now require an id --- src/pipecat/processors/frameworks/rtvi.py | 125 +++++++++++++--------- 1 file changed, 74 insertions(+), 51 deletions(-) diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 062dc63a0..376a4f034 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -83,35 +83,47 @@ class RTVIMessageData(BaseModel): class RTVIMessage(BaseModel): label: Literal["realtime-ai"] = "realtime-ai" type: str + id: str data: Optional[RTVIMessageData] = None class RTVIResponseData(BaseModel): - type: str success: bool error: Optional[str] = None + class RTVIResponse(BaseModel): label: Literal["realtime-ai"] = "realtime-ai" type: Literal["response"] = "response" + id: str data: RTVIResponseData +class RTVIErrorData(BaseModel): + message: str + + +class RTVIError(BaseModel): + label: Literal["realtime-ai"] = "realtime-ai" + type: Literal["error"] = "error" + data: RTVIErrorData + + class RTVILLMContextMessageData(BaseModel): messages: List[dict] -class RTVIBotReady(BaseModel): - label: Literal["realtime-ai"] = "realtime-ai" - type: Literal["bot-ready"] = "bot-ready" - - class RTVILLMContextMessage(BaseModel): label: Literal["realtime-ai"] = "realtime-ai" type: Literal["llm-context"] = "llm-context" data: RTVILLMContextMessageData +class RTVIBotReady(BaseModel): + label: Literal["realtime-ai"] = "realtime-ai" + type: Literal["bot-ready"] = "bot-ready" + + class RTVITranscriptionMessageData(BaseModel): text: str user_id: str @@ -178,7 +190,10 @@ class RTVIProcessor(FrameProcessor): if isinstance(frame, StartFrame): self._start_frame = frame - await self._handle_setup(self._setup) + try: + await self._handle_setup(self._setup) + except Exception as e: + await self._send_error(f"unable to setup RTVI: {e}") async def cleanup(self): self._frame_handler_task.cancel() @@ -239,16 +254,18 @@ class RTVIProcessor(FrameProcessor): try: message = RTVIMessage.model_validate(frame.message) except ValidationError as e: - await self._send_response("setup", False, f"invalid message: {e}") + await self._send_error(f"invalid message: {e}") return try: + success = True + error = None match message.type: case "setup": setup = None if message.data: setup = message.data.setup - await self._handle_setup(setup) + await self._handle_setup(message.id, setup) case "config-update": await self._handle_config_update(message.data.config) case "llm-get-context": @@ -261,59 +278,60 @@ class RTVIProcessor(FrameProcessor): await self._handle_tts_speak(message.data.tts) case "tts-interrupt": await self._handle_tts_interrupt() + case _: + success = False + error = f"unsupported type {message.type}" - # Send a message to indicate we successfully executed the command. - await self._send_response(message.type, True) + await self._send_response(message.id, success, error) except ValidationError as e: - await self._send_response(message.type, False, f"invalid message: {e}") + await self._send_response(message.id, False, f"invalid message: {e}") + except Exception as e: + await self._send_response(message.id, False, f"{e}") async def _handle_setup(self, setup: RTVISetup | None): - try: - model = DEFAULT_MODEL - if setup and setup.config and setup.config.llm and setup.config.llm.model: - model = setup.config.llm.model + model = DEFAULT_MODEL + if setup and setup.config and setup.config.llm and setup.config.llm.model: + model = setup.config.llm.model - messages = DEFAULT_MESSAGES - if setup and setup.config and setup.config.llm and setup.config.llm.messages: - messages = setup.config.llm.messages + messages = DEFAULT_MESSAGES + if setup and setup.config and setup.config.llm and setup.config.llm.messages: + messages = setup.config.llm.messages - voice = DEFAULT_VOICE - if setup and setup.config and setup.config.tts and setup.config.tts.voice: - voice = setup.config.tts.voice + voice = DEFAULT_VOICE + if setup and setup.config and setup.config.tts and setup.config.tts.voice: + voice = setup.config.tts.voice - self._tma_in = LLMUserResponseAggregator(messages) - self._tma_out = LLMAssistantResponseAggregator(messages) + self._tma_in = LLMUserResponseAggregator(messages) + self._tma_out = LLMAssistantResponseAggregator(messages) - self._llm = self._llm_cls( - base_url=self._llm_base_url, - api_key=self._llm_api_key, - model=model) + self._llm = self._llm_cls( + base_url=self._llm_base_url, + api_key=self._llm_api_key, + model=model) - self._tts = self._tts_cls(api_key=self._tts_api_key, voice_id=voice) + self._tts = self._tts_cls(api_key=self._tts_api_key, voice_id=voice) - pipeline = Pipeline([ - self._tma_in, - self._llm, - self._tts, - self._tma_out, - self._transport.output(), - ]) - self._pipeline = pipeline + pipeline = Pipeline([ + self._tma_in, + self._llm, + self._tts, + self._tma_out, + self._transport.output(), + ]) + self._pipeline = pipeline - parent = self.get_parent() - if parent and self._start_frame: - parent.link(pipeline) + parent = self.get_parent() + if parent and self._start_frame: + parent.link(pipeline) - # We need to initialize the new pipeline with the same settings - # as the initial one. - start_frame = dataclasses.replace(self._start_frame) - await self.push_frame(start_frame) + # We need to initialize the new pipeline with the same settings + # as the initial one. + start_frame = dataclasses.replace(self._start_frame) + await self.push_frame(start_frame) - message = RTVIBotReady() - frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) - await self.push_frame(frame) - except Exception as e: - await self._send_response("setup", False, f"unable to create pipeline: {e}") + message = RTVIBotReady() + frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) + await self.push_frame(frame) async def _handle_config_update(self, config: RTVIConfig): if config.llm and config.llm.model: @@ -352,7 +370,12 @@ class RTVIProcessor(FrameProcessor): async def _handle_tts_interrupt(self): await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM) - async def _send_response(self, type: str, success: bool, error: str | None = None): + async def _send_error(self, error: str): + message = RTVIError(data=RTVIErrorData(message=error)) + frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) + await self.push_frame(frame) + + async def _send_response(self, id: str, success: bool, error: str | None = None): # TODO(aleix): This is a bit hacky, but we might get invalid # configuration or something might going wrong during setup and we would # like to send the error to the client. However, if the pipeline is not @@ -367,6 +390,6 @@ class RTVIProcessor(FrameProcessor): if parent and self._start_frame: parent.link(pipeline) - message = RTVIResponse(data=RTVIResponseData(type=type, success=success, error=error)) + message = RTVIResponse(id=id, data=RTVIResponseData(success=success, error=error)) frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) await self.push_frame(frame) From 5d73db53a0110303f8e226523e4f0198efb603a1 Mon Sep 17 00:00:00 2001 From: Chad Bailey Date: Sun, 21 Jul 2024 01:15:44 +0000 Subject: [PATCH 27/37] initial pseudo function calling --- src/pipecat/processors/frameworks/rtvi.py | 97 ++++++++++++++++++++++- 1 file changed, 96 insertions(+), 1 deletion(-) diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 376a4f034..663359ad5 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -14,6 +14,8 @@ from pipecat.frames.frames import ( BotInterruptionFrame, Frame, InterimTranscriptionFrame, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, LLMMessagesAppendFrame, LLMMessagesUpdateFrame, LLMModelUpdateFrame, @@ -21,6 +23,7 @@ from pipecat.frames.frames import ( SystemFrame, TTSSpeakFrame, TTSVoiceUpdateFrame, + TextFrame, TranscriptionFrame, TransportMessageFrame, UserStartedSpeakingFrame, @@ -31,7 +34,7 @@ from pipecat.processors.aggregators.llm_response import ( from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.ai_services import AIService from pipecat.services.cartesia import CartesiaTTSService -from pipecat.services.openai import OpenAILLMService +from pipecat.services.openai import OpenAILLMService, OpenAILLMContext, OpenAILLMContextFrame from pipecat.transports.base_transport import BaseTransport DEFAULT_MESSAGES = [ @@ -152,6 +155,92 @@ class RTVIUserStoppedSpeakingMessage(BaseModel): type: Literal["user-stopped-speaking"] = "user-stopped-speaking" +class RTVIJSONCompletion(BaseModel): + label: Literal["realtime-ai"] = "realtime-ai" + type: Literal["json-completion"] = "json-completion" + data: str + + +class FunctionCaller(FrameProcessor): + def __init__(self, context): + super().__init__() + self._checking = False + self._aggregating = False + self._emitted_start = False + self._aggregation = "" + self._context = context + + self._callbacks = {} + self._start_callbacks = {} + + def register_function(self, function_name: str, callback, start_callback=None): + self._callbacks[function_name] = callback + if start_callback: + self._start_callbacks[function_name] = start_callback + + def unregister_function(self, function_name: str): + del self._callbacks[function_name] + if self._start_callbacks[function_name]: + del self._start_callbacks[function_name] + + def has_function(self, function_name: str): + return function_name in self._callbacks.keys() + + async def call_function(self, function_name: str, args): + if function_name in self._callbacks.keys(): + return await self._callbacks[function_name](self, args) + return None + + async def call_start_function(self, function_name: str): + if function_name in self._start_callbacks.keys(): + await self._start_callbacks[function_name](self) + + async def process_frame(self, frame: Frame, direction: FrameDirection): + if isinstance(frame, LLMFullResponseStartFrame): + self._checking = True + await self.push_frame(frame, direction) + elif isinstance(frame, TextFrame) and self._checking: + # TODO-CB: should we expand this to any non-text character to start the completion? + if frame.text.strip().startswith("{") or frame.text.strip().startswith("```"): + self._emitted_start = False + self._checking = False + self._aggregation = frame.text + self._aggregating = True + else: + self._checking = False + self._aggregating = False + self._aggregation = "" + self._emitted_start = False + await self.push_frame(frame, direction) + elif isinstance(frame, TextFrame) and self._aggregating: + self._aggregation += frame.text + # TODO-CB: We can probably ignore function start I think + # if not self._emitted_start: + # fn = re.search(r'{"function_name":\s*"(.*)",', self._aggregation) + # if fn and fn.group(1): + # await self.call_start_function(fn.group(1)) + # self._emitted_start = True + elif isinstance(frame, LLMFullResponseEndFrame) and self._aggregating: + try: + self._aggregation = self._aggregation.replace("```json", "").replace("```", "") + self._context.add_message({"role": "assistant", "content": self._aggregation}) + message = RTVIJSONCompletion(data=self._aggregation) + msg = message.model_dump(exclude_none=True) + await self.push_frame(TransportMessageFrame(message=msg)) + + except Exception as e: + print(f"Error parsing function call json: {e}") + print(f"aggregation was: {self._aggregation}") + + self._aggregating = False + self._aggregation = "" + self._emitted_start = False + elif isinstance(frame, LLMFullResponseEndFrame): + await self.push_frame(frame, direction) + else: + await self.push_frame(frame, direction) + + class RTVIProcessor(FrameProcessor): def __init__( @@ -311,9 +400,15 @@ class RTVIProcessor(FrameProcessor): self._tts = self._tts_cls(api_key=self._tts_api_key, voice_id=voice) + # TODO-CB: Eventually we'll need to switch the context aggregators to use the + # OpenAI context frames instead of message frames + context = OpenAILLMContext(messages=messages) + self._fc = FunctionCaller(context) + pipeline = Pipeline([ self._tma_in, self._llm, + self._fc, self._tts, self._tma_out, self._transport.output(), From 80baa0358dcf922fb422395e99c6ea28e704e065 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sun, 21 Jul 2024 17:13:18 -0700 Subject: [PATCH 28/37] processors(rtvi): lable is now rtvi --- src/pipecat/processors/frameworks/rtvi.py | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 663359ad5..23acb71bd 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -84,7 +84,7 @@ class RTVIMessageData(BaseModel): class RTVIMessage(BaseModel): - label: Literal["realtime-ai"] = "realtime-ai" + label: Literal["rtvi"] = "rtvi" type: str id: str data: Optional[RTVIMessageData] = None @@ -96,7 +96,7 @@ class RTVIResponseData(BaseModel): class RTVIResponse(BaseModel): - label: Literal["realtime-ai"] = "realtime-ai" + label: Literal["rtvi"] = "rtvi" type: Literal["response"] = "response" id: str data: RTVIResponseData @@ -107,7 +107,7 @@ class RTVIErrorData(BaseModel): class RTVIError(BaseModel): - label: Literal["realtime-ai"] = "realtime-ai" + label: Literal["rtvi"] = "rtvi" type: Literal["error"] = "error" data: RTVIErrorData @@ -117,13 +117,13 @@ class RTVILLMContextMessageData(BaseModel): class RTVILLMContextMessage(BaseModel): - label: Literal["realtime-ai"] = "realtime-ai" + label: Literal["rtvi"] = "rtvi" type: Literal["llm-context"] = "llm-context" data: RTVILLMContextMessageData class RTVIBotReady(BaseModel): - label: Literal["realtime-ai"] = "realtime-ai" + label: Literal["rtvi"] = "rtvi" type: Literal["bot-ready"] = "bot-ready" @@ -134,29 +134,29 @@ class RTVITranscriptionMessageData(BaseModel): class RTVITranscriptionMessage(BaseModel): - label: Literal["realtime-ai"] = "realtime-ai" + label: Literal["rtvi"] = "rtvi" type: Literal["user-transcription"] = "user-transcription" data: RTVITranscriptionMessageData class RTVIInterimTranscriptionMessage(BaseModel): - label: Literal["realtime-ai"] = "realtime-ai" + label: Literal["rtvi"] = "rtvi" type: Literal["user-interim-transcription"] = "user-interim-transcription" data: RTVITranscriptionMessageData class RTVIUserStartedSpeakingMessage(BaseModel): - label: Literal["realtime-ai"] = "realtime-ai" + label: Literal["rtvi"] = "rtvi" type: Literal["user-started-speaking"] = "user-started-speaking" class RTVIUserStoppedSpeakingMessage(BaseModel): - label: Literal["realtime-ai"] = "realtime-ai" + label: Literal["rtvi"] = "rtvi" type: Literal["user-stopped-speaking"] = "user-stopped-speaking" class RTVIJSONCompletion(BaseModel): - label: Literal["realtime-ai"] = "realtime-ai" + label: Literal["rtvi"] = "rtvi" type: Literal["json-completion"] = "json-completion" data: str From cea4d1894e485457ef85b77d64c136f973fafd59 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 22 Jul 2024 08:47:59 -0700 Subject: [PATCH 29/37] processors(rtvi): change voice before LLM updates --- src/pipecat/processors/frameworks/rtvi.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 23acb71bd..ca7995cb5 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -429,15 +429,16 @@ class RTVIProcessor(FrameProcessor): await self.push_frame(frame) async def _handle_config_update(self, config: RTVIConfig): + # Change voice before LLM updates, so we can hear the new vocie. + if config.tts and config.tts.voice: + frame = TTSVoiceUpdateFrame(config.tts.voice) + await self.push_frame(frame) if config.llm and config.llm.model: frame = LLMModelUpdateFrame(config.llm.model) await self.push_frame(frame) if config.llm and config.llm.messages: frame = LLMMessagesUpdateFrame(config.llm.messages) await self.push_frame(frame) - if config.tts and config.tts.voice: - frame = TTSVoiceUpdateFrame(config.tts.voice) - await self.push_frame(frame) async def _handle_llm_get_context(self): data = RTVILLMContextMessageData(messages=self._tma_in.messages) From 43932220f761789bc9e99280d05713d2eeba537a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 22 Jul 2024 09:40:16 -0700 Subject: [PATCH 30/37] processors(rtvi): use only user-transcription --- src/pipecat/processors/frameworks/rtvi.py | 17 ++++++++--------- 1 file changed, 8 insertions(+), 9 deletions(-) diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index ca7995cb5..4e5ed442b 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -131,6 +131,7 @@ class RTVITranscriptionMessageData(BaseModel): text: str user_id: str timestamp: str + final: bool class RTVITranscriptionMessage(BaseModel): @@ -139,12 +140,6 @@ class RTVITranscriptionMessage(BaseModel): data: RTVITranscriptionMessageData -class RTVIInterimTranscriptionMessage(BaseModel): - label: Literal["rtvi"] = "rtvi" - type: Literal["user-interim-transcription"] = "user-interim-transcription" - data: RTVITranscriptionMessageData - - class RTVIUserStartedSpeakingMessage(BaseModel): label: Literal["rtvi"] = "rtvi" type: Literal["user-started-speaking"] = "user-started-speaking" @@ -318,11 +313,15 @@ class RTVIProcessor(FrameProcessor): data=RTVITranscriptionMessageData( text=frame.text, user_id=frame.user_id, - timestamp=frame.timestamp)) + timestamp=frame.timestamp, + final=True)) elif isinstance(frame, InterimTranscriptionFrame): - message = RTVIInterimTranscriptionMessage( + message = RTVITranscriptionMessage( data=RTVITranscriptionMessageData( - text=frame.text, user_id=frame.user_id, timestamp=frame.timestamp)) + text=frame.text, + user_id=frame.user_id, + timestamp=frame.timestamp, + final=False)) if message: frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) From 9233bb490c46029d66deba5cfdb30535f9ea7420 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 22 Jul 2024 11:40:17 -0700 Subject: [PATCH 31/37] processors(rtvi): add support for "tts-text" messages --- src/pipecat/processors/frameworks/rtvi.py | 33 ++++++++++++++++++++++- 1 file changed, 32 insertions(+), 1 deletion(-) diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 4e5ed442b..1143499a2 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -34,7 +34,7 @@ from pipecat.processors.aggregators.llm_response import ( from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.ai_services import AIService from pipecat.services.cartesia import CartesiaTTSService -from pipecat.services.openai import OpenAILLMService, OpenAILLMContext, OpenAILLMContextFrame +from pipecat.services.openai import OpenAILLMService, OpenAILLMContext from pipecat.transports.base_transport import BaseTransport DEFAULT_MESSAGES = [ @@ -122,6 +122,16 @@ class RTVILLMContextMessage(BaseModel): data: RTVILLMContextMessageData +class RTVITTSTextMessageData(BaseModel): + text: str + + +class RTVITTSTextMessage(BaseModel): + label: Literal["rtvi"] = "rtvi" + type: Literal["tts-text"] = "tts-text" + data: RTVITTSTextMessageData + + class RTVIBotReady(BaseModel): label: Literal["rtvi"] = "rtvi" type: Literal["bot-ready"] = "bot-ready" @@ -157,6 +167,7 @@ class RTVIJSONCompletion(BaseModel): class FunctionCaller(FrameProcessor): + def __init__(self, context): super().__init__() self._checking = False @@ -191,6 +202,8 @@ class FunctionCaller(FrameProcessor): await self._start_callbacks[function_name](self) async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if isinstance(frame, LLMFullResponseStartFrame): self._checking = True await self.push_frame(frame, direction) @@ -236,6 +249,21 @@ class FunctionCaller(FrameProcessor): await self.push_frame(frame, direction) +class RTVITTSTextProcessor(FrameProcessor): + + def __init__(self): + super().__init__() + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + await self.push_frame(frame, direction) + + if isinstance(frame, TextFrame): + message = RTVITTSTextMessage(data=RTVITTSTextMessageData(text=frame.text)) + await self.push_frame(TransportMessageFrame(message=message.model_dump(exclude_none=True))) + + class RTVIProcessor(FrameProcessor): def __init__( @@ -404,11 +432,14 @@ class RTVIProcessor(FrameProcessor): context = OpenAILLMContext(messages=messages) self._fc = FunctionCaller(context) + self._tts_text = RTVITTSTextProcessor() + pipeline = Pipeline([ self._tma_in, self._llm, self._fc, self._tts, + self._tts_text, self._tma_out, self._transport.output(), ]) From 65cdf50774cb4163bef477b96469a2052af2bb7a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 22 Jul 2024 15:01:45 -0700 Subject: [PATCH 32/37] processors(rtvi): fix task cleanup --- src/pipecat/processors/async_frame_processor.py | 1 + src/pipecat/processors/frameworks/rtvi.py | 1 + src/pipecat/services/cartesia.py | 2 ++ src/pipecat/transports/base_input.py | 3 +++ src/pipecat/transports/base_output.py | 1 + 5 files changed, 8 insertions(+) diff --git a/src/pipecat/processors/async_frame_processor.py b/src/pipecat/processors/async_frame_processor.py index 24c016867..28a27d255 100644 --- a/src/pipecat/processors/async_frame_processor.py +++ b/src/pipecat/processors/async_frame_processor.py @@ -59,5 +59,6 @@ class AsyncFrameProcessor(FrameProcessor): (frame, direction) = await self._push_queue.get() await self.push_frame(frame, direction) running = not isinstance(frame, EndFrame) + self._push_queue.task_done() except asyncio.CancelledError: break diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 1143499a2..37142e2d3 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -316,6 +316,7 @@ class RTVIProcessor(FrameProcessor): try: (frame, direction) = await self._frame_queue.get() await self._handle_frame(frame, direction) + self._frame_queue.task_done() except asyncio.CancelledError: break diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index c672121be..367d0c400 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -114,9 +114,11 @@ class CartesiaTTSService(TTSService): try: if self._context_appending_task: self._context_appending_task.cancel() + await self._context_appending_task self._context_appending_task = None if self._receive_task: self._receive_task.cancel() + await self._receive_task self._receive_task = None if self._websocket: ws = self._websocket diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index f87772c01..c0a13d1ce 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -104,6 +104,7 @@ class BaseInputTransport(FrameProcessor): try: (frame, direction) = await self._push_queue.get() await self.push_frame(frame, direction) + self._push_queue.task_done() except asyncio.CancelledError: break @@ -185,6 +186,8 @@ class BaseInputTransport(FrameProcessor): # Push audio downstream if passthrough. if audio_passthrough: await self._internal_push_frame(frame) + + self._audio_in_queue.task_done() except asyncio.CancelledError: break except Exception as e: diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 41f2df79e..355008f89 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -204,6 +204,7 @@ class BaseOutputTransport(FrameProcessor): try: (frame, direction) = await self._push_queue.get() await self.push_frame(frame, direction) + self._push_queue.task_done() except asyncio.CancelledError: break From c1170260b500be2a8104cac2c44d213e3fc87843 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 22 Jul 2024 15:27:33 -0700 Subject: [PATCH 33/37] processors(rtvi): use generic LLM and TTS names --- src/pipecat/processors/frameworks/rtvi.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 37142e2d3..f9e76fb6d 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -422,11 +422,12 @@ class RTVIProcessor(FrameProcessor): self._tma_out = LLMAssistantResponseAggregator(messages) self._llm = self._llm_cls( + name="LLM", base_url=self._llm_base_url, api_key=self._llm_api_key, model=model) - self._tts = self._tts_cls(api_key=self._tts_api_key, voice_id=voice) + self._tts = self._tts_cls(name="TTS", api_key=self._tts_api_key, voice_id=voice) # TODO-CB: Eventually we'll need to switch the context aggregators to use the # OpenAI context frames instead of message frames From 087fe9a53754d5f4b57be33e4693663f57818f24 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 22 Jul 2024 15:30:16 -0700 Subject: [PATCH 34/37] services(cartesia): fix TTFB --- src/pipecat/services/cartesia.py | 12 +++--------- 1 file changed, 3 insertions(+), 9 deletions(-) diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 367d0c400..86ae10e99 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -82,7 +82,6 @@ class CartesiaTTSService(TTSService): self._timestamped_words_buffer = [] self._receive_task = None self._context_appending_task = None - self._waiting_for_ttfb = False def can_generate_metrics(self) -> bool: return True @@ -127,7 +126,6 @@ class CartesiaTTSService(TTSService): self._context_id = None self._context_id_start_timestamp = None self._timestamped_words_buffer = [] - self._waiting_for_ttfb = False await self.stop_all_metrics() except Exception as e: logger.exception(f"{self} error closing websocket: {e}") @@ -148,6 +146,7 @@ class CartesiaTTSService(TTSService): if not msg or msg["context_id"] != self._context_id: continue if msg["type"] == "done": + await self.stop_ttfb_metrics() # unset _context_id but not the _context_id_start_timestamp because we are likely still # playing out audio and need the timestamp to set send context frames self._context_id = None @@ -158,11 +157,9 @@ class CartesiaTTSService(TTSService): list(zip(msg["word_timestamps"]["words"], msg["word_timestamps"]["end"])) ) elif msg["type"] == "chunk": + await self.stop_ttfb_metrics() if not self._context_id_start_timestamp: self._context_id_start_timestamp = time.time() - if self._waiting_for_ttfb: - await self.stop_ttfb_metrics() - self._waiting_for_ttfb = False frame = AudioRawFrame( audio=base64.b64decode(msg["data"]), sample_rate=self._output_format["sample_rate"], @@ -198,11 +195,8 @@ class CartesiaTTSService(TTSService): if not self._websocket: await self._connect() - if not self._waiting_for_ttfb: - await self.start_ttfb_metrics() - self._waiting_for_ttfb = True - if not self._context_id: + await self.start_ttfb_metrics() self._context_id = str(uuid.uuid4()) msg = { From 87bc8a9da6fd3a57fefbfd8bef0b2f2218eb1ff3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 22 Jul 2024 15:53:39 -0700 Subject: [PATCH 35/37] examples: remove RTVI since there are full demos elsewhere --- examples/rtvi/bot.py | 78 ----------------------------------------- examples/rtvi/runner.py | 58 ------------------------------ 2 files changed, 136 deletions(-) delete mode 100644 examples/rtvi/bot.py delete mode 100644 examples/rtvi/runner.py diff --git a/examples/rtvi/bot.py b/examples/rtvi/bot.py deleted file mode 100644 index 6214a6c72..000000000 --- a/examples/rtvi/bot.py +++ /dev/null @@ -1,78 +0,0 @@ -# -# Copyright (c) 2024, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -import asyncio -import sys -import os - -from pipecat.pipeline.pipeline import Pipeline -from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.pipeline.runner import PipelineRunner -from pipecat.processors.frameworks.rtvi import ( - RTVIConfig, - RTVILLMConfig, - RTVIProcessor, - RTVISetup, - RTVITTSConfig) -from pipecat.transports.services.daily import DailyParams, DailyTransport -from pipecat.vad.silero import SileroVADAnalyzer - -from runner import configure - -from loguru import logger - -from dotenv import load_dotenv -load_dotenv(override=True) - -logger.remove(0) -logger.add(sys.stderr, level="DEBUG") - - -async def main(room_url, token): - transport = DailyTransport( - room_url, - token, - "Realtime AI", - DailyParams( - audio_out_enabled=True, - transcription_enabled=True, - vad_enabled=True, - vad_analyzer=SileroVADAnalyzer() - )) - - llm = RTVILLMConfig( - model="llama3-70b-8192", - messages=[{"role": "system", "content": "You are a helpful assistant named Gary. Briefly say hello!"}] - ) - tts = RTVITTSConfig(voice="79a125e8-cd45-4c13-8a67-188112f4dd22") - setup = RTVISetup(config=RTVIConfig(llm=llm, tts=tts)) - - rtai = RTVIProcessor( - transport=transport, - setup=setup, - llm_api_key=os.getenv("OPENAI_API_KEY"), - tts_api_key=os.getenv("CARTESIA_API_KEY")) - - runner = PipelineRunner() - - pipeline = Pipeline([transport.input(), rtai]) - - task = PipelineTask( - pipeline, - params=PipelineParams( - allow_interruptions=True, - enable_metrics=True, - report_only_initial_ttfb=True)) - - @transport.event_handler("on_participant_joined") - async def on_participant_joined(transport, participant): - transport.capture_participant_transcription(participant["id"]) - - await runner.run(task) - -if __name__ == "__main__": - (url, token) = configure() - asyncio.run(main(url, token)) diff --git a/examples/rtvi/runner.py b/examples/rtvi/runner.py deleted file mode 100644 index 6d1a8113d..000000000 --- a/examples/rtvi/runner.py +++ /dev/null @@ -1,58 +0,0 @@ -import argparse -import os -import time -import urllib -import requests - - -def configure(): - parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample") - parser.add_argument( - "-u", - "--url", - type=str, - required=False, - help="URL of the Daily room to join") - parser.add_argument( - "-k", - "--apikey", - type=str, - required=False, - help="Daily API Key (needed to create an owner token for the room)", - ) - - args, unknown = parser.parse_known_args() - - url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL") - key = args.apikey or os.getenv("DAILY_API_KEY") - - if not url: - raise Exception( - "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL.") - - if not key: - raise Exception("No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers.") - - # Create a meeting token for the given room with an expiration 1 hour in - # the future. - room_name: str = urllib.parse.urlparse(url).path[1:] - expiration: float = time.time() + 60 * 60 - - res: requests.Response = requests.post( - f"https://api.daily.co/v1/meeting-tokens", - headers={ - "Authorization": f"Bearer {key}"}, - json={ - "properties": { - "room_name": room_name, - "is_owner": True, - "exp": expiration}}, - ) - - if res.status_code != 200: - raise Exception( - f"Failed to create meeting token: {res.status_code} {res.text}") - - token: str = res.json()["token"] - - return (url, token) From 95ff1d141cbf0555391ce42ea877763960a8da06 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 22 Jul 2024 15:54:52 -0700 Subject: [PATCH 36/37] update CHANGELOG with RTVIProcessor --- CHANGELOG.md | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a2512d378..b07dde759 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added -- Added `RealtimeAIProcessor`... +- Added `RTVIProcessor` which implements the RTVI-AI standard. + See https://github.com/rtvi-ai - Added `BotInterruptionFrame` which allows interrupting the bot while talking. @@ -52,10 +53,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 leverages the new Cartesia contexts which maintains generated audio prosody when multiple inputs are sent, therefore improving audio quality a lot. -### Other - -- Added `examples/realtime-ai/bot.py`... - ## [0.0.36] - 2024-07-02 ### Added From 6dab0e9de7af8be4fed3271ee678ae32fa2764f7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 22 Jul 2024 16:00:06 -0700 Subject: [PATCH 37/37] update CHANGELOG for 0.0.37 --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b07dde759..412921b02 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,7 @@ 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). -## [Unreleased] +## [0.0.37] - 2024-07-22 ### Added