From 0b8f5ca6a55e7ffac6f61d9129db7faa7dbd2e4b Mon Sep 17 00:00:00 2001 From: Tico Ballagas Date: Sat, 8 Feb 2025 18:46:59 -0800 Subject: [PATCH] Initial implementation of AWS Transcribe TTS --- .../foundational/07m-interruptible-polly.py | 13 +- src/pipecat/services/aws/tts.py | 624 +++++++++++++++++- 2 files changed, 626 insertions(+), 11 deletions(-) diff --git a/examples/foundational/07m-interruptible-polly.py b/examples/foundational/07m-interruptible-polly.py index 286fe5128..63349360e 100644 --- a/examples/foundational/07m-interruptible-polly.py +++ b/examples/foundational/07m-interruptible-polly.py @@ -15,9 +15,9 @@ 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.services.aws.tts import PollyTTSService -from pipecat.services.deepgram.stt import DeepgramSTTService -from pipecat.services.openai.llm import OpenAILLMService +from pipecat.services.aws import PollyTTSService, TranscribeSTTService +from pipecat.services.openai import OpenAILLMService +from pipecat.transcriptions.language import Language from pipecat.transports.base_transport import TransportParams from pipecat.transports.network.small_webrtc import SmallWebRTCTransport from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection @@ -37,14 +37,11 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac ), ) - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + stt = TranscribeSTTService() tts = PollyTTSService( - api_key=os.getenv("AWS_SECRET_ACCESS_KEY"), - aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"), - region=os.getenv("AWS_REGION"), voice_id="Amy", - params=PollyTTSService.InputParams(engine="neural", language="en-GB", rate="1.05"), + params=PollyTTSService.InputParams(engine="standard", language=Language.EN_GB, rate="1.05"), ) llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) diff --git a/src/pipecat/services/aws/tts.py b/src/pipecat/services/aws/tts.py index db6e168ab..e90ea9220 100644 --- a/src/pipecat/services/aws/tts.py +++ b/src/pipecat/services/aws/tts.py @@ -5,7 +5,21 @@ # import asyncio -from typing import AsyncGenerator, Optional +from typing import AsyncGenerator, Optional, Dict +import os +import datetime +import time +from urllib.parse import urlencode +import json +import struct +from io import BytesIO +import urllib.parse +import hashlib +import hmac +import random +import string +import binascii +import numpy as np from loguru import logger from pydantic import BaseModel @@ -17,17 +31,27 @@ from pipecat.frames.frames import ( TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, + TranscriptionFrame, + InterimTranscriptionFrame, + StartFrame, + EndFrame, + CancelFrame, ) -from pipecat.services.tts_service import TTSService +from pipecat.services.ai_services import TTSService, STTService from pipecat.transcriptions.language import Language +from pipecat.utils.time import time_now_iso8601 try: import boto3 from botocore.exceptions import BotoCoreError, ClientError + import websockets + from botocore.auth import SigV4Auth + from botocore.awsrequest import AWSRequest + from botocore.credentials import Credentials except ModuleNotFoundError as e: logger.error(f"Exception: {e}") logger.error( - "In order to use Deepgram, you need to `pip install pipecat-ai[aws]`. Also, set `AWS_SECRET_ACCESS_KEY`, `AWS_ACCESS_KEY_ID`, and `AWS_REGION` environment variable." + "In order to use AWS services, you need to `pip install pipecat-ai[aws]`. Also, set `AWS_SECRET_ACCESS_KEY`, `AWS_ACCESS_KEY_ID`, and `AWS_REGION` environment variable." ) raise Exception(f"Missing module: {e}") @@ -151,6 +175,24 @@ class PollyTTSService(TTSService): self.set_voice(voice_id) + # Get credentials from environment variables if not provided + self._credentials = { + "aws_access_key_id": aws_access_key_id or os.getenv("AWS_ACCESS_KEY_ID"), + "aws_secret_access_key": api_key or os.getenv("AWS_SECRET_ACCESS_KEY"), + "aws_session_token": aws_session_token or os.getenv("AWS_SESSION_TOKEN"), + "region": region or os.getenv("AWS_REGION", "us-east-1"), + } + + # Validate that we have the required credentials + if ( + not self._credentials["aws_access_key_id"] + or not self._credentials["aws_secret_access_key"] + ): + raise ValueError( + "AWS credentials not found. Please provide them either through constructor parameters " + "or set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables." + ) + def can_generate_metrics(self) -> bool: return True @@ -248,3 +290,579 @@ class PollyTTSService(TTSService): finally: yield TTSStoppedFrame() + + +class AWSTTSService(PollyTTSService): + def __init__(self, **kwargs): + super().__init__(**kwargs) + + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "'AWSTTSService' is deprecated, use 'PollyTTSService' instead.", DeprecationWarning + ) + + +def get_presigned_url( + *, + region: str, + credentials: Dict[str, Optional[str]], + language_code: str, + media_encoding: str = "pcm", + sample_rate: int = 16000, + number_of_channels: int = 1, + enable_partial_results_stabilization: bool = True, + partial_results_stability: str = "high", + vocabulary_name: Optional[str] = None, + vocabulary_filter_name: Optional[str] = None, + show_speaker_label: bool = False, + enable_channel_identification: bool = False, +) -> str: + """Create a presigned URL for AWS Transcribe streaming.""" + access_key = credentials.get("access_key") + secret_key = credentials.get("secret_key") + session_token = credentials.get("session_token") + + if not access_key or not secret_key: + raise ValueError("AWS credentials are required") + + # Initialize the URL generator + url_generator = AWSTranscribePresignedURL( + access_key=access_key, secret_key=secret_key, session_token=session_token, region=region + ) + + # Get the presigned URL + return url_generator.get_request_url( + sample_rate=sample_rate, + language_code=language_code, + media_encoding=media_encoding, + vocabulary_name=vocabulary_name, + vocabulary_filter_name=vocabulary_filter_name, + show_speaker_label=show_speaker_label, + enable_channel_identification=enable_channel_identification, + number_of_channels=number_of_channels, + enable_partial_results_stabilization=enable_partial_results_stabilization, + partial_results_stability=partial_results_stability, + ) + + +class AWSTranscribePresignedURL: + def __init__( + self, access_key: str, secret_key: str, session_token: str, region: str = "us-east-1" + ): + self.access_key = access_key + self.secret_key = secret_key + self.session_token = session_token + self.method = "GET" + self.service = "transcribe" + self.region = region + self.endpoint = "" + self.host = "" + self.amz_date = "" + self.datestamp = "" + self.canonical_uri = "/stream-transcription-websocket" + self.canonical_headers = "" + self.signed_headers = "host" + self.algorithm = "AWS4-HMAC-SHA256" + self.credential_scope = "" + self.canonical_querystring = "" + self.payload_hash = "" + self.canonical_request = "" + self.string_to_sign = "" + self.signature = "" + self.request_url = "" + + def get_request_url( + self, + sample_rate: int, + language_code: str = "", + media_encoding: str = "pcm", + vocabulary_name: str = "", + vocabulary_filter_name: str = "", + show_speaker_label: bool = False, + enable_channel_identification: bool = False, + number_of_channels: int = 1, + enable_partial_results_stabilization: bool = False, + partial_results_stability: str = "", + ) -> str: + self.endpoint = f"wss://transcribestreaming.{self.region}.amazonaws.com:8443" + self.host = f"transcribestreaming.{self.region}.amazonaws.com:8443" + + now = datetime.datetime.utcnow() + self.amz_date = now.strftime("%Y%m%dT%H%M%SZ") + self.datestamp = now.strftime("%Y%m%d") + self.canonical_headers = f"host:{self.host}\n" + self.credential_scope = f"{self.datestamp}%2F{self.region}%2F{self.service}%2Faws4_request" + + # Create canonical querystring + self.canonical_querystring = "X-Amz-Algorithm=" + self.algorithm + self.canonical_querystring += ( + "&X-Amz-Credential=" + self.access_key + "%2F" + self.credential_scope + ) + self.canonical_querystring += "&X-Amz-Date=" + self.amz_date + self.canonical_querystring += "&X-Amz-Expires=300" + if self.session_token: + self.canonical_querystring += "&X-Amz-Security-Token=" + urllib.parse.quote( + self.session_token, safe="" + ) + self.canonical_querystring += "&X-Amz-SignedHeaders=" + self.signed_headers + + if enable_channel_identification: + self.canonical_querystring += "&enable-channel-identification=true" + if enable_partial_results_stabilization: + self.canonical_querystring += "&enable-partial-results-stabilization=true" + if language_code: + self.canonical_querystring += "&language-code=" + language_code + if media_encoding: + self.canonical_querystring += "&media-encoding=" + media_encoding + if number_of_channels > 1: + self.canonical_querystring += "&number-of-channels=" + str(number_of_channels) + if partial_results_stability: + self.canonical_querystring += "&partial-results-stability=" + partial_results_stability + if sample_rate: + self.canonical_querystring += "&sample-rate=" + str(sample_rate) + if show_speaker_label: + self.canonical_querystring += "&show-speaker-label=true" + if vocabulary_filter_name: + self.canonical_querystring += "&vocabulary-filter-name=" + vocabulary_filter_name + if vocabulary_name: + self.canonical_querystring += "&vocabulary-name=" + vocabulary_name + + # Create payload hash + self.payload_hash = hashlib.sha256("".encode("utf-8")).hexdigest() + + # Create canonical request + self.canonical_request = f"{self.method}\n{self.canonical_uri}\n{self.canonical_querystring}\n{self.canonical_headers}\n{self.signed_headers}\n{self.payload_hash}" + + # Create string to sign + credential_scope = f"{self.datestamp}/{self.region}/{self.service}/aws4_request" + string_to_sign = ( + f"{self.algorithm}\n{self.amz_date}\n{credential_scope}\n" + + hashlib.sha256(self.canonical_request.encode("utf-8")).hexdigest() + ) + + # Calculate signature + k_date = hmac.new( + f"AWS4{self.secret_key}".encode("utf-8"), self.datestamp.encode("utf-8"), hashlib.sha256 + ).digest() + k_region = hmac.new(k_date, self.region.encode("utf-8"), hashlib.sha256).digest() + k_service = hmac.new(k_region, self.service.encode("utf-8"), hashlib.sha256).digest() + k_signing = hmac.new(k_service, b"aws4_request", hashlib.sha256).digest() + self.signature = hmac.new( + k_signing, string_to_sign.encode("utf-8"), hashlib.sha256 + ).hexdigest() + + # Add signature to query string + self.canonical_querystring += "&X-Amz-Signature=" + self.signature + + # Create request URL + self.request_url = self.endpoint + self.canonical_uri + "?" + self.canonical_querystring + return self.request_url + + +def get_headers(header_name: str, header_value: str) -> bytearray: + """Build a header following AWS event stream format.""" + name = header_name.encode("utf-8") + name_byte_length = bytes([len(name)]) + value_type = bytes([7]) # 7 represents a string + value = header_value.encode("utf-8") + value_byte_length = struct.pack(">H", len(value)) + + # Construct the header + header_list = bytearray() + header_list.extend(name_byte_length) + header_list.extend(name) + header_list.extend(value_type) + header_list.extend(value_byte_length) + header_list.extend(value) + return header_list + + +def build_event_message(payload: bytes) -> bytes: + """ + Build an event message for AWS Transcribe streaming. + Matches AWS sample: https://github.com/aws-samples/amazon-transcribe-streaming-python-websockets/blob/main/eventstream.py + """ + # Build headers + content_type_header = get_headers(":content-type", "application/octet-stream") + event_type_header = get_headers(":event-type", "AudioEvent") + message_type_header = get_headers(":message-type", "event") + + headers = bytearray() + headers.extend(content_type_header) + headers.extend(event_type_header) + headers.extend(message_type_header) + + # Calculate total byte length and headers byte length + # 16 accounts for 8 byte prelude, 2x 4 byte CRCs + total_byte_length = struct.pack(">I", len(headers) + len(payload) + 16) + headers_byte_length = struct.pack(">I", len(headers)) + + # Build the prelude + prelude = bytearray([0] * 8) + prelude[:4] = total_byte_length + prelude[4:] = headers_byte_length + + # Calculate checksum for prelude + prelude_crc = struct.pack(">I", binascii.crc32(prelude) & 0xFFFFFFFF) + + # Construct the message + message_as_list = bytearray() + message_as_list.extend(prelude) + message_as_list.extend(prelude_crc) + message_as_list.extend(headers) + message_as_list.extend(payload) + + # Calculate checksum for message + message = bytes(message_as_list) + message_crc = struct.pack(">I", binascii.crc32(message) & 0xFFFFFFFF) + + # Add message checksum + message_as_list.extend(message_crc) + + return bytes(message_as_list) + + +def decode_event(message): + # Extract the prelude, headers, payload and CRC + prelude = message[:8] + total_length, headers_length = struct.unpack(">II", prelude) + prelude_crc = struct.unpack(">I", message[8:12])[0] + headers = message[12 : 12 + headers_length] + payload = message[12 + headers_length : -4] + message_crc = struct.unpack(">I", message[-4:])[0] + + # Check the CRCs + assert prelude_crc == binascii.crc32(prelude) & 0xFFFFFFFF, "Prelude CRC check failed" + assert message_crc == binascii.crc32(message[:-4]) & 0xFFFFFFFF, "Message CRC check failed" + + # Parse the headers + headers_dict = {} + while headers: + name_len = headers[0] + name = headers[1 : 1 + name_len].decode("utf-8") + value_type = headers[1 + name_len] + value_len = struct.unpack(">H", headers[2 + name_len : 4 + name_len])[0] + value = headers[4 + name_len : 4 + name_len + value_len].decode("utf-8") + headers_dict[name] = value + headers = headers[4 + name_len + value_len :] + + return headers_dict, json.loads(payload) + + +class TranscribeSTTService(STTService): + def __init__( + self, + *, + api_key: Optional[str] = None, + aws_access_key_id: Optional[str] = None, + aws_session_token: Optional[str] = None, + region: Optional[str] = "us-east-1", + sample_rate: int = 16000, + language: Language = Language.EN, + **kwargs, + ): + super().__init__(**kwargs) + + self._settings = { + "sample_rate": sample_rate, + "language": language, + "media_encoding": "linear16", # AWS expects raw PCM + "number_of_channels": 1, + "show_speaker_label": False, + "enable_channel_identification": False, + } + + # Validate sample rate - AWS Transcribe only supports 8000 Hz or 16000 Hz + if sample_rate not in [8000, 16000]: + logger.warning( + f"AWS Transcribe only supports 8000 Hz or 16000 Hz sample rates. Converting from {sample_rate} Hz to 16000 Hz." + ) + self._settings["sample_rate"] = 16000 + + self._credentials = { + "aws_access_key_id": aws_access_key_id or os.getenv("AWS_ACCESS_KEY_ID"), + "aws_secret_access_key": api_key or os.getenv("AWS_SECRET_ACCESS_KEY"), + "aws_session_token": aws_session_token or os.getenv("AWS_SESSION_TOKEN"), + "region": region or os.getenv("AWS_REGION", "us-east-1"), + } + + self._ws_client = None + self._connection_lock = asyncio.Lock() + self._connecting = False + self._receive_task = None + + def get_service_encoding(self, encoding: str) -> str: + """Convert internal encoding format to AWS Transcribe format.""" + encoding_map = { + "linear16": "pcm", # AWS expects "pcm" for 16-bit linear PCM + } + return encoding_map.get(encoding, encoding) + + async def start(self, frame: StartFrame): + """Initialize the connection when the service starts.""" + await super().start(frame) + logger.info("Starting AWS Transcribe service...") + retry_count = 0 + max_retries = 3 + + while retry_count < max_retries: + try: + await self._connect() + if self._ws_client and self._ws_client.open: + logger.info("Successfully established WebSocket connection") + return + logger.warning("WebSocket connection not established after connect") + except Exception as e: + logger.error(f"Failed to connect (attempt {retry_count + 1}/{max_retries}): {e}") + retry_count += 1 + if retry_count < max_retries: + await asyncio.sleep(1) # Wait before retrying + + raise RuntimeError("Failed to establish WebSocket connection after multiple attempts") + + async def run_stt(self, frame: Frame) -> AsyncGenerator[Frame, None]: + """Process audio data and send to AWS Transcribe""" + try: + # Skip if no speech detected + if hasattr(frame, "is_speech") and not frame.is_speech: + logger.debug("Skipping non-speech frame") + return + + # Ensure WebSocket is connected + if not self._ws_client or not self._ws_client.open: + logger.info("WebSocket not connected, attempting to reconnect...") + try: + await self._connect() + except Exception as e: + logger.error(f"Failed to reconnect: {e}") + yield ErrorFrame("Failed to reconnect to AWS Transcribe", fatal=False) + return + + # Get the audio data - if frame is bytes, use directly, otherwise get audio attribute + audio_data = frame if isinstance(frame, bytes) else frame.audio + + # Format the audio data according to AWS event stream format + event_message = build_event_message(audio_data) + # logger.debug(f"Sending audio chunk of size {len(audio_data)} bytes") + + # Send the formatted event message + try: + await self._ws_client.send(event_message) + # Start metrics after first chunk sent + await self.start_processing_metrics() + await self.start_ttfb_metrics() + except websockets.exceptions.ConnectionClosed as e: + logger.warning(f"Connection closed while sending: {e}") + await self._disconnect() + # Don't yield error here - we'll retry on next frame + except Exception as e: + logger.error(f"Error sending audio: {e}") + yield ErrorFrame(f"AWS Transcribe error: {str(e)}", fatal=False) + await self._disconnect() + + except Exception as e: + logger.error(f"Error in run_stt: {e}") + yield ErrorFrame(f"AWS Transcribe error: {str(e)}", fatal=False) + await self._disconnect() + + async def _connect(self): + """Connect to AWS Transcribe with connection state management.""" + if ( + self._ws_client + and self._ws_client.open + and self._receive_task + and not self._receive_task.done() + ): + logger.debug("Already connected") + return + + async with self._connection_lock: + if self._connecting: + logger.debug("Connection already in progress") + return + + try: + self._connecting = True + logger.debug("Starting connection process...") + + if self._ws_client: + await self._disconnect() + + language_code = self.language_to_service_language( + Language(self._settings["language"]) + ) + if not language_code: + raise ValueError(f"Unsupported language: {self._settings['language']}") + + # Generate random websocket key + websocket_key = "".join( + random.choices( + string.ascii_uppercase + string.ascii_lowercase + string.digits, k=20 + ) + ) + + # Add required headers + extra_headers = { + "Origin": "https://localhost", + "Sec-WebSocket-Key": websocket_key, + "Sec-WebSocket-Version": "13", + "Connection": "keep-alive", + } + + # Get presigned URL + presigned_url = get_presigned_url( + region=self._credentials["region"], + credentials={ + "access_key": self._credentials["aws_access_key_id"], + "secret_key": self._credentials["aws_secret_access_key"], + "session_token": self._credentials["aws_session_token"], + }, + language_code=language_code, + media_encoding=self.get_service_encoding( + self._settings["media_encoding"] + ), # Convert to AWS format + sample_rate=self._settings["sample_rate"], + number_of_channels=self._settings["number_of_channels"], + enable_partial_results_stabilization=True, + partial_results_stability="high", + show_speaker_label=self._settings["show_speaker_label"], + enable_channel_identification=self._settings["enable_channel_identification"], + ) + + logger.debug(f"Connecting to WebSocket with URL: {presigned_url[:100]}...") + + # Connect with the required headers and settings + self._ws_client = await websockets.connect( + presigned_url, + extra_headers=extra_headers, + subprotocols=["mqtt"], + ping_interval=None, + ping_timeout=None, + compression=None, + ) + logger.debug("WebSocket connected, starting receive task...") + + # Start receive task + self._receive_task = asyncio.create_task(self._receive_loop()) + + logger.info("Successfully connected to AWS Transcribe") + + except Exception as e: + logger.error(f"Failed to connect to AWS Transcribe: {e}") + await self._disconnect() + raise + + finally: + self._connecting = False + + async def _disconnect(self): + """Disconnect from AWS Transcribe.""" + if self._receive_task: + self._receive_task.cancel() + try: + await self._receive_task + except asyncio.CancelledError: + pass + self._receive_task = None + + if self._ws_client: + try: + if self._ws_client.open: + # Send end-stream message + end_stream = {"message-type": "event", "event": "end"} + await self._ws_client.send(json.dumps(end_stream)) + await self._ws_client.close() + except Exception as e: + logger.warning(f"Error closing WebSocket connection: {e}") + finally: + self._ws_client = None + + def language_to_service_language(self, language: Language) -> str | None: + """Convert internal language enum to AWS Transcribe language code.""" + language_map = { + Language.EN: "en-US", + Language.ES: "es-US", + Language.FR: "fr-FR", + Language.DE: "de-DE", + Language.IT: "it-IT", + Language.PT: "pt-BR", + Language.JA: "ja-JP", + Language.KO: "ko-KR", + Language.ZH: "zh-CN", + } + return language_map.get(language) + + async def _receive_loop(self): + """Background task to receive and process messages from AWS Transcribe.""" + try: + logger.debug("Receive loop started") + while True: + if not self._ws_client or not self._ws_client.open: + logger.warning("WebSocket closed in receive loop") + break + + try: + response = await self._ws_client.recv() + headers, payload = decode_event(response) + + # logger.debug(f"Received message type: {headers.get(':message-type')}") + + if headers.get(":message-type") == "event": + # Process transcription results + results = payload.get("Transcript", {}).get("Results", []) + if results: + result = results[0] + alternatives = result.get("Alternatives", []) + if alternatives: + transcript = alternatives[0].get("Transcript", "") + is_final = not result.get("IsPartial", True) + + if transcript: + await self.stop_ttfb_metrics() + if is_final: + await self.push_frame( + TranscriptionFrame( + transcript, + "", + time_now_iso8601(), + self._settings["language"], + ) + ) + await self.stop_processing_metrics() + else: + await self.push_frame( + InterimTranscriptionFrame( + transcript, + "", + time_now_iso8601(), + self._settings["language"], + ) + ) + elif headers.get(":message-type") == "exception": + error_msg = payload.get("Message", "Unknown error") + logger.error(f"Exception from AWS: {error_msg}") + await self.push_frame( + ErrorFrame(f"AWS Transcribe error: {error_msg}", fatal=False) + ) + else: + logger.debug(f"Other message type received: {headers}") + logger.debug(f"Payload: {payload}") + + except websockets.exceptions.ConnectionClosed as e: + logger.error( + f"WebSocket connection closed in receive loop with code {e.code}: {e.reason}" + ) + break + except Exception as e: + logger.error(f"Error in receive loop: {e}") + break + + except asyncio.CancelledError: + logger.debug("Receive loop cancelled") + except Exception as e: + logger.error(f"Unexpected error in receive loop: {e}") + finally: + logger.debug("Receive loop ended")