翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
API を使用してストリーミングの会話をスタートする
Amazon Lex V2 ボットへのストリーミングをスタートすると、次のタスクが実行されます。
-
サーバーへの初期接続を作成します。
-
セキュリティ認証情報およびボットの詳細を設定します。ボットの詳細には、ボットが DTMF と音声入力、またはテキスト入力を受け取るかどうかが含まれています。
-
イベントをサーバーに送信します。これらのイベントは、ユーザーからのテキストデータまたは音声データです。
-
サーバーから送信されたイベントを処理します。このステップでは、ボットの出力がテキストまたはスピーチとしてユーザーに提示されるかどうかを決定します。
次のコード例では、Amazon Lex V2 ボットとローカルマシンとのストリーミングの会話を初期化します。必要に応じてコードを変更できます。
次のコードは、AWS SDK for Java を使用したリクエストの例であり、ボットへの接続をスタートし、ボットの詳細と認証情報を設定します。
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();
}
}
}
次のコードは、AWS SDK for Java を使用したリクエストの例であり、ボットにイベントを送信します。この例のコードは、コンピュータのマイクを使用して音声イベントを送信します。
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();
}
}
}
次のコードは、AWS SDK for Java を使用したリクエストの例であり、ボットからのレスポンスを処理します。この例のコードは、Amazon Lex V2 が音声レスポンスを再生するように設定します。
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;
}
}
音声で入力イベントに応答するようにボットを設定するには、最初に Amazon Lex V2 の音声イベントを受信し、ユーザーからの入力イベントに対する音声レスポンスを提供するようにボットを設定する必要があります。
次のコードは、Amazon Lex V2 からの音声イベントを受信する AWS SDK for Java の例です。
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);
}
}
}
}
次の AWS SDK for Java の例は、入力イベントに音声レスポンスを提供するように Amazon Lex V2 ボットを設定します。
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();
}
}
