From 995c5fba0edd5a6112a237c2ed7f897acfa13c9e Mon Sep 17 00:00:00 2001 From: subochev Date: Sun, 30 Aug 2026 13:12:15 +0300 Subject: [PATCH] fix --- .gitignore | 3 + README.md | 33 +- build.gradle.kts | 65 +- podman-compose.yaml | 19 +- .../kotlin/pw/binom/llmproxy/ConfigFile.kt | 17 + .../kotlin/pw/binom/llmproxy/Env.kt | 3 + .../kotlin/pw/binom/llmproxy/Logging.kt | 13 + .../kotlin/pw/binom/llmproxy/Main.kt | 568 ++++++++++++++++++ .../kotlin/pw/binom/llmproxy/Platform.kt | 10 + .../pw/binom/llmproxy/ConfigLogicTest.kt | 249 ++++++++ src/jvmMain/kotlin/pw/binom/llmproxy/Env.kt | 3 + .../kotlin/pw/binom/llmproxy/Platform.kt | 15 + .../kotlin/pw/binom/llmproxy/Env.kt | 8 + .../kotlin/pw/binom/llmproxy/Platform.kt | 15 + src/main/kotlin/pw/binom/llmproxy/Main.kt | 323 ---------- 15 files changed, 969 insertions(+), 375 deletions(-) create mode 100644 src/commonMain/kotlin/pw/binom/llmproxy/ConfigFile.kt create mode 100644 src/commonMain/kotlin/pw/binom/llmproxy/Env.kt create mode 100644 src/commonMain/kotlin/pw/binom/llmproxy/Logging.kt create mode 100644 src/commonMain/kotlin/pw/binom/llmproxy/Main.kt create mode 100644 src/commonMain/kotlin/pw/binom/llmproxy/Platform.kt create mode 100644 src/commonTest/kotlin/pw/binom/llmproxy/ConfigLogicTest.kt create mode 100644 src/jvmMain/kotlin/pw/binom/llmproxy/Env.kt create mode 100644 src/jvmMain/kotlin/pw/binom/llmproxy/Platform.kt create mode 100644 src/linuxX64Main/kotlin/pw/binom/llmproxy/Env.kt create mode 100644 src/linuxX64Main/kotlin/pw/binom/llmproxy/Platform.kt delete mode 100644 src/main/kotlin/pw/binom/llmproxy/Main.kt diff --git a/.gitignore b/.gitignore index b19dc3b..371ef37 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,6 @@ build/ .gradle/ *.jar !gradle/wrapper/gradle-wrapper.jar +# локальный конфиг с секретами (ключ API) — не коммитим +config.yaml + diff --git a/README.md b/README.md index a93feba..1a8fcfe 100644 --- a/README.md +++ b/README.md @@ -1,33 +1,22 @@ # llm-proxy -Прозрачная прослойка OpenAI API перед RouterAI (routerai.ru). Принимает -`/v1/chat/completions` и `/v1/models`, модифицирует запрос и проксирует дальше. +Прозрачная прослойка/роутер OpenAI API. Принимает `/v1/chat/completions` и +`/v1/models`, маршрутизирует между объявленными в YAML апстримами по внутреннему +учёту конкурентности и мержит `patch`-слои (`provider` → `upstream` → `model`) в +тело запроса. Подробная спецификация конфига — в `CONFIG.md`. -## Что добавляет в запрос +## Конфиг -1. **`provider.ignore`** — исключение дорогих провайдеров (список из env - `EXCLUDED_PROVIDERS`, напр. `deepseek`) + `allow_fallbacks: false` - (иначе ignore не жёсткий — RouterAI может уйти на исключённого резервной - попыткой). -2. **`reasoning: {"enabled": false}`** — для моделей с суффиксом `-no-think` - (отключение думанья; проверено на `deepseek/deepseek-v4-flash-0731`: - работает `reasoning.enabled=false`, не работает `include_reasoning=false`). - Суффикс снимается перед отправкой — RouterAI видит оригинальное имя. - -## Каталог /v1/models - -Модели, чей id содержит любую из подстрок `THINKING_MODELS`, дублируются в -каталоге с суффиксом `-no-think` (напр. `deepseek/deepseek-v4-flash-0731-no-think`). - -## Конфиг (env) +Конфигурация — в YAML-файле (по умолчанию `config.yaml` в каталоге проекта, +CWD; переопределяется env `CONFIG_PATH`). Три блока: `providers`, `upstreams`, +`models`. Старый механизм env-переменных (`UPSTREAM_URL`, `ROUTER_API_KEY`, +`EXCLUDED_PROVIDERS`, `THINKING_MODELS`) удалён — его поведение теперь в +декларативном `patch` и блоке `upstreams`. | Переменная | Default | Описание | |---|---|---| | `PORT` | 8100 | Порт сервера | -| `UPSTREAM_URL` | `https://routerai.ru/api/v1` | Куда проксировать | -| `ROUTER_API_KEY` | — (обязателен) | Bearer-ключ RouterAI | -| `EXCLUDED_PROVIDERS` | пусто | slug'и провайдеров через запятую | -| `THINKING_MODELS` | `deepseek/deepseek-v4-flash-0731,deepseek/deepseek-v4-flash` | подстроки id «думающих» моделей | +| `CONFIG_PATH` | `config.yaml` (CWD) | Путь к YAML-конфигу | ## Сборка и деплой diff --git a/build.gradle.kts b/build.gradle.kts index ce4fca1..d15979c 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -1,7 +1,6 @@ plugins { - kotlin("jvm") version "2.4.10" - application - id("com.gradleup.shadow") version "8.3.5" + alias(libs.plugins.kotlinMultiplatform) + alias(libs.plugins.kotlinSerialization) } group = "pw.binom" @@ -11,31 +10,51 @@ repositories { mavenCentral() } -dependencies { - implementation("io.ktor:ktor-server-core:3.5.2") - implementation("io.ktor:ktor-server-cio:3.5.2") - implementation("io.ktor:ktor-server-content-negotiation:3.5.2") - implementation("io.ktor:ktor-serialization-kotlinx-json:3.5.2") - implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.8.1") - implementation("ch.qos.logback:logback-classic:1.5.18") -} - -application { - mainClass.set("pw.binom.llmproxy.MainKt") -} - kotlin { - compilerOptions { - jvmTarget.set(org.jetbrains.kotlin.gradle.dsl.JvmTarget.JVM_17) + jvm() + linuxX64() + sourceSets { + val commonMain by getting { + dependencies { + implementation(libs.ktor.server.core) + implementation(libs.ktor.server.cio) + implementation(libs.ktor.client.core) + implementation(libs.ktor.client.cio) + implementation(libs.ktor.serialization.kotlinx.json) + implementation(libs.kotlinx.serialization.json) + implementation(libs.yamlkt) + implementation(libs.kotlinx.coroutines.core) + implementation(libs.kotlinx.datetime) + implementation(libs.kotlinx.io.core) + } + } + val commonTest by getting { + dependencies { + implementation(libs.kotlin.test) + } + } + val jvmMain by getting { + dependencies { + runtimeOnly(libs.logback.classic) + } + } + val jvmTest by getting { + dependencies { + implementation(libs.junit.jupiter) + runtimeOnly(libs.junit.platform.launcher) + } + } } } -java { - sourceCompatibility = JavaVersion.VERSION_17 - targetCompatibility = JavaVersion.VERSION_17 +tasks.withType { + useJUnitPlatform() } -tasks.shadowJar { +tasks.register("fatJar") { archiveFileName.set("llm-proxy.jar") - mergeServiceFiles() + manifest { attributes("Main-Class" to "pw.binom.llmproxy.MainKt") } + duplicatesStrategy = DuplicatesStrategy.EXCLUDE + from(kotlin.jvm().compilations["main"].output) + from(configurations.getByName("jvmRuntimeClasspath").map { if (it.isDirectory) it else zipTree(it) }) } diff --git a/podman-compose.yaml b/podman-compose.yaml index 0fe7c62..34e7a25 100644 --- a/podman-compose.yaml +++ b/podman-compose.yaml @@ -1,6 +1,6 @@ -# llm-proxy — прослойка OpenAI API: добавляет provider.ignore (исключение -# дорогих провайдеров) + allow_fallbacks:false и reasoning.enabled:false для -# моделей с суффиксом "-no-think", проксирует на RouterAI. Ответ — как есть. +# llm-proxy — YAML-декларативный роутер OpenAI API (подробности в CONFIG.md). +# Принимает /v1/chat/completions и /v1/models, маршрутизирует между апстримами +# по внутреннему учёту конкурентности и мержит patch-слои в тело запроса. # # Запуск на 76.179 (llm-router): podman-compose up -d # Образ тянется из нашего реестра (zot): images.binom.pw/llm-proxy: @@ -13,10 +13,15 @@ services: - bifrost environment: - PORT=8100 - - UPSTREAM_URL=https://routerai.ru/api/v1 - - ROUTER_API_KEY=__SET_FROM_CONFIG_DB__ - - EXCLUDED_PROVIDERS=deepseek - - THINKING_MODELS=deepseek/deepseek-v4-flash-0731,deepseek/deepseek-v4-flash + # Путь к конфигу внутри контейнера (файл монтируется ниже). + - CONFIG_PATH=/config/config.yaml + volumes: + # Монтируем свой config.yaml (секреты — через ${...} внутри файла, + # резолвятся из env контейнера, напр. ROUTER_API_KEY). + - ./config.yaml:/config/config.yaml:ro + # Если ключи в config.yaml ссылаются на env (${ROUTER_API_KEY}), пробросим их: + # environment: + # - ROUTER_API_KEY=__SET_FROM_CONFIG_DB__ networks: bifrost: diff --git a/src/commonMain/kotlin/pw/binom/llmproxy/ConfigFile.kt b/src/commonMain/kotlin/pw/binom/llmproxy/ConfigFile.kt new file mode 100644 index 0000000..b656428 --- /dev/null +++ b/src/commonMain/kotlin/pw/binom/llmproxy/ConfigFile.kt @@ -0,0 +1,17 @@ +package pw.binom.llmproxy + +import kotlinx.io.Buffer +import kotlinx.io.files.Path +import kotlinx.io.files.SystemFileSystem +import kotlinx.io.readByteArray + +internal fun readConfigText(path: String): String { + val src = SystemFileSystem.source(Path(path)) + try { + val buf = Buffer() + while (src.readAtMostTo(buf, Long.MAX_VALUE) > 0L) { } + return buf.readByteArray().decodeToString() + } finally { + src.close() + } +} diff --git a/src/commonMain/kotlin/pw/binom/llmproxy/Env.kt b/src/commonMain/kotlin/pw/binom/llmproxy/Env.kt new file mode 100644 index 0000000..854136c --- /dev/null +++ b/src/commonMain/kotlin/pw/binom/llmproxy/Env.kt @@ -0,0 +1,3 @@ +package pw.binom.llmproxy + +expect fun getEnv(name: String): String? diff --git a/src/commonMain/kotlin/pw/binom/llmproxy/Logging.kt b/src/commonMain/kotlin/pw/binom/llmproxy/Logging.kt new file mode 100644 index 0000000..1c8da92 --- /dev/null +++ b/src/commonMain/kotlin/pw/binom/llmproxy/Logging.kt @@ -0,0 +1,13 @@ +package pw.binom.llmproxy + +enum class LogLevel { INFO, WARN, ERROR } + +fun emitLog(level: LogLevel, message: String) { + println("[llm-proxy] ${level.name.lowercase()} $message") +} + +object log { + inline fun info(lazyMessage: () -> Any?) = emitLog(LogLevel.INFO, lazyMessage()?.toString() ?: "null") + inline fun warn(lazyMessage: () -> Any?) = emitLog(LogLevel.WARN, lazyMessage()?.toString() ?: "null") + inline fun error(lazyMessage: () -> Any?) = emitLog(LogLevel.ERROR, lazyMessage()?.toString() ?: "null") +} diff --git a/src/commonMain/kotlin/pw/binom/llmproxy/Main.kt b/src/commonMain/kotlin/pw/binom/llmproxy/Main.kt new file mode 100644 index 0000000..ca1c707 --- /dev/null +++ b/src/commonMain/kotlin/pw/binom/llmproxy/Main.kt @@ -0,0 +1,568 @@ +package pw.binom.llmproxy + +import io.ktor.client.HttpClient +import io.ktor.client.call.body +import io.ktor.client.request.header +import io.ktor.utils.io.readAvailable +import io.ktor.client.request.preparePost +import io.ktor.client.request.setBody +import io.ktor.client.statement.HttpResponse +import io.ktor.http.ContentType +import io.ktor.http.HttpStatusCode +import io.ktor.server.application.Application +import io.ktor.server.application.ApplicationCall +import io.ktor.server.application.call +import io.ktor.server.application.install +import io.ktor.server.request.receiveText +import io.ktor.server.response.respondBytesWriter +import io.ktor.server.response.respondText +import io.ktor.server.routing.get +import io.ktor.server.routing.post +import io.ktor.server.routing.routing +import io.ktor.utils.io.ByteReadChannel +import io.ktor.utils.io.ByteWriteChannel +import io.ktor.utils.io.writeFully +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.sync.Mutex +import kotlinx.datetime.Clock +import kotlinx.serialization.Serializable +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonArray +import kotlinx.serialization.json.JsonElement +import kotlinx.serialization.json.JsonNull +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.JsonPrimitive +import kotlinx.serialization.json.jsonArray +import kotlinx.serialization.json.jsonObject +import kotlinx.serialization.json.jsonPrimitive +import net.mamoe.yamlkt.Yaml +import net.mamoe.yamlkt.YamlElement +import net.mamoe.yamlkt.YamlList +import net.mamoe.yamlkt.YamlLiteral +import net.mamoe.yamlkt.YamlMap +import kotlin.concurrent.Volatile +import kotlin.time.Duration.Companion.seconds +import kotlin.time.TimeSource + +/** + * Прослойка OpenAI API (YAML-декларативный роутер, Kotlin Multiplatform). + * + * Конфигурация целиком в YAML-файле (см. CONFIG.md): объявляются провайдеры + * (`providers`), каталог апстрим-моделей (`upstreams`) и модели, видимые клиенту + * (`models`). Маршрутизация между апстримами модели — по внутреннему учёту + * конкурентности (AtomicInt на апстрим); в тело запроса послойно + * вмердживаются `patch` (provider → upstream → model). + * + * Env (только эти): + * - PORT (default 8100) + * - CONFIG_PATH путь к YAML; default `config.yaml` в каталоге проекта (CWD) + */ +fun main() { + val port = getEnv("PORT")?.toIntOrNull() ?: 8100 + val path = getEnv("CONFIG_PATH") ?: "config.yaml" + val root = Yaml.decodeYamlFromString(readConfigText(path)) + val config = parseConfig(root) + val providersById = config.providers.associateBy { it.id } + val upstreamsById = config.upstreams.associateBy { it.id } + + config.upstreams.forEach { up -> + if (up.provider !in providersById) { + log.warn { "[llm-proxy] upstream '${up.id}' ссылается на несуществующего provider '${up.provider}'" } + } + } + config.models.forEach { m -> + m.upstreams.forEach { ref -> + if (ref !in upstreamsById) { + log.warn { "[llm-proxy] model '${m.name}' ссылается на несуществующий upstream '$ref'" } + } + } + } + + val active = config.upstreams.associate { up -> up.id to UpstreamCounter(up.max_concurrency ?: Int.MAX_VALUE) } + + log.info { + "[llm-proxy] загружено: providers=${config.providers.size}, " + + "upstreams=${config.upstreams.size}, models=${config.models.size} (config=$path)" + } + + val http = createHttpClient() + startServer(port) { + proxyModule(config, providersById, upstreamsById, active, http) + } +} + +private val json = Json { ignoreUnknownKeys = true } + +fun Application.proxyModule( + config: Config, + providersById: Map, + upstreamsById: Map, + active: Map, + http: HttpClient, +) { + routing { + post("/v1/chat/completions") { + handleChat(call, config, providersById, upstreamsById, active, http) + } + get("/v1/models") { + handleModels(call, config) + } + } +} + +private suspend fun handleChat( + call: ApplicationCall, + config: Config, + providersById: Map, + upstreamsById: Map, + active: Map, + http: HttpClient, +) { + val raw = call.receiveText() + if (raw.isBlank()) { + call.respondText(errorJson("empty body"), ContentType.Application.Json, HttpStatusCode.BadRequest) + return + } + val bodyJson = try { + json.parseToJsonElement(raw).jsonObject + } catch (e: Exception) { + call.respondText(errorJson("bad json: ${e.message}"), ContentType.Application.Json, HttpStatusCode.BadRequest) + return + } + + val modelName = bodyJson["model"]?.jsonPrimitive?.content ?: "?" + val modelConf = config.models.firstOrNull { it.name == modelName } + if (modelConf == null) { + call.respondText(errorJson("unknown model: $modelName"), ContentType.Application.Json, HttpStatusCode.NotFound) + return + } + + val pool = modelConf.upstreams.mapNotNull { upstreamsById[it] } + if (pool.isEmpty()) { + call.respondText(errorJson("model $modelName has no valid upstreams"), ContentType.Application.Json, HttpStatusCode.BadGateway) + return + } + + val clientWantsStream = bodyJson["stream"]?.jsonPrimitive?.content == "true" + val failed = mutableSetOf() + var anyClaimed = false + + while (true) { + val up = pickFreeUpstream(pool, active, failed) ?: break + anyClaimed = true + val start = TimeSource.Monotonic.markNow() + try { + val provider = providersById[up.provider] + if (provider == null) { + log.error { "[llm-proxy] upstream '${up.id}': provider '${up.provider}' не найден (конфиг)" } + failed.add(up.id) + continue + } + + val patched = buildBody(bodyJson, provider, up, modelConf) + val forwarded = if (clientWantsStream) { + patched + } else { + val m = patched.toMutableMap().apply { + this["stream"] = JsonPrimitive(true) + val so = (this["stream_options"] as? JsonObject)?.toMutableMap() ?: mutableMapOf() + so["include_usage"] = JsonPrimitive(true) + this["stream_options"] = JsonObject(so) + } + JsonObject(m) + } + + val url = provider.url.trimEnd('/') + "/chat/completions" + var failover = false + var responded = false + var upstreamStatus = 0 + + http.preparePost(url) { + header("Authorization", "Bearer ${resolveEnv(provider.key)}") + header("Content-Type", "application/json") + setBody(forwarded.toString()) + }.execute { resp -> + upstreamStatus = resp.status.value + if (upstreamStatus >= 500 || upstreamStatus == 429) { + log.warn { "[llm-proxy] model=$modelName upstream=${up.id} вернул $upstreamStatus → фейловер" } + failed.add(up.id) + failover = true + return@execute + } + responded = true + if (clientWantsStream) { + val ct = resp.headers["Content-Type"] ?: "text/event-stream" + val status = HttpStatusCode.fromValue(upstreamStatus) + call.respondBytesWriter(ContentType.parse(ct), status) { + val ch = resp.body() + val buf = ByteArray(8192) + while (true) { + val n = ch.readAvailable(buf) + if (n == -1) break + if (n > 0) { + writeFully(buf, 0, n) + flush() + } + } + } + log.info { + "[llm-proxy] chat model=$modelName upstream=${up.id} provider=${provider.id} " + + "status=$upstreamStatus за ${start.elapsedNow().inWholeMilliseconds}ms stream=true" + } + } else { + val ct = resp.headers["Content-Type"] ?: "application/json" + val full = resp.body() + val out = if (ct.contains("text/event-stream")) rebuildFromChunks(full) else full + val outCt = runCatching { + if (ct.contains("text/event-stream") && + Json.parseToJsonElement(out).jsonObject["error"] != null + ) { + ContentType.Application.Json + } else { + ContentType.parse(ct) + } + }.getOrDefault(ContentType.parse(ct)) + call.respondText(out, outCt, HttpStatusCode.fromValue(upstreamStatus)) + log.info { + "[llm-proxy] chat model=$modelName upstream=${up.id} provider=${provider.id} " + + "status=$upstreamStatus за ${start.elapsedNow().inWholeMilliseconds}ms stream=false" + } + } + } + + if (failover) continue + if (responded) return + } catch (e: CancellationException) { + log.info { "[llm-proxy] chat model=$modelName upstream=${up.id} ОТМЕНЕНО клиентом за ${start.elapsedNow().inWholeMilliseconds}ms" } + throw e + } catch (e: Exception) { + log.error { "[llm-proxy] chat model=$modelName upstream=${up.id} ОШИБКА: ${e.message} за ${start.elapsedNow().inWholeMilliseconds}ms → фейловер" } + failed.add(up.id) + continue + } finally { + release(up, active) + } + } + + if (anyClaimed) { + call.respondText(errorJson("all upstreams failed"), ContentType.Application.Json, HttpStatusCode.BadGateway) + } else { + call.respondText(errorJson("all upstreams busy"), ContentType.Application.Json, HttpStatusCode.ServiceUnavailable) + } +} + +private fun errorJson(msg: String): String = + """{"error":{"message":"$msg"}}""" + +/** + * Сборка тела запроса: подмена `model` на реальное имя апстрима + глубокий + * послойный мерж `patch` в порядке provider → upstream → model. + */ +internal fun buildBody(base: JsonObject, provider: ProviderConf, up: UpstreamConf, model: ModelConf): JsonObject { + val withModel = base.toMutableMap().apply { this["model"] = JsonPrimitive(up.model) } + var acc = JsonObject(withModel) + listOf(provider.patch, up.patch, model.patch).filterNotNull().forEach { patch -> + acc = merge(acc, patch) + } + return acc +} + +/** Конвертация YAML-дерева (YamlElement) в kotlinx JsonElement для мержа в тело. */ +internal fun yamlToJson(el: YamlElement): JsonElement = when (el) { + is YamlMap -> JsonObject(el.mapKeys { (k, _) -> yamlKey(k) }.mapValues { (_, v) -> yamlToJson(v) }) + is YamlList -> JsonArray(el.map { yamlToJson(it) }) + is YamlLiteral -> yamlLiteralToJson(el) + else -> JsonNull +} + +internal fun yamlKey(key: YamlElement): String = + if (key is YamlLiteral) key.content else key.toString() + +internal fun yamlLiteralToJson(lit: YamlLiteral): JsonPrimitive { + val c = lit.content + val bool = c.toBooleanStrictOrNull() + val lng = c.toLongOrNull() + val dbl = c.toDoubleOrNull() + return when { + bool != null -> JsonPrimitive(bool) + lng != null -> JsonPrimitive(lng) + dbl != null -> JsonPrimitive(dbl) + else -> JsonPrimitive(c) + } +} + +/** Глубокий мерж: вложенные объекты сливаются рекурсивно, скаляры/массивы заменяются. */ +internal fun merge(base: JsonObject, patch: JsonObject): JsonObject { + val merged = base.toMutableMap() + for ((k, v) in patch) { + merged[k] = when { + v is JsonObject && merged[k] is JsonObject -> + merge(merged[k] as JsonObject, v) + else -> v + } + } + return JsonObject(merged) +} + +/** Подстановка `${ENV}` в строках ключей/url из переменных среды. */ +internal fun resolveEnv(s: String): String = + """\$\{([^}]+)\}""".toRegex().replace(s) { m -> getEnv(m.groupValues[1]) ?: "" } + +/** Атомарно занять слот у апстрима (по max_concurrency); false, если все заняты. */ +internal fun tryClaim(up: UpstreamConf, active: Map): Boolean { + val counter = active.getValue(up.id) + val limit = up.max_concurrency ?: Int.MAX_VALUE + return counter.tryClaim(limit) +} + +/** Освободить слот апстрима (в finally по завершении проксирования). */ +internal fun release(up: UpstreamConf, active: Map) { + active.getValue(up.id).release() +} + +class UpstreamCounter(private val limit: Int, initial: Int = 0) { + private val lock = Mutex() + @Volatile + private var count: Int = initial + + fun tryClaim(max: Int): Boolean { + if (!lock.tryLock()) return false + try { + return if (count >= max) false else { + count++ + true + } + } finally { + lock.unlock() + } + } + + fun release(): Unit { + if (!lock.tryLock()) return + try { + if (count > 0) count-- + } finally { + lock.unlock() + } + } + + val current: Int + get() { + if (!lock.tryLock()) return -1 + try { + return count + } finally { + lock.unlock() + } + } +} + +/** + * Выбор апстрима для попытки: первый по порядку (приоритету) апстрим из `pool`, + * у которого свободен слот и который ещё не в `excluded` (не упал ранее). + * Сразу занимает слот (через [tryClaim]). Если свободных нет — возвращает null. + */ +internal fun pickFreeUpstream( + pool: List, + active: Map, + excluded: Set, +): UpstreamConf? = + pool.firstOrNull { up -> up.id !in excluded && tryClaim(up, active) } + +private suspend fun handleModels(call: ApplicationCall, config: Config) { + val created = TimeSource.Monotonic.markNow().elapsedNow().inWholeSeconds + val data = config.models.map { m -> + JsonObject( + mapOf( + "id" to JsonPrimitive(m.name), + "object" to JsonPrimitive("model"), + "created" to JsonPrimitive(created), + "owned_by" to JsonPrimitive("llm-proxy"), + ), + ) + } + val body = JsonObject(mapOf("object" to JsonPrimitive("list"), "data" to JsonArray(data))) + call.respondText(body.toString(), ContentType.Application.Json, HttpStatusCode.OK) +} + +/** + * Собрать полный chat.completion из SSE-чанков апстрима (для non-stream клиентов). + * Сохраняются все значимые поля, которые отдаёт апстрим: id/created/model/ + * system_fingerprint/service_tier, usage, несколько choices (по index) с + * role/content/reasoning_content/tool_calls/finish_reason/logprobs. + */ +internal fun rebuildFromChunks(sse: String): String { + var id = "" + var created: Long? = null + var model = "" + var systemFingerprint: String? = null + var serviceTier: String? = null + var usage: JsonObject? = null + var provider: JsonElement? = null + var error: JsonObject? = null + + class MutableChoice { + var role: String? = null + val content = StringBuilder() + val reasoning = StringBuilder() + var finishReason: String? = null + var logprobs: JsonElement? = null + val toolCalls = mutableListOf() + } + + val choices = mutableMapOf() + + sse.lineSequence().forEach { line -> + if (!line.startsWith("data:")) return@forEach + val data = line.removePrefix("data:").trim() + if (data.isEmpty() || data == "[DONE]") return@forEach + val obj = runCatching { json.parseToJsonElement(data).jsonObject }.getOrNull() ?: return@forEach + if (id.isEmpty()) id = obj["id"]?.jsonPrimitive?.content ?: "" + if (created == null) created = obj["created"]?.jsonPrimitive?.content?.toLongOrNull() + if (model.isEmpty()) model = obj["model"]?.jsonPrimitive?.content ?: "" + if (systemFingerprint == null) systemFingerprint = obj["system_fingerprint"]?.jsonPrimitive?.content + if (serviceTier == null) serviceTier = obj["service_tier"]?.jsonPrimitive?.content + if (provider == null) provider = obj["provider"] + (obj["error"] as? JsonObject)?.let { error = it } + (obj["usage"] as? JsonObject)?.let { usage = it } + val chArr = obj["choices"]?.jsonArray ?: return@forEach + for (ch in chArr) { + val c = ch.jsonObject + val idx = c["index"]?.jsonPrimitive?.content?.toIntOrNull() ?: 0 + val mc = choices.getOrPut(idx) { MutableChoice() } + val delta = c["delta"]?.jsonObject + if (delta != null) { + if (mc.role == null) mc.role = delta["role"]?.jsonPrimitive?.content + delta["content"]?.jsonPrimitive?.content?.takeIf { it != "null" }?.let { mc.content.append(it) } + delta["reasoning_content"]?.jsonPrimitive?.content?.takeIf { it != "null" }?.let { mc.reasoning.append(it) } + delta["tool_calls"]?.jsonArray?.forEach { tc -> (tc as? JsonObject)?.let { mc.toolCalls.add(it) } } + } + c["finish_reason"]?.jsonPrimitive?.content?.takeIf { it.isNotEmpty() && it != "null" }?.let { mc.finishReason = it } + c["logprobs"]?.let { mc.logprobs = it } + } + } + + // Апстрим вернул ошибку (в SSE она приходит чанком data: {"error":{...}}). + // Не «проглатываем» — отдаём клиенту как есть. + if (error != null) { + return JsonObject(mapOf("error" to error)).toString() + } + + val choiceArr = choices.entries.sortedBy { it.key }.map { (idx, mc) -> + val msg = mutableMapOf( + "role" to JsonPrimitive(mc.role ?: "assistant"), + "content" to JsonPrimitive(mc.content.toString()), + ) + if (mc.reasoning.isNotEmpty()) msg["reasoning_content"] = JsonPrimitive(mc.reasoning.toString()) + if (mc.toolCalls.isNotEmpty()) msg["tool_calls"] = JsonArray(mc.toolCalls) + val ch = mutableMapOf( + "index" to JsonPrimitive(idx), + "message" to JsonObject(msg), + "finish_reason" to JsonPrimitive(mc.finishReason ?: "stop"), + ) + mc.logprobs?.let { ch["logprobs"] = it } + JsonObject(ch) + } + + val root = mutableMapOf( + "id" to JsonPrimitive(id), + "object" to JsonPrimitive("chat.completion"), + "created" to JsonPrimitive(created ?: Clock.System.now().epochSeconds), + "model" to JsonPrimitive(model), + "choices" to JsonArray(choiceArr), + ) + systemFingerprint?.let { root["system_fingerprint"] = JsonPrimitive(it) } + serviceTier?.let { root["service_tier"] = JsonPrimitive(it) } + provider?.let { root["provider"] = it } + usage?.let { root["usage"] = it } + return JsonObject(root).toString() +} + +@Serializable +data class ProviderConf( + val id: String, + val url: String, + val key: String = "", + val patch: JsonObject? = null, +) + +data class UpstreamConf( + val id: String, + val provider: String, + val model: String, + val max_concurrency: Int? = null, + val patch: JsonObject? = null, +) + +data class ModelConf( + val name: String, + val upstreams: List, + val patch: JsonObject? = null, +) + +data class Config( + val providers: List, + val upstreams: List, + val models: List, +) + +/** + * Ручной разбор YAML-дерева (YamlElement) в конфиг. Поле `patch` сразу + * конвертируется в kotlinx JsonObject (merge`ить в тело тривиально). + * Декодирование всего документа как YamlElement надёжнее, чем вложенный + * `@Serializable`-класс с полем YamlElement (yamlkt криво читает списки). + */ +internal fun parseConfig(root: YamlElement): Config { + val rootMap = root as? YamlMap ?: error("config root must be a map") + val top = rootMap.toMap() + + fun list(key: String): List { + val v = top[key] ?: return emptyList() + return (v as? YamlList)?.map { it } ?: emptyList() + } + + val providers = list("providers").map { entry -> + val m = (entry as YamlMap).toMap() + ProviderConf( + id = m.str("id"), + url = m.str("url"), + key = m.strOrNull("key") ?: "", + patch = m.yamlMapOrNull("patch")?.let { yamlToJson(it) as JsonObject }, + ) + } + + val upstreams = list("upstreams").map { entry -> + val m = (entry as YamlMap).toMap() + UpstreamConf( + id = m.str("id"), + provider = m.str("provider"), + model = m.str("model"), + max_concurrency = m.strOrNull("max_concurrency")?.toIntOrNull(), + patch = m.yamlMapOrNull("patch")?.let { yamlToJson(it) as JsonObject }, + ) + } + + val models = list("models").map { entry -> + val m = (entry as YamlMap).toMap() + ModelConf( + name = m.str("name"), + upstreams = (m["upstreams"] as? YamlList)?.map { (it as YamlLiteral).content } ?: emptyList(), + patch = m.yamlMapOrNull("patch")?.let { yamlToJson(it) as JsonObject }, + ) + } + + return Config(providers, upstreams, models) +} + +/** YamlMap -> Map (ключи YAML — строковые скаляры). */ +internal fun YamlMap.toMap(): Map = + entries.associate { (it.key as YamlLiteral).content to it.value } + +internal fun Map.str(key: String): String = + (this[key] as? YamlLiteral)?.content ?: error("config: missing string '$key'") + +internal fun Map.strOrNull(key: String): String? = + (this[key] as? YamlLiteral)?.content + +internal fun Map.yamlMapOrNull(key: String): YamlMap? = + this[key] as? YamlMap diff --git a/src/commonMain/kotlin/pw/binom/llmproxy/Platform.kt b/src/commonMain/kotlin/pw/binom/llmproxy/Platform.kt new file mode 100644 index 0000000..51d29ef --- /dev/null +++ b/src/commonMain/kotlin/pw/binom/llmproxy/Platform.kt @@ -0,0 +1,10 @@ +package pw.binom.llmproxy + +import io.ktor.client.HttpClient +import io.ktor.server.application.Application + +/** Платформенное создание HTTP-клиента (движок задаётся в actual). */ +expect fun createHttpClient(): HttpClient + +/** Платформенный запуск Ktor-сервера (движок задаётся в actual). */ +expect fun startServer(port: Int, module: Application.() -> Unit) diff --git a/src/commonTest/kotlin/pw/binom/llmproxy/ConfigLogicTest.kt b/src/commonTest/kotlin/pw/binom/llmproxy/ConfigLogicTest.kt new file mode 100644 index 0000000..61c0cc1 --- /dev/null +++ b/src/commonTest/kotlin/pw/binom/llmproxy/ConfigLogicTest.kt @@ -0,0 +1,249 @@ +package pw.binom.llmproxy + +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertTrue +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.jsonArray +import kotlinx.serialization.json.jsonObject +import kotlinx.serialization.json.jsonPrimitive +import net.mamoe.yamlkt.Yaml + +class ConfigLogicTest { + + @Test + fun mergeDeepMergesNestedObjectsAndReplacesScalars() { + val base = Json.parseToJsonElement("""{"a":{"x":1,"y":2},"b":1}""").jsonObject + val patch = Json.parseToJsonElement("""{"a":{"y":3,"z":4},"c":5}""").jsonObject + val merged = merge(base, patch) + assertEquals("""{"a":{"x":1,"y":3,"z":4},"b":1,"c":5}""", merged.toString()) + } + + @Test + fun mergeReplacesArraysByKey() { + val base = Json.parseToJsonElement("""{"provider":{"ignore":["a"]}}""").jsonObject + val patch = Json.parseToJsonElement("""{"provider":{"ignore":["b","c"]}}""").jsonObject + assertEquals("""{"provider":{"ignore":["b","c"]}}""", merge(base, patch).toString()) + } + + @Test + fun parseConfigReadsAllBlocksAndConvertsPatchToJsonObject() { + val yaml = """ + providers: + - id: p1 + url: "https://x.ru/api/v1" + key: "k" + patch: + provider: + allow_fallbacks: false + upstreams: + - id: u1 + provider: p1 + model: real-1 + max_concurrency: 2 + patch: + provider: + ignore: [deepseek] + - id: bad + provider: missing + model: x + models: + - name: m1 + upstreams: [u1, bad] + patch: + reasoning: + enabled: false + """.trimIndent() + + val cfg = parseConfig(Yaml.decodeYamlFromString(yaml)) + + assertEquals(1, cfg.providers.size) + assertEquals("https://x.ru/api/v1", cfg.providers[0].url) + assertEquals(2, cfg.upstreams.size) + assertEquals(2, cfg.upstreams[0].max_concurrency) + assertEquals(listOf("u1", "bad"), cfg.models[0].upstreams) + + assertEquals("""{"provider":{"allow_fallbacks":false}}""", cfg.providers[0].patch.toString()) + assertEquals("""{"provider":{"ignore":["deepseek"]}}""", cfg.upstreams[0].patch.toString()) + assertEquals("""{"reasoning":{"enabled":false}}""", cfg.models[0].patch.toString()) + } + + @Test + fun parseConfigOmitsPatchWhenAbsent() { + val yaml = """ + providers: + - id: p1 + url: "https://x.ru/api/v1" + upstreams: + - id: u1 + provider: p1 + model: real-1 + models: + - name: m1 + upstreams: [u1] + """.trimIndent() + val cfg = parseConfig(Yaml.decodeYamlFromString(yaml)) + assertEquals(null, cfg.providers[0].patch) + assertEquals(null, cfg.models[0].patch) + assertEquals(null, cfg.upstreams[0].max_concurrency) + } + + @Test + fun buildBodySubstitutesModelAndAppliesLayersInOrder() { + val provider = ProviderConf( + "p1", "https://x", "", + Json.parseToJsonElement("""{"provider":{"allow_fallbacks":false}}""").jsonObject, + ) + val up = UpstreamConf( + "u1", "p1", "real-1", null, + Json.parseToJsonElement("""{"provider":{"ignore":["deepseek"]}}""").jsonObject, + ) + val model = ModelConf( + "m1", listOf("u1"), + Json.parseToJsonElement("""{"reasoning":{"enabled":false}}""").jsonObject, + ) + val body = Json.parseToJsonElement( + """{"model":"m1","messages":[],"temperature":0.7}""", + ).jsonObject + + val out = buildBody(body, provider, up, model) + assertEquals( + """{"model":"real-1","messages":[],"temperature":0.7,"provider":{"allow_fallbacks":false,"ignore":["deepseek"]},"reasoning":{"enabled":false}}""", + out.toString(), + ) + } + + @Test + fun buildBodyWithNoPatchesReturnsClientBodyWithSubstitutedModel() { + val provider = ProviderConf("p1", "https://x", "") + val up = UpstreamConf("u1", "p1", "real-1", null, null) + val model = ModelConf("m1", listOf("u1"), null) + val body = Json.parseToJsonElement("""{"model":"m1","a":1}""").jsonObject + assertEquals("""{"model":"real-1","a":1}""", buildBody(body, provider, up, model).toString()) + } + + @Test + fun resolveEnvSubstitutesMissingVarWithEmptyAndPassesLiteralThrough() { + assertEquals("", resolveEnv("\${LLM_PROXY_TEST_MISSING_VAR}")) + assertEquals("plain", resolveEnv("plain")) + assertEquals("pre--post", resolveEnv("pre-\${LLM_PROXY_TEST_MISSING_VAR}-post")) + } + + @Test + fun tryClaimRespectsMaxConcurrencyAndReleaseFreesSlot() { + val active = mapOf("u1" to UpstreamCounter(0)) + val up = UpstreamConf("u1", "p", "m", 1, null) + assertTrue(tryClaim(up, active)) + assertFalse(tryClaim(up, active)) + release(up, active) + assertTrue(tryClaim(up, active)) + } + + @Test + fun tryClaimUnlimitedWhenMaxConcurrencyIsNull() { + val active = mapOf("u2" to UpstreamCounter(0)) + val up = UpstreamConf("u2", "p", "m", null, null) + assertTrue(tryClaim(up, active)) + assertTrue(tryClaim(up, active)) + } + + @Test + fun pickFreeUpstreamReturnsFirstFreeInDeclarationOrder() { + val active = mapOf("u1" to UpstreamCounter(0), "u2" to UpstreamCounter(0)) + val pool = listOf( + UpstreamConf("u1", "p", "m", 1, null), + UpstreamConf("u2", "p", "m", 2, null), + ) + val up = pickFreeUpstream(pool, active, emptySet()) + assertEquals("u1", up?.id) + // слот реально занят + assertEquals(1, active.getValue("u1").current) + } + + @Test + fun pickFreeUpstreamSkipsExcluded() { + val active = mapOf("u1" to UpstreamCounter(0), "u2" to UpstreamCounter(0)) + val pool = listOf( + UpstreamConf("u1", "p", "m", 1, null), + UpstreamConf("u2", "p", "m", 2, null), + ) + val up = pickFreeUpstream(pool, active, setOf("u1")) + assertEquals("u2", up?.id) + } + + @Test + fun pickFreeUpstreamReturnsNullWhenAllBusy() { + val active = mapOf("u1" to UpstreamCounter(1, 1)) // уже на лимите 1 + val pool = listOf(UpstreamConf("u1", "p", "m", 1, null)) + assertEquals(null, pickFreeUpstream(pool, active, emptySet())) + } + + @Test + fun pickFreeUpstreamImplementsFailoverOrder() { + // dead исключён (упал ранее) — выбирается следующий живой u1 + val active = mapOf("dead" to UpstreamCounter(0), "u1" to UpstreamCounter(0)) + val pool = listOf( + UpstreamConf("dead", "p", "m", 1, null), + UpstreamConf("u1", "p", "m", 1, null), + ) + val up = pickFreeUpstream(pool, active, setOf("dead")) + assertEquals("u1", up?.id) + } + + @Test + fun pickFreeUpstreamExhaustsConcurrencyThenReturnsNull() { + val active = mapOf("u1" to UpstreamCounter(0), "u2" to UpstreamCounter(0)) + val pool = listOf( + UpstreamConf("u1", "p", "m", 1, null), + UpstreamConf("u2", "p", "m", 1, null), + ) + assertEquals("u1", pickFreeUpstream(pool, active, emptySet())?.id) + assertEquals("u2", pickFreeUpstream(pool, active, emptySet())?.id) + assertEquals(null, pickFreeUpstream(pool, active, emptySet())?.id) + } + + @Test + fun rebuildFromChunksPreservesAllUpstreamFields() { + val sse = """ + data: {"id":"c1","object":"chat.completion.chunk","created":123,"model":"m","provider":"Baidu","system_fingerprint":"sf1","choices":[{"index":0,"delta":{"role":"assistant","content":"He"},"finish_reason":null}]} + + data: {"id":"c1","created":123,"model":"m","provider":"Baidu","choices":[{"index":0,"delta":{"content":"llo"},"finish_reason":null}]} + + data: {"id":"c1","created":123,"model":"m","provider":"Baidu","choices":[{"index":1,"delta":{"content":"B"},"finish_reason":null}]} + + data: {"id":"c1","created":123,"model":"m","provider":"Baidu","choices":[{"index":0,"delta":{},"finish_reason":"stop"},{"index":1,"delta":{},"finish_reason":"stop"}]} + + data: {"id":"c1","created":123,"model":"m","provider":"Baidu","usage":{"prompt_tokens":1,"completion_tokens":2,"total_tokens":3}} + + data: [DONE] + """.trimIndent() + + val out = Json.parseToJsonElement(rebuildFromChunks(sse)).jsonObject + assertEquals("c1", out["id"]?.jsonPrimitive?.content) + assertEquals("m", out["model"]?.jsonPrimitive?.content) + assertEquals("Baidu", out["provider"]?.jsonPrimitive?.content) + assertEquals("sf1", out["system_fingerprint"]?.jsonPrimitive?.content) + assertEquals("chat.completion", out["object"]?.jsonPrimitive?.content) + assertEquals(123, out["created"]?.jsonPrimitive?.content?.toLong()) + assertEquals(3, out["usage"]?.jsonObject?.get("total_tokens")?.jsonPrimitive?.content?.toInt()) + val choices = out["choices"]?.jsonArray ?: error("no choices") + assertEquals(2, choices.size) + assertEquals("Hello", choices[0].jsonObject["message"]?.jsonObject?.get("content")?.jsonPrimitive?.content) + assertEquals("B", choices[1].jsonObject["message"]?.jsonObject?.get("content")?.jsonPrimitive?.content) + assertEquals("stop", choices[0].jsonObject["finish_reason"]?.jsonPrimitive?.content) + } + + @Test + fun rebuildFromChunksReturnsUpstreamErrorVerbatim() { + val sse = """ + data: {"id":"c1","error":{"message":"Unsupported field 'foo'","type":"invalid_request"},"choices":[]} + + data: [DONE] + """.trimIndent() + val out = Json.parseToJsonElement(rebuildFromChunks(sse)).jsonObject + val err = out["error"]?.jsonObject ?: error("error block missing") + assertEquals("Unsupported field 'foo'", err["message"]?.jsonPrimitive?.content) + } +} diff --git a/src/jvmMain/kotlin/pw/binom/llmproxy/Env.kt b/src/jvmMain/kotlin/pw/binom/llmproxy/Env.kt new file mode 100644 index 0000000..545b7c4 --- /dev/null +++ b/src/jvmMain/kotlin/pw/binom/llmproxy/Env.kt @@ -0,0 +1,3 @@ +package pw.binom.llmproxy + +actual fun getEnv(name: String): String? = System.getenv(name) diff --git a/src/jvmMain/kotlin/pw/binom/llmproxy/Platform.kt b/src/jvmMain/kotlin/pw/binom/llmproxy/Platform.kt new file mode 100644 index 0000000..e2091fc --- /dev/null +++ b/src/jvmMain/kotlin/pw/binom/llmproxy/Platform.kt @@ -0,0 +1,15 @@ +package pw.binom.llmproxy + +import io.ktor.client.HttpClient +import io.ktor.client.engine.cio.CIO +import io.ktor.server.application.Application +import io.ktor.server.cio.CIO as ServerCIO +import io.ktor.server.engine.embeddedServer + +actual fun createHttpClient(): HttpClient = HttpClient(CIO) + +actual fun startServer(port: Int, module: Application.() -> Unit) { + embeddedServer(ServerCIO, port = port, host = "0.0.0.0") { + module() + }.start(wait = true) +} diff --git a/src/linuxX64Main/kotlin/pw/binom/llmproxy/Env.kt b/src/linuxX64Main/kotlin/pw/binom/llmproxy/Env.kt new file mode 100644 index 0000000..0a1e2b0 --- /dev/null +++ b/src/linuxX64Main/kotlin/pw/binom/llmproxy/Env.kt @@ -0,0 +1,8 @@ +package pw.binom.llmproxy + +import kotlinx.cinterop.ExperimentalForeignApi +import kotlinx.cinterop.toKString +import platform.posix.getenv + +@OptIn(ExperimentalForeignApi::class) +actual fun getEnv(name: String): String? = getenv(name)?.toKString() diff --git a/src/linuxX64Main/kotlin/pw/binom/llmproxy/Platform.kt b/src/linuxX64Main/kotlin/pw/binom/llmproxy/Platform.kt new file mode 100644 index 0000000..e2091fc --- /dev/null +++ b/src/linuxX64Main/kotlin/pw/binom/llmproxy/Platform.kt @@ -0,0 +1,15 @@ +package pw.binom.llmproxy + +import io.ktor.client.HttpClient +import io.ktor.client.engine.cio.CIO +import io.ktor.server.application.Application +import io.ktor.server.cio.CIO as ServerCIO +import io.ktor.server.engine.embeddedServer + +actual fun createHttpClient(): HttpClient = HttpClient(CIO) + +actual fun startServer(port: Int, module: Application.() -> Unit) { + embeddedServer(ServerCIO, port = port, host = "0.0.0.0") { + module() + }.start(wait = true) +} diff --git a/src/main/kotlin/pw/binom/llmproxy/Main.kt b/src/main/kotlin/pw/binom/llmproxy/Main.kt deleted file mode 100644 index 1d168dd..0000000 --- a/src/main/kotlin/pw/binom/llmproxy/Main.kt +++ /dev/null @@ -1,323 +0,0 @@ -package pw.binom.llmproxy - -import io.ktor.http.ContentType -import io.ktor.http.HttpStatusCode -import io.ktor.server.application.Application -import io.ktor.server.application.ApplicationCall -import io.ktor.server.application.install -import io.ktor.server.cio.CIO -import io.ktor.server.engine.embeddedServer -import io.ktor.server.request.receiveText -import io.ktor.server.response.respondBytes -import io.ktor.server.response.respondOutputStream -import io.ktor.server.routing.get -import io.ktor.server.routing.post -import io.ktor.server.routing.routing -import io.ktor.server.http.HttpRequestLifecycle -import kotlinx.serialization.json.Json -import kotlinx.serialization.json.JsonArray -import kotlinx.serialization.json.JsonObject -import kotlinx.serialization.json.JsonPrimitive -import kotlinx.serialization.json.jsonArray -import kotlinx.coroutines.CancellationException -import kotlinx.coroutines.suspendCancellableCoroutine -import kotlinx.coroutines.currentCoroutineContext -import kotlinx.coroutines.ensureActive -import kotlin.coroutines.resume -import kotlin.coroutines.resumeWithException -import java.util.concurrent.CompletableFuture -import kotlinx.serialization.json.jsonObject -import kotlinx.serialization.json.jsonPrimitive -import java.net.URI -import java.net.http.HttpClient -import java.net.http.HttpRequest -import java.net.http.HttpResponse -import java.time.Duration - -/** - * Прослойка OpenAI API: принимает chat/completions, добавляет в JSON - * `provider.ignore` (исключение дорогих провайдеров) + `allow_fallbacks: false` - * и прозрачно проксирует на RouterAI (routerai.ru/api/v1). Ответ — как есть, - * включая SSE-стрим. - * - * Трюк «-no-think»: модели из [THINKING_MODELS] в каталоге /v1/models дублируются - * с суффиксом `-no-think`; запрос на такой id получает `reasoning: {"enabled": false}` - * (модель не думает — не жрёт токены на reasoning). Проверено на - * deepseek/deepseek-v4-flash-0731: reasoning.enabled=false глушит думанье. - * - * Конфиг (env): - * - PORT (default 8100) - * - UPSTREAM_URL (default https://routerai.ru/api/v1) - * - ROUTER_API_KEY — Bearer-ключ RouterAI (обязателен) - * - EXCLUDED_PROVIDERS — slug'и провайдеров через запятую (напр. "deepseek") - * - THINKING_MODELS — подстроки id «думающих» моделей через запятую - * (напр. "deepseek/deepseek-v4-flash-0731,deepseek/deepseek-r1") - */ -fun main() { - val port = System.getenv("PORT")?.toIntOrNull() ?: 8100 - val upstream = System.getenv("UPSTREAM_URL") ?: "https://routerai.ru/api/v1" - val apiKey = System.getenv("ROUTER_API_KEY") - ?: throw IllegalStateException("ROUTER_API_KEY required") - val excluded = (System.getenv("EXCLUDED_PROVIDERS") ?: "") - .split(",").map { it.trim() }.filter { it.isNotEmpty() }.distinct() - val thinking = (System.getenv("THINKING_MODELS") ?: "deepseek/deepseek-v4-flash-0731,deepseek/deepseek-v4-flash") - .split(",").map { it.trim() }.filter { it.isNotEmpty() }.distinct() - - embeddedServer(CIO, port = port, host = "0.0.0.0") { - install(HttpRequestLifecycle) { - cancelCallOnClose = true - } - proxyModule(upstream, apiKey, excluded, thinking) - }.start(wait = true) -} - -private val json = Json { ignoreUnknownKeys = true } - -/** Прокси-клиент: один на процесс (HttpClient потокобезопасен). */ -private val http = HttpClient.newBuilder() - .connectTimeout(Duration.ofSeconds(15)) - .build() - -fun Application.proxyModule(upstream: String, apiKey: String, excluded: List, thinking: List) { - routing { - post("/v1/chat/completions") { - handleChat(call, upstream, apiKey, excluded, thinking) - } - get("/v1/models") { - handleModels(call, upstream, apiKey, thinking) - } - } -} - -private suspend fun handleChat( - call: ApplicationCall, - upstream: String, - apiKey: String, - excluded: List, - thinking: List, -) { - val raw = call.receiveText() - if (raw.isBlank()) { - call.respondBytes( - """{"error":{"message":"empty body"}}""".toByteArray(), - ContentType.Application.Json, HttpStatusCode.BadRequest, - ) - return - } - val patched = try { - patchBody(raw, excluded, thinking) - } catch (e: Exception) { - call.respondBytes( - """{"error":{"message":"bad json: ${e.message}"}}""".toByteArray(), - ContentType.Application.Json, HttpStatusCode.BadRequest, - ) - return - } - - val start = System.currentTimeMillis() - val model = try { - json.parseToJsonElement(patched).jsonObject["model"]?.jsonPrimitive?.content ?: "?" - } catch (e: Exception) { "?" } - val req = HttpRequest.newBuilder() - .uri(URI.create(upstream.trimEnd('/') + "/chat/completions")) - .header("Authorization", "Bearer $apiKey") - .header("Content-Type", "application/json") - .POST(HttpRequest.BodyPublishers.ofString(patched)) - .build() - - val stream = json.parseToJsonElement(raw).jsonObject["stream"]?.jsonPrimitive?.content == "true" - try { - if (stream) { - // SSE-стрим: транслируем как есть, чанк за чанком. - val resp = http.send(req, HttpResponse.BodyHandlers.ofInputStream()) - val ct = resp.headers().firstValue("content-type").orElse("text/event-stream") - call.respondOutputStream(ContentType.parse(ct), HttpStatusCode.fromValue(resp.statusCode())) { - resp.body().use { input -> input.copyTo(this, 8192) } - } - println("[llm-proxy] chat model=$model status=${resp.statusCode()} в ${System.currentTimeMillis() - start}ms stream=true") - } else { - // Клиент ждёт non-stream ответ, но апстриму шлём stream=true: - // не-стрим генерация у llama.cpp НЕ отменяется обрывом соединения, - // а стрим — отменяется. Так отмена клиента реально рвёт генерацию. - val streamed = json.parseToJsonElement(patched).jsonObject.toMutableMap().apply { - this["stream"] = JsonPrimitive(true) - } - val req2 = HttpRequest.newBuilder() - .uri(URI.create(upstream.trimEnd('/') + "/chat/completions")) - .header("Authorization", "Bearer $apiKey") - .header("Content-Type", "application/json") - .POST(HttpRequest.BodyPublishers.ofString(JsonObject(streamed).toString())) - .build() - val future = http.sendAsync(req2, HttpResponse.BodyHandlers.ofInputStream()) - val resp = future.awaitOrCancel() - val body = resp.body() - // Читаем с проверкой отмены: при обрыве клиента ensureActive() бросит - // CancellationException, а finally закроет входной поток — это рвёт - // апстрим-соединение, и llama.cpp отменяет генерацию. - val full = try { - val sb = StringBuilder() - val reader = body.bufferedReader() - while (true) { - currentCoroutineContext().ensureActive() - val line = reader.readLine() ?: break - sb.append(line).append('\n') - } - sb.toString() - } finally { - body.close() - } - val ct = resp.headers().firstValue("content-type").orElse("application/json") - val out = if (ct.contains("text/event-stream")) rebuildFromChunks(full) else full - call.respondBytes(out.toByteArray(), ContentType.parse(ct), HttpStatusCode.fromValue(resp.statusCode())) - println("[llm-proxy] chat model=$model status=${resp.statusCode()} в ${System.currentTimeMillis() - start}ms stream=false") - } - } catch (e: CancellationException) { - // Клиент оборвал соединение: апстрим-запрос уже отменён через awaitOrCancel. - println("[llm-proxy] chat model=$model ОТМЕНЕНО клиентом в ${System.currentTimeMillis() - start}ms") - throw e - } catch (e: Exception) { - println("[llm-proxy] chat model=$model ОШИБКА: ${e.message} в ${System.currentTimeMillis() - start}ms") - call.respondBytes( - """{"error":{"message":"upstream: ${e.message}"}}""".toByteArray(), - ContentType.Application.Json, HttpStatusCode.BadGateway, - ) - } -} - -/** - * Патч запроса: (1) если модель оканчивается на "-no-think" — снять суффикс и - * добавить `reasoning: {"enabled": false}` (не думать); (2) добавить - * `provider.ignore` (объединяя с присланным клиентом) и `allow_fallbacks: false` — - * иначе ignore не жёсткий: RouterAI может уйти на исключённого провайдера - * резервной попыткой (см. гайд provider-selection). - */ -internal fun patchBody(raw: String, excluded: List, thinking: List): String { - val root = json.parseToJsonElement(raw).jsonObject.toMutableMap() - - val model = root["model"]?.jsonPrimitive?.content ?: "" - if (model.endsWith("-no-think")) { - root["model"] = JsonPrimitive(model.removeSuffix("-no-think")) - if (root["reasoning"] == null) { - root["reasoning"] = JsonObject(mapOf("enabled" to JsonPrimitive(false))) - } - // SGLANG_COMPAT: Qwen3.8 глушится только через chat_template_kwargs.enable_thinking=false - if (System.getenv("SGLANG_COMPAT") == "true") { - val ctk = root["chat_template_kwargs"]?.jsonObject?.toMutableMap() ?: mutableMapOf() - ctk["enable_thinking"] = JsonPrimitive(false) - root["chat_template_kwargs"] = JsonObject(ctk) - } - } - - val provider = root["provider"]?.jsonObject?.toMutableMap() ?: mutableMapOf() - val existing = provider["ignore"]?.jsonArray?.map { it.jsonPrimitive.content } ?: emptyList() - provider["ignore"] = JsonArray((existing + excluded).distinct().map { JsonPrimitive(it) }) - if (provider["allow_fallbacks"] == null) { - provider["allow_fallbacks"] = JsonPrimitive(false) - } - root["provider"] = JsonObject(provider) - return JsonObject(root).toString() -} - -private suspend fun handleModels( - call: ApplicationCall, - upstream: String, - apiKey: String, - thinking: List, -) { - val req = HttpRequest.newBuilder() - .uri(URI.create(upstream.trimEnd('/') + "/models")) - .header("Authorization", "Bearer $apiKey") - .GET() - .build() - val resp = http.send(req, HttpResponse.BodyHandlers.ofByteArray()) - val ct = resp.headers().firstValue("content-type").orElse("application/json") - val body = if (thinking.isNotEmpty()) { - patchModelsCatalog(String(resp.body(), Charsets.UTF_8), thinking).toByteArray(Charsets.UTF_8) - } else { - resp.body() - } - call.respondBytes(body, ContentType.parse(ct), HttpStatusCode.fromValue(resp.statusCode())) -} - -/** - * Дублировать «думающие» модели в каталоге с суффиксом "-no-think": - * каждая модель, чей id содержит любую из подстрок [thinking], получает копию - * с id = "<оригинал>-no-think". - */ -internal fun patchModelsCatalog(raw: String, thinking: List): String { - val root = json.parseToJsonElement(raw).jsonObject.toMutableMap() - val data = root["data"]?.jsonArray?.map { it.jsonObject } ?: emptyList() - if (data.isEmpty()) return raw - val copies = data.filter { m -> - val id = m["id"]?.jsonPrimitive?.content ?: "" - thinking.any { id.contains(it) } - }.map { m -> - val id = m["id"]?.jsonPrimitive?.content ?: "" - JsonObject(m.toMutableMap().apply { this["id"] = JsonPrimitive(id + "-no-think") }) - } - if (copies.isEmpty()) return raw - root["data"] = JsonArray(data + copies) - return JsonObject(root).toString() -} - -/** - * Ожидание CompletableFuture с пробросом отмены корутины на апстрим-запрос: - * если клиент оборвал соединение (Ktor отменяет корутину), рвём и апстрим — - * upstream (llama.cpp/sglang) видит обрыв и отменяет генерацию (слот свободен). - */ -private suspend fun CompletableFuture.awaitOrCancel(): T = - suspendCancellableCoroutine { cont -> - this.whenComplete { res, err -> - if (err != null) cont.resumeWithException(err) else cont.resume(res) - } - cont.invokeOnCancellation { this.cancel(true) } - } - -/** - * Собрать полный chat.completion из SSE-чанков апстрима (для non-stream клиентов). - */ -private fun rebuildFromChunks(sse: String): String { - var content = StringBuilder() - var reasoning = StringBuilder() - var finish = "stop" - var id = "" - var model = "" - val created = System.currentTimeMillis() / 1000 - sse.lineSequence().forEach { line -> - if (!line.startsWith("data:")) return@forEach - val data = line.removePrefix("data:").trim() - if (data.isEmpty() || data == "[DONE]") return@forEach - try { - val obj = json.parseToJsonElement(data).jsonObject - if (id.isEmpty()) id = obj["id"]?.jsonPrimitive?.content ?: "" - if (model.isEmpty()) model = obj["model"]?.jsonPrimitive?.content ?: "" - val choice = obj["choices"]?.jsonArray?.firstOrNull()?.jsonObject - if (choice != null) { - choice["finish_reason"]?.jsonPrimitive?.content - ?.takeIf { it.isNotEmpty() && it != "null" }?.let { finish = it } - val delta = choice["delta"]?.jsonObject - delta?.get("content")?.jsonPrimitive?.content - ?.takeIf { it != "null" }?.let { content.append(it) } - delta?.get("reasoning_content")?.jsonPrimitive?.content - ?.takeIf { it != "null" }?.let { reasoning.append(it) } - } - } catch (_: Exception) {} - } - val msg = JsonObject(mutableMapOf( - "role" to JsonPrimitive("assistant"), - "content" to JsonPrimitive(content.toString()), - "reasoning" to JsonPrimitive(reasoning.toString()), - )) - val choice = JsonObject(mutableMapOf( - "index" to JsonPrimitive(0), - "message" to msg, - "finish_reason" to JsonPrimitive(finish), - )) - return JsonObject(mutableMapOf( - "id" to JsonPrimitive(id), - "object" to JsonPrimitive("chat.completion"), - "created" to JsonPrimitive(created), - "model" to JsonPrimitive(model), - "choices" to JsonArray(listOf(choice)), - )).toString() -}