DRAFT: Feat/poc OpenAI realtime api

foundation-models/openai/pom.xml

@@ -66,6 +66,10 @@
       <groupId>com.sap.ai.sdk</groupId>
       <artifactId>core</artifactId>
     </dependency>
+    <dependency>
+      <groupId>com.openai</groupId>
+      <artifactId>openai-java</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.google.code.findbugs</groupId>
       <artifactId>jsr305</artifactId>

foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/AudioInputChannel.java

@@ -0,0 +1,17 @@
+package com.sap.ai.sdk.foundationmodels.openai;
+
+/**
+ * Functional interface representing audio input channel (audio data consumer)
+ * <p/>
+ * 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);
+}

foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/AudioOutputChannel.java

@@ -0,0 +1,22 @@
+package com.sap.ai.sdk.foundationmodels.openai;
+
+import java.io.IOException;
+
+/**
+ * Functional interface representing audio output channel (audio data consumer)
+ * <p/>
+ * Should be closed by application (try-with-resources) when not needed anymore
+ */
+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);
+}

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 lombok.extern.slf4j.Slf4j;
+
+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;
+
+@Slf4j
+public class BufferedWebSocketListener implements WebSocket.Listener {
+
+    private final Consumer<WebSocket> onOpen;
+    private final BiConsumer<WebSocket, CharSequence> onText;
+    private final StringBuilder buffer;
+
+    public BufferedWebSocketListener(Consumer<WebSocket> onOpen, BiConsumer<WebSocket, CharSequence> 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 last) {
+        buffer.append(data);
+        webSocket.request(1);
+        if (last) {
+            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 last) {
+        log.warn("Received unexpected binary bytes for WebSocket connection");
+        return CompletableFuture.completedStage(null);
+    }
+}

foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/OpenAiClient.java

@@ -33,6 +33,7 @@
 import com.sap.cloud.sdk.cloudplatform.connectivity.HttpDestination;
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.stream.Stream;
 import javax.annotation.Nonnull;
@@ -50,6 +51,8 @@
 @RequiredArgsConstructor(access = AccessLevel.PRIVATE)
 public final class OpenAiClient {
   private static final String DEFAULT_API_VERSION = "2024-02-01";
+  private static final int PATH_BUFFER_SIZE = 400; // existing URLs are ~120 symbols long, 400 has reasonable margin
+
 
   private static final ObjectMapper JACKSON = getOpenAiObjectMapper();
 
@@ -260,6 +263,76 @@ public Stream<String> streamChatCompletion(@Nonnull final String prompt)
         .map(OpenAiChatCompletionDelta::getDeltaContent);
   }
 
+  /**
+   * Creates realtime channel allowing to input text and voice it (receive audio output)
+   *
+   * <p>The input channel should be used with a try-with-resources block to ensure that the underlying
+   * connection is closed.
+   *
+   * <p>Example:
+   *
+   * <pre>{@code
+   * try (var textInputChannel = client.textToSpeech(audioOutputConsumer)) {
+   *       textInputChannel.sendText("...");
+   *       ....
+   * }
+   * }</pre>
+   *
+   * 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) {
+    var extraHeaders = destination.asHttp().getHeaders();
+    var headers = new HashMap<String, String>(extraHeaders.size() + 1);
+    for (Header header : extraHeaders) {
+      headers.put(header.getName(), header.getValue());
+    }
+    var endpoint = getRealtimeEndpoint();
+
+    return new TextToSpeechRealtimeClient(endpoint, headers, audioOutputConsumer);
+  }
+
+  public AudioInputChannel speechToText(@Nonnull final TextOutputChannel textOutputConsumer) {
+    var extraHeaders = destination.asHttp().getHeaders();
+    var headers = new HashMap<String, String>(extraHeaders.size() + 1);
+    for (Header header : extraHeaders) {
+      headers.put(header.getName(), header.getValue());
+    }
+    var endpoint = getRealtimeEndpoint() + "?intent=transcription";
+
+    return new SpeechToTextRealtimeClient(endpoint, headers, textOutputConsumer);
+  }
+
+  public AudioInputChannel speechToSpeech(@Nonnull final AudioOutputChannel audioOutputChannel) {
+    var extraHeaders = destination.asHttp().getHeaders();
+    var headers = new HashMap<String, String>(extraHeaders.size() + 1);
+    for (Header header : extraHeaders) {
+      headers.put(header.getName(), header.getValue());
+    }
+    var endpoint = getRealtimeEndpoint();
+
+    return new SpeechToSpeechRealtimeClient(endpoint, headers, audioOutputChannel);
+  }
+
+  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();
+  }
+
   private static void throwOnContentFilter(@Nonnull final OpenAiChatCompletionDelta delta) {
     final String finishReason = delta.getFinishReason();
     if (finishReason != null && finishReason.equals("content_filter")) {

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 */

foundation-models/openai/src/main/java/com/sap/ai/sdk/foundationmodels/openai/SpeechToSpeechRealtimeClient.java

@@ -0,0 +1,109 @@
+package com.sap.ai.sdk.foundationmodels.openai;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.openai.models.realtime.*;
+import com.openai.models.realtime.clientsecrets.ClientSecretCreateParams;
+import java.util.*;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public class SpeechToSpeechRealtimeClient extends WSSOpenAIRealtimeClient implements AudioInputChannel {
+
+    private static final int MAX_DATA_CHUNK_SIZE_BYTES = 8192;
+    private static final byte[] EMPTY_BYTE_ARRAY = new byte[0];
+
+    private static final Set<String> HANDLED_RESPONSE_TYPES = Set.of(
+            "response.output_audio.delta", "response.output_audio.done"
+    );
+
+    private static final List<RealtimeSessionCreateRequest.OutputModality> OUTPUT_MODALITIES = List.of(
+            RealtimeSessionCreateRequest.OutputModality.AUDIO
+    );
+
+    private static final String TASK = "";
+
+
+    private final AudioOutputChannel outputConsumer;
+
+    public SpeechToSpeechRealtimeClient(String url, Map<String, String> httpHeaders, AudioOutputChannel outputConsumer) {
+        super(url, httpHeaders, HANDLED_RESPONSE_TYPES);
+        this.outputConsumer = outputConsumer;
+    }
+
+    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;
+        }
+
+        // TODO: make turn detection configurable and uncomment for explicit turn control
+        // var commitAudioMessage = InputAudioBufferCommitEvent.builder().build();
+        // super.sendMessage(commitAudioMessage);
+        // askForResponse();
+    }
+
+    @Override
+    protected String getSystemPrompt() {
+        return TASK;
+    }
+
+    @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);
+        }
+    }
+
+    @Override
+    protected SessionUpdateEvent sessionConfiguration() {
+    SessionUpdateEvent sessionUpdateEvent =
+        SessionUpdateEvent.builder()
+            .session(
+                ClientSecretCreateParams.Session.ofRealtime(
+                        RealtimeSessionCreateRequest.builder()
+                            .outputModalities(OUTPUT_MODALITIES)
+                            .audio(
+                                RealtimeAudioConfig.builder()
+                                    .input(
+                                        RealtimeAudioConfigInput.builder()
+                                            .turnDetection(
+                                                RealtimeAudioInputTurnDetection.ofSemanticVad(RealtimeAudioInputTurnDetection.SemanticVad.builder().build()))
+                                            .format(
+                                                RealtimeAudioFormats.AudioPcm.builder()
+                                                    .type(
+                                                        RealtimeAudioFormats.AudioPcm.Type
+                                                            .AUDIO_PCM)
+                                                    .rate(RealtimeAudioFormats.AudioPcm.Rate._24000)
+                                                    .build())
+                                            .build())
+                                    .output(
+                                        RealtimeAudioConfigOutput.builder()
+                                            .format(
+                                                RealtimeAudioFormats.AudioPcm.builder()
+                                                    .type(
+                                                        RealtimeAudioFormats.AudioPcm.Type
+                                                            .AUDIO_PCM)
+                                                    .rate(RealtimeAudioFormats.AudioPcm.Rate._24000)
+                                                    .build())
+                                            .voice(
+                                                RealtimeAudioConfigOutput.Voice.UnionMember1.ALLOY)
+                                            .build())
+                                    .build())
+                            .build())
+                    .asRealtime())
+            .build();
+        return sessionUpdateEvent;
+    }
+}