New OpenAIServiceStreamedExtra function/endpoint - createChatCompletionStreamed (with impl)

Author: peterbanda

openai-client-stream/src/main/scala/io/cequence/openaiscala/service/OpenAIServiceStreamedExtra.scala

@@ -2,8 +2,11 @@ package io.cequence.openaiscala.service
 
 import akka.NotUsed
 import akka.stream.scaladsl.Source
-import io.cequence.openaiscala.domain.response.{FineTuneEvent, TextCompletionResponse}
-import io.cequence.openaiscala.domain.settings.CreateCompletionSettings
+import io.cequence.openaiscala.domain.MessageSpec
+import io.cequence.openaiscala.domain.response.{ChatCompletionChunkResponse, ChatCompletionResponse, FineTuneEvent, TextCompletionResponse}
+import io.cequence.openaiscala.domain.settings.{CreateChatCompletionSettings, CreateCompletionSettings}
+
+import scala.concurrent.Future
 
 trait OpenAIServiceStreamedExtra extends OpenAIServiceConsts {
 
@@ -23,6 +26,20 @@ trait OpenAIServiceStreamedExtra extends OpenAIServiceConsts {
     settings: CreateCompletionSettings = DefaultSettings.CreateCompletion
   ): Source[TextCompletionResponse, NotUsed]
 
+  /**
+   * Creates a completion for the chat message(s) with streamed results.
+   *
+   * @param messages The messages to generate chat completions.
+   * @param settings
+   * @return chat completion response
+   *
+   * @see <a href="https://platform.openai.com/docs/api-reference/chat/create">OpenAI Doc</a>
+   */
+  def createChatCompletionStreamed(
+    messages: Seq[MessageSpec],
+    settings: CreateChatCompletionSettings = DefaultSettings.CreateChatCompletion
+  ): Source[ChatCompletionChunkResponse, NotUsed]
+
   /**
    * Get fine-grained status updates for a fine-tune job with streamed results.
    *

openai-client-stream/src/main/scala/io/cequence/openaiscala/service/OpenAIServiceStreamedImpl.scala

@@ -9,6 +9,7 @@ import io.cequence.openaiscala.domain.settings._
 import io.cequence.openaiscala.domain.response._
 import io.cequence.openaiscala.service.ws.{Timeouts, WSStreamRequestHelper}
 import io.cequence.openaiscala.OpenAIScalaClientException
+import io.cequence.openaiscala.domain.MessageSpec
 import play.api.libs.json.{JsValue, Json}
 
 import scala.concurrent.ExecutionContext
@@ -42,6 +43,22 @@ private trait OpenAIServiceStreamedExtraImpl extends OpenAIServiceStreamedExtra
       )
     }
 
+  override def createChatCompletionStreamed(
+    messages: Seq[MessageSpec],
+    settings: CreateChatCompletionSettings = DefaultSettings.CreateChatCompletion
+  ): Source[ChatCompletionChunkResponse, NotUsed] =
+    execJsonStreamAux(
+      Command.chat_completions,
+      "POST",
+      bodyParams = createBodyParamsForChatCompletion(messages, settings, stream = true)
+    ).map { (json: JsValue) =>
+      (json \ "error").toOption.map { error =>
+        throw new OpenAIScalaClientException(error.toString())
+      }.getOrElse(
+        json.asSafe[ChatCompletionChunkResponse]
+      )
+    }
+
   override def listFineTuneEventsStreamed(
     fineTuneId: String
   ): Source[FineTuneEvent, NotUsed] =