feat: Add StreamingChatClient

Author: alexcheng1982

pom.xml

@@ -71,6 +71,11 @@
       <artifactId>jackson-databind</artifactId>
       <version>${jackson.version}</version>
     </dependency>
+    <dependency>
+      <groupId>io.projectreactor.addons</groupId>
+      <artifactId>reactor-adapter</artifactId>
+      <version>3.5.1</version>
+    </dependency>
 
     <dependency>
       <groupId>org.junit.jupiter</groupId>

src/main/java/io/github/alexcheng1982/springai/dashscope/DashscopeChatClient.java

@@ -19,6 +19,7 @@
 import java.util.Set;
 import org.springframework.ai.chat.ChatClient;
 import org.springframework.ai.chat.ChatResponse;
+import org.springframework.ai.chat.StreamingChatClient;
 import org.springframework.ai.chat.messages.MessageType;
 import org.springframework.ai.chat.metadata.ChatGenerationMetadata;
 import org.springframework.ai.chat.prompt.ChatOptions;
@@ -29,13 +30,16 @@
 import org.springframework.ai.model.function.FunctionCallbackContext;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
+import reactor.adapter.rxjava.RxJava2Adapter;
+import reactor.core.publisher.Flux;
 
 /**
- * Spring AI {@linkplain ChatClient} for Aliyun Dashscope
+ * Spring AI {@linkplain ChatClient} and {@linkplain StreamingChatClient} for
+ * Aliyun Dashscope
  */
 public class DashscopeChatClient extends
     AbstractFunctionCallSupport<Message, ChatCompletionRequest, GenerationResult> implements
-    ChatClient {
+    ChatClient, StreamingChatClient {
 
   private static final DashscopeChatOptions DEFAULT_OPTIONS = DashscopeChatOptions.builder()
       .withModel(DashscopeModelName.QWEN_MAX)
@@ -79,6 +83,23 @@ public static DashscopeChatClient createDefault() {
   @Override
   public ChatResponse call(Prompt prompt) {
     var generationResult = callWithFunctionSupport(createRequest(prompt));
+    return generationResultToChatResponse(generationResult);
+  }
+
+  @Override
+  public Flux<ChatResponse> stream(Prompt prompt) {
+    var request = createRequest(prompt);
+    return RxJava2Adapter.flowableToFlux(
+        dashscopeApi.chatCompletionStream(request.messages(), request.options())
+            .map(result -> {
+              var generationResult = handleFunctionCallOrReturn(request,
+                  result);
+              return generationResultToChatResponse(generationResult);
+            }));
+  }
+
+  private ChatResponse generationResultToChatResponse(
+      GenerationResult generationResult) {
     List<org.springframework.ai.chat.Generation> generations = generationResult.getOutput()
         .getChoices().stream()
         .map(choice -> new org.springframework.ai.chat.Generation(

src/main/java/io/github/alexcheng1982/springai/dashscope/DashscopeEmbeddingClient.java

@@ -3,7 +3,9 @@
 import com.alibaba.dashscope.embeddings.TextEmbedding;
 import com.alibaba.dashscope.embeddings.TextEmbeddingParam;
 import com.alibaba.dashscope.embeddings.TextEmbeddingParam.TextType;
+import com.alibaba.dashscope.exception.ApiException;
 import com.alibaba.dashscope.exception.NoApiKeyException;
+import io.github.alexcheng1982.springai.dashscope.api.DashscopeApiException;
 import io.github.alexcheng1982.springai.dashscope.api.DashscopeModelName;
 import java.util.List;
 import org.springframework.ai.document.Document;
@@ -12,6 +14,9 @@
 import org.springframework.ai.embedding.EmbeddingRequest;
 import org.springframework.ai.embedding.EmbeddingResponse;
 
+/**
+ * Spring AI {@linkplain EmbeddingClient} for Aliyun Dashscope
+ */
 public class DashscopeEmbeddingClient implements EmbeddingClient {
 
   @Override
@@ -32,11 +37,12 @@ public EmbeddingResponse call(EmbeddingRequest request) {
     TextEmbedding embedding = new TextEmbedding();
     try {
       var result = embedding.call(builder.build());
-      return new EmbeddingResponse(result.getOutput().getEmbeddings().stream().map(item ->
-          new Embedding(item.getEmbedding(), item.getTextIndex())
-      ).toList());
-    } catch (NoApiKeyException e) {
-      throw new RuntimeException(e);
+      return new EmbeddingResponse(
+          result.getOutput().getEmbeddings().stream().map(item ->
+              new Embedding(item.getEmbedding(), item.getTextIndex())
+          ).toList());
+    } catch (ApiException | NoApiKeyException e) {
+      throw new DashscopeApiException(e);
     }
   }
 

src/main/java/io/github/alexcheng1982/springai/dashscope/api/DashscopeApi.java

@@ -6,9 +6,11 @@
 import com.alibaba.dashscope.aigc.generation.GenerationParam;
 import com.alibaba.dashscope.aigc.generation.GenerationResult;
 import com.alibaba.dashscope.common.Message;
+import com.alibaba.dashscope.exception.ApiException;
 import com.alibaba.dashscope.exception.InputRequiredException;
 import com.alibaba.dashscope.exception.NoApiKeyException;
 import io.github.alexcheng1982.springai.dashscope.DashscopeChatOptions;
+import io.reactivex.Flowable;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
@@ -27,6 +29,25 @@ public DashscopeApi() {
   public GenerationResult chatCompletion(
       List<Message> messages,
       DashscopeChatOptions options) {
+    try {
+      return generation.call(buildGenerationParam(messages, options, false));
+    } catch (ApiException | NoApiKeyException | InputRequiredException e) {
+      throw new DashscopeApiException(e);
+    }
+  }
+
+  public Flowable<GenerationResult> chatCompletionStream(List<Message> messages,
+      DashscopeChatOptions options) {
+    try {
+      return generation.streamCall(
+          buildGenerationParam(messages, options, true));
+    } catch (ApiException | NoApiKeyException | InputRequiredException e) {
+      throw new DashscopeApiException(e);
+    }
+  }
+
+  private GenerationParam buildGenerationParam(List<Message> messages,
+      DashscopeChatOptions options, boolean streaming) {
     var builder = GenerationParam.builder()
         .model(options.getModel())
         .topP(Optional.ofNullable(options.getTopP()).map(Double::valueOf)
@@ -39,20 +60,15 @@ public GenerationResult chatCompletion(
         .maxTokens(options.getMaxTokens())
         .messages(messages)
         .tools(options.getTools())
-        .resultFormat(MESSAGE);
+        .resultFormat(MESSAGE)
+        .incrementalOutput(streaming);
 
     if (options.getStops() != null) {
       builder.stopStrings(options.getStops());
     }
-
-    try {
-      return generation.call(builder.build());
-    } catch (NoApiKeyException | InputRequiredException e) {
-      throw new RuntimeException(e);
-    }
+    return builder.build();
   }
 
-
   public record ChatCompletionRequest(
       List<Message> messages,
       DashscopeChatOptions options) {

src/main/java/io/github/alexcheng1982/springai/dashscope/api/DashscopeApiException.java

@@ -0,0 +1,8 @@
+package io.github.alexcheng1982.springai.dashscope.api;
+
+public class DashscopeApiException extends RuntimeException {
+
+  public DashscopeApiException(Throwable cause) {
+    super(cause);
+  }
+}

src/main/resources/META-INF/native-image/io.github.alexcheng1982/spring-ai-dashscope-client/resource-config.json

@@ -1,23 +1,6 @@
 {
-  "resources":{
-  "includes":[{
-    "pattern":"\\QMETA-INF/services/java.lang.System$LoggerFinder\\E"
-  }, {
-    "pattern":"\\QMETA-INF/services/java.net.spi.InetAddressResolverProvider\\E"
-  }, {
-    "pattern":"\\QMETA-INF/services/java.net.spi.URLStreamHandlerProvider\\E"
-  }, {
-    "pattern":"\\QMETA-INF/services/java.time.zone.ZoneRulesProvider\\E"
-  }, {
-    "pattern":"\\QMETA-INF/services/org.slf4j.spi.SLF4JServiceProvider\\E"
-  }, {
-    "pattern":"\\Qsimplelogger.properties\\E"
-  }, {
-    "pattern":"java.base:\\Qjdk/internal/icu/impl/data/icudt72b/nfkc.nrm\\E"
-  }, {
-    "pattern":"java.base:\\Qjdk/internal/icu/impl/data/icudt72b/uprops.icu\\E"
-  }, {
-    "pattern":"java.base:\\Qsun/net/idn/uidna.spp\\E"
-  }]},
-  "bundles":[]
+  "resources": {
+    "includes": []
+  },
+  "bundles": []
 }

src/test/java/io/github/alexcheng1982/springai/dashscope/DashscopeChatClientTest.java

@@ -16,4 +16,11 @@ void smokeTest() {
     var response = client.call("hello");
     assertNotNull(response);
   }
+
+  @Test
+  void streamSmokeTest() {
+    var client = DashscopeChatClient.createDefault();
+    var response = client.stream("如何做西红柿炖牛腩？");
+    response.toIterable().forEach(System.out::println);
+  }
 }