OpenAI Streaming

e2e-test-app/pom.xml

@@ -95,6 +95,10 @@
       <groupId>org.springframework</groupId>
       <artifactId>spring-web</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.springframework</groupId>
+      <artifactId>spring-webmvc</artifactId>
+    </dependency>
     <dependency>
       <groupId>com.google.code.findbugs</groupId>
       <artifactId>jsr305</artifactId>

e2e-test-app/src/main/java/com/sap/ai/sdk/app/controllers/OpenAiController.java

@@ -14,13 +14,21 @@
 import com.sap.ai.sdk.foundationmodels.openai.model.OpenAiChatMessage.OpenAiChatUserMessage.ImageDetailLevel;
 import com.sap.ai.sdk.foundationmodels.openai.model.OpenAiEmbeddingOutput;
 import com.sap.ai.sdk.foundationmodels.openai.model.OpenAiEmbeddingParameters;
+import com.sap.cloud.sdk.cloudplatform.thread.ThreadContextExecutors;
+import java.io.IOException;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import javax.annotation.Nonnull;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;
 
 /** Endpoints for OpenAI operations */
+@Slf4j
 @RestController
 class OpenAiController {
   /**
@@ -38,6 +46,47 @@ public static OpenAiChatCompletionOutput chatCompletion() {
     return OpenAiClient.forModel(GPT_35_TURBO).chatCompletion(request);
   }
 
+  /**
+   * Stream chat request to OpenAI
+   *
+   * @return the emitter that streams the assistant message response
+   */
+  @GetMapping("/stream")
+  @Nonnull
+  public static ResponseEntity<ResponseBodyEmitter> stream() {
+    final var request =
+        new OpenAiChatCompletionParameters()
+            .setMessages(
+                List.of(
+                    new OpenAiChatUserMessage()
+                        .addText(
+                            "Can you give me the first 100 number of the Fibonacci sequence?")));
+
+    ResponseBodyEmitter emitter = new ResponseBodyEmitter();
+    // Start streaming the content asynchronously
+    ThreadContextExecutors.getExecutor()
+        .submit(
+            () -> {
+              try (var stream = OpenAiClient.forModel(GPT_35_TURBO).stream(request)) {
+                stream
+                    .getDeltaStream()
+                    .filter(delta -> delta.getDeltaContent() != null)
+                    .forEach(
+                        delta -> {
+                          try {
+                            emitter.send(delta.getDeltaContent());
+                          } catch (IOException e) {
+                            log.error(Arrays.toString(e.getStackTrace()));
+                            emitter.completeWithError(e);
+                          }
+                        });
+              } finally {
+                emitter.complete();
+              }
+            });
+    return ResponseEntity.ok().contentType(MediaType.TEXT_EVENT_STREAM).body(emitter);
+  }
+
   /**
    * Chat request to OpenAI with an image
    *

e2e-test-app/src/main/resources/static/index.html

@@ -71,6 +71,7 @@ <h2>Endpoints</h2>
             <li><h4>OpenAI</h4></li>
             <ul>
                 <li><a href="/chatCompletion">/chatCompletion</a></li>
+                <li><a href="/stream">/stream</a></li>
                 <li><a href="/chatCompletionTool">/chatCompletionTool</a></li>
                 <li><a href="/chatCompletionImage">/chatCompletionImage</a></li>
                 <li><a href="/embedding">/embedding</a></li>

e2e-test-app/src/test/java/com/sap/ai/sdk/app/controllers/OpenAiTest.java

@@ -23,6 +23,12 @@ void chatCompletionImage() {
     assertThat(message.getContent()).isNotEmpty();
   }
 
+  @Test
+  void stream() {
+    final var emitter = OpenAiController.stream();
+    // TODO: assert on the emitter
+  }
+
   @Test
   void chatCompletionTools() {
     final var completion = OpenAiController.chatCompletionTools();

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

@@ -9,8 +9,10 @@
 import com.sap.ai.sdk.core.Core;
 import com.sap.ai.sdk.foundationmodels.openai.model.OpenAiChatCompletionOutput;
 import com.sap.ai.sdk.foundationmodels.openai.model.OpenAiChatCompletionParameters;
+import com.sap.ai.sdk.foundationmodels.openai.model.OpenAiChatCompletionStream;
 import com.sap.ai.sdk.foundationmodels.openai.model.OpenAiEmbeddingOutput;
 import com.sap.ai.sdk.foundationmodels.openai.model.OpenAiEmbeddingParameters;
+import com.sap.ai.sdk.foundationmodels.openai.model.OpenAiStreamOutput;
 import com.sap.cloud.sdk.cloudplatform.connectivity.ApacheHttpClient5Accessor;
 import com.sap.cloud.sdk.cloudplatform.connectivity.DefaultHttpDestination;
 import com.sap.cloud.sdk.cloudplatform.connectivity.Destination;
@@ -30,7 +32,7 @@
 @RequiredArgsConstructor(access = AccessLevel.PRIVATE)
 public final class OpenAiClient {
   private static final String DEFAULT_API_VERSION = "2024-02-01";
-  private static final ObjectMapper JACKSON;
+  static final ObjectMapper JACKSON;
 
   static {
     JACKSON =
@@ -105,6 +107,20 @@ public OpenAiChatCompletionOutput chatCompletion(
     return execute("/chat/completions", parameters, OpenAiChatCompletionOutput.class);
   }
 
+  /**
+   * Generate a completion for the given prompt.
+   *
+   * @param parameters the prompt, including messages and other parameters.
+   * @return the completion output
+   * @throws OpenAiClientException if the request fails
+   */
+  @Nonnull
+  public OpenAiChatCompletionStream<OpenAiStreamOutput> stream(
+      @Nonnull final OpenAiChatCompletionParameters parameters) throws OpenAiClientException {
+    parameters.setStream(true);
+    return stream("/chat/completions", parameters, OpenAiStreamOutput.class);
+  }
+
   /**
    * Get a vector representation of a given input that can be easily consumed by machine learning
    * models and algorithms.
@@ -129,6 +145,16 @@ private <T> T execute(
     return executeRequest(request, responseType);
   }
 
+  @Nonnull
+  private <T> OpenAiChatCompletionStream<T> stream(
+      @Nonnull final String path,
+      @Nonnull final Object payload,
+      @Nonnull final Class<T> responseType) {
+    final var request = new HttpPost(path);
+    serializeAndSetHttpEntity(request, payload);
+    return streamRequest(request, responseType);
+  }
+
   private static void serializeAndSetHttpEntity(
       @Nonnull final BasicClassicHttpRequest request, @Nonnull final Object payload) {
     try {
@@ -145,9 +171,22 @@ private <T> T executeRequest(
     try {
       @SuppressWarnings("UnstableApiUsage")
       final var client = ApacheHttpClient5Accessor.getHttpClient(destination);
-      return client.execute(request, new OpenAiResponseHandler<>(responseType, JACKSON));
+      return client.execute(request, new OpenAiResponseHandler<>(responseType));
+    } catch (final IOException e) {
+      throw new OpenAiClientException(e);
+    }
+  }
+
+  @Nonnull
+  private <T> OpenAiChatCompletionStream<T> streamRequest(
+      final BasicClassicHttpRequest request, @Nonnull final Class<T> responseType) {
+    try {
+      @SuppressWarnings("UnstableApiUsage")
+      final var client = ApacheHttpClient5Accessor.getHttpClient(destination);
+      return new OpenAiStreamingHandler<>(responseType)
+          .handleResponse(client.executeOpen(null, request, null));
     } catch (final IOException e) {
-      throw new OpenAiClientException("Request to OpenAI model failed.", e);
+      throw new OpenAiClientException(e);
     }
   }
 }

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

@@ -5,8 +5,23 @@
 
 /** Generic exception for errors occurring when using OpenAI foundation models. */
 public class OpenAiClientException extends RuntimeException {
+  static final String BASE_ERROR_MESSAGE = "Request to OpenAI model failed";
   @Serial private static final long serialVersionUID = -7345541120979974432L;
 
+  /** Create a new exception with the base message: {@code Request to OpenAI model failed} */
+  public OpenAiClientException() {
+    super(BASE_ERROR_MESSAGE);
+  }
+
+  /**
+   * Create a new exception with the base message: {@code Request to OpenAI model failed}
+   *
+   * @param e the cause
+   */
+  public OpenAiClientException(@Nonnull final Exception e) {
+    super(BASE_ERROR_MESSAGE, e);
+  }
+
   /**
    * Create a new exception with the given message.
    *

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

@@ -1,7 +1,8 @@
 package com.sap.ai.sdk.foundationmodels.openai;
 
+import static com.sap.ai.sdk.foundationmodels.openai.OpenAiClient.JACKSON;
+
 import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.sap.ai.sdk.foundationmodels.openai.model.OpenAiError;
 import io.vavr.control.Try;
 import java.io.IOException;
@@ -21,8 +22,14 @@
 class OpenAiResponseHandler<T> implements HttpClientResponseHandler<T> {
 
   @Nonnull private final Class<T> responseType;
-  @Nonnull private final ObjectMapper jackson;
 
+  /**
+   * Processes a {@link ClassicHttpResponse} and returns some value corresponding to that response.
+   *
+   * @param response The response to process
+   * @return A model class instantiated from the response
+   * @throws OpenAiClientException in case of a problem or the connection was aborted
+   */
   @Override
   public T handleResponse(@Nonnull final ClassicHttpResponse response)
       throws OpenAiClientException {
@@ -35,14 +42,15 @@ public T handleResponse(@Nonnull final ClassicHttpResponse response)
   // The InputStream of the HTTP entity is closed by EntityUtils.toString
   @SuppressWarnings("PMD.CloseResource")
   @Nonnull
-  private T parseResponse(@Nonnull final ClassicHttpResponse response) {
+  private T parseResponse(@Nonnull final ClassicHttpResponse response)
+      throws OpenAiClientException {
     final HttpEntity responseEntity = response.getEntity();
     if (responseEntity == null) {
       throw new OpenAiClientException("Response from OpenAI model was empty.");
     }
     final var content = getContent(responseEntity);
     try {
-      return jackson.readValue(content, responseType);
+      return JACKSON.readValue(content, responseType);
     } catch (final JsonProcessingException e) {
       log.error("Failed to parse the following response from OpenAI model: {}", content);
       throw new OpenAiClientException("Failed to parse response from OpenAI model", e);
@@ -60,7 +68,8 @@ private static String getContent(@Nonnull final HttpEntity entity) {
 
   // The InputStream of the HTTP entity is closed by EntityUtils.toString
   @SuppressWarnings("PMD.CloseResource")
-  private void buildExceptionAndThrow(@Nonnull final ClassicHttpResponse response) {
+  static void buildExceptionAndThrow(@Nonnull final ClassicHttpResponse response)
+      throws OpenAiClientException {
     final var exception =
         new OpenAiClientException(
             "Request to OpenAI model failed with status %s %s "
@@ -85,19 +94,28 @@ private void buildExceptionAndThrow(@Nonnull final ClassicHttpResponse response)
       throw exception;
     }
 
-    final var maybeError = Try.of(() -> jackson.readValue(content, OpenAiError.class));
+    parseErrorAndThrow(content, exception);
+  }
+
+  /**
+   * Parse the error response and throw an exception.
+   *
+   * @param errorResponse the error response, most likely a JSON of {@link OpenAiError}.
+   * @param baseException a base exception to add the error message to.
+   */
+  static void parseErrorAndThrow(String errorResponse, OpenAiClientException baseException)
+      throws OpenAiClientException {
+    final var maybeError = Try.of(() -> JACKSON.readValue(errorResponse, OpenAiError.class));
     if (maybeError.isFailure()) {
-      exception.addSuppressed(maybeError.getCause());
-      throw exception;
+      baseException.addSuppressed(maybeError.getCause());
+      throw baseException;
     }
 
     final var error = maybeError.get().getError();
     if (error == null) {
-      throw exception;
+      throw baseException;
     }
-    final var message =
-        "Request to OpenAI model failed with %s %s and error message: '%s'"
-            .formatted(response.getCode(), response.getReasonPhrase(), error.getMessage());
-    throw new OpenAiClientException(message);
+    throw new OpenAiClientException(
+        baseException.getMessage() + "and error message: '%s'".formatted(error.getMessage()));
   }
 }

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

@@ -0,0 +1,82 @@
+package com.sap.ai.sdk.foundationmodels.openai;
+
+import static com.sap.ai.sdk.foundationmodels.openai.OpenAiClient.JACKSON;
+import static com.sap.ai.sdk.foundationmodels.openai.OpenAiResponseHandler.buildExceptionAndThrow;
+import static com.sap.ai.sdk.foundationmodels.openai.OpenAiResponseHandler.parseErrorAndThrow;
+
+import com.sap.ai.sdk.foundationmodels.openai.model.OpenAiChatCompletionStream;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.util.stream.Stream;
+import javax.annotation.Nonnull;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.hc.core5.http.ClassicHttpResponse;
+
+@Slf4j
+@RequiredArgsConstructor
+class OpenAiStreamingHandler<T> {
+
+  @Nonnull private final Class<T> responseType;
+
+  /**
+   * Processes a {@link ClassicHttpResponse} and returns some value corresponding to that response.
+   *
+   * @param response The response to process
+   * @return A {@link OpenAiChatCompletionStream} of a model class instantiated from the response
+   * @throws OpenAiClientException in case of a problem or the connection was aborted
+   */
+  @Nonnull
+  public OpenAiChatCompletionStream<T> handleResponse(@Nonnull final ClassicHttpResponse response)
+      throws OpenAiClientException {
+    if (response.getCode() >= 300) {
+      buildExceptionAndThrow(response);
+    }
+    return parseResponse(response);
+  }
+
+  /**
+   * @param response The response to process
+   * @return A {@link OpenAiChatCompletionStream} of a model class instantiated from the response
+   * @author stippi
+   */
+  private OpenAiChatCompletionStream<T> parseResponse(@Nonnull final ClassicHttpResponse response)
+      throws OpenAiClientException {
+
+    InputStream inputStream;
+    try {
+      inputStream = response.getEntity().getContent();
+    } catch (IOException e) {
+      throw new OpenAiClientException("Failed to read response content.", e);
+    }
+    var output = new OpenAiChatCompletionStream<T>();
+    BufferedReader br =
+        new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8));
+
+    // TODO: set total
+    Stream<T> deltaStream =
+        br.lines()
+            .filter(
+                responseLine ->
+                    !responseLine.isEmpty() && !"data: [DONE]".equals(responseLine.trim()))
+            .peek(
+                responseLine -> {
+                  if (!responseLine.startsWith("data: ")) {
+                    parseErrorAndThrow(responseLine, new OpenAiClientException());
+                  }
+                })
+            .map(
+                responseLine -> {
+                  String data = responseLine.substring(5).replace("delta", "message");
+                  try {
+                    return JACKSON.readValue(data, responseType);
+                  } catch (IOException e) {
+                    throw new RuntimeException(e);
+                  }
+                });
+    return output.setDeltaStream(deltaStream);
+  }
+}