Introducing bidirectional streaming for real-time inference on Amazon SageMaker AI

In 2025, generative AI has developed from textual content era to multi-modal use instances starting from audio transcription and translation to voice brokers that require real-time knowledge streaming. At the moment’s purposes demand one thing extra: steady, real-time dialogue between customers and fashions—the power for knowledge to circulate each methods, concurrently, over a single persistent connection. Think about a speech to textual content use-case, the place you’ll need to stream the audio stream as enter and obtain the transcripted textual content as a steady stream. Such use-cases would require bi-directional streaming functionality.

We’re introducing bidirectional streaming for Amazon SageMaker AI Inference, which transforms inference from a transactional trade right into a steady dialog. Speech works greatest with real-time AI when conversations circulate naturally with out interruptions. With bidirectional streaming, speech to textual content turns into rapid. The mannequin listens and transcribes on the identical time, so phrases seem the second they’re spoken. Image a caller describing a problem to a assist line. As they communicate, the dwell transcript seems in entrance of the decision heart agent, giving the agent immediate context and letting them reply with out ready for the caller to complete. This sort of steady trade makes voice experiences really feel fluid, responsive and human.

This put up reveals you the best way to construct and deploy a container with bidirectional streaming functionality to a SageMaker AI endpoint. We additionally display how one can deliver your individual container or use our accomplice Deepgram’s pre-built fashions and containers on SageMaker AI to allow bi-directional streaming characteristic for real-time inference.

Bidirectional streaming: Deep dive

With bidirectional streaming, knowledge flows each methods directly by way of a single, persistent connection.

Within the conventional strategy to inference requests, the consumer sends an entire query and waits, whereas the mannequin processes the request and returns an entire reply earlier than the consumer can ship the subsequent query.

Shopper: [sends complete question] → waits...
Mannequin: ...processes... [returns complete answer] 
Shopper: [sends next question] → waits... 
Mannequin: ...processes... [returns complete answer]

In bidirectional streaming, the consumer’s speech begins flowing whereas the mannequin concurrently begins processing and transcribing the reply instantly.

Shopper: [question starts flowing with enough context] →                                    
Mannequin: ← [answer starts flowing immediately]
Shopper: → [continues/adjusts question]
                                    ↓
Mannequin: ← [adapts answer in real-time]

Customers see outcomes as quickly because the mannequin begins producing them. Sustaining one persistent connection replaces a whole bunch of short-lived connections. This reduces overhead on networking infrastructure, TLS handshakes, and connection administration. Fashions can preserve context throughout a steady stream, enabling multi-turn interactions with out resending dialog historical past every time.
 

SageMaker AI Inference bidirectional streaming functionality

SageMaker AI Inference combines HTTP/2 and WebSocket protocols for real-time, two-way communication between shoppers and fashions. Once you invoke a SageMaker AI Inference endpoint with bidirectional streaming, your request travels by way of the three-layer infrastructure in SageMaker AI:

As soon as the connection is established, knowledge flows freely in each instructions:

The WebSocket connection between the Sidecar and mannequin container stays open all through your session, with built-in well being monitoring. To keep up connection well being, SageMaker AI sends WebSocket ping frames each 60 seconds to confirm the connection is energetic, and your mannequin container responds with pong frames to verify it’s wholesome. If 5 consecutive pings go unanswered, the connection is gracefully closed.

Constructing your individual container for implementing bidirectional streaming

If you want to make use of open supply or your individual fashions, you may customise your container to assist bidirectional streaming. Your container should implement the WebSocket protocol to deal with incoming knowledge frames and ship response frames again to SageMaker AI.

To get began, allow us to construct an instance bi-directional streaming software with bring-your-own container use case. With this instance we’ll:

Conditions

Construct docker container with bi-directional streaming functionality

First, clone our demo repository and arrange your atmosphere as outlined within the README.md. The steps beneath will create a easy demo docker picture and push it Amazon ECR repository in your account.

# The applying makes use of atmosphere variables for AWS authentication. Set these earlier than operating the applying:
# export AWS_ACCESS_KEY_ID="your-access-key"
# export AWS_SECRET_ACCESS_KEY="your-secret-key"
# export AWS_DEFAULT_REGION="us-west-2"

container_name="sagemaker-bidirectional-streaming"
container_tag="newest"

cd container

account=$(aws sts get-caller-identity --query Account --output textual content)

# Get the area outlined within the present configuration (default to us-west-2 if none outlined)
area=$(aws configure get area)
area=${area:-us-west-2}

container_image_uri="${account}.dkr.ecr.${area}.amazonaws.com/${container_name}:${container_tag}"

# If the repository does not exist in ECR, create it.
aws ecr describe-repositories --repository-names "${container_name}" --region "${area}" > /dev/null 2>&1

if [ $? -ne 0 ]
then
    aws ecr create-repository --repository-name "${container_name}" --region "${area}" > /dev/null
fi

# Get the login command from ECR and execute it instantly
aws ecr get-login-password --region ${area} | docker login --username AWS --password-stdin ${account}.dkr.ecr.${area}.amazonaws.com/${container_name}

# Construct the docker picture regionally with the picture title after which push it to ECR
# with the total title.
docker construct --platform linux/amd64 --provenance=false -t ${container_name} .
docker tag ${container_name} ${container_image_uri}

docker push ${container_image_uri}

This creates a container with a Docker label indicating to SageMaker AI that bidirectional streaming functionality is supported on this container.

com.amazonaws.sagemaker.capabilities.bidirectional-streaming=true

Deploy the demo bi-directional streaming container to the SageMaker AI endpoint

The next instance script creates the SageMaker AI endpoint with the created container:

import boto3
from datetime import datetime

sagemaker_client = boto3.consumer('sagemaker', region_name=REGION)

# Create mannequin
sagemaker_client.create_model(
    ModelName=MODEL_NAME,
    PrimaryContainer={'Picture': IMAGE_URI, 'Mode': 'SingleModel'},
    ExecutionRoleArn=ROLE
)

# Create config
sagemaker_client.create_endpoint_config(
    EndpointConfigName=ENDPOINT_CONFIG_NAME,
    ProductionVariants=[{
        'VariantName': 'AllTraffic',
        'ModelName': MODEL_NAME,
        'InitialInstanceCount': 1,
        'InstanceType': INSTANCE_TYPE,
        'InitialVariantWeight': 1.0
    }]
)

# Create endpoint
sagemaker_client.create_endpoint(
    EndpointName=ENDPOINT_NAME,
    EndpointConfigName=ENDPOINT_CONFIG_NAME
)

print(f"Endpoint '{ENDPOINT_NAME}' creation initiated")

Invoke the SageMaker AI endpoint with the brand new bidirectional streaming API

As soon as the SageMaker AI endpoint is InService, we will proceed to invoke the endpoint to check the bidirectional streaming performance of the check container.

#!/usr/bin/env python3
"""
SageMaker AI Bidirectional Streaming Python SDK Script.
This script connects to a SageMaker AI endpoint for bidirectional streaming communication.
"""

import argparse
import asyncio
import sys
from aws_sdk_sagemaker_runtime_http2.consumer import SageMakerRuntimeHTTP2Client
from aws_sdk_sagemaker_runtime_http2.config import Config, HTTPAuthSchemeResolver
from aws_sdk_sagemaker_runtime_http2.fashions import InvokeEndpointWithBidirectionalStreamInput, RequestStreamEventPayloadPart, RequestPayloadPart
from smithy_aws_core.identification import EnvironmentCredentialsResolver
from smithy_aws_core.auth.sigv4 import SigV4AuthScheme
import logging


def parse_arguments():
    """Parse command-line arguments."""
    parser = argparse.ArgumentParser(
        description="Hook up with SageMaker AI endpoint for bidirectional streaming"
    )
    parser.add_argument(
        "ENDPOINT_NAME",
        assist="Identify of the SageMaker AI endpoint to connect with"
    )
    return parser.parse_args()


# Configuration
AWS_REGION = "us-west-2"
BIDI_ENDPOINT = f"https://runtime.sagemaker.{AWS_REGION}.amazonaws.com:8443"

logging.basicConfig(stage=logging.INFO)
logger = logging.getLogger(__name__)


class SimpleClient:
    def __init__(self, endpoint_name, area=AWS_REGION):
        self.endpoint_name = endpoint_name
        self.area = area
        self.consumer = None
        self.stream = None
        self.response = None
        self.is_active = False

    def _initialize_client(self):
        config = Config(
            endpoint_uri=BIDI_ENDPOINT,
            area=self.area,
            aws_credentials_identity_resolver=EnvironmentCredentialsResolver(),
            auth_scheme_resolver=HTTPAuthSchemeResolver(),
            auth_schemes={"aws.auth#sigv4": SigV4AuthScheme(service="sagemaker")}
        )
        self.consumer = SageMakerRuntimeHTTP2Client(config=config)

    async def start_session(self):
        if not self.consumer:
            self._initialize_client()

        logger.information(f"Beginning session with endpoint: {self.endpoint_name}")
        self.stream = await self.consumer.invoke_endpoint_with_bidirectional_stream(
            InvokeEndpointWithBidirectionalStreamInput(endpoint_name=self.endpoint_name)
        )
        self.is_active = True

        self.response = asyncio.create_task(self._process_responses())

    async def send_words(self, phrases):
        for i, phrase in enumerate(phrases):
            logger.information(f"Sending payload: {phrase}")
            await self.send_event(phrase.encode('utf-8'))
            await asyncio.sleep(1)

    async def send_event(self, data_bytes):
        payload = RequestPayloadPart(bytes_=data_bytes)
        occasion = RequestStreamEventPayloadPart(worth=payload)
        await self.stream.input_stream.ship(occasion)

    async def end_session(self):
        if not self.is_active:
            return

        await self.stream.input_stream.shut()
        logger.information("Stream closed")

    async def _process_responses(self):
        strive:
            output = await self.stream.await_output()
            output_stream = output[1]

            whereas self.is_active:
                consequence = await output_stream.obtain()

                if result's None:
                    logger.information("No extra responses")
                    break

                if consequence.worth and consequence.worth.bytes_:
                    response_data = consequence.worth.bytes_.decode('utf-8')
                    logger.information(f"Obtained: {response_data}")
        besides Exception as e:
            logger.error(f"Error processing responses: {e}")


def essential():
    """Foremost operate to parse arguments and run the streaming consumer."""
    args = parse_arguments()
    
    print("=" * 60)
    print("SageMaker AI Bidirectional Streaming Shopper")
    print("=" * 60)
    print(f"Endpoint Identify: {args.ENDPOINT_NAME}")
    print(f"AWS Area: {AWS_REGION}")
    print("=" * 60)
    
    async def run_client():
        sagemaker_client = SimpleClient(endpoint_name=args.ENDPOINT_NAME)

        strive:
            await sagemaker_client.start_session()

            phrases = ["I need help with", "my account balance", "I can help with that", "and recent charges"]
            await sagemaker_client.send_words(phrases)

            await asyncio.sleep(2)

            await sagemaker_client.end_session()
            sagemaker_client.is_active = False

            if sagemaker_client.response and never sagemaker_client.response.finished():
                sagemaker_client.response.cancel()

            logger.information("Session ended efficiently")
            return 0
            
        besides Exception as e:
            logger.error(f"Shopper error: {e}")
            return 1

    strive:
        exit_code = asyncio.run(run_client())
        sys.exit(exit_code)
    besides KeyboardInterrupt:
        logger.information("Interrupted by consumer")
        sys.exit(1)
    besides Exception as e:
        logger.error(f"Sudden error: {e}")
        sys.exit(1)


if __name__ == "__main__":
    essential()

The next is pattern output displaying the enter and output streams generated by the earlier script. The container echoes incoming knowledge to the output stream, demonstrating bidirectional streaming functionality.

SageMaker AI integration with Deepgram fashions

SageMaker AI and Deepgram have collaborated to construct bidirectional streaming assist for SageMaker AI endpoints. Deepgram, an AWS Superior Tier Accomplice, delivers enterprise-grade voice AI fashions with industry-leading accuracy and velocity. Their fashions energy real-time transcription, text-to-speech and voice brokers for contact facilities, media platforms, and conversational AI purposes.

For patrons with strict compliance necessities that require audio processing to by no means go away their AWS VPC, conventional self-hosted choices have required important operational overhead to setup and preserve. Amazon SageMaker bidirectional streaming transforms this expertise so prospects can deploy and scale real-time AI purposes with only a few actions within the AWS Administration Console.

Deepgram Nova-3 speech-to-text mannequin is offered in the present day within the AWS Market for deployment as a SageMaker AI endpoint with further fashions coming quickly. Capabilities of Deepgram Nova-3 embody multi-lingual transcription, enterprise scale efficiency and area particular recognition. Deepgram is providing a 14 day free trial on Amazon SageMaker AI for builders to prototype purposes with out incurring software program license charges. Infrastructure costs of the chosen machine kind will nonetheless be incurred throughout this time. For extra particulars, see the Amazon SageMaker AI Pricing documentation.

A high-level overview and pattern code is supplied within the following part. Confer with the detailed fast begin information on the Deepgram documentation page for extra info and examples. Join with the Deepgram Developer Community should you want further assist with arrange.

Arrange a Deepgram SageMaker AI real-time inference endpoint

To arrange a Deepgram SageMaker AI endpoint:

Utilizing the Deepgram SageMaker AI real-time inference endpoint

We’ll now stroll by way of a pattern typescript software that streams an audio file to the Deepgram mannequin hosted on a SageMaker AI real-time inference endpoint and prints a transcription streamed again in real-time.

import * as fs from "fs";
import * as path from "path";
import { RequestStreamEvent } from '@aws-sdk/client-sagemaker-runtime-http2';

operate sleep(ms: quantity): Promise<void> {
 return new Promise(resolve => setTimeout(resolve, ms));
}

async operate* streamWavFile(filePath: string): AsyncIterable<RequestStreamEvent> {
 const full = path.resolve(filePath);

 if (!fs.existsSync(full)) {
 throw new Error(`Audio file not discovered: ${full}`);
 }

 console.log(`Streaming audio: ${full}`);

 const readStream = fs.createReadStream(full, { highWaterMark: 512_000 }); // 512 KB

 for await (const chunk of readStream) {
 yield {
 PayloadPart: {
 Bytes: chunk,
 DataType: "BINARY"
 }
 };
 }
 
 // Maintain the stream alive to obtain transcription responses after complete audio file is shipped
 console.log("Audio despatched, ready for transcription to complete..."); 
 await sleep(15000); // Wait 15 seconds for processing last audio chunk.
 //Lengthy audio information might require sending hold alive packets whereas the transcript is being processed. see https://builders.deepgram.com/docs/audio-keep-alive for extra info.

 // Inform the container we're finished
 yield {
 PayloadPart: {
 Bytes: new TextEncoder().encode('{"kind":"CloseStream"}'),
 DataType: "UTF8"
 }
 };
}

import {
    SageMakerRuntimeHTTP2Client,
    InvokeEndpointWithBidirectionalStreamCommand
} from '@aws-sdk/client-sagemaker-runtime-http2';

const area = "us-east-1";              // AWS Area
const endpointName = "REPLACEME";        // Your SageMaker Deepgram endpoint title
const audioFile = "check.wav";            // Native audio file

// Deepgram WebSocket path contained in the mannequin container
const modelInvocationPath = "v1/pay attention";
const modelQueryString = "mannequin=nova-3";

const consumer = new SageMakerRuntimeHTTP2Client({
    area
});

async operate run() {
    console.log("Sending audio to Deepgram by way of SageMaker...");

    const command = new InvokeEndpointWithBidirectionalStreamCommand({
        EndpointName: endpointName,
        Physique: streamWavFile(audioFile),
        ModelInvocationPath: modelInvocationPath,
        ModelQueryString: modelQueryString
    });

    const response = await consumer.ship(command);

    if (!response.Physique) {
        console.log("No streaming response acquired.");
        return;
    }

    const decoder = new TextDecoder();

    for await (const msg of response.Physique) {
        if (msg.PayloadPart?.Bytes) {
            const textual content = decoder.decode(msg.PayloadPart.Bytes);

            strive {
                const parsed = JSON.parse(textual content);
                
                // Extract and show the transcript
                if (parsed.channel?.options?.[0]?.transcript) {
                    const transcript = parsed.channel.options[0].transcript;
                    if (transcript.trim()) {
                        console.log("Transcript:", transcript);
                    }
                }
                    
                console.debug("Deepgram (uncooked):", parsed);

            } catch {
                console.error("Deepgram (error):", textual content);
            }
        }
    }

    console.log("Streaming completed.");
}

run().catch(console.error);

Conclusion

On this put up, we supplied an outline of constructing actual time brokers with generative AI, the challenges, and the way SageMaker AI bidirectional streaming helps you tackle these challenges. We additionally supplied particulars on the best way to construct your individual container to leverage bidirectional streaming characteristic. We then walked you thru the steps to construct a pattern chatbot container and the real-time speech-to-text mannequin provided by our accomplice Deepgram which is a core part in a real-time voice AI agent software.

Begin constructing bidirectional streaming purposes with LLMs and SageMaker AI in the present day.

In regards to the authors

Lingran Xia is a software program improvement engineer at AWS. He at present focuses on bettering inference efficiency of machine studying fashions. In his free time, he enjoys touring and snowboarding.

Vivek Gangasani is a Worldwide Lead GenAI Specialist Options Architect for SageMaker Inference. He drives Go-to-Market (GTM) and Outbound Product technique for SageMaker Inference. He additionally helps enterprises and startups deploy, handle, and scale their GenAI fashions with SageMaker and GPUs. At the moment, he’s targeted on growing methods and options for optimizing inference efficiency and GPU effectivity for internet hosting Giant Language Fashions. In his free time, Vivek enjoys climbing, watching films, and attempting completely different cuisines.

Victor Wang is a Sr. Options Architect at Amazon Internet Providers, based mostly in San Francisco, CA, supporting GenAI Startups together with Deepgram. Victor has spent 7 years at Amazon; earlier roles embody software program developer for AWS Website-to-Website VPN, AWS ProServe Guide for Public Sector Companions, and Technical Program Supervisor for Amazon Aurora MySQL. His ardour is studying new applied sciences and touring the world. Victor has flown over two million miles and plans to proceed his everlasting journey of exploration.

Chinmay Bapat is an Engineering Supervisor within the Amazon SageMaker AI Inference group at AWS, the place he leads engineering efforts targeted on constructing scalable infrastructure for generative AI inference. His work permits prospects to deploy and serve massive language fashions and different AI fashions effectively at scale. Exterior of labor, he enjoys enjoying board video games and is studying to ski.

Deepti Ragha is a Senior Software program Improvement Engineer on the Amazon SageMaker AI group, specializing in ML inference infrastructure and mannequin internet hosting optimization. She builds options that enhance deployment efficiency, cut back inference prices, and make ML accessible to organizations of all sizes. Exterior of labor, she enjoys touring, climbing, and gardening.

Kareem Syed-Mohammed is a Product Supervisor at AWS. He’s focuses on enabling Gen AI mannequin improvement and governance on SageMaker HyperPod. Previous to this, at Amazon QuickSight, he led embedded analytics, and developer expertise. Along with QuickSight, he has been with AWS Market and Amazon retail as a Product Supervisor. Kareem began his profession as a developer for name heart applied sciences, Native Skilled and Advertisements for Expedia, and administration guide at McKinsey.

Xu Deng is a Software program Engineer Supervisor with the SageMaker group. He focuses on serving to prospects construct and optimize their AI/ML inference expertise on Amazon SageMaker. In his spare time, he loves touring and snowboarding.