Server support stream

gradle/libs.versions.toml

@@ -58,10 +58,13 @@ kotlinx-coroutines-jdk8 = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-j
 ktor-utils = { module = "io.ktor:ktor-utils", version.ref = "ktor" }
 ktor-http = { module = "io.ktor:ktor-http", version.ref = "ktor" }
 ktor-client ={ module = "io.ktor:ktor-client-core", version.ref = "ktor" }
+ktor-client-auth = { module = "io.ktor:ktor-client-auth", version.ref = "ktor" }
 ktor-client-content-negotiation ={ module = "io.ktor:ktor-client-content-negotiation", version.ref = "ktor" }
 ktor-client-serialization = { module = "io.ktor:ktor-serialization-kotlinx-json", version.ref = "ktor" }
 ktor-client-cio = { module = "io.ktor:ktor-client-cio", version.ref = "ktor" }
+ktor-client-logging = { module = "io.ktor:ktor-client-logging", version.ref = "ktor" }
 ktor-client-js = { module = "io.ktor:ktor-client-js", version.ref = "ktor" }
+ktor-client-json = { module = "io.ktor:ktor-client-json", version.ref = "ktor" }
 ktor-client-winhttp = { module = "io.ktor:ktor-client-winhttp", version.ref = "ktor" }
 ktor-server-auth = { module = "io.ktor:ktor-server-auth", version.ref = "ktor" }
 ktor-server-core = { module = "io.ktor:ktor-server-core", version.ref = "ktor" }

server/build.gradle.kts

@@ -22,6 +22,11 @@ dependencies {
     implementation(libs.kotlinx.serialization.json)
     implementation(libs.kotlinx.serialization.hocon)
     implementation(libs.ktor.serialization.json)
+    implementation(libs.ktor.client)
+    implementation(libs.ktor.client.auth)
+    implementation(libs.ktor.client.content.negotiation)
+    implementation(libs.ktor.client.logging)
+    implementation(libs.ktor.client.json)
     implementation(libs.ktor.server.auth)
     implementation(libs.ktor.server.netty)
     implementation(libs.ktor.server.core)

server/src/main/kotlin/com/xebia/functional/xef/server/Main.kt

@@ -10,6 +10,11 @@ import com.xebia.functional.xef.server.db.psql.Migrate
 import com.xebia.functional.xef.server.db.psql.XefVectorStoreConfig
 import com.xebia.functional.xef.server.db.psql.XefVectorStoreConfig.Companion.getPersistenceService
 import com.xebia.functional.xef.server.http.routes.routes
+import io.ktor.client.*
+import io.ktor.client.engine.cio.*
+import io.ktor.client.plugins.contentnegotiation.ContentNegotiation as ClientContentNegotiation
+import io.ktor.client.plugins.auth.*
+import io.ktor.client.plugins.logging.*
 import io.ktor.serialization.kotlinx.json.*
 import io.ktor.server.application.*
 import io.ktor.server.auth.*
@@ -32,6 +37,14 @@ object Main {
             val persistenceService = vectorStoreConfig.getPersistenceService(config)
             persistenceService.addCollection()
 
+            val ktorClient = HttpClient(CIO){
+                install(Auth)
+                install(Logging) {
+                    level = LogLevel.INFO
+                }
+                install(ClientContentNegotiation)
+            }
+
             server(factory = Netty, port = 8080, host = "0.0.0.0") {
                 install(CORS) {
                     allowNonSimpleContentTypes = true
@@ -46,7 +59,7 @@ object Main {
                         }
                     }
                 }
-                routing { routes(persistenceService) }
+                routing { routes(ktorClient, persistenceService) }
             }
             awaitCancellation()
         }

server/src/main/kotlin/com/xebia/functional/xef/server/http/routes/Routes.kt

@@ -5,65 +5,124 @@ import com.aallam.openai.api.chat.ChatCompletionRequest
 import com.xebia.functional.xef.conversation.Conversation
 import com.xebia.functional.xef.prompt.configuration.PromptConfiguration
 import com.xebia.functional.xef.conversation.llm.openai.*
+import com.xebia.functional.xef.llm.StreamedFunction
+import com.xebia.functional.xef.llm.models.chat.ChatCompletionRequest as XefChatCompletionRequest
+import com.xebia.functional.xef.llm.models.chat.ChatCompletionResponse
 import com.xebia.functional.xef.prompt.Prompt
 import com.xebia.functional.xef.server.services.PersistenceService
+import io.ktor.client.*
+import io.ktor.client.call.*
+import io.ktor.client.request.*
 import io.ktor.http.*
 import io.ktor.server.application.*
 import io.ktor.server.auth.*
 import io.ktor.server.request.*
 import io.ktor.server.response.*
 import io.ktor.server.routing.*
+import io.ktor.util.cio.*
 import io.ktor.util.pipeline.*
+import io.ktor.utils.io.*
+import io.ktor.utils.io.core.*
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.runBlocking
+import kotlinx.serialization.encodeToString
+import kotlinx.serialization.json.Json
+import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.boolean
+import kotlinx.serialization.json.jsonPrimitive
 
 enum class Provider {
-  OPENAI, GPT4ALL, GCP
+    OPENAI, GPT4ALL, GCP
 }
 
 fun String.toProvider(): Provider? = when (this) {
-  "openai" -> Provider.OPENAI
-  "gpt4all" -> Provider.GPT4ALL
-  "gcp" -> Provider.GCP
-  else -> null
+    "openai" -> Provider.OPENAI
+    "gpt4all" -> Provider.GPT4ALL
+    "gcp" -> Provider.GCP
+    else -> null
 }
 
 
 @OptIn(BetaOpenAI::class)
-fun Routing.routes(persistenceService: PersistenceService) {
-  authenticate("auth-bearer") {
-    post("/chat/completions") {
-      val provider: Provider = call.request.headers["xef-provider"]?.toProvider()
-        ?: throw IllegalArgumentException("Not a valid provider")
-      val token = call.principal<UserIdPrincipal>()?.name ?: throw IllegalArgumentException("No token found")
-      val scope = Conversation(
-        persistenceService.getVectorStore(provider, token)
-      )
-      val data = call.receive<ChatCompletionRequest>().toCore()
-      val model: OpenAIModel = data.model.toOpenAIModel(token)
-      response<String, Throwable> {
-        model.promptMessage(
-          prompt = Prompt(
-            messages = data.messages,
-            configuration = PromptConfiguration(
-              temperature = data.temperature,
-              numberOfPredictions = data.n,
-              user = data.user ?: ""
-            )
-          ),
-          scope = scope
-        )
-      }
+fun Routing.routes(
+    client: HttpClient,
+    persistenceService: PersistenceService
+) {
+    val openAiUrl = "https://api.openai.com/v1"
+
+    authenticate("auth-bearer") {
+        post("/chat/completions") {
+            val provider: Provider = call.getProvider()
+            val token = call.getToken()
+            val scope = Conversation(persistenceService.getVectorStore(provider, token))
+            val darta = call.receive<String>()
+            val data = Json.decodeFromString<JsonObject>(darta)// call.receive<JsonObject>()
+            if (!data.containsKey("model")) {
+                call.respondText("No model found", status = HttpStatusCode.BadRequest)
+                return@post
+            }
+            val model: OpenAIModel = data["model"]?.jsonPrimitive?.content?.toOpenAIModel(token) ?: run {
+                call.respondText("No model found", status = HttpStatusCode.BadRequest)
+                return@post
+            }
+
+            println(darta)
+
+            val isStream = data["stream"]?.jsonPrimitive?.boolean ?: false
+
+            if (!isStream) {
+                val response = client.request("$openAiUrl/chat/completions") {
+                    headers {
+                        bearerAuth(token)
+                    }
+                    contentType(ContentType.Application.Json)
+                    method = HttpMethod.Post
+                    setBody(darta)
+                }
+                call.respond(response.body<String>())
+            } else {
+                runBlocking {
+                    client.preparePost("$openAiUrl/chat/completions") {
+                        headers {
+                            bearerAuth(token)
+                        }
+                        contentType(ContentType.Application.Json)
+                        method = HttpMethod.Post
+                        setBody(darta)
+                    }.execute { httpResponse ->
+                        val channel: ByteReadChannel = httpResponse.body()
+                        call.respondBytesWriter(contentType = ContentType.Application.Json) {
+                            while (!channel.isClosedForRead) {
+                                val packet = channel.readRemaining(DEFAULT_BUFFER_SIZE.toLong())
+                                while (!packet.isEmpty) {
+                                    val bytes = packet.readBytes()
+                                    writeStringUtf8(bytes.decodeToString())
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+        }
     }
-  }
 }
 
+private fun ApplicationCall.getProvider(): Provider =
+    request.headers["xef-provider"]?.toProvider()
+        ?: throw IllegalArgumentException("Not a valid provider")
+
+private fun ApplicationCall.getToken(): String =
+    principal<UserIdPrincipal>()?.name ?: throw IllegalArgumentException("No token found")
+
 
 /**
  * Responds with the data and converts any potential Throwable into a 404.
  */
 private suspend inline fun <reified T : Any, E : Throwable> PipelineContext<*, ApplicationCall>.response(
-  block: () -> T
+    block: () -> T
 ) = arrow.core.raise.recover<E, Unit>({
-  call.respond(block())
+    call.respond(block())
 }) {
-  call.respondText(it.message ?: "Response not found", status = HttpStatusCode.NotFound)
+    call.respondText(it.message ?: "Response not found", status = HttpStatusCode.NotFound)
 }