Merge pull request #312 from pipecat-ai/aleix/rtvi-support

RTVI support

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

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

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)

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)
 

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

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

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)

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)
 

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)
 

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}]")
 

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 = {

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}]")
 

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}]")
 

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)
 

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}]")
 

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]

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.
+                await self._start_interruption()
-                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()
             elif isinstance(frame, UserStoppedSpeakingFrame):
                 logger.debug("User stopped speaking")
-                await self.push_frame(StopInterruptionFrame())
+                await self._stop_interruption()
-        await self._internal_push_frame(frame)
+
+        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:

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
 

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):