From ab8dcd6edead6686f07ba9d7a6476f07310f4fb2 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sat, 22 Nov 2025 07:07:21 -0500 Subject: [PATCH 1/4] Add SageMaker BiDi client --- CHANGELOG.md | 3 + pyproject.toml | 3 +- src/pipecat/services/aws/__init__.py | 1 + .../services/aws/sagemaker/__init__.py | 0 .../services/aws/sagemaker/bidi_client.py | 283 ++++++++++++++++++ uv.lock | 86 +++--- 6 files changed, 341 insertions(+), 35 deletions(-) create mode 100644 src/pipecat/services/aws/sagemaker/__init__.py create mode 100644 src/pipecat/services/aws/sagemaker/bidi_client.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 699a946d0..3a767a777 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `SageMakerBidiClient` to connect to SageMaker hosted BiDi compatible + services. + - Added support for `include_timestamps` and `enable_logging` in `ElevenLabsRealtimeSTTService`. When `include_timestamps` is enabled, timestamp data is included in the `TranscriptionFrame`'s `result` diff --git a/pyproject.toml b/pyproject.toml index 73da9083a..e4b0a380b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -50,7 +50,7 @@ anthropic = [ "anthropic~=0.49.0" ] assemblyai = [ "pipecat-ai[websockets-base]" ] asyncai = [ "pipecat-ai[websockets-base]" ] aws = [ "aioboto3~=15.0.0", "pipecat-ai[websockets-base]" ] -aws-nova-sonic = [ "aws_sdk_bedrock_runtime~=0.1.1; python_version>='3.12'" ] +aws-nova-sonic = [ "aws_sdk_bedrock_runtime~=0.2.0; python_version>='3.12'" ] azure = [ "azure-cognitiveservices-speech~=1.42.0"] cartesia = [ "cartesia~=2.0.3", "pipecat-ai[websockets-base]" ] cerebras = [] @@ -98,6 +98,7 @@ sentry = [ "sentry-sdk>=2.28.0,<3" ] local-smart-turn = [ "coremltools>=8.0", "transformers", "torch>=2.5.0,<3", "torchaudio>=2.5.0,<3" ] local-smart-turn-v3 = [ "transformers", "onnxruntime>=1.20.1,<2" ] remote-smart-turn = [] +sagemaker = ["aws_sdk_sagemaker_runtime_http2; python_version>='3.12'"] silero = [ "onnxruntime>=1.20.1,<2" ] simli = [ "simli-ai~=1.0.3"] soniox = [ "pipecat-ai[websockets-base]" ] diff --git a/src/pipecat/services/aws/__init__.py b/src/pipecat/services/aws/__init__.py index 3cdd4cc5a..6f6903f75 100644 --- a/src/pipecat/services/aws/__init__.py +++ b/src/pipecat/services/aws/__init__.py @@ -10,6 +10,7 @@ from pipecat.services import DeprecatedModuleProxy from .llm import * from .nova_sonic import * +from .sagemaker import * from .stt import * from .tts import * diff --git a/src/pipecat/services/aws/sagemaker/__init__.py b/src/pipecat/services/aws/sagemaker/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/services/aws/sagemaker/bidi_client.py b/src/pipecat/services/aws/sagemaker/bidi_client.py new file mode 100644 index 000000000..5e02af03d --- /dev/null +++ b/src/pipecat/services/aws/sagemaker/bidi_client.py @@ -0,0 +1,283 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""AWS SageMaker bidirectional streaming client. + +This module provides a client for streaming bidirectional communication with +SageMaker endpoints using the HTTP/2 protocol. Supports sending audio, text, +and JSON data to SageMaker model endpoints and receiving streaming responses. +""" + +import os +from typing import Optional + +from loguru import logger + +try: + from aws_sdk_sagemaker_runtime_http2.client import SageMakerRuntimeHTTP2Client + from aws_sdk_sagemaker_runtime_http2.config import Config, HTTPAuthSchemeResolver + from aws_sdk_sagemaker_runtime_http2.models import ( + InvokeEndpointWithBidirectionalStreamInput, + RequestPayloadPart, + RequestStreamEventPayloadPart, + ResponseStreamEvent, + ) + from smithy_aws_core.auth.sigv4 import SigV4AuthScheme + from smithy_aws_core.identity import EnvironmentCredentialsResolver + from smithy_core.aio.eventstream import DuplexEventStream +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error( + "In order to use SageMaker BiDi client, you need to `pip install pipecat-ai[sagemaker]`." + ) + raise Exception(f"Missing module: {e}") + + +class SageMakerBidiClient: + """Client for bidirectional streaming with AWS SageMaker endpoints. + + Handles low-level HTTP/2 bidirectional streaming protocol for communicating + with SageMaker model endpoints. Provides methods for sending various data + types (audio, text, JSON) and receiving streaming responses. + + This client uses AWS SigV4 authentication and supports credential resolution + from environment variables, AWS CLI configuration, and instance metadata. + + Example:: + + client = SageMakerBidiClient( + endpoint_name="my-deepgram-endpoint", + region="us-east-2", + model_invocation_path="v1/listen", + model_query_string="model=nova-3&language=en" + ) + await client.start_session() + await client.send_audio_chunk(audio_bytes) + response = await client.receive_response() + await client.close_session() + """ + + def __init__( + self, + endpoint_name: str, + region: str, + model_invocation_path: str = "", + model_query_string: str = "", + ): + """Initialize the SageMaker BiDi client. + + Args: + endpoint_name: Name of the SageMaker endpoint to connect to. + region: AWS region where the endpoint is deployed. + model_invocation_path: API path for the model invocation (e.g., "v1/listen"). + model_query_string: Query string parameters for the model (e.g., "model=nova-3"). + """ + self.endpoint_name = endpoint_name + self.region = region + self.model_invocation_path = model_invocation_path + self.model_query_string = model_query_string + self.bidi_endpoint = f"https://runtime.sagemaker.{region}.amazonaws.com:8443" + self._client: Optional[SageMakerRuntimeHTTP2Client] = None + self._stream: Optional[ + DuplexEventStream[RequestStreamEventPayloadPart, ResponseStreamEvent, any] + ] = None + self._output_stream = None + self._is_active = False + + def _initialize_client(self): + """Initialize the SageMaker Runtime HTTP2 client with AWS credentials. + + Creates and configures the SageMaker Runtime HTTP2 client with SigV4 + authentication. Attempts to resolve AWS credentials from environment + variables, AWS CLI configuration, or instance metadata. + """ + logger.debug(f"Initializing SageMaker BiDi client for region: {self.region}") + logger.debug(f"Using endpoint URI: {self.bidi_endpoint}") + + # Check for AWS credentials + has_env_creds = bool(os.getenv("AWS_ACCESS_KEY_ID") and os.getenv("AWS_SECRET_ACCESS_KEY")) + + if not has_env_creds: + logger.warning( + "AWS credentials not found in environment variables. " + "Attempting to use EnvironmentCredentialsResolver which will check " + "AWS CLI configuration and instance metadata." + ) + + config = Config( + endpoint_uri=self.bidi_endpoint, + region=self.region, + aws_credentials_identity_resolver=EnvironmentCredentialsResolver(), + auth_scheme_resolver=HTTPAuthSchemeResolver(), + auth_schemes={"aws.auth#sigv4": SigV4AuthScheme(service="sagemaker")}, + ) + self._client = SageMakerRuntimeHTTP2Client(config=config) + + async def start_session(self): + """Start a bidirectional streaming session with the SageMaker endpoint. + + Initializes the client if needed, creates the bidirectional stream, and + establishes the connection to the SageMaker endpoint. Must be called + before sending or receiving data. + + Returns: + The output stream for receiving responses. + + Raises: + RuntimeError: If client initialization or connection fails. + """ + if not self._client: + self._initialize_client() + + logger.debug(f"Starting BiDi session with endpoint: {self.endpoint_name}") + logger.debug(f"Model invocation path: {self.model_invocation_path}") + logger.debug(f"Model query string: {self.model_query_string}") + + # Create the bidirectional stream + stream_input = InvokeEndpointWithBidirectionalStreamInput( + endpoint_name=self.endpoint_name, + model_invocation_path=self.model_invocation_path, + model_query_string=self.model_query_string, + ) + + try: + self._stream = await self._client.invoke_endpoint_with_bidirectional_stream( + stream_input + ) + self._is_active = True + + # Get output stream + output = await self._stream.await_output() + self._output_stream = output[1] + + logger.debug("BiDi session started successfully") + return self._output_stream + + except Exception as e: + logger.error(f"Failed to start BiDi session: {e}") + self._is_active = False + raise RuntimeError(f"Failed to start SageMaker BiDi session: {e}") + + async def send_data(self, data_bytes: bytes, data_type: Optional[str] = None): + """Send a chunk of data to the stream. + + Generic method for sending any type of data to the SageMaker endpoint. + Use the convenience methods (send_audio_chunk, send_text, send_json) + for common data types. + + Args: + data_bytes: Raw bytes to send. + data_type: Optional data type header. Common values are "BINARY" for + audio/binary data and "UTF8" for text/JSON data. + + Raises: + RuntimeError: If session is not active or send fails. + """ + if not self._is_active or not self._stream: + raise RuntimeError("BiDi session not active") + + try: + payload = RequestPayloadPart(bytes_=data_bytes, data_type=data_type) + event = RequestStreamEventPayloadPart(value=payload) + await self._stream.input_stream.send(event) + except Exception as e: + logger.error(f"Failed to send data: {e}") + raise + + async def send_audio_chunk(self, audio_bytes: bytes): + """Send a chunk of audio data to the stream. + + Convenience method for sending audio data. Automatically sets the data + type to "BINARY". + + Args: + audio_bytes: Raw audio bytes to send (e.g., PCM audio data). + + Raises: + RuntimeError: If session is not active or send fails. + """ + await self.send_data(audio_bytes, data_type="BINARY") + + async def send_text(self, text: str): + """Send text data to the stream. + + Convenience method for sending text data. Automatically encodes the text + as UTF-8 and sets the data type to "UTF8". + + Args: + text: Text string to send. + + Raises: + RuntimeError: If session is not active or send fails. + """ + await self.send_data(text.encode("utf-8"), data_type="UTF8") + + async def send_json(self, data: dict): + """Send JSON data to the stream. + + Convenience method for sending JSON-encoded messages. Useful for control + messages like KeepAlive or CloseStream. Automatically serializes the + dictionary to JSON, encodes as UTF-8, and sets the data type to "UTF8". + + Args: + data: Dictionary to send as JSON (e.g., {"type": "KeepAlive"}). + + Raises: + RuntimeError: If session is not active or send fails. + """ + import json + + await self.send_data(json.dumps(data).encode("utf-8"), data_type="UTF8") + + async def receive_response(self) -> Optional[ResponseStreamEvent]: + """Receive a response from the stream. + + Blocks until a response is available from the SageMaker endpoint. Returns + None when the stream is closed. + + Returns: + The response event containing payload data, or None if stream is closed. + + Raises: + RuntimeError: If session is not active. + """ + if not self._is_active or not self._output_stream: + raise RuntimeError("BiDi session not active") + + try: + result = await self._output_stream.receive() + return result + except Exception as e: + logger.error(f"Failed to receive response: {e}") + raise + + async def close_session(self): + """Close the bidirectional streaming session. + + Gracefully closes the input stream and marks the session as inactive. + Safe to call multiple times. + """ + if not self._is_active: + return + + logger.debug("Closing BiDi session...") + self._is_active = False + + try: + if self._stream: + await self._stream.input_stream.close() + logger.debug("BiDi session closed successfully") + except Exception as e: + logger.warning(f"Error closing BiDi session: {e}") + + @property + def is_active(self) -> bool: + """Check if the session is currently active. + + Returns: + True if session is active, False otherwise. + """ + return self._is_active diff --git a/uv.lock b/uv.lock index 662150b5b..eb2fca39c 100644 --- a/uv.lock +++ b/uv.lock @@ -419,16 +419,30 @@ wheels = [ [[package]] name = "aws-sdk-bedrock-runtime" -version = "0.1.1" +version = "0.2.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "smithy-aws-core", extra = ["eventstream", "json"], marker = "python_full_version >= '3.12'" }, { name = "smithy-core", marker = "python_full_version >= '3.12'" }, { name = "smithy-http", extra = ["awscrt"], marker = "python_full_version >= '3.12'" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/1d/78/48574454b3cac869df67665e4a403ebfc3abfcfba2c2ff01ccfd67d55f8f/aws_sdk_bedrock_runtime-0.1.1.tar.gz", hash = "sha256:c896f99e675c3a1ab600633a07b785f3dc9fe8ab94f640b1f992b63da2dfc784", size = 82446, upload-time = "2025-10-21T20:25:25.845Z" } +sdist = { url = "https://files.pythonhosted.org/packages/db/94/f2451bb09c106e5690bbb88fc366637cdcec942b352ed9bb788804c877e0/aws_sdk_bedrock_runtime-0.2.0.tar.gz", hash = "sha256:8de52dd4492e74c73244d4b41a52304e1db368814a10e49dbbf8f4e8e412cd0e", size = 88156, upload-time = "2025-11-22T00:35:44.978Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/83/07/62c0b70223d178c138f29124ac2f7973a6ba803abc7735b6a01a85217f3d/aws_sdk_bedrock_runtime-0.1.1-py3-none-any.whl", hash = "sha256:c0336b377b2112cf88197d3d44302fbeb3efb1101989fa49ae55e78f49cfe345", size = 74954, upload-time = "2025-10-21T20:25:24.973Z" }, + { url = "https://files.pythonhosted.org/packages/eb/6b/07fbddd31dd6e38c967fe088b5e91a7cc3a2bc0f645f18b4e5d45bc03f1f/aws_sdk_bedrock_runtime-0.2.0-py3-none-any.whl", hash = "sha256:19594de50a52d199d73efca153c0a2328bd781827715a6e012d50b11085236cc", size = 79875, upload-time = "2025-11-22T00:35:44.092Z" }, +] + +[[package]] +name = "aws-sdk-sagemaker-runtime-http2" +version = "0.1.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "smithy-aws-core", extra = ["eventstream", "json"], marker = "python_full_version >= '3.12'" }, + { name = "smithy-core", marker = "python_full_version >= '3.12'" }, + { name = "smithy-http", extra = ["awscrt"], marker = "python_full_version >= '3.12'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/6e/ca/00f9c55887fc0f3fa345995dd871d40ff81473ab1591e56b4b4483d99d00/aws_sdk_sagemaker_runtime_http2-0.1.0.tar.gz", hash = "sha256:5077ec0c4440495b15004bbf04e27bc0bc137f1f8950d32195c6b45d7788d837", size = 20863, upload-time = "2025-11-22T00:20:56.358Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/9c/24/2e2f727c51c20f4625cd19364d9421dbd7c893fe2b53a46eb0caaf6263a2/aws_sdk_sagemaker_runtime_http2-0.1.0-py3-none-any.whl", hash = "sha256:1aebb728ba6c6d14e58e29ecf89b51f7abbe8786d34144f8a7d59a419e80bd2f", size = 21911, upload-time = "2025-11-22T00:20:55.054Z" }, ] [[package]] @@ -4569,6 +4583,9 @@ runner = [ { name = "python-dotenv" }, { name = "uvicorn" }, ] +sagemaker = [ + { name = "aws-sdk-sagemaker-runtime-http2", marker = "python_full_version >= '3.12'" }, +] sarvam = [ { name = "sarvamai" }, { name = "websockets" }, @@ -4654,7 +4671,8 @@ requires-dist = [ { name = "aiortc", marker = "extra == 'webrtc'", specifier = ">=1.13.0,<2" }, { name = "anthropic", marker = "extra == 'anthropic'", specifier = "~=0.49.0" }, { name = "audioop-lts", marker = "python_full_version >= '3.13'", specifier = "~=0.2.1" }, - { name = "aws-sdk-bedrock-runtime", marker = "python_full_version >= '3.12' and extra == 'aws-nova-sonic'", specifier = "~=0.1.1" }, + { name = "aws-sdk-bedrock-runtime", marker = "python_full_version >= '3.12' and extra == 'aws-nova-sonic'", specifier = "~=0.2.0" }, + { name = "aws-sdk-sagemaker-runtime-http2", marker = "python_full_version >= '3.12' and extra == 'sagemaker'" }, { name = "azure-cognitiveservices-speech", marker = "extra == 'azure'", specifier = "~=1.42.0" }, { name = "cartesia", marker = "extra == 'cartesia'", specifier = "~=2.0.3" }, { name = "coremltools", marker = "extra == 'local-smart-turn'", specifier = ">=8.0" }, @@ -4745,7 +4763,7 @@ requires-dist = [ { name = "wait-for2", marker = "python_full_version < '3.12'", specifier = ">=0.4.1" }, { name = "websockets", marker = "extra == 'websockets-base'", specifier = ">=13.1,<16.0" }, ] -provides-extras = ["aic", "anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "deepseek", "daily", "deepgram", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "hume", "inworld", "krisp", "koala", "langchain", "livekit", "lmnt", "local", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "nim", "neuphonic", "noisereduce", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "rime", "riva", "runner", "sambanova", "sarvam", "sentry", "local-smart-turn", "local-smart-turn-v3", "remote-smart-turn", "silero", "simli", "soniox", "soundfile", "speechmatics", "strands", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "websockets-base", "whisper"] +provides-extras = ["aic", "anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "deepseek", "daily", "deepgram", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "hume", "inworld", "krisp", "koala", "langchain", "livekit", "lmnt", "local", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "nim", "neuphonic", "noisereduce", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "rime", "riva", "runner", "sambanova", "sarvam", "sentry", "local-smart-turn", "local-smart-turn-v3", "remote-smart-turn", "sagemaker", "silero", "simli", "soniox", "soundfile", "speechmatics", "strands", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "websockets-base", "whisper"] [package.metadata.requires-dev] dev = [ @@ -6522,16 +6540,16 @@ wheels = [ [[package]] name = "smithy-aws-core" -version = "0.1.1" +version = "0.2.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "aws-sdk-signers", marker = "python_full_version >= '3.12'" }, { name = "smithy-core", marker = "python_full_version >= '3.12'" }, { name = "smithy-http", marker = "python_full_version >= '3.12'" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/56/d3/f847e0fd36b95aa36ce3a4c9ce1a08e16b2aa9a56b71714045c9c924e282/smithy_aws_core-0.1.1.tar.gz", hash = "sha256:78dfd7040fc2bc72b6af293096642fc9a7bfd2db28ddbdf7c4110535eab9d662", size = 11196, upload-time = "2025-10-21T20:21:18.648Z" } +sdist = { url = "https://files.pythonhosted.org/packages/c1/c8/5970c869527972b23a1733de3993d54283c84a2340e84acdd48a11aa0ff4/smithy_aws_core-0.2.0.tar.gz", hash = "sha256:dfa1ecd311d6f0a16f48c86d793085e2a0a33a46de897d129dd1f142a43bf7f6", size = 11344, upload-time = "2025-11-21T18:33:01.928Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/d0/04/87cb06f0f6d664b5cffdef6d4042dd52c11c138436084d30ffdaa3543031/smithy_aws_core-0.1.1-py3-none-any.whl", hash = "sha256:0d1634f276c2999dc2a04fafef63b9d28309de50d939d1d49df952773a7063c4", size = 18963, upload-time = "2025-10-21T20:21:17.692Z" }, + { url = "https://files.pythonhosted.org/packages/88/25/739c0005a6cb4effbc2d37fe23590660b508fe314200f4acf94410a8f315/smithy_aws_core-0.2.0-py3-none-any.whl", hash = "sha256:d112082ef77758e1977f8694cf690ac35c76570c12a6590fccd5da085a3ce507", size = 18966, upload-time = "2025-11-21T18:33:00.812Z" }, ] [package.optional-dependencies] @@ -6544,35 +6562,35 @@ json = [ [[package]] name = "smithy-aws-event-stream" -version = "0.1.0" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "smithy-core", marker = "python_full_version >= '3.12'" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/49/26/8ff24194efed60b2df18f610ea05fa2a4c6546858b80a0a51335a4943b9b/smithy_aws_event_stream-0.1.0.tar.gz", hash = "sha256:6634691a3bf5d4801a2c29f0761db2dc4771f3ae43cdee50c10d4b4bb2f86475", size = 12216, upload-time = "2025-09-29T19:37:14.659Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/90/c4/2b63d31af58fc359577e5515bf730348a235f2f2fa10e17af8640495c81c/smithy_aws_event_stream-0.1.0-py3-none-any.whl", hash = "sha256:17a7300a85cb90df4c6c23f895ca6343361fa419203c3cf80019edd7d3b5f036", size = 15581, upload-time = "2025-09-29T19:37:13.589Z" }, -] - -[[package]] -name = "smithy-core" -version = "0.1.0" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/b9/8d/16028d03456071d21de7591f1e1e6a1cc81b2389e53ef8663dbf59caf9cd/smithy_core-0.1.0.tar.gz", hash = "sha256:b159b8905264e1e4c613eab9f74cec0b2f5b8119c42fbadddb4da0a8ed8050e9", size = 48415, upload-time = "2025-09-29T19:37:16.873Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/ca/5b/563cb2beadcfa40597b0c3ff3f2d42e21f065b14782c4ba9cb41a44b745f/smithy_core-0.1.0-py3-none-any.whl", hash = "sha256:cb44e9355fb89e89f2c6ba6a1d59c5db4f2f7282c72d31d9307b6202d66cd0fa", size = 62895, upload-time = "2025-09-29T19:37:15.917Z" }, -] - -[[package]] -name = "smithy-http" version = "0.2.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "smithy-core", marker = "python_full_version >= '3.12'" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/3c/1c/44e99a7dfb8c39bf0c3d998accdf4573a7a3488863b90f21af260cec2d45/smithy_http-0.2.0.tar.gz", hash = "sha256:2382562fa9af326be455f14b18615f16ffe9db756e51b2a4ca0d23e1b881cff8", size = 26729, upload-time = "2025-10-21T20:21:06.146Z" } +sdist = { url = "https://files.pythonhosted.org/packages/31/90/78283c21484f8cf9862982e53bc2769b784910735fb5fb2400a17bfb5fdd/smithy_aws_event_stream-0.2.0.tar.gz", hash = "sha256:99700a11346e7ab1435ff2e53e6f6d60a1e857f2b2ee1941d40b54270adf3323", size = 12278, upload-time = "2025-11-21T18:33:03.79Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/d4/e2/d475fad81ac74ec0e145cb6d72afe5ecde4e2358bd632c2fd5d3f4bc87dc/smithy_http-0.2.0-py3-none-any.whl", hash = "sha256:49ee2402d7737798d70f99f491fbfb2a5767283ae562e21b6f86e3fd14f3e3e0", size = 37328, upload-time = "2025-10-21T20:21:05.362Z" }, + { url = "https://files.pythonhosted.org/packages/ca/f5/08b997eee81b55150496ce565f0e03c72d0c80e5b218170bdeae7c46a5a4/smithy_aws_event_stream-0.2.0-py3-none-any.whl", hash = "sha256:679a0c7d944e67d3a55d287541b3ca1e61f9d6a62e13401367dcc034e75aa55d", size = 15567, upload-time = "2025-11-21T18:33:02.711Z" }, +] + +[[package]] +name = "smithy-core" +version = "0.2.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/c7/f6/140f0be9331dd7cd8fa012b3ca4735df39a1a81d03eea89728f997249116/smithy_core-0.2.0.tar.gz", hash = "sha256:05c3e3309df5dcb9cf53e241bd57a96510e4575186443ea157db9dbb59b6c85e", size = 50334, upload-time = "2025-11-21T18:33:05.697Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/16/e3/d0defa2acf50b91625fe15e3ddb0c8e41ff64363a1f4cd9b8f19ae2ec0c6/smithy_core-0.2.0-py3-none-any.whl", hash = "sha256:db4620da3497abb60f79ac1d8a738d3eac46d7e820bfb50c777c36e932915239", size = 64777, upload-time = "2025-11-21T18:33:04.591Z" }, +] + +[[package]] +name = "smithy-http" +version = "0.3.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "smithy-core", marker = "python_full_version >= '3.12'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/1c/c7/4d8be56e897f99f3b6ffcdf52ba00a468febc939fca85b90f1c122450830/smithy_http-0.3.0.tar.gz", hash = "sha256:55dcc3af315eee6863d2f3f58ada1d9cb4bcc3a57faac10a1b21d4a93722f520", size = 28674, upload-time = "2025-11-21T18:33:07.387Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/2d/e5/59ae79ecdc9a935ad10512c581b3054ebb1afd90498ecc8afaf141dbc22b/smithy_http-0.3.0-py3-none-any.whl", hash = "sha256:972924304febd77c7134a7cffab83ce3b48423ff966dcc1f257e2c0d58fa9b18", size = 40520, upload-time = "2025-11-21T18:33:06.312Z" }, ] [package.optional-dependencies] @@ -6582,15 +6600,15 @@ awscrt = [ [[package]] name = "smithy-json" -version = "0.1.0" +version = "0.2.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "ijson", marker = "python_full_version >= '3.12'" }, { name = "smithy-core", marker = "python_full_version >= '3.12'" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/e2/5b/0ecb10007475e1b8faca3bbff1be2fc6edb3ea12ffc5e939e2249be95325/smithy_json-0.1.0.tar.gz", hash = "sha256:84fb48e445b87d850c240d837702c16b259ea53bad76b655ac1bbd8094d48912", size = 7086, upload-time = "2025-09-29T19:37:20.432Z" } +sdist = { url = "https://files.pythonhosted.org/packages/89/cf/e319a2a299b27bc0addf46ee3d4b9c25ec0817e3a0507b2b7a33eddc19f1/smithy_json-0.2.0.tar.gz", hash = "sha256:0946066fdda15d6a579dfdd4b61a547ab915eb057bd176fc2bc17d01dc789499", size = 7157, upload-time = "2025-11-21T18:33:08.968Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/62/95/e11c04e56aae12b62e38c49000004a1dc598a64dc207018c08448efde322/smithy_json-0.1.0-py3-none-any.whl", hash = "sha256:80ff64734dccdabf1ba6a2908555b97e60f62c07c3a27df48e421ee058413cb9", size = 9914, upload-time = "2025-09-29T19:37:19.459Z" }, + { url = "https://files.pythonhosted.org/packages/2e/b1/33012ac5b2e5940a00b6e1ccc313330e6f8692152a151f72a398cd6be0e0/smithy_json-0.2.0-py3-none-any.whl", hash = "sha256:5018a4e61731afa3094a02d737d4f956dbf270c271410c089045a17d86fc3b3b", size = 9911, upload-time = "2025-11-21T18:33:08.267Z" }, ] [[package]] From 782b257bbb2b6cf9f77fff6e6a8ab243ff8bd235 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sat, 22 Nov 2025 07:09:54 -0500 Subject: [PATCH 2/4] Add DeepgramSageMakerSTTService --- CHANGELOG.md | 3 + src/pipecat/services/deepgram/__init__.py | 1 + .../services/deepgram/stt_sagemaker.py | 447 ++++++++++++++++++ 3 files changed, 451 insertions(+) create mode 100644 src/pipecat/services/deepgram/stt_sagemaker.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 3a767a777..93e0ac4dc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `DeepgramSageMakerSTTService` which connects to a SageMaker hosted + Deepgram STT model. + - Added `SageMakerBidiClient` to connect to SageMaker hosted BiDi compatible services. diff --git a/src/pipecat/services/deepgram/__init__.py b/src/pipecat/services/deepgram/__init__.py index c23ebbec9..227ac5c64 100644 --- a/src/pipecat/services/deepgram/__init__.py +++ b/src/pipecat/services/deepgram/__init__.py @@ -10,6 +10,7 @@ from pipecat.services import DeprecatedModuleProxy from .flux import * from .stt import * +from .stt_sagemaker import * from .tts import * sys.modules[__name__] = DeprecatedModuleProxy(globals(), "deepgram", "deepgram.[stt,tts]") diff --git a/src/pipecat/services/deepgram/stt_sagemaker.py b/src/pipecat/services/deepgram/stt_sagemaker.py new file mode 100644 index 000000000..6d28feefa --- /dev/null +++ b/src/pipecat/services/deepgram/stt_sagemaker.py @@ -0,0 +1,447 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Deepgram speech-to-text service for AWS SageMaker. + +This module provides a Pipecat STT service that connects to Deepgram models +deployed on AWS SageMaker endpoints. Uses HTTP/2 bidirectional streaming for +low-latency real-time transcription with support for interim results, multiple +languages, and various Deepgram features. +""" + +import asyncio +import json +from typing import AsyncGenerator, Optional + +from loguru import logger + +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + ErrorFrame, + Frame, + InterimTranscriptionFrame, + StartFrame, + TranscriptionFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, +) +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.aws.sagemaker.bidi_client import SageMakerBidiClient +from pipecat.services.stt_service import STTService +from pipecat.transcriptions.language import Language +from pipecat.utils.time import time_now_iso8601 +from pipecat.utils.tracing.service_decorators import traced_stt + +try: + from deepgram import LiveOptions +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error( + "In order to use DeepgramSageMakerSTTService, you need to `pip install pipecat-ai[deepgram,sagemaker]`." + ) + raise Exception(f"Missing module: {e}") + + +class DeepgramSageMakerSTTService(STTService): + """Deepgram speech-to-text service for AWS SageMaker. + + Provides real-time speech recognition using Deepgram models deployed on + AWS SageMaker endpoints. Uses HTTP/2 bidirectional streaming for low-latency + transcription with support for interim results, speaker diarization, and + multiple languages. + + Requirements: + + - AWS credentials configured (via environment variables, AWS CLI, or instance metadata) + - A deployed SageMaker endpoint with Deepgram model: https://developers.deepgram.com/docs/deploy-amazon-sagemaker + - Deepgram SDK for LiveOptions configuration + + Example:: + + stt = DeepgramSageMakerSTTService( + endpoint_name="my-deepgram-endpoint", + region="us-east-2", + live_options=LiveOptions( + model="nova-3", + language="en", + interim_results=True, + punctuate=True, + ), + ) + """ + + def __init__( + self, + *, + endpoint_name: str, + region: str, + sample_rate: Optional[int] = None, + live_options: Optional[LiveOptions] = None, + **kwargs, + ): + """Initialize the Deepgram SageMaker STT service. + + Args: + endpoint_name: Name of the SageMaker endpoint with Deepgram model + deployed (e.g., "my-deepgram-nova-3-endpoint"). + region: AWS region where the endpoint is deployed (e.g., "us-east-2"). + sample_rate: Audio sample rate in Hz. If None, uses value from + live_options or defaults to the value from StartFrame. + live_options: Deepgram LiveOptions for detailed configuration. If None, + uses sensible defaults (nova-3 model, English, interim results enabled). + **kwargs: Additional arguments passed to the parent STTService. + """ + sample_rate = sample_rate or (live_options.sample_rate if live_options else None) + super().__init__(sample_rate=sample_rate, **kwargs) + + self._endpoint_name = endpoint_name + self._region = region + + # Create default options similar to DeepgramSTTService + default_options = LiveOptions( + encoding="linear16", + language=Language.EN, + model="nova-3", + channels=1, + interim_results=True, + punctuate=True, + ) + + # Merge with provided options + merged_options = default_options.to_dict() + if live_options: + default_model = default_options.model + merged_options.update(live_options.to_dict()) + # Handle the "None" string bug from deepgram-sdk + if "model" in merged_options and merged_options["model"] == "None": + merged_options["model"] = default_model + + # Convert Language enum to string if needed + if "language" in merged_options and isinstance(merged_options["language"], Language): + merged_options["language"] = merged_options["language"].value + + self.set_model_name(merged_options["model"]) + self._settings = merged_options + + self._client: Optional[SageMakerBidiClient] = None + self._response_task: Optional[asyncio.Task] = None + self._keepalive_task: Optional[asyncio.Task] = None + + def can_generate_metrics(self) -> bool: + """Check if this service can generate processing metrics. + + Returns: + True, as Deepgram SageMaker service supports metrics generation. + """ + return True + + async def set_model(self, model: str): + """Set the Deepgram model and reconnect. + + Disconnects from the current session, updates the model setting, and + establishes a new connection with the updated model. + + Args: + model: The Deepgram model name to use (e.g., "nova-3"). + """ + await super().set_model(model) + logger.info(f"Switching STT model to: [{model}]") + self._settings["model"] = model + await self._disconnect() + await self._connect() + + async def set_language(self, language: Language): + """Set the recognition language and reconnect. + + Disconnects from the current session, updates the language setting, and + establishes a new connection with the updated language. + + Args: + language: The language to use for speech recognition (e.g., Language.EN, + Language.ES). + """ + logger.info(f"Switching STT language to: [{language}]") + self._settings["language"] = language + await self._disconnect() + await self._connect() + + async def start(self, frame: StartFrame): + """Start the Deepgram SageMaker STT service. + + Args: + frame: The start frame containing initialization parameters. + """ + await super().start(frame) + self._settings["sample_rate"] = self.sample_rate + await self._connect() + + async def stop(self, frame: EndFrame): + """Stop the Deepgram SageMaker STT service. + + Args: + frame: The end frame. + """ + await super().stop(frame) + await self._disconnect() + + async def cancel(self, frame: CancelFrame): + """Cancel the Deepgram SageMaker STT service. + + Args: + frame: The cancel frame. + """ + await super().cancel(frame) + await self._disconnect() + + async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: + """Send audio data to Deepgram for transcription. + + Args: + audio: Raw audio bytes to transcribe. + + Yields: + Frame: None (transcription results come via BiDi stream callbacks). + """ + if self._client and self._client.is_active: + try: + await self._client.send_audio_chunk(audio) + except Exception as e: + logger.error(f"Error sending audio to SageMaker: {e}") + await self.push_error(ErrorFrame(error=f"SageMaker STT error: {e}")) + yield None + + async def _connect(self): + """Connect to the SageMaker endpoint and start the BiDi session. + + Builds the Deepgram query string from settings, creates the BiDi client, + starts the streaming session, and launches background tasks for processing + responses and sending KeepAlive messages. + """ + logger.debug("Connecting to Deepgram on SageMaker...") + + # Update sample rate in settings + self._settings["sample_rate"] = self.sample_rate + + # Build query string from settings, converting booleans to strings + query_params = {} + for key, value in self._settings.items(): + if value is not None: + # Convert boolean values to lowercase strings for Deepgram API + if isinstance(value, bool): + query_params[key] = str(value).lower() + else: + query_params[key] = str(value) + + query_string = "&".join(f"{k}={v}" for k, v in query_params.items()) + + # Create BiDi client + self._client = SageMakerBidiClient( + endpoint_name=self._endpoint_name, + region=self._region, + model_invocation_path="v1/listen", + model_query_string=query_string, + ) + + try: + # Start the session + await self._client.start_session() + + # Start processing responses in the background + self._response_task = self.create_task(self._process_responses()) + + # Start keepalive task to maintain connection + self._keepalive_task = self.create_task(self._send_keepalive()) + + logger.debug("Connected to Deepgram on SageMaker") + await self._call_event_handler("on_connected") + + except Exception as e: + logger.error(f"Failed to connect to SageMaker: {e}") + await self.push_error(ErrorFrame(error=f"SageMaker connection error: {e}")) + await self._call_event_handler("on_connection_error", str(e)) + + async def _disconnect(self): + """Disconnect from the SageMaker endpoint. + + Sends a CloseStream message to Deepgram, cancels background tasks + (KeepAlive and response processing), and closes the BiDi session. + Safe to call multiple times. + """ + if self._client and self._client.is_active: + logger.debug("Disconnecting from Deepgram on SageMaker...") + + # Send CloseStream message to Deepgram + try: + await self._client.send_json({"type": "CloseStream"}) + except Exception as e: + logger.warning(f"Failed to send CloseStream message: {e}") + + # Cancel keepalive task + if self._keepalive_task and not self._keepalive_task.done(): + await self.cancel_task(self._keepalive_task) + + # Cancel response processing task + if self._response_task and not self._response_task.done(): + await self.cancel_task(self._response_task) + + # Close the BiDi session + await self._client.close_session() + + logger.debug("Disconnected from Deepgram on SageMaker") + await self._call_event_handler("on_disconnected") + + async def _send_keepalive(self): + """Send periodic KeepAlive messages to maintain the connection. + + Sends a KeepAlive JSON message to Deepgram every 5 seconds while the + connection is active. This prevents the connection from timing out during + periods of silence. + """ + while self._client and self._client.is_active: + await asyncio.sleep(5) + if self._client and self._client.is_active: + try: + await self._client.send_json({"type": "KeepAlive"}) + except Exception as e: + logger.warning(f"Failed to send KeepAlive: {e}") + + async def _process_responses(self): + """Process streaming responses from Deepgram on SageMaker. + + Continuously receives responses from the BiDi stream, decodes the payload, + parses JSON responses from Deepgram, and processes transcription results. + Runs as a background task until the connection is closed or cancelled. + """ + try: + while self._client and self._client.is_active: + result = await self._client.receive_response() + + if result is None: + break + + # Check if this is a PayloadPart with bytes + if hasattr(result, "value") and hasattr(result.value, "bytes_"): + if result.value.bytes_: + response_data = result.value.bytes_.decode("utf-8") + + try: + # Parse JSON response from Deepgram + parsed = json.loads(response_data) + + # Extract and process transcript if available + if "channel" in parsed: + await self._handle_transcript_response(parsed) + + except json.JSONDecodeError: + logger.warning(f"Non-JSON response: {response_data}") + + except asyncio.CancelledError: + logger.debug("Response processor cancelled") + except Exception as e: + logger.error(f"Error processing responses: {e}", exc_info=True) + await self.push_error(ErrorFrame(error=f"SageMaker response error: {e}")) + finally: + logger.debug("Response processor stopped") + + async def _handle_transcript_response(self, parsed: dict): + """Handle a transcript response from Deepgram. + + Extracts the transcript text, determines if it's final or interim, extracts + language information, and pushes the appropriate frame (TranscriptionFrame + or InterimTranscriptionFrame) downstream. + + Args: + parsed: The parsed JSON response from Deepgram containing channel, + alternatives, transcript, and metadata. + """ + alternatives = parsed.get("channel", {}).get("alternatives", []) + if not alternatives or not alternatives[0].get("transcript"): + return + + transcript = alternatives[0]["transcript"] + if not transcript.strip(): + return + + # Stop TTFB metrics on first transcript + await self.stop_ttfb_metrics() + + is_final = parsed.get("is_final", False) + speech_final = parsed.get("speech_final", False) + + # Extract language if available + language = None + if alternatives[0].get("languages"): + language = alternatives[0]["languages"][0] + language = Language(language) + + if is_final and speech_final: + # Final transcription + await self.push_frame( + TranscriptionFrame( + transcript, + self._user_id, + time_now_iso8601(), + language, + result=parsed, + ) + ) + await self._handle_transcription(transcript, is_final, language) + await self.stop_processing_metrics() + else: + # Interim transcription + await self.push_frame( + InterimTranscriptionFrame( + transcript, + self._user_id, + time_now_iso8601(), + language, + result=parsed, + ) + ) + + @traced_stt + async def _handle_transcription( + self, transcript: str, is_final: bool, language: Optional[Language] = None + ): + """Handle a transcription result with tracing. + + This method is decorated with @traced_stt for observability and tracing + integration. The actual transcription processing is handled by the parent + class and observers. + + Args: + transcript: The transcribed text. + is_final: Whether this is a final transcription result. + language: The detected language of the transcription, if available. + """ + pass + + async def start_metrics(self): + """Start TTFB and processing metrics collection.""" + await self.start_ttfb_metrics() + await self.start_processing_metrics() + + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process frames with Deepgram SageMaker-specific handling. + + Args: + frame: The frame to process. + direction: The direction of frame processing. + """ + await super().process_frame(frame, direction) + + # Start metrics when user starts speaking (if VAD is not provided by Deepgram) + if isinstance(frame, UserStartedSpeakingFrame): + await self.start_metrics() + elif isinstance(frame, UserStoppedSpeakingFrame): + # Send finalize message to Deepgram when user stops speaking + # This tells Deepgram to flush any remaining audio and return final results + if self._client and self._client.is_active: + try: + await self._client.send_json({"type": "Finalize"}) + except Exception as e: + logger.warning(f"Error sending Finalize message: {e}") From 0ece8b5894470997ca17cd4066c4907849afd50d Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sat, 22 Nov 2025 07:10:47 -0500 Subject: [PATCH 3/4] Add 07c Deepgram SageMaker example --- CHANGELOG.md | 3 +- env.example | 1 + .../07c-interruptible-deepgram-sagemaker.py | 137 ++++++++++++++++++ 3 files changed, 140 insertions(+), 1 deletion(-) create mode 100644 examples/foundational/07c-interruptible-deepgram-sagemaker.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 93e0ac4dc..aaf7ec85a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - Added `DeepgramSageMakerSTTService` which connects to a SageMaker hosted - Deepgram STT model. + Deepgram STT model. Added `07c-interruptible-deepgram-sagemaker.py` + foundational example. - Added `SageMakerBidiClient` to connect to SageMaker hosted BiDi compatible services. diff --git a/env.example b/env.example index 2865772ea..33c699259 100644 --- a/env.example +++ b/env.example @@ -44,6 +44,7 @@ DAILY_SAMPLE_ROOM_URL=https://... # Deepgram DEEPGRAM_API_KEY=... +SAGEMAKER_ENDPOINT_NAME=... # DeepSeek DEEPSEEK_API_KEY=... diff --git a/examples/foundational/07c-interruptible-deepgram-sagemaker.py b/examples/foundational/07c-interruptible-deepgram-sagemaker.py new file mode 100644 index 000000000..db230a8ba --- /dev/null +++ b/examples/foundational/07c-interruptible-deepgram-sagemaker.py @@ -0,0 +1,137 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +import os + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams +from pipecat.audio.turn.smart_turn.local_smart_turn_v3 import LocalSmartTurnAnalyzerV3 +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.audio.vad.vad_analyzer import VADParams +from pipecat.frames.frames import LLMRunFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.aws.llm import AWSBedrockLLMService +from pipecat.services.deepgram.stt_sagemaker import DeepgramSageMakerSTTService +from pipecat.services.deepgram.tts import DeepgramTTSService +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.daily.transport import DailyParams +from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams + +load_dotenv(override=True) + + +# We store functions so objects (e.g. SileroVADAnalyzer) don't get +# instantiated. The function will be called when the desired transport gets +# selected. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), + turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()), + ), + "twilio": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), + turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()), + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), + turn_analyzer=LocalSmartTurnAnalyzerV3(params=SmartTurnParams()), + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + # Initialize Deepgram SageMaker STT Service + # This requires: + # - AWS credentials configured (via environment variables or AWS CLI) + # - A deployed SageMaker endpoint with Deepgram model + stt = DeepgramSageMakerSTTService( + endpoint_name=os.getenv("SAGEMAKER_ENDPOINT_NAME"), + region=os.getenv("AWS_REGION"), + ) + + tts = DeepgramTTSService(api_key=os.getenv("DEEPGRAM_API_KEY"), voice="aura-2-andromeda-en") + + llm = AWSBedrockLLMService( + aws_region=os.getenv("AWS_REGION"), + model="us.amazon.nova-pro-v1:0", + params=AWSBedrockLLMService.InputParams(temperature=0.8), + ) + + 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 spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. Respond to what the user said in a creative and helpful way.", + }, + ] + + context = LLMContext(messages) + context_aggregator = LLMContextAggregatorPair(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, # STT + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + messages.append({"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([LLMRunFrame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() + + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + + await runner.run(task) + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud.""" + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +if __name__ == "__main__": + from pipecat.runner.run import main + + main() From a357ff0205aea7da2987a3a11573037b55f96834 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sat, 22 Nov 2025 07:20:37 -0500 Subject: [PATCH 4/4] Alphabetize the project.optional-dependencies --- pyproject.toml | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index e4b0a380b..cf83e53ee 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,9 +54,9 @@ aws-nova-sonic = [ "aws_sdk_bedrock_runtime~=0.2.0; python_version>='3.12'" ] azure = [ "azure-cognitiveservices-speech~=1.42.0"] cartesia = [ "cartesia~=2.0.3", "pipecat-ai[websockets-base]" ] cerebras = [] -deepseek = [] daily = [ "daily-python~=0.22.0" ] deepgram = [ "deepgram-sdk~=4.7.0" ] +deepseek = [] elevenlabs = [ "pipecat-ai[websockets-base]" ] fal = [ "fal-client~=0.5.9" ] fireworks = [] @@ -69,19 +69,21 @@ gstreamer = [ "pygobject~=3.50.0" ] heygen = [ "livekit>=1.0.13", "pipecat-ai[websockets-base]" ] hume = [ "hume>=0.11.2" ] inworld = [] -krisp = [ "pipecat-ai-krisp~=0.4.0" ] koala = [ "pvkoala~=2.0.3" ] +krisp = [ "pipecat-ai-krisp~=0.4.0" ] langchain = [ "langchain~=0.3.20", "langchain-community~=0.3.20", "langchain-openai~=0.3.9" ] livekit = [ "livekit~=1.0.13", "livekit-api~=1.0.5", "tenacity>=8.2.3,<10.0.0" ] lmnt = [ "pipecat-ai[websockets-base]" ] local = [ "pyaudio~=0.2.14" ] +local-smart-turn = [ "coremltools>=8.0", "transformers", "torch>=2.5.0,<3", "torchaudio>=2.5.0,<3" ] +local-smart-turn-v3 = [ "transformers", "onnxruntime>=1.20.1,<2" ] mcp = [ "mcp[cli]>=1.11.0,<2" ] mem0 = [ "mem0ai~=0.1.94" ] mistral = [] mlx-whisper = [ "mlx-whisper~=0.4.2" ] moondream = [ "accelerate~=1.10.0", "einops~=0.8.0", "pyvips[binary]~=3.0.0", "timm~=1.0.13", "transformers>=4.48.0" ] -nim = [] neuphonic = [ "pipecat-ai[websockets-base]" ] +nim = [] noisereduce = [ "noisereduce~=3.0.3" ] openai = [ "pipecat-ai[websockets-base]" ] openpipe = [ "openpipe>=4.50.0,<6" ] @@ -89,16 +91,14 @@ openrouter = [] perplexity = [] playht = [ "pipecat-ai[websockets-base]" ] qwen = [] +remote-smart-turn = [] rime = [ "pipecat-ai[websockets-base]" ] riva = [ "nvidia-riva-client~=2.21.1" ] runner = [ "python-dotenv>=1.0.0,<2.0.0", "uvicorn>=0.32.0,<1.0.0", "fastapi>=0.115.6,<0.122.0", "pipecat-ai-small-webrtc-prebuilt>=1.0.0"] +sagemaker = ["aws_sdk_sagemaker_runtime_http2; python_version>='3.12'"] sambanova = [] sarvam = [ "sarvamai==0.1.21", "pipecat-ai[websockets-base]" ] sentry = [ "sentry-sdk>=2.28.0,<3" ] -local-smart-turn = [ "coremltools>=8.0", "transformers", "torch>=2.5.0,<3", "torchaudio>=2.5.0,<3" ] -local-smart-turn-v3 = [ "transformers", "onnxruntime>=1.20.1,<2" ] -remote-smart-turn = [] -sagemaker = ["aws_sdk_sagemaker_runtime_http2; python_version>='3.12'"] silero = [ "onnxruntime>=1.20.1,<2" ] simli = [ "simli-ai~=1.0.3"] soniox = [ "pipecat-ai[websockets-base]" ]