diff --git a/core/src/main/java/com/sap/ai/sdk/core/RealtimeParam.java b/core/src/main/java/com/sap/ai/sdk/core/RealtimeParam.java new file mode 100644 index 000000000..9c1e4d57e --- /dev/null +++ b/core/src/main/java/com/sap/ai/sdk/core/RealtimeParam.java @@ -0,0 +1,27 @@ +package com.sap.ai.sdk.core; + +import javax.annotation.Nonnull; + +/** Represents possible configuration params of realtime client */ +public interface RealtimeParam { + enum SpeechOutputParamName { + VOICE, + TURN_DETECTION, + } + + /** + * Returns param name + * + * @return name + */ + @Nonnull + SpeechOutputParamName getParamName(); + + /** + * Returns string value representation of the param + * + * @return string value + */ + @Nonnull + String getValueAsString(); +} diff --git a/core/src/main/java/com/sap/ai/sdk/core/RealtimeParamTurnDetection.java b/core/src/main/java/com/sap/ai/sdk/core/RealtimeParamTurnDetection.java new file mode 100644 index 000000000..78fbdfa3a --- /dev/null +++ b/core/src/main/java/com/sap/ai/sdk/core/RealtimeParamTurnDetection.java @@ -0,0 +1,48 @@ +package com.sap.ai.sdk.core; + +import java.util.Objects; +import javax.annotation.Nonnull; + +/** Allows to configure turn detection (how model responds). */ +public final class RealtimeParamTurnDetection implements RealtimeParam { + + /** Model tries to recognize if/when it should respond automatically */ + public static final RealtimeParamTurnDetection BY_MODEL_AUTO = + new RealtimeParamTurnDetection("BY_MODEL_AUTO"); + + /** + * Each call to the provided realtime client is considered a turn (eager explicit turn detection). + * Less convenient than the automatic option but may give lower latency in some cases (model does + * not need to perform additional turn detection analysis). + */ + public static final RealtimeParamTurnDetection EACH_CALL_IS_A_TURN = + new RealtimeParamTurnDetection("EACH_CALL_IS_A_TURN"); + + private final String turnDetectionKind; + + private RealtimeParamTurnDetection(String turnDetectionKind) { + this.turnDetectionKind = turnDetectionKind; + } + + @Override + public @Nonnull SpeechOutputParamName getParamName() { + return SpeechOutputParamName.TURN_DETECTION; + } + + @Override + public @Nonnull String getValueAsString() { + return turnDetectionKind; + } + + @Override + public boolean equals(Object o) { + if (o == null || getClass() != o.getClass()) return false; + RealtimeParamTurnDetection that = (RealtimeParamTurnDetection) o; + return Objects.equals(turnDetectionKind, that.turnDetectionKind); + } + + @Override + public int hashCode() { + return Objects.hashCode(turnDetectionKind); + } +} diff --git a/core/src/main/java/com/sap/ai/sdk/core/RealtimeParamVoice.java b/core/src/main/java/com/sap/ai/sdk/core/RealtimeParamVoice.java new file mode 100644 index 000000000..1743430b0 --- /dev/null +++ b/core/src/main/java/com/sap/ai/sdk/core/RealtimeParamVoice.java @@ -0,0 +1,53 @@ +package com.sap.ai.sdk.core; + +import java.util.Objects; +import javax.annotation.Nonnull; + +/** Allows to configure model output voice */ +public final class RealtimeParamVoice implements RealtimeParam { + + /** Standard voice 1 */ + public static final RealtimeParamVoice DEFAULT_1 = new RealtimeParamVoice("DEFAULT_1"); + + /** Standard voice 2 */ + public static final RealtimeParamVoice DEFAULT_2 = new RealtimeParamVoice("DEFAULT_2"); + + private final String voice; + + private RealtimeParamVoice(@Nonnull String voice) { + this.voice = voice; + } + + /** + * Allows to configure raw voice name as named by model provider. Unsafe because SDK cannot verify + * in advance if the provided voice name is correct and supported by the chosen model and use case + * + * @param voiceName as named by model provider + * @return typed voice client configuration param + */ + public static RealtimeParamVoice unsafeWithExplicitVoice(String voiceName) { + return new RealtimeParamVoice(voiceName); + } + + @Override + public @Nonnull SpeechOutputParamName getParamName() { + return SpeechOutputParamName.VOICE; + } + + @Override + public @Nonnull String getValueAsString() { + return voice; + } + + @Override + public boolean equals(Object o) { + if (o == null || getClass() != o.getClass()) return false; + RealtimeParamVoice that = (RealtimeParamVoice) o; + return Objects.equals(voice, that.voice); + } + + @Override + public int hashCode() { + return Objects.hashCode(voice); + } +} diff --git a/foundation-models/openai/pom.xml b/foundation-models/openai/pom.xml index 9d5280f45..91a93f52c 100644 --- a/foundation-models/openai/pom.xml +++ b/foundation-models/openai/pom.xml @@ -70,6 +70,10 @@ com.sap.ai.sdk core + + com.openai + openai-java + com.google.code.findbugs jsr305 diff --git a/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/AudioInputChannel.java b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/AudioInputChannel.java new file mode 100644 index 000000000..50bea1504 --- /dev/null +++ b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/AudioInputChannel.java @@ -0,0 +1,18 @@ +package com.sap.ai.sdk.foundationmodels.openai; + +/** + * Functional interface representing audio input channel (audio data consumer) + * + *

Should be closed by application (try-with-resources) when not needed anymore + */ +public interface AudioInputChannel extends AutoCloseable { + + /** + * This method is sequentially invoked by audio data provider to supply implementer (consumer) + * with the audio data. Exact audio format (encoding, sampling rate, etc.) depends on the usage + * context + * + * @param rawBytesChunk binary data in the depending on the use case format + */ + void inputAudio(byte[] rawBytesChunk); +} diff --git a/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/AudioOutputChannel.java b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/AudioOutputChannel.java new file mode 100644 index 000000000..81217d100 --- /dev/null +++ b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/AudioOutputChannel.java @@ -0,0 +1,17 @@ +package com.sap.ai.sdk.foundationmodels.openai; + +/** Functional interface representing audio output channel (audio data consumer) */ +public interface AudioOutputChannel { + + /** + * This method is sequentially invoked by audio data provider to supply implementer (consumer) + * with the audio data. Exact audio format (encoding, sampling rate, etc.) depends on the usage + * context + * + * @param rawBytesChunk binary data in the depending on the use case format + * @param isLast true if this call logically concludes previous and this passed bytes data into a + * single logical entity (e.g. gets called at the end when all byte parts of a single message + * get passed) + */ + void outputAudio(byte[] rawBytesChunk, boolean isLast); +} diff --git a/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/BufferedWebSocketListener.java b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/BufferedWebSocketListener.java new file mode 100644 index 000000000..1eb449229 --- /dev/null +++ b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/BufferedWebSocketListener.java @@ -0,0 +1,54 @@ +package com.sap.ai.sdk.foundationmodels.openai; + +import java.net.http.WebSocket; +import java.nio.ByteBuffer; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.function.BiConsumer; +import java.util.function.Consumer; +import lombok.extern.slf4j.Slf4j; + +@Slf4j +class BufferedWebSocketListener implements WebSocket.Listener { + + private final Consumer onOpen; + private final BiConsumer onText; + private final StringBuilder buffer; + + BufferedWebSocketListener( + Consumer onOpen, BiConsumer onText) { + this.onOpen = onOpen; + this.onText = onText; + this.buffer = new StringBuilder(1024 * 1024); + } + + @Override + public void onOpen(WebSocket webSocket) { + this.onOpen.accept(webSocket); + webSocket.request(1); + } + + @Override + public CompletionStage onText(WebSocket webSocket, CharSequence data, boolean isLast) { + buffer.append(data); + webSocket.request(1); + if (isLast) { + var completeMessage = buffer.toString(); + buffer.setLength(0); + this.onText.accept(webSocket, completeMessage); + } + + return CompletableFuture.completedStage(null); + } + + @Override + public void onError(WebSocket webSocket, Throwable error) { + log.error("Websocket error occurred during realtime communication", error); + } + + @Override + public CompletionStage onBinary(WebSocket webSocket, ByteBuffer data, boolean isLast) { + log.warn("Received unexpected binary bytes for WebSocket connection"); + return CompletableFuture.completedStage(null); + } +} diff --git a/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/OpenAiClient.java b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/OpenAiClient.java index a76c3d89f..f34eb1622 100644 --- a/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/OpenAiClient.java +++ b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/OpenAiClient.java @@ -76,6 +76,17 @@ public static OpenAiClient forModel(@Nonnull final OpenAiModel foundationModel) return client.withApiVersion(DEFAULT_API_VERSION); } + /** + * Creates and configures OpenAI Realtime API client + * + * @return created client + */ + @Beta + public static OpenAiRealtimeClient realtimeClient() { + var withResolvedDestination = OpenAiClient.forModel(OpenAiModel.GPT_REALTIME); + return new OpenAiRealtimeClient(withResolvedDestination.destination); + } + /** * Create a new OpenAI client targeting the specified API version. * diff --git a/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/OpenAiModel.java b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/OpenAiModel.java index 1926f56ce..62c435bcc 100644 --- a/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/OpenAiModel.java +++ b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/OpenAiModel.java @@ -121,7 +121,7 @@ public record OpenAiModel(@Nonnull String name, @Nullable String version) implem /** Azure OpenAI GPT-5-nano model */ public static final OpenAiModel GPT_5_NANO = new OpenAiModel("gpt-5-nano", null); - /** Azure OpenAI GPT-5-nano model */ + /** Azure OpenAI GPT-realtime model */ public static final OpenAiModel GPT_REALTIME = new OpenAiModel("gpt-realtime", null); /** Azure OpenAI GPT-5.2 model */ diff --git a/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/OpenAiRealtimeClient.java b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/OpenAiRealtimeClient.java new file mode 100644 index 000000000..4ccbbd12c --- /dev/null +++ b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/OpenAiRealtimeClient.java @@ -0,0 +1,113 @@ +package com.sap.ai.sdk.foundationmodels.openai; + +import com.google.common.annotations.Beta; +import com.sap.ai.sdk.core.RealtimeParam; +import com.sap.cloud.sdk.cloudplatform.connectivity.Destination; +import com.sap.cloud.sdk.cloudplatform.connectivity.Header; +import java.util.HashMap; +import javax.annotation.Nonnull; + +public class OpenAiRealtimeClient { + + private static final int PATH_BUFFER_SIZE = + 400; // existing URLs are ~120 symbols long, 400 has reasonable margin + + private final Destination destination; + + OpenAiRealtimeClient(Destination destination) { + this.destination = destination; + } + + /** + * Creates realtime channel allowing to input text and voice it (receive audio output) + * + *

The input channel should be used with a try-with-resources block to ensure that the + * underlying connection is closed. + * + *

Example: + * + * 

{@code
+   * try (var textInputChannel = client.textToSpeech(audioOutputConsumer)) {
+   *       textInputChannel.sendText("...");
+   *       ....
+   * }
+   * }
+ * + * This API implements full duplex (input + output) communication channels. Application should + * logically synchronize their state and close input channel when it is appropriate (e.g. last + * part of the response has been received via output channel and application does not need to send + * any other input). When input channel is closed, output channel will be closed automatically and + * output consumer will not be called anymore. + * + * @param audioOutputConsumer - audio consumer of raw PCM mono 24000 Hz little endian output, 16 + * bit depth + * @return input channel, allowing for text input + */ + @Nonnull + @Beta + public TextInputChannel textToSpeech( + @Nonnull final AudioOutputChannel audioOutputConsumer, RealtimeParam... params) { + var extraHeaders = destination.asHttp().getHeaders(); + var headers = new HashMap(extraHeaders.size() + 1); + for (Header header : extraHeaders) { + headers.put(header.getName(), header.getValue()); + } + var endpoint = getRealtimeEndpoint(); + + return new TextToSpeechRealtimeClient(endpoint, headers, audioOutputConsumer, params); + } + + /** + * Creates realtime channel allowing for audio conversation with a model + * + *

The input channel should be used with a try-with-resources block to ensure that the + * underlying connection is closed. + * + *

Example: + * + * 

{@code
+   * try (var audioInputChannel = client.speechToSpeech(audioOutputConsumer)) {
+   *       audioInputChannel.inputAudio(audioBytesData);
+   *       ....
+   * }
+   * }
+ * + * This API implements full duplex (input + output) communication channels. Application should + * logically synchronize their state and close input channel when it is appropriate (e.g. last + * part of the response has been received via output channel and application does not need to send + * any other input). When input channel is closed, output channel will be closed automatically and + * output consumer will not be called anymore. + * + * @param audioOutputConsumer - audio consumer of raw PCM mono 24000 Hz little endian output, 16 + * bit depth + * @param params - optional configuration params + * @return input channel, allowing for audio data input (bytes, PCM mono 24000 Hz little endian 16 + * bit) + */ + @Nonnull + @Beta + public AudioInputChannel speechToSpeech( + @Nonnull final AudioOutputChannel audioOutputConsumer, RealtimeParam... params) { + var extraHeaders = destination.asHttp().getHeaders(); + var headers = new HashMap(extraHeaders.size() + 1); + for (Header header : extraHeaders) { + headers.put(header.getName(), header.getValue()); + } + var endpoint = getRealtimeEndpoint(); + + return new SpeechToSpeechRealtimeClient(endpoint, headers, audioOutputConsumer, params); + } + + private String getRealtimeEndpoint() { + var sb = new StringBuilder(PATH_BUFFER_SIZE); + sb.append("wss://"); + var pathParts = destination.asHttp().getUri().toString().split("//"); + if (pathParts.length != 2) { + throw new IllegalArgumentException( + "Invalid destination URI: " + destination.asHttp().getUri()); + } + sb.append(pathParts[1].replaceFirst("^api\\.", "realtime.")); + sb.append("v1/realtime"); + return sb.toString(); + } +} diff --git a/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/SpeechToSpeechRealtimeClient.java b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/SpeechToSpeechRealtimeClient.java new file mode 100644 index 000000000..9de369923 --- /dev/null +++ b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/SpeechToSpeechRealtimeClient.java @@ -0,0 +1,92 @@ +package com.sap.ai.sdk.foundationmodels.openai; + +import com.openai.models.realtime.InputAudioBufferAppendEvent; +import com.openai.models.realtime.InputAudioBufferCommitEvent; +import com.openai.models.realtime.RealtimeAudioConfigInput; +import com.openai.models.realtime.RealtimeAudioFormats; +import com.openai.models.realtime.RealtimeAudioInputTurnDetection; +import com.sap.ai.sdk.core.RealtimeParam; +import com.sap.ai.sdk.core.RealtimeParamTurnDetection; +import java.util.Arrays; +import java.util.Base64; +import java.util.Map; +import lombok.extern.slf4j.Slf4j; + +@Slf4j +class SpeechToSpeechRealtimeClient extends ToAudioRealtimeClient implements AudioInputChannel { + + private static final int MAX_DATA_CHUNK_SIZE_BYTES = 8192; + + private static final String TASK = ""; + + private final boolean eagerTurnDetection; + + public SpeechToSpeechRealtimeClient( + String url, + Map httpHeaders, + AudioOutputChannel outputConsumer, + RealtimeParam... params) { + super(url, httpHeaders, outputConsumer, params); + + var turnDetectionEager = false; + for (RealtimeParam param : params) { + if (param.getParamName() == RealtimeParam.SpeechOutputParamName.TURN_DETECTION) { + if (RealtimeParamTurnDetection.EACH_CALL_IS_A_TURN.equals(param)) { + turnDetectionEager = true; + } else if (RealtimeParamTurnDetection.BY_MODEL_AUTO.equals(param)) { + turnDetectionEager = false; + } + } + } + this.eagerTurnDetection = turnDetectionEager; + } + + @Override + protected RealtimeAudioConfigInput inputConfig() { + RealtimeAudioInputTurnDetection turnDetection; + if (eagerTurnDetection) { + turnDetection = null; + } else { + turnDetection = + RealtimeAudioInputTurnDetection.ofSemanticVad( + RealtimeAudioInputTurnDetection.SemanticVad.builder().build()); + } + + return RealtimeAudioConfigInput.builder() + .turnDetection(turnDetection) + .format( + RealtimeAudioFormats.AudioPcm.builder() + .type(RealtimeAudioFormats.AudioPcm.Type.AUDIO_PCM) + .rate(RealtimeAudioFormats.AudioPcm.Rate._24000) + .build()) + .build(); + } + + public void inputAudio(byte[] rawAudioChunk) { + if (rawAudioChunk.length == 0) { + return; + } + var cursorLeft = 0; + while (cursorLeft < rawAudioChunk.length) { + var cursorRight = Math.min(cursorLeft + MAX_DATA_CHUNK_SIZE_BYTES, rawAudioChunk.length); + var part = Arrays.copyOfRange(rawAudioChunk, cursorLeft, cursorRight); + var audioInputMessage = + InputAudioBufferAppendEvent.builder() + .audio(Base64.getEncoder().encodeToString(part)) + .build(); + super.sendMessage(audioInputMessage); + cursorLeft += MAX_DATA_CHUNK_SIZE_BYTES; + } + + if (eagerTurnDetection) { + var commitAudioMessage = InputAudioBufferCommitEvent.builder().build(); + super.sendMessage(commitAudioMessage); + askForResponse(); + } + } + + @Override + protected String getSystemPrompt() { + return TASK; + } +} diff --git a/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/TextInputChannel.java b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/TextInputChannel.java new file mode 100644 index 000000000..e1cfe32db --- /dev/null +++ b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/TextInputChannel.java @@ -0,0 +1,5 @@ +package com.sap.ai.sdk.foundationmodels.openai; + +public interface TextInputChannel extends AutoCloseable { + void sendText(String text); +} diff --git a/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/TextOutputChannel.java b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/TextOutputChannel.java new file mode 100644 index 000000000..add9c2460 --- /dev/null +++ b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/TextOutputChannel.java @@ -0,0 +1,16 @@ +package com.sap.ai.sdk.foundationmodels.openai; + +/** Functional interface representing text output channel (text data consumer) */ +public interface TextOutputChannel { + + /** + * This method is sequentially invoked by text data provider to supply implementer (consumer) with + * the text data. + * + * @param textChunk chunk of the text (possibly partial content) + * @param isLast true if this call logically concludes previous and this passed text data into a + * single logical entity (e.g. gets called at the end when all parts of a single message get + * passed) + */ + void outputText(CharSequence textChunk, boolean isLast); +} diff --git a/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/TextToSpeechRealtimeClient.java b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/TextToSpeechRealtimeClient.java new file mode 100644 index 000000000..6bde180ff --- /dev/null +++ b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/TextToSpeechRealtimeClient.java @@ -0,0 +1,73 @@ +package com.sap.ai.sdk.foundationmodels.openai; + +import com.openai.models.realtime.*; +import com.sap.ai.sdk.core.RealtimeParam; +import com.sap.ai.sdk.core.RealtimeParamTurnDetection; +import java.util.*; +import lombok.extern.slf4j.Slf4j; + +@Slf4j +class TextToSpeechRealtimeClient extends ToAudioRealtimeClient implements TextInputChannel { + + private static final String TASK = + "you are a speaker and your role is to read (produce audio) of the user input speech. voice user text input, " + + "do not answer questions, just read them"; + + private final boolean eagerTurnDetection; + + public TextToSpeechRealtimeClient( + String url, + Map httpHeaders, + AudioOutputChannel outputConsumer, + RealtimeParam... params) { + super(url, httpHeaders, outputConsumer, params); + var turnDetectionEager = true; + for (RealtimeParam param : params) { + if (param.getParamName() == RealtimeParam.SpeechOutputParamName.TURN_DETECTION) { + if (RealtimeParamTurnDetection.EACH_CALL_IS_A_TURN.equals(param)) { + turnDetectionEager = true; + } else if (RealtimeParamTurnDetection.BY_MODEL_AUTO.equals(param)) { + turnDetectionEager = false; + } + } + } + this.eagerTurnDetection = turnDetectionEager; + } + + @Override + protected RealtimeAudioConfigInput inputConfig() { + return RealtimeAudioConfigInput.builder() + .turnDetection(Optional.empty()) + .format( + RealtimeAudioFormats.AudioPcm.builder() + .type(RealtimeAudioFormats.AudioPcm.Type.AUDIO_PCM) + .rate(RealtimeAudioFormats.AudioPcm.Rate._24000) + .build()) + .build(); + } + + public void sendText(String text) { + var message = + ConversationItemCreateEvent.builder() + .item( + ConversationItem.ofRealtimeConversationItemUserMessage( + RealtimeConversationItemUserMessage.builder() + .addContent( + RealtimeConversationItemUserMessage.Content.builder() + .text(text) + .type(RealtimeConversationItemUserMessage.Content.Type.INPUT_TEXT) + .build()) + .build())) + .build(); + + super.sendMessage(message); + if (eagerTurnDetection) { + askForResponse(); + } + } + + @Override + protected String getSystemPrompt() { + return TASK; + } +} diff --git a/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/ToAudioRealtimeClient.java b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/ToAudioRealtimeClient.java new file mode 100644 index 000000000..7279ba465 --- /dev/null +++ b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/ToAudioRealtimeClient.java @@ -0,0 +1,92 @@ +package com.sap.ai.sdk.foundationmodels.openai; + +import com.fasterxml.jackson.databind.JsonNode; +import com.openai.models.realtime.RealtimeAudioConfig; +import com.openai.models.realtime.RealtimeAudioConfigInput; +import com.openai.models.realtime.RealtimeAudioConfigOutput; +import com.openai.models.realtime.RealtimeAudioFormats; +import com.openai.models.realtime.RealtimeSessionCreateRequest; +import com.openai.models.realtime.SessionUpdateEvent; +import com.openai.models.realtime.clientsecrets.ClientSecretCreateParams; +import com.sap.ai.sdk.core.RealtimeParam; +import com.sap.ai.sdk.core.RealtimeParamVoice; +import java.util.Base64; +import java.util.List; +import java.util.Map; +import java.util.Set; +import lombok.extern.slf4j.Slf4j; + +@Slf4j +abstract class ToAudioRealtimeClient extends WSOpenAiRealtimeClient { + + private static final Set HANDLED_RESPONSE_TYPES = + Set.of("response.output_audio.delta", "response.output_audio.done"); + private static final List OUTPUT_MODALITIES = + List.of(RealtimeSessionCreateRequest.OutputModality.AUDIO); + private static final byte[] EMPTY_BYTE_ARRAY = new byte[0]; + + private final AudioOutputChannel outputConsumer; + private final RealtimeAudioConfigOutput.Voice.UnionMember1 voice; + + public ToAudioRealtimeClient( + String url, + Map httpHeaders, + AudioOutputChannel outputConsumer, + RealtimeParam... params) { + super(url, httpHeaders, HANDLED_RESPONSE_TYPES); + var voice = RealtimeAudioConfigOutput.Voice.UnionMember1.MARIN; + for (RealtimeParam param : params) { + switch (param.getParamName()) { + case VOICE -> { + if (RealtimeParamVoice.DEFAULT_2.equals(param)) { + voice = RealtimeAudioConfigOutput.Voice.UnionMember1.MARIN; + } else if (RealtimeParamVoice.DEFAULT_1.equals(param)) { + voice = RealtimeAudioConfigOutput.Voice.UnionMember1.ECHO; + } + } + } + } + this.outputConsumer = outputConsumer; + this.voice = voice; + } + + protected abstract RealtimeAudioConfigInput inputConfig(); + + @Override + protected void onResponse(String eventType, JsonNode event) { + if ("response.output_audio.delta".equals(eventType)) { + var base64Audio = event.get("delta").asText(); + byte[] audio = Base64.getDecoder().decode(base64Audio); + this.outputConsumer.outputAudio(audio, Boolean.FALSE); + } else if ("response.output_audio.done".equals(eventType)) { + this.outputConsumer.outputAudio(EMPTY_BYTE_ARRAY, Boolean.TRUE); + } else { + log.warn("skipping message type: {}", eventType); + } + } + + @Override + protected SessionUpdateEvent sessionConfiguration() { + return SessionUpdateEvent.builder() + .session( + ClientSecretCreateParams.Session.ofRealtime( + RealtimeSessionCreateRequest.builder() + .outputModalities(OUTPUT_MODALITIES) + .audio( + RealtimeAudioConfig.builder() + .input(inputConfig()) + .output( + RealtimeAudioConfigOutput.builder() + .format( + RealtimeAudioFormats.AudioPcm.builder() + .type(RealtimeAudioFormats.AudioPcm.Type.AUDIO_PCM) + .rate(RealtimeAudioFormats.AudioPcm.Rate._24000) + .build()) + .voice(voice) + .build()) + .build()) + .build()) + .asRealtime()) + .build(); + } +} diff --git a/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/WSOpenAiRealtimeClient.java b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/WSOpenAiRealtimeClient.java new file mode 100644 index 000000000..90cf63bbb --- /dev/null +++ b/foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/WSOpenAiRealtimeClient.java @@ -0,0 +1,210 @@ +package com.sap.ai.sdk.foundationmodels.openai; + +import com.fasterxml.jackson.annotation.JsonAutoDetect; +import com.fasterxml.jackson.annotation.PropertyAccessor; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.openai.models.realtime.ConversationItem; +import com.openai.models.realtime.ConversationItemCreateEvent; +import com.openai.models.realtime.RealtimeConversationItemSystemMessage; +import com.openai.models.realtime.ResponseCreateEvent; +import com.openai.models.realtime.SessionUpdateEvent; +import com.sap.ai.sdk.core.common.ClientException; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.WebSocket; +import java.nio.ByteBuffer; +import java.util.Map; +import java.util.Set; +import java.util.Timer; +import java.util.TimerTask; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import lombok.AccessLevel; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +@Slf4j +@RequiredArgsConstructor(access = AccessLevel.PACKAGE) +abstract class WSOpenAiRealtimeClient implements AutoCloseable { + + private static final int SUCCESS_FINISH_WSS_CODE = 1000; + private static final ObjectMapper JACKSON = + new ObjectMapper().setVisibility(PropertyAccessor.IS_GETTER, JsonAutoDetect.Visibility.NONE); + + /** + * WebSocket keeps TCP connection to the server. If connection idles, depending on the provider in + * cloud environments, cloud provider sends hard TCP RST (reset) after 60-300 seconds of + * inactivity, this is normally not configurable and would leave realtime conversation unusable on + * a pause. + * + *

In mobile networks this period is even shorter and typical CGNAT session can only idle for + * 10-15 seconds before it gets terminated + * + *

In realtime client we rely on explicit lifetime management (create/close), heartbeat + * mechanism is required to prevent unintended closing of connections while conversation is still + * expected to continue + */ + private static final int HEARTBEAT_INTERVAL_MILLIS = 4500; + + private static final String HEARTBEAT_TIMER_NAME = "wss_realtime_heartbeat"; + + private final HttpClient client; + private final CompletableFuture ws; + private final Timer heartbeatTimer; + private final Set handleMessageTypes; + + public WSOpenAiRealtimeClient( + String url, Map httpHeaders, Set handleMessageTypes) { + this.client = HttpClient.newHttpClient(); + var wsBuilder = this.client.newWebSocketBuilder(); + for (Map.Entry entry : httpHeaders.entrySet()) { + wsBuilder = wsBuilder.header(entry.getKey(), entry.getValue()); + } + this.ws = + wsBuilder.buildAsync( + URI.create(url), new BufferedWebSocketListener(this::onSocketOpen, this::onText)); + this.handleMessageTypes = handleMessageTypes; + this.heartbeatTimer = new Timer(HEARTBEAT_TIMER_NAME, true); + } + + public void askForResponse() { + WebSocket ws; + try { + ws = this.ws.join(); + } catch (CompletionException e) { + throw new ClientException("failed to establish web socket connection", e); + } + synchronized (this) { + try { + ws.sendText(JACKSON.writeValueAsString(ResponseCreateEvent.builder().build()), true); + } catch (JsonProcessingException e) { + throw new RuntimeException("Failed to serialize ask for response", e); + } + ws.request(1); + } + } + + @Override + public void close() { + WebSocket ws; + try { + ws = this.ws.join(); + } catch (CompletionException e) { + throw new ClientException("failed to establish web socket connection", e); + } + synchronized (this) { + heartbeatTimer.cancel(); + try { + ws.sendClose(SUCCESS_FINISH_WSS_CODE, "done").join(); + } catch (Exception e) { + log.error("Error while closing WebSocket", e); + } + // this.client.close(); // exists only since java 21 + } + } + + protected abstract String getSystemPrompt(); + + protected abstract void onResponse(String eventType, JsonNode event); + + protected abstract SessionUpdateEvent sessionConfiguration(); + + protected void sendMessage(Object message) { + WebSocket ws; + try { + ws = this.ws.join(); + } catch (CompletionException e) { + throw new ClientException("failed to establish web socket connection", e); + } + synchronized (this) { + try { + ws.sendText(JACKSON.writeValueAsString(message), true); + ws.request(1); + } catch (JsonProcessingException e) { + throw new ClientException("Failed to serialize message", e); + } + } + } + + private synchronized void onSocketOpen(WebSocket ws) { + configureSession(ws); + configureConversation(ws); + scheduleHeartbeat(ws); + } + + private void onText(WebSocket webSocket, CharSequence data) { + final JsonNode event; + try { + event = JACKSON.readTree(data.toString()); + } catch (JsonProcessingException e) { + throw new ClientException("Error parsing JSON response from speech API", e); + } + var eventType = event.get("type").asText(); + if (handleMessageTypes.contains(eventType)) { + onResponse(eventType, event); + } else { + log.trace("Unhandled event type: {}", eventType); + } + + webSocket.request(1); + } + + private synchronized void sendPing(WebSocket ws) { + if (ws.isInputClosed()) { + return; + } + ws.sendPing(ByteBuffer.wrap("ping".getBytes())).join(); + ws.request(1); + } + + private void configureSession(WebSocket ws) { + SessionUpdateEvent sue = sessionConfiguration(); + try { + ws.sendText(JACKSON.writeValueAsString(sue), true); + } catch (JsonProcessingException e) { + throw new ClientException("Failed to serialize session request", e); + } + + ws.request(1); + } + + private void configureConversation(WebSocket ws) { + var systemPrompt = getSystemPrompt(); + if (systemPrompt == null || systemPrompt.isEmpty()) { + return; + } + var systemConversationItem = + ConversationItemCreateEvent.builder() + .item( + ConversationItem.ofRealtimeConversationItemSystemMessage( + RealtimeConversationItemSystemMessage.builder() + .addContent( + RealtimeConversationItemSystemMessage.Content.builder() + .text(systemPrompt) + .type(RealtimeConversationItemSystemMessage.Content.Type.INPUT_TEXT) + .build()) + .build())) + .build(); + + try { + var json = JACKSON.writeValueAsString(systemConversationItem); + ws.sendText(json, true); + } catch (JsonProcessingException e) { + throw new ClientException("Failed to serialize message", e); + } + ws.request(1); + } + + private void scheduleHeartbeat(WebSocket ws) { + TimerTask task = + new TimerTask() { + @Override + public void run() { + sendPing(ws); + } + }; + heartbeatTimer.scheduleAtFixedRate(task, 0, HEARTBEAT_INTERVAL_MILLIS); + } +} diff --git a/pom.xml b/pom.xml index dbe2342a5..ec3bb4f6d 100644 --- a/pom.xml +++ b/pom.xml @@ -66,6 +66,7 @@ 2.1.3 3.5.6 1.1.8 + 4.41.0 3.8.6 3.2.0 5.23.0 @@ -119,6 +120,11 @@ pom import + + com.openai + openai-java + ${com.openai.openai-java.version} + io.projectreactor @@ -274,11 +280,6 @@ openai-java-core ${openai-java.version} - - com.openai - openai-java - ${openai-java.version} - diff --git a/sample-code/spring-app/pom.xml b/sample-code/spring-app/pom.xml index 78c5b57ac..75f32a353 100644 --- a/sample-code/spring-app/pom.xml +++ b/sample-code/spring-app/pom.xml @@ -139,6 +139,10 @@ org.springframework.ai spring-ai-commons + + org.springframework.boot + spring-boot-starter-websocket + org.springframework.ai spring-ai-model diff --git a/sample-code/spring-app/src/main/java/com/sap/ai/sdk/app/WebsocketConfig.java b/sample-code/spring-app/src/main/java/com/sap/ai/sdk/app/WebsocketConfig.java new file mode 100644 index 000000000..9273342c8 --- /dev/null +++ b/sample-code/spring-app/src/main/java/com/sap/ai/sdk/app/WebsocketConfig.java @@ -0,0 +1,28 @@ +package com.sap.ai.sdk.app; + +import com.sap.ai.sdk.app.realtime.SpeechToSpeechWebsocketHandler; +import com.sap.ai.sdk.app.realtime.TextToSpeechWebsocketHandler; +import org.springframework.context.annotation.Configuration; +import org.springframework.web.socket.config.annotation.EnableWebSocket; +import org.springframework.web.socket.config.annotation.WebSocketConfigurer; +import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry; + +@Configuration +@EnableWebSocket +public class WebsocketConfig implements WebSocketConfigurer { + + private final TextToSpeechWebsocketHandler textToSpeech; + private final SpeechToSpeechWebsocketHandler speechToSpeech; + + public WebsocketConfig( + TextToSpeechWebsocketHandler textToSpeech, SpeechToSpeechWebsocketHandler speechToSpeech) { + this.textToSpeech = textToSpeech; + this.speechToSpeech = speechToSpeech; + } + + @Override + public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { + registry.addHandler(textToSpeech, "/text-to-speech"); + registry.addHandler(speechToSpeech, "/speech-to-speech"); + } +} diff --git a/sample-code/spring-app/src/main/java/com/sap/ai/sdk/app/realtime/SpeechToSpeechWebsocketHandler.java b/sample-code/spring-app/src/main/java/com/sap/ai/sdk/app/realtime/SpeechToSpeechWebsocketHandler.java new file mode 100644 index 000000000..64911e08b --- /dev/null +++ b/sample-code/spring-app/src/main/java/com/sap/ai/sdk/app/realtime/SpeechToSpeechWebsocketHandler.java @@ -0,0 +1,65 @@ +package com.sap.ai.sdk.app.realtime; + +import com.sap.ai.sdk.app.services.OpenAiService; +import com.sap.ai.sdk.foundationmodels.openai.AudioInputChannel; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import javax.annotation.Nonnull; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; +import org.springframework.web.socket.BinaryMessage; +import org.springframework.web.socket.CloseStatus; +import org.springframework.web.socket.WebSocketSession; +import org.springframework.web.socket.handler.BinaryWebSocketHandler; + +@Component +@Slf4j +public class SpeechToSpeechWebsocketHandler extends BinaryWebSocketHandler { + + private final OpenAiService service; + private final Map channels; + + @Autowired + public SpeechToSpeechWebsocketHandler(OpenAiService service) { + this.service = service; + channels = new ConcurrentHashMap<>(); + } + + @Override + protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) { + ByteBuffer payload = message.getPayload(); + byte[] chunkBytes = payload.array(); + AudioInputChannel channel = + channels.computeIfAbsent( + session.getId(), + sessionId -> + service.speechToSpeech( + (rawBytesChunk, isLast) -> { + try { + session.sendMessage(new BinaryMessage(rawBytesChunk, isLast)); + } catch (IOException e) { + log.error("failed to send audio data to realtime api", e); + } + })); + channel.inputAudio(chunkBytes); + } + + @Override + public void afterConnectionClosed(WebSocketSession session, @Nonnull CloseStatus status) + throws Exception { + channels.computeIfPresent( + session.getId(), + (sessionId, inputChannel) -> { + try { + inputChannel.close(); + } catch (Exception e) { + log.warn("failed to close input channel for session {}", sessionId, e); + } + return null; + }); + super.afterConnectionClosed(session, status); + } +} diff --git a/sample-code/spring-app/src/main/java/com/sap/ai/sdk/app/realtime/TextToSpeechWebsocketHandler.java b/sample-code/spring-app/src/main/java/com/sap/ai/sdk/app/realtime/TextToSpeechWebsocketHandler.java new file mode 100644 index 000000000..88ce6109b --- /dev/null +++ b/sample-code/spring-app/src/main/java/com/sap/ai/sdk/app/realtime/TextToSpeechWebsocketHandler.java @@ -0,0 +1,65 @@ +package com.sap.ai.sdk.app.realtime; + +import com.sap.ai.sdk.app.services.OpenAiService; +import com.sap.ai.sdk.foundationmodels.openai.TextInputChannel; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import javax.annotation.Nonnull; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; +import org.springframework.web.socket.BinaryMessage; +import org.springframework.web.socket.CloseStatus; +import org.springframework.web.socket.WebSocketSession; +import org.springframework.web.socket.handler.BinaryWebSocketHandler; + +@Component +@Slf4j +public class TextToSpeechWebsocketHandler extends BinaryWebSocketHandler { + + private final OpenAiService service; + private final Map channels; + + @Autowired + public TextToSpeechWebsocketHandler(OpenAiService service) { + this.service = service; + channels = new ConcurrentHashMap<>(); + } + + @Override + protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) { + ByteBuffer payload = message.getPayload(); + byte[] textBytes = payload.array(); + TextInputChannel channel = + channels.computeIfAbsent( + session.getId(), + sessionId -> + service.textToSpeech( + (rawBytesChunk, isLast) -> { + try { + session.sendMessage(new BinaryMessage(rawBytesChunk, isLast)); + } catch (IOException e) { + log.error("failed to send text message to realtime api", e); + } + })); + channel.sendText(new String(textBytes)); + } + + @Override + public void afterConnectionClosed(WebSocketSession session, @Nonnull CloseStatus status) + throws Exception { + channels.computeIfPresent( + session.getId(), + (sessionId, inputChannel) -> { + try { + inputChannel.close(); + } catch (Exception e) { + log.warn("failed to close input channel for session {}", sessionId, e); + } + return null; + }); + super.afterConnectionClosed(session, status); + } +} diff --git a/sample-code/spring-app/src/main/java/com/sap/ai/sdk/app/services/OpenAiService.java b/sample-code/spring-app/src/main/java/com/sap/ai/sdk/app/services/OpenAiService.java index 261cf9a9c..e8c83a5f8 100644 --- a/sample-code/spring-app/src/main/java/com/sap/ai/sdk/app/services/OpenAiService.java +++ b/sample-code/spring-app/src/main/java/com/sap/ai/sdk/app/services/OpenAiService.java @@ -5,6 +5,8 @@ import static com.sap.ai.sdk.foundationmodels.openai.OpenAiModel.TEXT_EMBEDDING_3_SMALL; import com.sap.ai.sdk.core.AiCoreService; +import com.sap.ai.sdk.foundationmodels.openai.AudioInputChannel; +import com.sap.ai.sdk.foundationmodels.openai.AudioOutputChannel; import com.sap.ai.sdk.foundationmodels.openai.OpenAiChatCompletionDelta; import com.sap.ai.sdk.foundationmodels.openai.OpenAiChatCompletionRequest; import com.sap.ai.sdk.foundationmodels.openai.OpenAiChatCompletionResponse; @@ -14,6 +16,7 @@ import com.sap.ai.sdk.foundationmodels.openai.OpenAiImageItem; import com.sap.ai.sdk.foundationmodels.openai.OpenAiMessage; import com.sap.ai.sdk.foundationmodels.openai.OpenAiTool; +import com.sap.ai.sdk.foundationmodels.openai.TextInputChannel; import java.util.ArrayList; import java.util.List; import java.util.stream.Stream; @@ -74,6 +77,65 @@ public Stream streamChatCompletionDeltas( return OpenAiClient.forModel(GPT_5_MINI).streamChatCompletionDeltas(request); } + /** + * Creates realtime channel allowing to input text and voice it (receive audio output) + * + *

The input channel should be used with a try-with-resources block to ensure that the + * underlying connection is closed. + * + *

Example: + * + * 

{@code
+   * try (var textInputChannel = client.textToSpeech(audioOutputConsumer)) {
+   *       textInputChannel.sendText("...");
+   *       ....
+   * }
+   * }
+ * + * This API implements full duplex (input + output) communication channels. Application should + * logically synchronize their state and close input channel when it is appropriate (e.g. last + * part of the response has been received via output channel and application does not need to send + * any other input). When input channel is closed, output channel will be closed automatically and + * output consumer will not be called anymore. + * + * @param audioOutputConsumer - audio consumer of raw PCM mono 24000 Hz little endian output + * @return input channel, allowing for text input + */ + @Nonnull + public TextInputChannel textToSpeech(@Nonnull final AudioOutputChannel audioOutputConsumer) { + return OpenAiClient.realtimeClient().textToSpeech(audioOutputConsumer); + } + + /** + * Creates realtime channel allowing for audio conversation with a model + * + *

The input channel should be used with a try-with-resources block to ensure that the + * underlying connection is closed. + * + *

Example: + * + * 

{@code
+   * try (var audioInputChannel = client.speechToSpeech(audioOutputConsumer)) {
+   *       audioInputChannel.inputAudio(audioBytesData);
+   *       ....
+   * }
+   * }
+ * + * This API implements full duplex (input + output) communication channels. Application should + * logically synchronize their state and close input channel when it is appropriate (e.g. last + * part of the response has been received via output channel and application does not need to send + * any other input). When input channel is closed, output channel will be closed automatically and + * output consumer will not be called anymore. + * + * @param audioOutputConsumer - audio consumer of raw PCM mono 24000 Hz little endian output, 16 + * bit depth + * @return input channel, allowing for audio data input (bytes, PCM mono 24000 Hz little endian 16 + * bit) + */ + public AudioInputChannel speechToSpeech(@Nonnull final AudioOutputChannel audioOutputConsumer) { + return OpenAiClient.realtimeClient().speechToSpeech(audioOutputConsumer); + } + /** * Asynchronous stream of an OpenAI chat request * diff --git a/sample-code/spring-app/src/main/resources/static/index.html b/sample-code/spring-app/src/main/resources/static/index.html index be8accffe..014fddcf1 100644 --- a/sample-code/spring-app/src/main/resources/static/index.html +++ b/sample-code/spring-app/src/main/resources/static/index.html @@ -1445,6 +1445,46 @@

📂 Batch API

+ +
+
+
+
+

⏰ Realtime API

+
+ Realtime API allows for various real time interactions TBD Documentation +
+
+
    +
  • +
    +

    Text to speech

    + disabled +
    + + + +
    +
    + +
    • +
  • +
    +

    Speech to speech

    +
    disconnected
    +
    disabled
    +
    + +
    +
    + +
    • +
+
+
+
diff --git a/sample-code/spring-app/src/main/resources/static/speech-to-speech.js b/sample-code/spring-app/src/main/resources/static/speech-to-speech.js new file mode 100644 index 000000000..71d224a9c --- /dev/null +++ b/sample-code/spring-app/src/main/resources/static/speech-to-speech.js @@ -0,0 +1,128 @@ +const SPEECH_TO_SPEECH_URL = 'ws://localhost:8080/speech-to-speech'; +const SPEECH_TO_SPEECH_SAMPLE_RATE = 24000; + +const wsStatusEl = document.getElementById('speech-to-speech-websocket-status') +const micStatusEl = document.getElementById('speech-to-speech-mic-status') +const speechBtn = document.getElementById('speech-to-speech-btn') + +let sts_ws = null; +let sts_audioCtx = null; +let sts_mediaStream = null; +let sts_workletNode = null; +let sts_nextStartTime = 0; +let sts_started = false; + +async function startSession() { + sts_ws = new WebSocket(SPEECH_TO_SPEECH_URL); + sts_ws.binaryType = 'arraybuffer'; + + sts_ws.onopen = async () => { + wsStatusEl.textContent = `connected to ${SPEECH_TO_SPEECH_URL}`; + wsStatusEl.style.color = 'green'; + + await startMicrophone(); + speechBtn.innerText = 'Stop'; + sts_started = true; + } + + sts_ws.onmessage = (event) => { + if (event.data instanceof ArrayBuffer) { + sts_playPcmAudio(event.data) + } + } + + sts_ws.onclose = () => stopSession(); +} + +async function startMicrophone() { + try { + sts_mediaStream = await navigator.mediaDevices.getUserMedia({audio: true}); + console.log("mediaStream is: ", sts_mediaStream) + micStatusEl.textContent = 'active'; + micStatusEl.style.color = 'green'; + + sts_audioCtx = new (window.AudioContext || window.webkitAudioContext)({sampleRate: SPEECH_TO_SPEECH_SAMPLE_RATE}); + + const workletCode = ` + class MicProcessor extends AudioWorkletProcessor { + process(inputs) { + const input = inputs[0]; + if (input && input[0]) { + const float32Input = input[0]; + const int16Buffer = new Int16Array(float32Input.length); + for (let i = 0; i < float32Input.length; i++) { + const s = Math.max(-1, Math.min(1, float32Input[i])); + int16Buffer[i] = s < 0 ? s * 0x8000 : s * 0x7FFF; + } + this.port.postMessage(int16Buffer.buffer, [int16Buffer.buffer]); + } + return true; + } + } + registerProcessor('mic-processor', MicProcessor); + `; + const blob = new Blob([workletCode], {type: 'application/javascript'}); + const workletUrl = URL.createObjectURL(blob); + await sts_audioCtx.audioWorklet.addModule(workletUrl); + URL.revokeObjectURL(workletUrl); + + const source = sts_audioCtx.createMediaStreamSource(sts_mediaStream); + sts_workletNode = new AudioWorkletNode(sts_audioCtx, 'mic-processor'); + sts_workletNode.port.onmessage = (e) => { + if (sts_ws && sts_ws.readyState === WebSocket.OPEN) { + sts_ws.send(e.data); + } + }; + + source.connect(sts_workletNode); + + } catch (err) { + console.error('failed to find or bind microphone: ', err); + micStatusEl.innerText = 'mic binding error'; + micStatusEl.style.color = 'red'; + } +} + +function sts_playPcmAudio(arrayBuffer) { + if (!sts_audioCtx) return; + + const int16Data = new Int16Array(arrayBuffer); + const float32Data = new Float32Array(int16Data.length); + for (let i = 0; i < int16Data.length; i++) { + float32Data[i] = int16Data[i] / 32768.0; + } + + const audioBuffer = sts_audioCtx.createBuffer(1, float32Data.length, SPEECH_TO_SPEECH_SAMPLE_RATE); + audioBuffer.copyToChannel(float32Data, 0); + const source = sts_audioCtx.createBufferSource(); + source.buffer = audioBuffer; + source.connect(sts_audioCtx.destination); + + if (sts_nextStartTime < sts_audioCtx.currentTime) { + sts_nextStartTime = sts_audioCtx.currentTime; + } + + source.start(sts_nextStartTime); + sts_nextStartTime += audioBuffer.duration; +} + +function stopSession() { + if (sts_ws) sts_ws.close(); + if (sts_mediaStream) sts_mediaStream.getTracks().forEach(track => track.stop()); + if (sts_workletNode) sts_workletNode.disconnect(); + if (sts_audioCtx) { + sts_audioCtx.close(); + sts_audioCtx = null; + } + + wsStatusEl.textContent = 'disconnected'; + wsStatusEl.style.color = 'red'; + micStatusEl.textContent = 'disabled'; + micStatusEl.style.color = 'red'; + speechBtn.innerText = 'Start'; + sts_started = false; +} + +speechBtn.addEventListener('click', () => { + sts_started ? stopSession() : startSession(); +}) \ No newline at end of file diff --git a/sample-code/spring-app/src/main/resources/static/text-to-speech.js b/sample-code/spring-app/src/main/resources/static/text-to-speech.js new file mode 100644 index 000000000..80bd75841 --- /dev/null +++ b/sample-code/spring-app/src/main/resources/static/text-to-speech.js @@ -0,0 +1,77 @@ +const WS_URL = 'ws://localhost:8080/text-to-speech'; +const SAMPLE_RATE = 24000; + +const statusEl = document.getElementById('text-to-speech-status') +const textInput = document.getElementById('text-to-speech-input') +const sendBtn = document.getElementById('text-to-speech-send-btn') + +let audioCtx; +let nextStartTime = 0; + +let ws = new WebSocket(WS_URL); +ws.binaryType = 'arraybuffer'; + +ws.onopen = () => { + statusEl.textContent = `connected to ${WS_URL}`; + statusEl.style.color = 'green'; + textInput.disabled = false; + sendBtn.disabled = false; +}; + +ws.onclose = () => { + statusEl.textContent = 'disconnected'; + statusEl.style.color = 'red'; + textInput.disabled = true; + sendBtn.disabled = true; +} + +ws.onerror = (error) => { + console.error('text-to-speech WebSocket error:', error); + statusEl.textContent = 'connection error' + statusEl.style.color = 'red'; +} + +ws.onmessage = (event) => { + if (event.data instanceof ArrayBuffer) { + playPcmAudio(event.data); + } else { + console.log('text data received:', event.data) + } +} + +sendBtn.addEventListener('click', () => { + const text = textInput.value.trim(); + if (text && ws.readyState === WebSocket.OPEN) { + if (!audioCtx) { + audioCtx = new (window.AudioContext || window.webkitAudioContext)({ sampleRate: SAMPLE_RATE}); + } + + const encoder = new TextEncoder(); + const binaryData = encoder.encode(text); + ws.send(binaryData); + textInput.value = ''; + } +}) + +function playPcmAudio(arrayBuffer) { + if (!audioCtx) return; + + const int16Data = new Int16Array(arrayBuffer); + const float32Data = new Float32Array(int16Data.length); + for (let i = 0; i < int16Data.length; i++) { + float32Data[i] = int16Data[i] / 32768.0; + } + + const audioBuffer = audioCtx.createBuffer(1, float32Data.length, SAMPLE_RATE); + audioBuffer.copyToChannel(float32Data, 0); + const source = audioCtx.createBufferSource(); + source.buffer = audioBuffer; + source.connect(audioCtx.destination); + + if (nextStartTime < audioCtx.currentTime) { + nextStartTime = audioCtx.currentTime; + } + + source.start(nextStartTime); + nextStartTime += audioBuffer.duration; +} \ No newline at end of file diff --git a/sample-code/spring-app/src/test/java/com/sap/ai/sdk/app/controllers/RealtimeApiTest.java b/sample-code/spring-app/src/test/java/com/sap/ai/sdk/app/controllers/RealtimeApiTest.java new file mode 100644 index 000000000..10f6027e4 --- /dev/null +++ b/sample-code/spring-app/src/test/java/com/sap/ai/sdk/app/controllers/RealtimeApiTest.java @@ -0,0 +1,153 @@ +package com.sap.ai.sdk.app.controllers; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; + +import com.sap.ai.sdk.core.RealtimeParamTurnDetection; +import com.sap.ai.sdk.foundationmodels.openai.AudioInputChannel; +import com.sap.ai.sdk.foundationmodels.openai.OpenAiClient; +import com.sap.ai.sdk.foundationmodels.openai.TextInputChannel; +import java.io.ByteArrayOutputStream; +import java.io.FileInputStream; +import java.io.IOException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import lombok.extern.slf4j.Slf4j; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +@Slf4j +public class RealtimeApiTest { + + private static final double LOG_2 = Math.log(2.0); + + private static byte[] QUESTION_FIXTURE_PCM; + + @BeforeAll + public static void setUp() { + try (var fis = new FileInputStream("src/test/resources/fixtures/question.pcm")) { + QUESTION_FIXTURE_PCM = fis.readAllBytes(); + } catch (IOException e) { + fail(e.getMessage()); + } + } + + @Test + @Timeout(value = 30, unit = TimeUnit.SECONDS) + void testTextToSpeech() { + var outputBuffer = new ByteArrayOutputStream(300000); + var monitor = new CountDownLatch(1); + var client = OpenAiClient.realtimeClient(); + + try (TextInputChannel input = + client.textToSpeech( + (byteChunk, isLast) -> { + outputBuffer.writeBytes(byteChunk); + if (isLast) { + monitor.countDown(); + } + }, + RealtimeParamTurnDetection.EACH_CALL_IS_A_TURN)) { + input.sendText("Ordnung muss sein!"); + monitor.await(); + } catch (Exception e) { + if (!(e instanceof InterruptedException)) { + fail(e); + } + // do nothing, test has either been interrupted by user (intended) or by jupiter if timeout is + // reached + return; + } + assertThat(monitor.getCount()).isEqualTo(0); + assertThat(outputBuffer.size()).isGreaterThan(0); + + var metrics = pcm16AudioMetrics(outputBuffer.toByteArray()); + + // root mean squire (measures the deviation from an average value) + assertThat(metrics.rms).isGreaterThan(500d); + // asserts that variety of deviation is sufficient (not a trivial repeating pattern) + assertThat(metrics.entropy).isGreaterThan(4); + } + + @Test + @Timeout(value = 60, unit = TimeUnit.SECONDS) + void testSpeechToSpeech() { + var outputBuffer = new ByteArrayOutputStream(300000); + var monitor = new CountDownLatch(1); + var client = OpenAiClient.realtimeClient(); + + try (AudioInputChannel input = + client.speechToSpeech( + (byteChunk, isLast) -> { + outputBuffer.writeBytes(byteChunk); + if (isLast) { + monitor.countDown(); + } + }, + RealtimeParamTurnDetection.EACH_CALL_IS_A_TURN)) { + input.inputAudio(QUESTION_FIXTURE_PCM); + monitor.await(); + } catch (Exception e) { + if (!(e instanceof InterruptedException)) { + fail(e); + } + // do nothing, test has either been interrupted by user (intended) or by jupiter if timeout is + // reached + return; + } + + assertThat(monitor.getCount()).isEqualTo(0); + assertThat(outputBuffer.size()).isGreaterThan(0); + + var metrics = pcm16AudioMetrics(outputBuffer.toByteArray()); + + // root mean squire (measures the deviation from an average value) + assertThat(metrics.rms).isGreaterThan(500d); + // asserts that variety of deviation is sufficient (not a trivial repeating pattern) + assertThat(metrics.entropy).isGreaterThan(4); + } + + private record AudioMetrics(double rms, double entropy) {} + ; + + /** + * Audio quality metrics for signed 16-bit little-endian PCM, computed on the mean-subtracted + * (DC-offset-removed) signal so they reflect only the varying, audible part of the audio. + * + * @param pcm - The raw PCM audio buffer. + * @return The RMS amplitude and Shannon entropy (bits) of the zero-mean samples. + */ + AudioMetrics pcm16AudioMetrics(byte[] pcm) { + if (pcm.length < 1) { + return new AudioMetrics(0, 0); + } + var sum = 0L; + for (var i = 0; i < pcm.length; i += 2) { + sum += (pcm[i]) | ((pcm[i + 1]) << 8); + } + var mean = sum / (pcm.length / 2); + + var sumOfSquares = 0d; + var counts = new char[Character.MAX_VALUE]; + for (var i = 0; i < pcm.length; i += 2) { + var residual = ((pcm[i]) | ((pcm[i + 1]) << 8)) - mean; + sumOfSquares += residual * residual; + var key = (int) (residual + Short.MAX_VALUE + 1); + counts[key]++; + } + + var rms = Math.sqrt(sumOfSquares / (double) (pcm.length / 2)); + var entropy = 0d; + for (var i = 0; i < counts.length; i++) { + var p = counts[i] / counts.length; + entropy -= p * log2(p); + } + + return new AudioMetrics(rms, entropy); + } + + private static double log2(double logNumber) { + return Math.log(logNumber) / LOG_2; + } +} diff --git a/sample-code/spring-app/src/test/resources/fixtures/question.pcm b/sample-code/spring-app/src/test/resources/fixtures/question.pcm new file mode 100644 index 000000000..d0707295b Binary files /dev/null and b/sample-code/spring-app/src/test/resources/fixtures/question.pcm differ