[Stream] Support chat completions

Author: qianmoQ

src/main/java/org/devlive/sdk/openai/DefaultClient.java

@@ -75,8 +75,16 @@ public CompleteResponse createCompletion(CompletionEntity configure)
 
     public ChatResponse createChatCompletion(ChatEntity configure)
     {
-        return this.api.fetchChatCompletions(ProviderUtils.getUrl(provider, UrlModel.FETCH_CHAT_COMPLETIONS), configure)
-                .blockingGet();
+        String url = ProviderUtils.getUrl(provider, UrlModel.FETCH_CHAT_COMPLETIONS);
+        if (ObjectUtils.isNotEmpty(this.listener)) {
+            configure.setStream(true);
+            this.createEventSource(url, configure);
+            return null;
+        }
+        else {
+            return this.api.fetchChatCompletions(url, configure)
+                    .blockingGet();
+        }
     }
 
     public UserKeyResponse getKeys()

src/main/java/org/devlive/sdk/openai/entity/ChatEntity.java

@@ -38,6 +38,13 @@ public class ChatEntity
     @JsonProperty(value = "top_p")
     private Double topP;
 
+    /**
+     * Whether to stream back partial progress. If set, tokens will be sent as data-only server-sent events as they become available, with the stream terminated by a data: [DONE] message.
+     * 是否流回部分进度。如果设置，令牌将在可用时作为仅数据服务器发送事件发送，流由 data: [DONE] 消息终止。
+     */
+    @JsonProperty(value = "stream")
+    private boolean stream = false;
+
     private ChatEntity(ChatEntityBuilder builder)
     {
         if (ObjectUtils.isEmpty(builder.model)) {
@@ -113,6 +120,11 @@ public ChatEntityBuilder maxTokens(Integer maxTokens)
             return this;
         }
 
+        private ChatEntityBuilder stream()
+        {
+            return this;
+        }
+
         public ChatEntity build()
         {
             return new ChatEntity(this);

src/main/java/org/devlive/sdk/openai/listener/ConsoleEventSourceListener.java

@@ -1,14 +1,11 @@
 package org.devlive.sdk.openai.listener;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
 import lombok.Builder;
 import lombok.extern.slf4j.Slf4j;
 import okhttp3.Response;
 import okhttp3.sse.EventSource;
 import okhttp3.sse.EventSourceListener;
 import org.apache.commons.lang3.ObjectUtils;
-import org.devlive.sdk.openai.response.CompleteResponse;
-import org.devlive.sdk.openai.utils.JsonUtils;
 
 import java.time.LocalDateTime;
 import java.util.concurrent.CountDownLatch;
@@ -19,13 +16,11 @@ public class ConsoleEventSourceListener
         extends EventSourceListener
 {
     private CountDownLatch countDownLatch;
-    private JsonUtils<CompleteResponse> jsonUtils;
 
     @Override
     public void onOpen(EventSource eventSource, Response response)
     {
         log.info("Console listener opened on time {}", LocalDateTime.now());
-        this.jsonUtils = JsonUtils.getInstance();
     }
 
     @Override
@@ -45,13 +40,7 @@ public void onEvent(EventSource eventSource, String id, String type, String data
             this.close();
         }
         else {
-            try {
-                CompleteResponse completeResponse = jsonUtils.getObject(data, CompleteResponse.class);
-                log.info("Console event received on time {} id {} type {} data {}", LocalDateTime.now(), id, type, completeResponse.getChoices().get(0).getContent());
-            }
-            catch (JsonProcessingException e) {
-                log.warn("Console event error on time {} id {} type {} data {}", LocalDateTime.now(), id, type, data, e);
-            }
+            log.info("Console event received on time {} id {} type {} data {}", LocalDateTime.now(), id, type, data);
         }
     }
 

src/main/java/org/devlive/sdk/openai/listener/HttpServletEventSourceListener.java

@@ -67,7 +67,7 @@ public void onEvent(EventSource eventSource, String id, String type, String data
         else {
             try {
                 CompleteResponse completeResponse = jsonUtils.getObject(data, CompleteResponse.class);
-                log.debug("HttpServlet event received on time {} id {} type {} data {}", LocalDateTime.now(), id, type, completeResponse.getChoices().get(0).getContent());
+                log.info("HttpServlet event received on time {} id {} type {} data {}", LocalDateTime.now(), id, type, completeResponse.getChoices().get(0).getContent());
                 this.response.getWriter().write(completeResponse.getChoices().get(0).getContent());
                 this.response.flushBuffer();
             }

src/test/java/org/devlive/sdk/openai/StreamClientTest.java

@@ -1,11 +1,15 @@
 package org.devlive.sdk.openai;
 
+import com.google.common.collect.Lists;
 import lombok.extern.slf4j.Slf4j;
+import org.devlive.sdk.openai.entity.ChatEntity;
 import org.devlive.sdk.openai.entity.CompletionEntity;
+import org.devlive.sdk.openai.entity.MessageEntity;
 import org.devlive.sdk.openai.listener.ConsoleEventSourceListener;
 import org.junit.Before;
 import org.junit.Test;
 
+import java.util.List;
 import java.util.concurrent.CountDownLatch;
 
 @Slf4j
@@ -42,4 +46,25 @@ public void testCreateCompletion()
             log.error("Interrupted while waiting", e);
         }
     }
+
+    @Test
+    public void testCreateChatCompletion()
+    {
+        List<MessageEntity> messages = Lists.newArrayList();
+        messages.add(MessageEntity.builder()
+                .content("Hello, my name is openai-java-sdk")
+                .build());
+
+        ChatEntity configure = ChatEntity.builder()
+                .messages(messages)
+                .build();
+
+        client.createChatCompletion(configure);
+        try {
+            countDownLatch.await();
+        }
+        catch (InterruptedException e) {
+            log.error("Interrupted while waiting", e);
+        }
+    }
 }