feat: add dial-tone detection and generation

Use Google Gemini message format when adding message to the LLM context

CHANGELOG.md

@@ -20,6 +20,10 @@ 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 Google Gemini message handling to properly convert appended messages to Gemini's required format
+
 ## [0.0.49] - 2024-11-17
 
 ### Added

examples/foundational/tones.py

@@ -0,0 +1,162 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+
+import aiohttp
+import numpy as np
+from dotenv import load_dotenv
+from dtmf import detect, generate, model, parse
+from loguru import logger
+from runner import configure
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.frames.frames import (
+    BotSpeakingFrame,
+    Frame,
+    InputAudioRawFrame,
+    LLMMessagesFrame,
+    OutputAudioRawFrame,
+    TextFrame,
+    TTSAudioRawFrame,
+    UserStoppedSpeakingFrame,
+)
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.services.cartesia import CartesiaTTSService
+from pipecat.services.openai import OpenAILLMService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+class DebugProcessor(FrameProcessor):
+    def __init__(self, name, **kwargs):
+        self._name = name
+        super().__init__(**kwargs)
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        if not (
+            isinstance(frame, InputAudioRawFrame)
+            or isinstance(frame, BotSpeakingFrame)
+            or isinstance(frame, UserStoppedSpeakingFrame)
+            or isinstance(frame, TTSAudioRawFrame)
+            or isinstance(frame, TextFrame)
+        ):
+            logger.debug(f"--- DebugProcessor {self._name}: {frame} {direction}")
+        await self.push_frame(frame, direction)
+
+
+class DTMFProcessor(FrameProcessor):
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+        tones = model.String(
+            [
+                model.Tone("1"),
+                model.Tone("2"),
+                model.Tone("3"),
+                model.Tone("4"),
+                model.Pause(),
+                model.Tone("5"),
+                model.Tone("6"),
+                model.Tone("7"),
+                model.Tone("8"),
+                model.Tone("9"),
+            ]
+        )
+
+        tone_audio = generate(tones)
+
+        # Convert the generated audio to a numpy array (assuming the generate function returns an iterable of floats)
+        audio_data = np.array(list(tone_audio), dtype=np.float32)
+
+        # Create an AudioRawFrame with the audio data
+        audio_frame = OutputAudioRawFrame(audio_data, sample_rate=8000, num_channels=1)
+
+        await self.push_frame(audio_frame)
+
+
+async def main():
+    print(detect, generate, parse)
+    async with aiohttp.ClientSession() as session:
+        (room_url, token) = await configure(session)
+
+        transport = DailyTransport(
+            room_url,
+            token,
+            "Respond bot",
+            DailyParams(
+                audio_out_enabled=True,
+                transcription_enabled=True,
+                vad_enabled=True,
+                vad_analyzer=SileroVADAnalyzer(),
+            ),
+        )
+
+        tts = CartesiaTTSService(
+            api_key=os.getenv("CARTESIA_API_KEY"),
+            voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22",  # British Lady
+        )
+
+        llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o")
+
+        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.",
+            },
+        ]
+
+        context = OpenAILLMContext(messages)
+        context_aggregator = llm.create_context_aggregator(context)
+
+        dtmf = DTMFProcessor()
+        dp = DebugProcessor("dp")
+
+        pipeline = Pipeline(
+            [
+                transport.input(),
+                context_aggregator.user(),
+                dtmf,
+                dp,
+                # llm,
+                # tts,
+                transport.output(),
+                context_aggregator.assistant(),
+            ]
+        )
+
+        task = PipelineTask(
+            pipeline,
+            PipelineParams(enable_metrics=True, enable_usage_metrics=True),
+        )
+
+        @transport.event_handler("on_first_participant_joined")
+        async def on_first_participant_joined(transport, participant):
+            await transport.capture_participant_transcription(participant["id"])
+            # Kick off the conversation.
+            messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+            await task.queue_frames([LLMMessagesFrame(messages)])
+
+        runner = PipelineRunner()
+
+        await runner.run(task)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

pyproject.toml

@@ -29,6 +29,7 @@ dependencies = [
     "pydantic~=2.8.2",
     "pyloudnorm~=0.1.1",
     "resampy~=0.4.3",
+    "dtmf @ git+https://github.com/jamsea/dtmf.git@hush/updateDeps",
 ]
 
 [project.urls]

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

@@ -332,6 +332,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: