diff --git a/CHANGELOG.md b/CHANGELOG.md index 286325679..412921b02 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,27 @@ 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 + +- Added `RTVIProcessor` which implements the RTVI-AI standard. + See https://github.com/rtvi-ai + +- Added `BotInterruptionFrame` which allows interrupting the bot while talking. + +- 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 @@ -24,6 +44,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/frames/frames.py b/src/pipecat/frames/frames.py index b8a2a6d06..46e2408bc 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -158,6 +158,34 @@ 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 + replace the current context LLM messages and should generate a new + LLMMessagesFrame. + + """ + 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 @@ -240,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 @@ -335,3 +373,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/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/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index aa85c5ade..6939a70c4 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -14,7 +14,9 @@ from pipecat.frames.frames import ( InterimTranscriptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, + LLMMessagesAppendFrame, LLMMessagesFrame, + LLMMessagesUpdateFrame, StartInterruptionFrame, TranscriptionFrame, TextFrame, @@ -120,6 +122,19 @@ 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. + 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/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/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 diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py new file mode 100644 index 000000000..f9e76fb6d --- /dev/null +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -0,0 +1,523 @@ +# +# Copyright (c) 2024, Daily +# +# 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, + LLMFullResponseEndFrame, + LLMFullResponseStartFrame, + LLMMessagesAppendFrame, + LLMMessagesUpdateFrame, + LLMModelUpdateFrame, + StartFrame, + SystemFrame, + TTSSpeakFrame, + TTSVoiceUpdateFrame, + TextFrame, + TranscriptionFrame, + TransportMessageFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame) +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.openai import OpenAILLMService, OpenAILLMContext +from pipecat.transports.base_transport import BaseTransport + +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 RTVILLMConfig(BaseModel): + model: Optional[str] = None + messages: Optional[List[dict]] = None + + +class RTVITTSConfig(BaseModel): + voice: Optional[str] = None + + +class RTVIConfig(BaseModel): + llm: Optional[RTVILLMConfig] = None + tts: Optional[RTVITTSConfig] = None + + +class RTVISetup(BaseModel): + config: Optional[RTVIConfig] = None + + +class RTVILLMMessageData(BaseModel): + messages: List[dict] + + +class RTVITTSMessageData(BaseModel): + text: str + interrupt: Optional[bool] = False + + +class RTVIMessageData(BaseModel): + setup: Optional[RTVISetup] = None + config: Optional[RTVIConfig] = None + llm: Optional[RTVILLMMessageData] = None + tts: Optional[RTVITTSMessageData] = None + + +class RTVIMessage(BaseModel): + label: Literal["rtvi"] = "rtvi" + type: str + id: str + data: Optional[RTVIMessageData] = None + + +class RTVIResponseData(BaseModel): + success: bool + error: Optional[str] = None + + +class RTVIResponse(BaseModel): + label: Literal["rtvi"] = "rtvi" + type: Literal["response"] = "response" + id: str + data: RTVIResponseData + + +class RTVIErrorData(BaseModel): + message: str + + +class RTVIError(BaseModel): + label: Literal["rtvi"] = "rtvi" + type: Literal["error"] = "error" + data: RTVIErrorData + + +class RTVILLMContextMessageData(BaseModel): + messages: List[dict] + + +class RTVILLMContextMessage(BaseModel): + label: Literal["rtvi"] = "rtvi" + type: Literal["llm-context"] = "llm-context" + 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" + + +class RTVITranscriptionMessageData(BaseModel): + text: str + user_id: str + timestamp: str + final: bool + + +class RTVITranscriptionMessage(BaseModel): + label: Literal["rtvi"] = "rtvi" + type: Literal["user-transcription"] = "user-transcription" + data: RTVITranscriptionMessageData + + +class RTVIUserStartedSpeakingMessage(BaseModel): + label: Literal["rtvi"] = "rtvi" + type: Literal["user-started-speaking"] = "user-started-speaking" + + +class RTVIUserStoppedSpeakingMessage(BaseModel): + label: Literal["rtvi"] = "rtvi" + type: Literal["user-stopped-speaking"] = "user-stopped-speaking" + + +class RTVIJSONCompletion(BaseModel): + label: Literal["rtvi"] = "rtvi" + 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): + await super().process_frame(frame, direction) + + 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 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__( + self, + *, + transport: BaseTransport, + setup: RTVISetup | None = None, + llm_api_key: str = "", + llm_base_url: str = "https://api.groq.com/openai/v1", + tts_api_key: str = "", + 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 + self._start_frame: Frame | None = None + self._llm: FrameProcessor | None = None + 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 + 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() + 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) + self._frame_queue.task_done() + 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, TranscriptionFrame) or isinstance(frame, InterimTranscriptionFrame): + await self._handle_transcriptions(frame) + elif isinstance(frame, UserStartedSpeakingFrame) or isinstance(frame, UserStoppedSpeakingFrame): + await self._handle_interruptions(frame) + + 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 = RTVITranscriptionMessage( + data=RTVITranscriptionMessageData( + text=frame.text, + user_id=frame.user_id, + timestamp=frame.timestamp, + final=True)) + elif isinstance(frame, InterimTranscriptionFrame): + message = RTVITranscriptionMessage( + data=RTVITranscriptionMessageData( + text=frame.text, + user_id=frame.user_id, + timestamp=frame.timestamp, + final=False)) + + 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): + message = RTVIUserStartedSpeakingMessage() + elif isinstance(frame, UserStoppedSpeakingFrame): + message = RTVIUserStoppedSpeakingMessage() + + if message: + frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) + await self.push_frame(frame) + + async def _handle_message(self, frame: TransportMessageFrame): + try: + message = RTVIMessage.model_validate(frame.message) + except ValidationError as 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(message.id, setup) + case "config-update": + 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.llm) + case "llm-update-context": + 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() + case _: + success = False + error = f"unsupported type {message.type}" + + await self._send_response(message.id, success, error) + except ValidationError as 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): + 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: + voice = setup.config.tts.voice + + self._tma_in = LLMUserResponseAggregator(messages) + 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(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 + 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(), + ]) + 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) + + message = RTVIBotReady() + frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) + 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) + + async def _handle_llm_get_context(self): + 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: RTVILLMMessageData): + if data and data.messages: + frame = LLMMessagesAppendFrame(data.messages) + await self.push_frame(frame) + + 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: RTVITTSMessageData): + 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_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 + # 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: + pipeline = Pipeline([self._transport.output()]) + self._pipeline = pipeline + + parent = self.get_parent() + if parent and self._start_frame: + parent.link(pipeline) + + message = RTVIResponse(id=id, data=RTVIResponseData(success=success, error=error)) + frame = TransportMessageFrame(message=message.model_dump(exclude_none=True)) + await self.push_frame(frame) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 8647d3f77..bc00accdf 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -19,8 +19,10 @@ from pipecat.frames.frames import ( LLMFullResponseEndFrame, StartFrame, StartInterruptionFrame, + TTSSpeakFrame, TTSStartedFrame, TTSStoppedFrame, + TTSVoiceUpdateFrame, TextFrame, VisionImageRawFrame, ) @@ -148,6 +150,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]: @@ -173,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 @@ -196,13 +202,18 @@ 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) 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: 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..86ae10e99 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -82,11 +82,14 @@ 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 + 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() @@ -110,9 +113,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 @@ -121,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}") @@ -142,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 @@ -152,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"], @@ -192,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 = { 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] diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 828dc3e40..c0a13d1ce 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) @@ -101,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 @@ -108,24 +112,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 +164,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 @@ -171,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 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):