Add a NIM LLM service

transports(websockets): use frame serializers during interruptions

CHANGELOG.md

@@ -20,6 +20,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Updated STT and TTS services with language options that match the supported
   languages for each service.
 
+### Fixed
+
+- Fixed a `WebsocketServerTransport` issue that would prevent interruptions with
+  `TwilioSerializer` from working.
+
+- `DailyTransport.capture_participant_video` now allows capturing user's screen
+  share by simply passing `video_source="screenVideo"`.
+
+- Fixed Google Gemini message handling to properly convert appended messages to
+  Gemini's required format.
+
 ## [0.0.49] - 2024-11-17
 
 ### Added

examples/README.md

@@ -42,6 +42,7 @@ Next, follow the steps in the README for each demo.
 | [Dialin Chatbot](dialin-chatbot)             | A chatbot that connects to an incoming phone call from Daily or Twilio.                                                                    | Deepgram, ElevenLabs, OpenAI, Daily, Twilio                       |
 | [Twilio Chatbot](twilio-chatbot)             | A chatbot that connects to an incoming phone call from Twilio.                                                                             | Deepgram, ElevenLabs, OpenAI, Daily, Twilio                       |
 | [studypal](studypal)                         | A chatbot to have a conversation about any article on the web                                                                              |                                                                   |
+| [WebSocket Chatbot Server](websocket-server) | A real-time websocket server that handles audio streaming and bot interactions with speech-to-text and text-to-speech capabilities | `python-websockets`, `openai`, `deepgram`, `silero-tts`, `numpy` |
 
 > [!IMPORTANT]
 > These example projects use Daily as a WebRTC transport and can be joined using their hosted Prebuilt UI.

examples/twilio-chatbot/requirements.txt

@@ -1,4 +1,4 @@
-pipecat-ai[daily,cartesia,openai,silero,deepgram]
+pipecat-ai[cartesia,openai,silero,deepgram]
 fastapi
 uvicorn
 python-dotenv

src/pipecat/processors/frame_processor.py

@@ -6,10 +6,11 @@
 
 import asyncio
 import inspect
-
 from enum import Enum
 from typing import Awaitable, Callable, Optional
 
+from loguru import logger
+
 from pipecat.clocks.base_clock import BaseClock
 from pipecat.frames.frames import (
     EndFrame,
@@ -24,8 +25,6 @@ from pipecat.metrics.metrics import LLMTokenUsage, MetricsData
 from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMetrics
 from pipecat.utils.utils import obj_count, obj_id
 
-from loguru import logger
-
 
 class FrameDirection(Enum):
     DOWNSTREAM = 1
@@ -220,11 +219,16 @@ class FrameProcessor:
     #
 
     async def _start_interruption(self):
-        # Cancel the push frame task. This will stop pushing frames downstream.
-        await self.__cancel_push_task()
+        try:
+            # Cancel the push frame task. This will stop pushing frames downstream.
+            await self.__cancel_push_task()
 
-        # Cancel the input task. This will stop processing queued frames.
-        await self.__cancel_input_task()
+            # Cancel the input task. This will stop processing queued frames.
+            await self.__cancel_input_task()
+        except Exception as e:
+            logger.exception(f"Uncaught exception in {self}: {e}")
+            await self.push_error(ErrorFrame(str(e)))
+            raise
 
         # Create a new input queue and task.
         self.__create_input_task()
@@ -281,7 +285,11 @@ class FrameProcessor:
 
                 self.__input_queue.task_done()
             except asyncio.CancelledError:
+                logger.trace(f"Cancelled input task in {self}")
                 break
+            except Exception as e:
+                logger.exception(f"Uncaught exception in {self}: {e}")
+                await self.push_error(ErrorFrame(str(e)))
 
     def __create_push_task(self):
         self.__push_queue = asyncio.Queue()
@@ -300,7 +308,11 @@ class FrameProcessor:
                 running = not isinstance(frame, EndFrame)
                 self.__push_queue.task_done()
             except asyncio.CancelledError:
+                logger.trace(f"Cancelled push task in {self}")
                 break
+            except Exception as e:
+                logger.exception(f"Uncaught exception in {self}: {e}")
+                await self.push_error(ErrorFrame(str(e)))
 
     async def _call_event_handler(self, event_name: str, *args, **kwargs):
         try:

src/pipecat/services/google.py

@@ -16,13 +16,19 @@ from PIL import Image
 from pydantic import BaseModel, Field
 
 from pipecat.frames.frames import (
+<<<<<<< Updated upstream
     AudioRawFrame,
+=======
+    CancelFrame,
+    EndFrame,
+>>>>>>> Stashed changes
     ErrorFrame,
     Frame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
     LLMUpdateSettingsFrame,
+    StartFrame,
     TextFrame,
     TTSAudioRawFrame,
     TTSStartedFrame,
@@ -45,8 +51,12 @@ from pipecat.transcriptions.language import Language
 try:
     import google.ai.generativelanguage as glm
     import google.generativeai as gai
+<<<<<<< Updated upstream
     from google.cloud import texttospeech_v1
     from google.generativeai.types import GenerationConfig
+=======
+    from google.cloud import speech, texttospeech_v1
+>>>>>>> Stashed changes
     from google.oauth2 import service_account
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
@@ -332,6 +342,22 @@ class GoogleLLMContext(OpenAILLMContext):
         self._messages[:] = messages
         self._restructure_from_openai_messages()
 
+    def add_messages(self, messages: List):
+        # Convert each message individually
+        converted_messages = []
+        for msg in messages:
+            if isinstance(msg, glm.Content):
+                # Already in Gemini format
+                converted_messages.append(msg)
+            else:
+                # Convert from standard format to Gemini format
+                converted = self.from_standard_message(msg)
+                if converted is not None:
+                    converted_messages.append(converted)
+
+        # Add the converted messages to our existing messages
+        self._messages.extend(converted_messages)
+
     def get_messages_for_logging(self):
         msgs = []
         for message in self.messages:
@@ -811,3 +837,107 @@ class GoogleTTSService(TTSService):
             yield ErrorFrame(error=error_message)
         finally:
             yield TTSStoppedFrame()
+<<<<<<< Updated upstream
+=======
+
+
+class GoogleSTTService(STTService):
+    def __init__(
+        self,
+        *,
+        credentials_path: str,
+        language: Language = Language.EN,
+        sample_rate: int = 16000,
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self._credentials_path = credentials_path
+        self._language = language
+        self._sample_rate = sample_rate
+        self._client = None
+        self._streaming_config = None
+        self._requests_queue = asyncio.Queue()
+        self._responses = None
+
+    async def start(self, frame: StartFrame):
+        await super().start(frame)
+        credentials = service_account.Credentials.from_service_account_file(self._credentials_path)
+        self._client = speech.SpeechClient(credentials=credentials)
+
+        config = speech.RecognitionConfig(
+            encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
+            sample_rate_hertz=self._sample_rate,
+            language_code=self._language.value,
+            enable_automatic_punctuation=True,
+        )
+        self._streaming_config = speech.StreamingRecognitionConfig(
+            config=config, interim_results=True
+        )
+
+        # Start the recognition stream
+        self._responses = self._client.streaming_recognize(
+            self._streaming_config, self._request_generator()
+        )
+
+    async def stop(self, frame: EndFrame):
+        await super().stop(frame)
+        await self._requests_queue.put(None)  # Signal to stop the request generator
+        self._client = None
+        self._streaming_config = None
+        self._responses = None
+
+    async def cancel(self, frame: CancelFrame):
+        await super().cancel(frame)
+        await self.stop(EndFrame())
+
+    async def set_language(self, language: Language):
+        self._language = language
+        # Recreate the streaming config with the new language
+        if self._client:
+            config = speech.RecognitionConfig(
+                encoding=speech.RecognitionConfig.AudioEncoding.LINEAR16,
+                sample_rate_hertz=self._sample_rate,
+                language_code=self._language.value,
+                enable_automatic_punctuation=True,
+            )
+            self._streaming_config = speech.StreamingRecognitionConfig(
+                config=config, interim_results=True
+            )
+            # Restart the recognition stream
+            await self._requests_queue.put(None)  # Signal to stop the current request generator
+            self._responses = self._client.streaming_recognize(
+                self._streaming_config, self._request_generator()
+            )
+
+    async def _request_generator(self):
+        while True:
+            request = await self._requests_queue.get()
+            if request is None:
+                break
+            yield request
+
+    async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
+        if not self._client or not self._streaming_config or not self._responses:
+            raise RuntimeError("GoogleSTTService not started")
+
+        # Queue the audio content
+        await self._requests_queue.put(speech.StreamingRecognizeRequest(audio_content=audio))
+
+        # Process the responses
+        for response in self._responses:
+            for result in response.results:
+                if result.alternatives:
+                    transcript = result.alternatives[0].transcript
+                    if result.is_final:
+                        await self.push_frame(
+                            TranscriptionFrame(transcript, "", time_now_iso8601(), self._language)
+                        )
+                    else:
+                        await self.push_frame(
+                            InterimTranscriptionFrame(
+                                transcript, "", time_now_iso8601(), self._language
+                            )
+                        )
+
+        yield None
+>>>>>>> Stashed changes

src/pipecat/services/nim.py

@@ -0,0 +1,105 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+
+from pipecat.metrics.metrics import LLMTokenUsage
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.services.openai import OpenAILLMService
+
+
+class NimLLMService(OpenAILLMService):
+    """A service for interacting with NVIDIA's NIM (NVIDIA Inference Microservice) API.
+
+    This service extends OpenAILLMService to work with NVIDIA's NIM API while maintaining
+    compatibility with the OpenAI-style interface. It specifically handles the difference
+    in token usage reporting between NIM (incremental) and OpenAI (final summary).
+
+    Args:
+        api_key (str): The API key for accessing NVIDIA's NIM API
+        base_url (str, optional): The base URL for NIM API. Defaults to "https://integrate.api.nvidia.com/v1"
+        model (str, optional): The model identifier to use. Defaults to "nvidia/llama-3.1-nemotron-70b-instruct"
+        **kwargs: Additional keyword arguments passed to OpenAILLMService
+
+    Example:
+        ```python
+        service = NimLLMService(
+            api_key="your-api-key",
+            model="nvidia/llama-3.1-nemotron-70b-instruct"
+        )
+        ```
+    """
+
+    def __init__(
+        self,
+        *,
+        api_key: str,
+        base_url: str = "https://integrate.api.nvidia.com/v1",
+        model: str = "nvidia/llama-3.1-nemotron-70b-instruct",
+        **kwargs,
+    ):
+        super().__init__(api_key=api_key, base_url=base_url, model=model, **kwargs)
+        # Counters for accumulating token usage metrics
+        self._prompt_tokens = 0
+        self._completion_tokens = 0
+        self._total_tokens = 0
+        self._has_reported_prompt_tokens = False
+        self._is_processing = False
+
+    async def _process_context(self, context: OpenAILLMContext):
+        """Process a context through the LLM and accumulate token usage metrics.
+
+        This method overrides the parent class implementation to handle NVIDIA's
+        incremental token reporting style, accumulating the counts and reporting
+        them once at the end of processing.
+
+        Args:
+            context (OpenAILLMContext): The context to process, containing messages
+                and other information needed for the LLM interaction.
+        """
+        # Reset all counters and flags at the start of processing
+        self._prompt_tokens = 0
+        self._completion_tokens = 0
+        self._total_tokens = 0
+        self._has_reported_prompt_tokens = False
+        self._is_processing = True
+
+        try:
+            await super()._process_context(context)
+        finally:
+            self._is_processing = False
+            # Report final accumulated token usage at the end of processing
+            if self._prompt_tokens > 0 or self._completion_tokens > 0:
+                self._total_tokens = self._prompt_tokens + self._completion_tokens
+                tokens = LLMTokenUsage(
+                    prompt_tokens=self._prompt_tokens,
+                    completion_tokens=self._completion_tokens,
+                    total_tokens=self._total_tokens,
+                )
+                await super().start_llm_usage_metrics(tokens)
+
+    async def start_llm_usage_metrics(self, tokens: LLMTokenUsage):
+        """Accumulate token usage metrics during processing.
+
+        This method intercepts the incremental token updates from NVIDIA's API
+        and accumulates them instead of passing each update to the metrics system.
+        The final accumulated totals are reported at the end of processing.
+
+        Args:
+            tokens (LLMTokenUsage): The token usage metrics for the current chunk
+                of processing, containing prompt_tokens and completion_tokens counts.
+        """
+        # Only accumulate metrics during active processing
+        if not self._is_processing:
+            return
+
+        # Record prompt tokens the first time we see them
+        if not self._has_reported_prompt_tokens and tokens.prompt_tokens > 0:
+            self._prompt_tokens = tokens.prompt_tokens
+            self._has_reported_prompt_tokens = True
+
+        # Update completion tokens count if it has increased
+        if tokens.completion_tokens > self._completion_tokens:
+            self._completion_tokens = tokens.completion_tokens

src/pipecat/transports/network/fastapi_websocket.py

@@ -15,8 +15,6 @@ from pydantic.main import BaseModel
 
 from pipecat.frames.frames import (
     AudioRawFrame,
-    CancelFrame,
-    EndFrame,
     Frame,
     InputAudioRawFrame,
     StartFrame,

src/pipecat/transports/network/websocket_server.py

@@ -148,6 +148,7 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
         await super().process_frame(frame, direction)
 
         if isinstance(frame, StartInterruptionFrame):
+            await self._write_frame(frame)
             self._next_send_time = 0
 
     async def write_raw_audio_frames(self, frames: bytes):
@@ -189,6 +190,11 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
 
         self._websocket_audio_buffer = bytes()
 
+    async def _write_frame(self, frame: Frame):
+        payload = self._params.serializer.serialize(frame)
+        if payload and self._websocket:
+            await self._websocket.send(payload)
+
 
 class WebsocketServerTransport(BaseTransport):
     def __init__(

src/pipecat/transports/services/daily.py

@@ -328,7 +328,7 @@ class DailyTransportClient(EventHandler):
 
                 logger.info(f"Joined {self._room_url}")
 
-                if self._token and self._params.transcription_enabled:
+                if self._params.transcription_enabled:
                     await self._start_transcription()
 
                 await self._callbacks.on_joined(data)
@@ -342,6 +342,10 @@ class DailyTransportClient(EventHandler):
             await self._callbacks.on_error(error_msg)
 
     async def _start_transcription(self):
+        if not self._token:
+            logger.warning(f"Transcription can't be started without a room token")
+            return
+
         logger.info(f"Enabling transcription with settings {self._params.transcription_settings}")
 
         future = self._loop.create_future()
@@ -436,6 +440,8 @@ class DailyTransportClient(EventHandler):
             await self._callbacks.on_error(error_msg)
 
     async def _stop_transcription(self):
+        if not self._token:
+            return
         future = self._loop.create_future()
         self._client.stop_transcription(completion=completion_callback(future))
         error = await future
@@ -499,9 +505,9 @@ class DailyTransportClient(EventHandler):
         video_source: str = "camera",
         color_format: str = "RGB",
     ):
-        # Only enable camera subscription on this participant
+        # Only enable the desired video source subscription on this participant.
         await self.update_subscriptions(
-            participant_settings={participant_id: {"media": {"camera": "subscribed"}}}
+            participant_settings={participant_id: {"media": {video_source: "subscribed"}}}
         )
 
         self._video_renderers[participant_id] = callback