[docker run] Fix loading of OpenAI Async client - use the Netty provi…

…der instead of okhttp (LangStream#442)
vectorize-io · Sep 19, 2023 · 0eb9c36 · 0eb9c36
1 parent 7bd54e7
commit 0eb9c36
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 5 deletions.
diff --git a/...i-agents/src/main/java/ai/langstream/ai/agents/services/impl/OpenAICompletionService.java b/...i-agents/src/main/java/ai/langstream/ai/agents/services/impl/OpenAICompletionService.java
@@ -35,7 +35,6 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import reactor.core.publisher.Flux;
@@ -84,11 +83,21 @@ public CompletableFuture<ChatCompletions> getChatCompletions(
             Flux<com.azure.ai.openai.models.ChatCompletions> flux =
                     client.getChatCompletionsStream(
                             (String) options.get("model"), chatCompletionsOptions);
-            Stream<com.azure.ai.openai.models.ChatCompletions> model = flux.toStream();
+
             ChatCompletionsConsumer chatCompletionsConsumer =
                     new ChatCompletionsConsumer(
                             streamingChunksConsumer, minChunksPerMessage, finished);
-            model.forEach(chatCompletionsConsumer);
+
+            flux.doOnError(
+                            error -> {
+                                log.error(
+                                        "Internal error while processing the streaming response",
+                                        error);
+                                finished.completeExceptionally(error);
+                            })
+                    .doOnNext(chatCompletionsConsumer)
+                    .subscribe();
+
             return finished.thenApply(
                     ___ -> {
                         result.setChoices(
@@ -101,8 +110,7 @@ public CompletableFuture<ChatCompletions> getChatCompletions(
         } else {
             com.azure.ai.openai.models.ChatCompletions chatCompletions =
                     client.getChatCompletions((String) options.get("model"), chatCompletionsOptions)
-                            .toFuture()
-                            .get();
+                            .block();
             result.setChoices(
                     chatCompletions.getChoices().stream()
                             .map(c -> new ChatChoice(convertMessage(c)))

diff --git a/langstream-runtime/langstream-runtime-tester/pom.xml b/langstream-runtime/langstream-runtime-tester/pom.xml
@@ -47,13 +47,23 @@
           <groupId>${project.groupId}</groupId>
           <artifactId>langstream-pulsar-runtime</artifactId>
         </exclusion>
+        <exclusion>
+          <groupId>${project.groupId}</groupId>
+          <artifactId>langstream-codestorage-azure-blob-storage</artifactId>
+        </exclusion>
       </exclusions>
     </dependency>
     <dependency>
       <groupId>${project.groupId}</groupId>
       <artifactId>langstream-webservice</artifactId>
       <version>${project.version}</version>
       <classifier>original</classifier>
+      <exclusions>
+        <exclusion>
+          <groupId>${project.groupId}</groupId>
+          <artifactId>langstream-codestorage-azure-blob-storage</artifactId>
+        </exclusion>
+      </exclusions>
     </dependency>
     <dependency>
       <groupId>${project.groupId}</groupId>