Add streaming

src/Providers/ChatCompletion/Alibaba.js

@@ -1,6 +1,7 @@
 "use strict";
 import Provider from "./provider.js";
 import baseHeaders from "../../Utils/baseHeaders.js";
+import startStreaming from "../../Utils/stream.js";
 
 class AlibabaProvider extends Provider {
   async chatCompletion(messages, options) {
@@ -23,8 +24,15 @@ class AlibabaProvider extends Provider {
           method: "POST",
         }
       );
-      const text = await response.json();
-      return text.choices[0].message.content;
+
+      if (options.stream === true) {
+        startStreaming(response).then((chunk) => {
+          return chunk;
+        });
+      } else {
+        const text = await response.json();
+        return text.choices[0].message.content;
+      }
     } catch (error) {
       console.error("Error:", error);
       throw error;

src/Providers/ChatCompletion/ChatBotRu.js

@@ -1,6 +1,7 @@
 "use strict";
 import Provider from "./provider.js";
 import baseHeaders from "../../Utils/baseHeaders.js";
+import startStreaming from "../../Utils/stream.js";
 
 class ChatBotRuProvider extends Provider {
   async chatCompletion(messages, options) {
@@ -32,8 +33,14 @@ class ChatBotRuProvider extends Provider {
         }
       );
 
-      const text = await response.json();
-      return text.choices[0].message.content;
+      if (options.stream === true) {
+        startStreaming(response).then((chunk) => {
+          return chunk;
+        });
+      } else {
+        const text = await response.json();
+        return text.choices[0].delta.content;
+      }
     } catch (error) {
       console.error("Error:", error);
       throw error;

src/Utils/stream.js

@@ -0,0 +1,43 @@
+async function startStreaming(response) {
+  if (!response.ok) {
+    throw new Error(`HTTP error! Status: ${response.status}`);
+  }
+
+  const reader = response.body.getReader();
+  const decoder = new TextDecoder("utf-8");
+  const buffer = [];
+  const seenChunks = new Set();
+
+  while (true) {
+    const { done, value } = await reader.read();
+    if (done) {
+      break;
+    }
+
+    buffer.push(decoder.decode(value, { stream: true }));
+    let combinedChunks = buffer.join("");
+    buffer.length = 0; // Clear the buffer
+
+    let chunks = combinedChunks.split("\n");
+
+    chunks.forEach((chunk) => {
+      chunk = chunk.replace(/^data: /, "");
+      if (chunk.trim() !== "" && !seenChunks.has(chunk)) {
+        seenChunks.add(chunk);
+        if (chunk.trim() === "[DONE]") {
+          return; // Skip processing [DONE] responses
+        }
+        try {
+          let chunkObj = JSON.parse(chunk);
+          console.log(chunkObj.choices[0].delta.content);
+        } catch (error) {
+          console.error("Error parsing chunk:", error);
+        }
+      }
+    });
+  }
+
+  return "";
+}
+
+export default startStreaming;