Merge pull request #224 from pipecat-ai/aleix/deepgram-stt-simple

deepgram stt simple

CHANGELOG.md

@@ -5,6 +5,16 @@ 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]
+
+### Changed
+
+- Added `DeepgramSTTService`. This service has an ongoing websocket
+  connection. To handle this, it subclasses `AIService` instead of
+  `STTService`. The output of this service will be pushed from the same task,
+  except system frames like `StartFrame`, `CancelFrame` or
+  `StartInterruptionFrame`.
+
 ## [0.0.29] - 2024-06-07
 
 ### Added

examples/foundational/07c-interruptible-deepgram.py

@@ -15,7 +15,7 @@ from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantResponseAggregator, LLMUserResponseAggregator)
-from pipecat.services.deepgram import DeepgramTTSService
+from pipecat.services.deepgram import DeepgramSTTService, DeepgramTTSService
 from pipecat.services.openai import OpenAILLMService
 from pipecat.transports.services.daily import DailyParams, DailyTransport
 from pipecat.vad.silero import SileroVADAnalyzer
@@ -39,12 +39,14 @@ async def main(room_url: str, token):
             "Respond bot",
             DailyParams(
                 audio_out_enabled=True,
-                transcription_enabled=True,
                 vad_enabled=True,
-                vad_analyzer=SileroVADAnalyzer()
+                vad_analyzer=SileroVADAnalyzer(),
+                vad_audio_passthrough=True
             )
         )
 
+        stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
         tts = DeepgramTTSService(
             aiohttp_session=session,
             api_key=os.getenv("DEEPGRAM_API_KEY"),
@@ -67,6 +69,7 @@ async def main(room_url: str, token):
 
         pipeline = Pipeline([
             transport.input(),   # Transport user input
+            stt,                 # STT
             tma_in,              # User responses
             llm,                 # LLM
             tts,                 # TTS

examples/foundational/13b-deepgram-transcription.py

@@ -0,0 +1,58 @@
+#
+# Copyright (c) 2024, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import os
+import sys
+
+from pipecat.frames.frames import Frame, TranscriptionFrame
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineTask
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.services.deepgram import DeepgramSTTService
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+
+from runner import configure
+
+from loguru import logger
+
+from dotenv import load_dotenv
+load_dotenv(override=True)
+
+logger.remove(0)
+logger.add(sys.stderr, level="DEBUG")
+
+
+class TranscriptionLogger(FrameProcessor):
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, TranscriptionFrame):
+            print(f"Transcription: {frame.text}")
+
+
+async def main(room_url: str):
+    transport = DailyTransport(room_url, None, "Transcription bot",
+                               DailyParams(audio_in_enabled=True))
+
+    stt = DeepgramSTTService(os.getenv("DEEPGRAM_API_KEY"))
+
+    tl = TranscriptionLogger()
+
+    pipeline = Pipeline([transport.input(), stt, tl])
+
+    task = PipelineTask(pipeline)
+
+    runner = PipelineRunner()
+
+    await runner.run(task)
+
+
+if __name__ == "__main__":
+    (url, token) = configure()
+    asyncio.run(main(url))

src/pipecat/services/ai_services.py

@@ -16,6 +16,7 @@ from pipecat.frames.frames import (
     EndFrame,
     ErrorFrame,
     Frame,
+    StartFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
     TextFrame,
@@ -30,6 +31,25 @@ class AIService(FrameProcessor):
     def __init__(self):
         super().__init__()
 
+    async def start(self, frame: StartFrame):
+        pass
+
+    async def stop(self, frame: EndFrame):
+        pass
+
+    async def cancel(self, frame: CancelFrame):
+        pass
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, StartFrame):
+            await self.start(frame)
+        elif isinstance(frame, CancelFrame):
+            await self.cancel(frame)
+        elif isinstance(frame, EndFrame):
+            await self.stop(frame)
+
     async def process_generator(self, generator: AsyncGenerator[Frame, None]):
         async for f in generator:
             if isinstance(f, ErrorFrame):

src/pipecat/services/deepgram.py

@@ -5,11 +5,29 @@
 #
 
 import aiohttp
+import asyncio
+import time
 
 from typing import AsyncGenerator
 
-from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame
+from pipecat.frames.frames import (
-from pipecat.services.ai_services import TTSService
+    AudioRawFrame,
+    CancelFrame,
+    EndFrame,
+    ErrorFrame,
+    Frame,
+    InterimTranscriptionFrame,
+    StartFrame,
+    SystemFrame,
+    TranscriptionFrame)
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.ai_services import AIService, TTSService
+
+from deepgram import (
+    DeepgramClient,
+    LiveTranscriptionEvents,
+    LiveOptions,
+)
 
 from loguru import logger
 
@@ -55,3 +73,76 @@ class DeepgramTTSService(TTSService):
                     yield frame
         except Exception as e:
             logger.error(f"Deepgram exception: {e}")
+
+
+class DeepgramSTTService(AIService):
+    def __init__(self,
+                 api_key: str,
+                 live_options: LiveOptions = LiveOptions(
+                     encoding="linear16",
+                     language="en-US",
+                     model="nova-2-conversationalai",
+                     sample_rate=16000,
+                     channels=1,
+                     interim_results=True,
+                     smart_format=True,
+                 ),
+                 **kwargs):
+        super().__init__(**kwargs)
+
+        self._live_options = live_options
+
+        self._client = DeepgramClient(api_key)
+        self._connection = self._client.listen.asynclive.v("1")
+        self._connection.on(LiveTranscriptionEvents.Transcript, self._on_message)
+
+        self._create_push_task()
+
+    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)
+        elif isinstance(frame, AudioRawFrame):
+            await self._connection.send(frame.audio)
+        else:
+            await self._push_queue.put((frame, direction))
+
+    async def start(self, frame: StartFrame):
+        if await self._connection.start(self._live_options):
+            logger.debug(f"{self}: Connected to Deepgram")
+        else:
+            logger.error(f"{self}: Unable to connect to Deepgram")
+
+    async def stop(self, frame: EndFrame):
+        await self._connection.finish()
+        await self._push_queue.put((frame, FrameDirection.DOWNSTREAM))
+        await self._push_frame_task
+
+    async def cancel(self, frame: CancelFrame):
+        await self._connection.finish()
+        self._push_frame_task.cancel()
+
+    def _create_push_task(self):
+        self._push_frame_task = self.get_event_loop().create_task(self._push_frame_task_handler())
+        self._push_queue = asyncio.Queue()
+
+    async def _push_frame_task_handler(self):
+        running = True
+        while running:
+            try:
+                (frame, direction) = await self._push_queue.get()
+                await self.push_frame(frame, direction)
+                running = not isinstance(frame, EndFrame)
+            except asyncio.CancelledError:
+                break
+
+    async def _on_message(self, *args, **kwargs):
+        result = kwargs["result"]
+        is_final = result.is_final
+        transcript = result.channel.alternatives[0].transcript
+        if len(transcript) > 0:
+            if is_final:
+                await self._push_queue.put((TranscriptionFrame(transcript, "", int(time.time_ns() / 1000000)), FrameDirection.DOWNSTREAM))
+            else:
+                await self._push_queue.put((InterimTranscriptionFrame(transcript, "", int(time.time_ns() / 1000000)), FrameDirection.DOWNSTREAM))

src/pipecat/transports/base_input.py

@@ -83,9 +83,9 @@ class BaseInputTransport(FrameProcessor):
         await super().process_frame(frame, direction)
 
         if isinstance(frame, CancelFrame):
+            await self.stop()
             # We don't queue a CancelFrame since we want to stop ASAP.
             await self.push_frame(frame, direction)
-            await self.stop()
         elif isinstance(frame, StartFrame):
             await self.start(frame)
             await self._internal_push_frame(frame, direction)

src/pipecat/transports/base_output.py

@@ -28,6 +28,7 @@ from pipecat.frames.frames import (
     ImageRawFrame,
     StartInterruptionFrame,
     StopInterruptionFrame,
+    SystemFrame,
     TransportMessageFrame)
 from pipecat.transports.base_transport import TransportParams
 
@@ -53,6 +54,7 @@ class BaseOutputTransport(FrameProcessor):
         if self._params.camera_out_enabled:
             self._camera_out_queue = queue.Queue()
         self._sink_queue = queue.Queue()
+        self._sink_thread = None
 
         self._stopped_event = asyncio.Event()
         self._is_interrupted = threading.Event()
@@ -106,6 +108,7 @@ class BaseOutputTransport(FrameProcessor):
         if self._params.camera_out_enabled:
             await self._camera_out_thread
 
+        if self._sink_thread:
             await self._sink_thread
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -121,11 +124,13 @@ class BaseOutputTransport(FrameProcessor):
             self._sink_queue.put_nowait(frame)
         # EndFrame is managed in the queue handler.
         elif isinstance(frame, CancelFrame):
-            await self.push_frame(frame, direction)
             await self.stop()
-        elif isinstance(frame, StartInterruptionFrame) or isinstance(frame, StopInterruptionFrame):
             await self.push_frame(frame, direction)
+        elif isinstance(frame, StartInterruptionFrame) or isinstance(frame, StopInterruptionFrame):
             await self._handle_interruptions(frame)
+            await self.push_frame(frame, direction)
+        elif isinstance(frame, SystemFrame):
+            await self.push_frame(frame, direction)
         else:
             self._sink_queue.put_nowait(frame)