diff --git a/CHANGELOG.md b/CHANGELOG.md index 62d4bfc81..33cc37438 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/examples/foundational/07c-interruptible-deepgram.py b/examples/foundational/07c-interruptible-deepgram.py index 818c8bc93..25be49a1c 100644 --- a/examples/foundational/07c-interruptible-deepgram.py +++ b/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 diff --git a/examples/foundational/13b-deepgram-transcription.py b/examples/foundational/13b-deepgram-transcription.py new file mode 100644 index 000000000..ceefbaa86 --- /dev/null +++ b/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)) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index a46f41f35..82175b797 100644 --- a/src/pipecat/services/ai_services.py +++ b/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): diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index ce418916b..1c3b890ee 100644 --- a/src/pipecat/services/deepgram.py +++ b/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.services.ai_services import TTSService +from pipecat.frames.frames import ( + 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)) diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 1162708cc..9496cd3b4 100644 --- a/src/pipecat/transports/base_input.py +++ b/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) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 4c19a0b55..b484c8fce 100644 --- a/src/pipecat/transports/base_output.py +++ b/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,7 +108,8 @@ class BaseOutputTransport(FrameProcessor): if self._params.camera_out_enabled: await self._camera_out_thread - await self._sink_thread + if self._sink_thread: + await self._sink_thread async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -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)