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

In 2025, generative AI has developed from textual content technology to multi-modal use instances starting from audio transcription and translation to voice brokers that require real-time information streaming. At the moment’s functions demand one thing extra: steady, real-time dialogue between customers and fashions—the flexibility for information to circulate each methods, concurrently, over a single persistent connection. Think about a speech to textual content use-case, the place you will have 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 alternate 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 speedy. 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 stay transcript seems in entrance of the decision middle agent, giving the agent immediate context and letting them reply with out ready for the caller to complete. This sort of steady alternate makes voice experiences really feel fluid, responsive and human.

This submit exhibits you find out how to construct and deploy a container with bidirectional streaming functionality to a SageMaker AI endpoint. We additionally display how one can convey your personal 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, information flows each methods without delay 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.

Consumer: [sends complete question] → waits...
Mannequin: ...processes... [returns complete answer] 
Consumer: [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.

Consumer: [question starts flowing with enough context] →                                    
Mannequin: ← [answer starts flowing immediately]
Consumer: → [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 lot of short-lived connections. This reduces overhead on networking infrastructure, TLS handshakes, and connection administration. Fashions can keep 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 purchasers and fashions. If 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, information flows freely in each instructions:

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

Constructing your personal container for implementing bidirectional streaming

If you want to make use of open supply or your personal fashions, you’ll be able to customise your container to assist bidirectional streaming. Your container should implement the WebSocket protocol to deal with incoming information 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 are going to:

Stipulations

Construct docker container with bi-directional streaming functionality

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

# The appliance makes use of surroundings variables for AWS authentication. Set these earlier than operating the appliance:
# 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 would 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 immediately
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 complete 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 are able to proceed to invoke the endpoint to check the bidirectional streaming performance of the take a look at 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 hook up 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:
                end result = await output_stream.obtain()

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

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


def fundamental():
    """Foremost operate to parse arguments and run the streaming consumer."""
    args = parse_arguments()
    
    print("=" * 60)
    print("SageMaker AI Bidirectional Streaming Consumer")
    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.achieved():
                sagemaker_client.response.cancel()

            logger.information("Session ended efficiently")
            return 0
            
        besides Exception as e:
            logger.error(f"Consumer 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"Surprising error: {e}")
        sys.exit(1)


if __name__ == "__main__":
    fundamental()

The next is pattern output displaying the enter and output streams generated by the earlier script. The container echoes incoming information 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 Associate, 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 functions.

For purchasers with strict compliance necessities that require audio processing to by no means depart their AWS VPC, conventional self-hosted choices have required vital operational overhead to setup and keep. Amazon SageMaker bidirectional streaming transforms this expertise so prospects can deploy and scale real-time AI functions with just some actions within the AWS Administration Console.

Deepgram Nova-3 speech-to-text mannequin is accessible immediately within the AWS Market for deployment as a SageMaker AI endpoint with extra 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 functions with out incurring software program license charges. Infrastructure prices of the chosen machine sort 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 offered within the following part. Seek advice from the detailed fast begin information on the Deepgram documentation web page for added data and examples. Join with the Deepgram Developer Group if you happen to want extra 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 {
 return new Promise(resolve => setTimeout(resolve, ms));
}

async operate* streamWavFile(filePath: string): AsyncIterable {
 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 distributed
 console.log("Audio despatched, ready for transcription to complete..."); 
 await sleep(15000); // Wait 15 seconds for processing ultimate audio chunk.
 //Lengthy audio recordsdata could require sending maintain alive packets whereas the transcript is being processed. see https://builders.deepgram.com/docs/audio-keep-alive for extra data.

 // Inform the container we're achieved
 yield {
 PayloadPart: {
 Bytes: new TextEncoder().encode('{"sort":"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 = "take a look at.wav";            // Native audio file

// Deepgram WebSocket path contained in the mannequin container
const modelInvocationPath = "v1/hear";
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?.alternate options?.[0]?.transcript) {
                    const transcript = parsed.channel.alternate 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 submit, we offered an outline of constructing actual time brokers with generative AI, the challenges, and the way SageMaker AI bidirectional streaming helps you deal with these challenges. We additionally offered particulars on find out how to construct your personal 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 supplied by our accomplice Deepgram which is a core element in a real-time voice AI agent software.

Begin constructing bidirectional streaming functions with LLMs and SageMaker AI immediately.

In regards to the authors

Lingran Xia is a software program growth engineer at AWS. He at present focuses on enhancing 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 present, he’s targeted on growing methods and options for optimizing inference efficiency and GPU effectivity for internet hosting Massive Language Fashions. In his free time, Vivek enjoys mountaineering, watching films, and attempting totally different cuisines.

Victor Wang is a Sr. Options Architect at Amazon Net Companies, primarily based 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 Advisor 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 workforce at AWS, the place he leads engineering efforts targeted on constructing scalable infrastructure for generative AI inference. His work allows prospects to deploy and serve giant language fashions and different AI fashions effectively at scale. Outdoors of labor, he enjoys taking part in board video games and is studying to ski.

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

Kareem Syed-Mohammed is a Product Supervisor at AWS. He’s focuses on enabling Gen AI mannequin growth 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 middle applied sciences, Native Skilled and Adverts for Expedia, and administration marketing consultant at McKinsey.

Xu Deng is a Software program Engineer Supervisor with the SageMaker workforce. 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.