Select your cookie preferences

We use essential cookies and similar tools that are necessary to provide our site and services. We use performance cookies to collect anonymous statistics, so we can understand how customers use our site and make improvements. Essential cookies cannot be deactivated, but you can choose “Customize” or “Decline” to decline performance cookies.

If you agree, AWS and approved third parties will also use cookies to provide useful site features, remember your preferences, and display relevant content, including relevant advertising. To accept or decline all non-essential cookies, choose “Accept” or “Decline.” To make more detailed choices, choose “Customize.”

Using the API to start a streaming conversation

Focus mode
Using the API to start a streaming conversation - Amazon Lex

When you start a stream to an Amazon Lex V2 bot, you accomplish the following tasks:

  1. Create an initial connection to the server.

  2. Configure the security credentials and bot details. Bot details include whether the bot takes DTMF and audio input, or text input.

  3. Send events to the server. These events are text data or audio data from the user.

  4. Process events sent from the server. In this step, you determine whether the bot output is presented to the user as text or speech.

The following code examples initialize a streaming conversation with an Amazon Lex V2 bot and your local machine. You can modify the code to meet your needs.

The following code is an example request using the AWS SDK for Java to start the connection to a bot and configure the bot details and credentials.

package com.lex.streaming.sample; import software.amazon.awssdk.auth.credentials.AwsBasicCredentials; import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider; import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider; import software.amazon.awssdk.regions.Region; import software.amazon.awssdk.services.lexruntimev2.LexRuntimeV2AsyncClient; import software.amazon.awssdk.services.lexruntimev2.model.ConversationMode; import software.amazon.awssdk.services.lexruntimev2.model.StartConversationRequest; import java.net.URISyntaxException; import java.util.UUID; import java.util.concurrent.CompletableFuture; /** * The following code creates a connection with the Amazon Lex bot and configures the bot details and credentials. * Prerequisite: To use this example, you must be familiar with the Reactive streams programming model. * For more information, see * https://github.com/reactive-streams/reactive-streams-jvm. * This example uses AWS SDK for Java for Amazon Lex V2. * <p> * The following sample application interacts with an Amazon Lex bot with the streaming API. It uses the Audio * conversation mode to return audio responses to the user's input. * <p> * The code in this example accomplishes the following: * <p> * 1. Configure details about the conversation between the user and the Amazon Lex bot. These details include the conversation mode and the specific bot the user is speaking with. * 2. Create an events publisher that passes the audio events to the Amazon Lex bot after you establish the connection. The code we provide in this example tells your computer to pick up the audio from * your microphone and send that audio data to Amazon Lex. * 3. Create a response handler that handles the audio responses from the Amazon Lex bot and plays back the audio to you. */ public class LexBidirectionalStreamingExample { public static void main(String[] args) throws URISyntaxException, InterruptedException { String botId = ""; String botAliasId = ""; String localeId = ""; String accessKey = ""; String secretKey = ""; String sessionId = UUID.randomUUID().toString(); Region region = Region.region_name; // Choose an AWS Region where the Amazon Lex Streaming API is available. AwsCredentialsProvider awsCredentialsProvider = StaticCredentialsProvider .create(AwsBasicCredentials.create(accessKey, secretKey)); // Create a new SDK client. You need to use an asynchronous client. System.out.println("step 1: creating a new Lex SDK client"); LexRuntimeV2AsyncClient lexRuntimeServiceClient = LexRuntimeV2AsyncClient.builder() .region(region) .credentialsProvider(awsCredentialsProvider) .build(); // Configure the bot, alias and locale that you'll use to have a conversation. System.out.println("step 2: configuring bot details"); StartConversationRequest.Builder startConversationRequestBuilder = StartConversationRequest.builder() .botId(botId) .botAliasId(botAliasId) .localeId(localeId); // Configure the conversation mode of the bot. By default, the // conversation mode is audio. System.out.println("step 3: choosing conversation mode"); startConversationRequestBuilder = startConversationRequestBuilder.conversationMode(ConversationMode.AUDIO); // Assign a unique identifier for the conversation. System.out.println("step 4: choosing a unique conversation identifier"); startConversationRequestBuilder = startConversationRequestBuilder.sessionId(sessionId); // Start the initial request. StartConversationRequest startConversationRequest = startConversationRequestBuilder.build(); // Create a stream of audio data to the Amazon Lex bot. The stream will start after the connection is established with the bot. EventsPublisher eventsPublisher = new EventsPublisher(); // Create a class to handle responses from bot. After the server processes the user data you've streamed, the server responds // on another stream. BotResponseHandler botResponseHandler = new BotResponseHandler(eventsPublisher); // Start a connection and pass in the publisher that streams the audio and process the responses from the bot. System.out.println("step 5: starting the conversation ..."); CompletableFuture<Void> conversation = lexRuntimeServiceClient.startConversation( startConversationRequest, eventsPublisher, botResponseHandler); // Wait until the conversation finishes. The conversation finishes if the dialog state reaches the "Closed" state. // The client stops the connection. If an exception occurs during the conversation, the // client sends a disconnection event. conversation.whenComplete((result, exception) -> { if (exception != null) { eventsPublisher.disconnect(); } }); // The conversation finishes when the dialog state is closed and last prompt has been played. while (!botResponseHandler.isConversationComplete()) { Thread.sleep(100); } // Randomly sleep for 100 milliseconds to prevent JVM from exiting. // You won't need this in your production code because your JVM is // likely to always run. // When the conversation finishes, the following code block stops publishing more data and informs the Amazon Lex bot that there is no more data to send. if (botResponseHandler.isConversationComplete()) { System.out.println("conversation is complete."); eventsPublisher.stop(); } } }

The following code is an example request using the AWS SDK for Java to send events to the bot. The code in this example uses the microphone on your computer to send audio events.

package com.lex.streaming.sample; import org.reactivestreams.Publisher; import org.reactivestreams.Subscriber; import software.amazon.awssdk.services.lexruntimev2.model.StartConversationRequestEventStream; /** * You use the Events publisher to send events to the Amazon Lex bot. When you establish a connection, the bot uses the * subscribe() method and enables the events publisher starts sending events to * your computer. The bot uses the "request" method of the subscription to make more requests. For more information on the request method, see https://github.com/reactive-streams/reactive-streams-jvm. */ public class EventsPublisher implements Publisher<StartConversationRequestEventStream> { private AudioEventsSubscription audioEventsSubscription; @Override public void subscribe(Subscriber<? super StartConversationRequestEventStream> subscriber) { if (audioEventsSubscription == null) { audioEventsSubscription = new AudioEventsSubscription(subscriber); subscriber.onSubscribe(audioEventsSubscription); } else { throw new IllegalStateException("received unexpected subscription request"); } } public void disconnect() { if (audioEventsSubscription != null) { audioEventsSubscription.disconnect(); } } public void stop() { if (audioEventsSubscription != null) { audioEventsSubscription.stop(); } } public void playbackFinished() { if (audioEventsSubscription != null) { audioEventsSubscription.playbackFinished(); } } }

The following code is an example request using the AWS SDK for Java to handle responses from the bot. The code in this example configures Amazon Lex V2 to play an audio response back to you.

package com.lex.streaming.sample; import javazoom.jl.decoder.JavaLayerException; import javazoom.jl.player.advanced.AdvancedPlayer; import javazoom.jl.player.advanced.PlaybackEvent; import javazoom.jl.player.advanced.PlaybackListener; import software.amazon.awssdk.core.async.SdkPublisher; import software.amazon.awssdk.services.lexruntimev2.model.AudioResponseEvent; import software.amazon.awssdk.services.lexruntimev2.model.DialogActionType; import software.amazon.awssdk.services.lexruntimev2.model.IntentResultEvent; import software.amazon.awssdk.services.lexruntimev2.model.PlaybackInterruptionEvent; import software.amazon.awssdk.services.lexruntimev2.model.StartConversationResponse; import software.amazon.awssdk.services.lexruntimev2.model.StartConversationResponseEventStream; import software.amazon.awssdk.services.lexruntimev2.model.StartConversationResponseHandler; import software.amazon.awssdk.services.lexruntimev2.model.TextResponseEvent; import software.amazon.awssdk.services.lexruntimev2.model.TranscriptEvent; import java.io.IOException; import java.io.UncheckedIOException; import java.util.concurrent.CompletableFuture; /** * The following class is responsible for processing events sent from the Amazon Lex bot. The bot sends multiple audio events, * so the following code concatenates those audio events and uses a publicly available Java audio player to play out the message to * the user. */ public class BotResponseHandler implements StartConversationResponseHandler { private final EventsPublisher eventsPublisher; private boolean lastBotResponsePlayedBack; private boolean isDialogStateClosed; private AudioResponse audioResponse; public BotResponseHandler(EventsPublisher eventsPublisher) { this.eventsPublisher = eventsPublisher; this.lastBotResponsePlayedBack = false;// At the start, we have not played back last response from bot. this.isDialogStateClosed = false; // At the start, the dialog state is open. } @Override public void responseReceived(StartConversationResponse startConversationResponse) { System.out.println("successfully established the connection with server. request id:" + startConversationResponse.responseMetadata().requestId()); // would have 2XX, request id. } @Override public void onEventStream(SdkPublisher<StartConversationResponseEventStream> sdkPublisher) { sdkPublisher.subscribe(event -> { if (event instanceof PlaybackInterruptionEvent) { handle((PlaybackInterruptionEvent) event); } else if (event instanceof TranscriptEvent) { handle((TranscriptEvent) event); } else if (event instanceof IntentResultEvent) { handle((IntentResultEvent) event); } else if (event instanceof TextResponseEvent) { handle((TextResponseEvent) event); } else if (event instanceof AudioResponseEvent) { handle((AudioResponseEvent) event); } }); } @Override public void exceptionOccurred(Throwable throwable) { System.err.println("got an exception:" + throwable); } @Override public void complete() { System.out.println("on complete"); } private void handle(PlaybackInterruptionEvent event) { System.out.println("Got a PlaybackInterruptionEvent: " + event); } private void handle(TranscriptEvent event) { System.out.println("Got a TranscriptEvent: " + event); } private void handle(IntentResultEvent event) { System.out.println("Got an IntentResultEvent: " + event); isDialogStateClosed = DialogActionType.CLOSE.equals(event.sessionState().dialogAction().type()); } private void handle(TextResponseEvent event) { System.out.println("Got an TextResponseEvent: " + event); event.messages().forEach(message -> { System.out.println("Message content type:" + message.contentType()); System.out.println("Message content:" + message.content()); }); } private void handle(AudioResponseEvent event) {//Synthesize speech // System.out.println("Got a AudioResponseEvent: " + event); if (audioResponse == null) { audioResponse = new AudioResponse(); //Start an audio player in a different thread. CompletableFuture.runAsync(() -> { try { AdvancedPlayer audioPlayer = new AdvancedPlayer(audioResponse); audioPlayer.setPlayBackListener(new PlaybackListener() { @Override public void playbackFinished(PlaybackEvent evt) { super.playbackFinished(evt); // Inform the Amazon Lex bot that the playback has finished. eventsPublisher.playbackFinished(); if (isDialogStateClosed) { lastBotResponsePlayedBack = true; } } }); audioPlayer.play(); } catch (JavaLayerException e) { throw new RuntimeException("got an exception when using audio player", e); } }); } if (event.audioChunk() != null) { audioResponse.write(event.audioChunk().asByteArray()); } else { // The audio audio prompt has ended when the audio response has no // audio bytes. try { audioResponse.close(); audioResponse = null; // Prepare for the next audio prompt. } catch (IOException e) { throw new UncheckedIOException("got an exception when closing the audio response", e); } } } // The conversation with the Amazon Lex bot is complete when the bot marks the Dialog as DialogActionType.CLOSE // and any prompt playback is finished. For more information, see // https://docs.aws.amazon.com/lexv2/latest/dg/API_runtime_DialogAction.html. public boolean isConversationComplete() { return isDialogStateClosed && lastBotResponsePlayedBack; } }

To configure a bot to respond to input events with audio, you must first subscribe to audio events from Amazon Lex V2 and then configure the bot to provide an audio response to the input events from the user.

The following code is an AWS SDK for Java example for subscribing to audio events from Amazon Lex V2.

package com.lex.streaming.sample; import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; import software.amazon.awssdk.core.SdkBytes; import software.amazon.awssdk.services.lexruntimev2.model.AudioInputEvent; import software.amazon.awssdk.services.lexruntimev2.model.ConfigurationEvent; import software.amazon.awssdk.services.lexruntimev2.model.DisconnectionEvent; import software.amazon.awssdk.services.lexruntimev2.model.PlaybackCompletionEvent; import software.amazon.awssdk.services.lexruntimev2.model.StartConversationRequestEventStream; import javax.sound.sampled.AudioFormat; import javax.sound.sampled.AudioInputStream; import javax.sound.sampled.AudioSystem; import javax.sound.sampled.DataLine; import javax.sound.sampled.LineUnavailableException; import javax.sound.sampled.TargetDataLine; import java.io.IOException; import java.io.UncheckedIOException; import java.nio.ByteBuffer; import java.util.Arrays; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CompletableFuture; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.atomic.AtomicLong; public class AudioEventsSubscription implements Subscription { private static final AudioFormat MIC_FORMAT = new AudioFormat(8000, 16, 1, true, false); private static final String AUDIO_CONTENT_TYPE = "audio/lpcm; sample-rate=8000; sample-size-bits=16; channel-count=1; is-big-endian=false"; //private static final String RESPONSE_TYPE = "audio/pcm; sample-rate=8000"; private static final String RESPONSE_TYPE = "audio/mpeg"; private static final int BYTES_IN_AUDIO_CHUNK = 320; private static final AtomicLong eventIdGenerator = new AtomicLong(0); private final AudioInputStream audioInputStream; private final Subscriber<? super StartConversationRequestEventStream> subscriber; private final EventWriter eventWriter; private CompletableFuture eventWriterFuture; public AudioEventsSubscription(Subscriber<? super StartConversationRequestEventStream> subscriber) { this.audioInputStream = getMicStream(); this.subscriber = subscriber; this.eventWriter = new EventWriter(subscriber, audioInputStream); configureConversation(); } private AudioInputStream getMicStream() { try { DataLine.Info dataLineInfo = new DataLine.Info(TargetDataLine.class, MIC_FORMAT); TargetDataLine targetDataLine = (TargetDataLine) AudioSystem.getLine(dataLineInfo); targetDataLine.open(MIC_FORMAT); targetDataLine.start(); return new AudioInputStream(targetDataLine); } catch (LineUnavailableException e) { throw new RuntimeException(e); } } @Override public void request(long demand) { // If a thread to write events has not been started, start it. if (eventWriterFuture == null) { eventWriterFuture = CompletableFuture.runAsync(eventWriter); } eventWriter.addDemand(demand); } @Override public void cancel() { subscriber.onError(new RuntimeException("stream was cancelled")); try { audioInputStream.close(); } catch (IOException e) { throw new UncheckedIOException(e); } } public void configureConversation() { String eventId = "ConfigurationEvent-" + String.valueOf(eventIdGenerator.incrementAndGet()); ConfigurationEvent configurationEvent = StartConversationRequestEventStream .configurationEventBuilder() .eventId(eventId) .clientTimestampMillis(System.currentTimeMillis()) .responseContentType(RESPONSE_TYPE) .build(); System.out.println("writing config event"); eventWriter.writeConfigurationEvent(configurationEvent); } public void disconnect() { String eventId = "DisconnectionEvent-" + String.valueOf(eventIdGenerator.incrementAndGet()); DisconnectionEvent disconnectionEvent = StartConversationRequestEventStream .disconnectionEventBuilder() .eventId(eventId) .clientTimestampMillis(System.currentTimeMillis()) .build(); eventWriter.writeDisconnectEvent(disconnectionEvent); try { audioInputStream.close(); } catch (IOException e) { throw new UncheckedIOException(e); } } //Notify the subscriber that we've finished. public void stop() { subscriber.onComplete(); } public void playbackFinished() { String eventId = "PlaybackCompletion-" + String.valueOf(eventIdGenerator.incrementAndGet()); PlaybackCompletionEvent playbackCompletionEvent = StartConversationRequestEventStream .playbackCompletionEventBuilder() .eventId(eventId) .clientTimestampMillis(System.currentTimeMillis()) .build(); eventWriter.writePlaybackFinishedEvent(playbackCompletionEvent); } private static class EventWriter implements Runnable { private final BlockingQueue<StartConversationRequestEventStream> eventQueue; private final AudioInputStream audioInputStream; private final AtomicLong demand; private final Subscriber subscriber; private boolean conversationConfigured; public EventWriter(Subscriber subscriber, AudioInputStream audioInputStream) { this.eventQueue = new LinkedBlockingQueue<>(); this.demand = new AtomicLong(0); this.subscriber = subscriber; this.audioInputStream = audioInputStream; } public void writeConfigurationEvent(ConfigurationEvent configurationEvent) { eventQueue.add(configurationEvent); } public void writeDisconnectEvent(DisconnectionEvent disconnectionEvent) { eventQueue.add(disconnectionEvent); } public void writePlaybackFinishedEvent(PlaybackCompletionEvent playbackCompletionEvent) { eventQueue.add(playbackCompletionEvent); } void addDemand(long l) { this.demand.addAndGet(l); } @Override public void run() { try { while (true) { long currentDemand = demand.get(); if (currentDemand > 0) { // Try to read from queue of events. // If nothing is in queue at this point, read the audio events directly from audio stream. for (long i = 0; i < currentDemand; i++) { if (eventQueue.peek() != null) { subscriber.onNext(eventQueue.take()); demand.decrementAndGet(); } else { writeAudioEvent(); } } } } } catch (InterruptedException e) { throw new RuntimeException("interrupted when reading data to be sent to server"); } catch (Exception e) { e.printStackTrace(); } } private void writeAudioEvent() { byte[] bytes = new byte[BYTES_IN_AUDIO_CHUNK]; int numBytesRead = 0; try { numBytesRead = audioInputStream.read(bytes); if (numBytesRead != -1) { byte[] byteArrayCopy = Arrays.copyOf(bytes, numBytesRead); String eventId = "AudioEvent-" + String.valueOf(eventIdGenerator.incrementAndGet()); AudioInputEvent audioInputEvent = StartConversationRequestEventStream .audioInputEventBuilder() .audioChunk(SdkBytes.fromByteBuffer(ByteBuffer.wrap(byteArrayCopy))) .contentType(AUDIO_CONTENT_TYPE) .clientTimestampMillis(System.currentTimeMillis()) .eventId(eventId).build(); //System.out.println("sending audio event:" + audioInputEvent); subscriber.onNext(audioInputEvent); demand.decrementAndGet(); //System.out.println("sent audio event:" + audioInputEvent); } else { subscriber.onComplete(); System.out.println("audio stream has ended"); } } catch (IOException e) { System.out.println("got an exception when reading from audio stream"); System.err.println(e); subscriber.onError(e); } } } }

The following AWS SDK for Java example configures the Amazon Lex V2 bot to provide an audio response to the input events.

package com.lex.streaming.sample; import java.io.IOException; import java.io.InputStream; import java.io.UncheckedIOException; import java.util.Optional; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; public class AudioResponse extends InputStream{ // Used to convert byte, which is signed in Java, to positive integer (unsigned) private static final int UNSIGNED_BYTE_MASK = 0xFF; private static final long POLL_INTERVAL_MS = 10; private final LinkedBlockingQueue<Integer> byteQueue = new LinkedBlockingQueue<>(); private volatile boolean closed; @Override public int read() throws IOException { try { Optional<Integer> maybeInt; while (true) { maybeInt = Optional.ofNullable(this.byteQueue.poll(POLL_INTERVAL_MS, TimeUnit.MILLISECONDS)); // If we get an integer from the queue, return it. if (maybeInt.isPresent()) { return maybeInt.get(); } // If the stream is closed and there is nothing queued up, return -1. if (this.closed) { return -1; } } } catch (InterruptedException e) { throw new IOException(e); } } /** * Writes data into the stream to be offered on future read() calls. */ public void write(byte[] byteArray) { // Don't write into the stream if it is already closed. if (this.closed) { throw new UncheckedIOException(new IOException("Stream already closed when attempting to write into it.")); } for (byte b : byteArray) { this.byteQueue.add(b & UNSIGNED_BYTE_MASK); } } @Override public void close() throws IOException { this.closed = true; super.close(); } }
PrivacySite termsCookie preferences
© 2025, Amazon Web Services, Inc. or its affiliates. All rights reserved.