Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 subochev
|
558b2f925b | ||
|
subochev
|
334015a2bd |
@@ -9,6 +9,10 @@ jobs:
|
||||
steps:
|
||||
- name: 'Checkout'
|
||||
uses: https://github.com/actions/checkout@v4
|
||||
- name: 'Run tests'
|
||||
uses: https://git.binom.pw/subochev/devops/build-gradle@main
|
||||
with:
|
||||
target: jvmTest
|
||||
- name: 'Build jar'
|
||||
uses: https://git.binom.pw/subochev/devops/build-gradle@main
|
||||
with:
|
||||
|
||||
+49
@@ -0,0 +1,49 @@
|
||||
# Тестирование llm-proxy
|
||||
|
||||
Все тесты — обычные модульные, живут в `src/commonTest/kotlin/pw/binom/llmproxy/`.
|
||||
|
||||
## Запуск
|
||||
|
||||
- `./gradlew jvmTest` — прогнать все тесты;
|
||||
- `./gradlew clean jvmTest fatJar` — полная сборка с нуля;
|
||||
- результат смотреть в `build/test-results/jvmTest/*.xml` (атрибуты `tests`/`failures`/`errors`), потому что строки вида «N tests completed» печатаются только при падениях.
|
||||
|
||||
## Что покрыто
|
||||
|
||||
| Файл | Что проверяет |
|
||||
| --- | --- |
|
||||
| `ThinkTagSplitterTest` | Автомат рассечения think-тегов: passthrough при off, вырезание рассуждений при split, отбрасывание при strip, удержание разрезанного тега, несколько блоков, незакрытый блок; |
|
||||
| `ConfigLogicTest` | Разбор конфига, приоритет источников (апстрим важнее провайдера), слияние патчей, выбор апстрима и лимиты конкурентности, заголовки, сессии; |
|
||||
| `ThinkTagTransformTest` | Non-stream путь `transformThinkMessage`: перенос рассуждений в `reasoning_content`, дописывание к уже имеющемуся, strip, незакрытый блок, отсутствие изменений → null; |
|
||||
| `ThinkTagChunkTest` | SSE-чанки `transformThinkChunk`: удержание хвоста тега между чанками, независимые сплиттеры по index, удаление пустого `content`; |
|
||||
| `ThinkTagStreamTest` | Обвязка стрима `streamSseWithThinkTags`: разрез тега между data-событиями, сброс удержанного хвоста в финиш-чанке, прохождение служебных строк и `[DONE]`, битый JSON, чанк без choices, strip. |
|
||||
| `StreamDoneContractTest` | Контракт конца SSE: детектор `finish_reason`/`[DONE]` (в т.ч. разрезанных границей чтения), дописывание `data: [DONE]\n\n` в сыром passthrough и в think-обвязке при штатном закрытии без маркера, отсутствие маркера при обрыве без `finish_reason`, отсутствие дублирования. |
|
||||
|
||||
## Проверка качества тестов (мутационная приёмка)
|
||||
|
||||
Приём по шагам:
|
||||
|
||||
1. Забэкапить файл.
|
||||
2. Внести РОВНО одну поломку в боевой код.
|
||||
3. Прогнать `./gradlew cleanJvmTest jvmTest`.
|
||||
4. Посмотреть XML — тест, который не упал, считается пустым.
|
||||
5. Откатить (`git checkout -- <файл>`).
|
||||
|
||||
Обязательно: `cleanJvmTest` обязателен, иначе прогон не перезапустится.
|
||||
|
||||
Проверенные мутации, каждая из которых ДОЛЖНА ронять тесты:
|
||||
|
||||
- `transformThinkMessage` возвращает null → падают тесты non-stream;
|
||||
- `transformThinkChunk` возвращает null → падают тесты чанков и стрима;
|
||||
- блок финиш-чанка в `streamSseWithThinkTags` не выполняется → падает тест про удержанный хвост;
|
||||
- `holdableSuffix` всегда 0 (хвост тега не удерживается) → падают тесты автомата, чанков и стрима;
|
||||
- `strip` начинает отдавать рассуждения → падает тест автомата;
|
||||
- `missingDoneMarker` всегда `false` → падают тесты сырого пути (`rawStreamIsByteExactAndAppendsDone`) и тест разрезанного маркера;
|
||||
- `if (sawFinishReason && !sawDone)` → `if (!sawDone)` в `streamSseWithThinkTags` → падает `thinkStreamTruncatedNeedsNoMarker`;
|
||||
- `if (sawFinishReason && !sawDone)` → `if (false)` в `streamSseWithThinkTags` → падает `thinkStreamAppendsDoneWhenUpstreamClosedWithoutMarker`.
|
||||
|
||||
Правило: боевой код нельзя подгонять под тест; если тест не проходит, неверен тест.
|
||||
|
||||
## Известное ограничение
|
||||
|
||||
Живой стрим в реальном апстриме модульными тестами не проверяется: обвязка испытывается на синтетическом SSE через каналы ktor. Реальный апстрим проверяется только после деплоя.
|
||||
@@ -31,6 +31,7 @@ kotlin {
|
||||
val commonTest by getting {
|
||||
dependencies {
|
||||
implementation(libs.kotlin.test)
|
||||
implementation(libs.kotlinx.coroutines.test)
|
||||
}
|
||||
}
|
||||
val jvmMain by getting {
|
||||
|
||||
@@ -0,0 +1,280 @@
|
||||
# ТЗ: выравнивание контракта конца SSE (дописать `data: [DONE]`)
|
||||
|
||||
Задача-источник: Vikunja **#71** («в opencode модель додумала, идёт ожидание ~45 с»).
|
||||
|
||||
## 1. Что происходит сейчас (проверено живыми замерами 11.09.2026)
|
||||
|
||||
Цепочка: opencode → `llm.binom.pw` (Bifrost v1.5.0) → `llm-proxy:8100` → `api.minimax.io`.
|
||||
|
||||
- **MiniMax не присылает терминатор `data: [DONE]`.** После чанка с `finish_reason: "stop"` он просто закрывает
|
||||
сокет (сразу). Проверено и с `reasoning_split`, и с `stream_options.include_usage`.
|
||||
- **llm-proxy отдаёт поток корректно**, но клиент ждёт маркер: при `think_tags: off` (это как раз MiniMax-апстрим
|
||||
`minimax-m3` модели `codding-big`) работает сырой байтовый passthrough — что прислал апстрим, то и ушло.
|
||||
Маркера нет → нет маркера.
|
||||
- **Bifrost v1.5.0** для custom-провайдеров считает `[DONE]` обязательным (`ProviderSendsDoneMarker` = true;
|
||||
флага `custom_provider_config.does_not_send_done_marker` в этой версии нет). Маркера нет → его цикл чтения
|
||||
не завершается, и стрим закрывается только когда умирает keep-alive-сокет llm-proxy — через **45.00 с**
|
||||
(Ktor CIO `connectionIdleTimeoutSeconds = 45`). После этого Bifrost дописывает клиенту `[DONE]` и закрывает поток.
|
||||
Отсюда «модель додумала, а opencode ещё думает».
|
||||
- Контроль: тот же Bifrost и та же тула, апстрим со своим `[DONE]` (RouterAI `local/deepseek/deepseek-v4-flash-0731`)
|
||||
— закрытие за **0.00 с**.
|
||||
|
||||
## 2. Требуемое поведение (контракт)
|
||||
|
||||
Стрим, который отдаёт llm-proxy клиенту, **всегда** заканчивается маркером `data: [DONE]\n\n`, если апстрим
|
||||
завершил поток штатно (в потоке был непустой `finish_reason`), — даже когда сам апстрим маркер не прислал.
|
||||
|
||||
Правило ровно в двух частях:
|
||||
|
||||
1. Апстрим закрыл поток, и в потоке был непустой `finish_reason`, но `data: [DONE]` мы не видели →
|
||||
дописать клиенту `data: [DONE]\n\n` и `flush()`.
|
||||
2. Апстрим оборвался, **не** прислав `finish_reason` (усечённый/битый поток) → маркер **НЕ** дописывать,
|
||||
отдать как есть.
|
||||
|
||||
Пункт 2 — не забыть и не «улучшить»: именно по отсутствию маркера Bifrost отличает усечённый стрим от
|
||||
успешного (его детект `SSEStreamEndedOnMarker`). Если дописать маркер при обрыве без `finish_reason`,
|
||||
мы превратим обрыв в «успех» и сломаем диагностику.
|
||||
|
||||
Обязательно: **байты уже существующих чанков не меняются** (passthrough остаётся byte-exact), маркер
|
||||
дописывается только В КОНЦЕ, только один раз, и никогда не дублируется, если апстрим прислал свой.
|
||||
|
||||
## 3. Правки в коде
|
||||
|
||||
Файл **один**: `src/commonMain/kotlin/pw/binom/llmproxy/Main.kt`.
|
||||
|
||||
### 3.1. Константы и хелперы (добавить рядом с блоком think-хелперов)
|
||||
|
||||
```kotlin
|
||||
/** Терминальный маркер SSE, который клиенты (в т.ч. Bifrost) ждут как конец потока. */
|
||||
private const val SSE_DONE_MARKER = "data: [DONE]\n\n"
|
||||
|
||||
/** Размер скользящего окна детектора: маркер может разрезаться границей чтения. */
|
||||
private const val STREAM_SCAN_WINDOW = 128
|
||||
|
||||
/**
|
||||
* Есть ли в тексте `finish_reason` со значением, отличным от `null`. Экранированные
|
||||
* вхождения (`\"finish_reason\\\":\\\"stop\\\"` внутри содержимого ответа) не считаются:
|
||||
* смотрим только на неэкранированную кавычку.
|
||||
*/
|
||||
internal fun hasNonNullFinishReasonText(text: String): Boolean {
|
||||
var from = 0
|
||||
while (true) {
|
||||
val i = text.indexOf("finish_reason", from)
|
||||
if (i < 0) return false
|
||||
val quote = i - 1
|
||||
val escaped = quote >= 0 && text[quote] == '"' && quote > 0 && text[quote - 1] == '\\'
|
||||
if (!escaped) {
|
||||
val rest = text.substring(i + "finish_reason".length)
|
||||
.dropWhile { it == ' ' || it == ':' || it == '"' }
|
||||
if (!rest.startsWith("null")) return true
|
||||
}
|
||||
from = i + 1
|
||||
}
|
||||
}
|
||||
|
||||
/** Структурная проверка чанка: в `choices[*].finish_reason` есть непустая строка. */
|
||||
internal fun hasFinishReason(obj: JsonObject): Boolean =
|
||||
obj["choices"]?.jsonArray?.any { el ->
|
||||
val fr = (el.jsonObject["finish_reason"] as? JsonPrimitive)?.takeIf { it.isString }?.content
|
||||
!fr.isNullOrEmpty()
|
||||
} ?: false
|
||||
|
||||
/**
|
||||
* Детектор контракта конца SSE для сырого passthrough-потока (`think_tags: off`).
|
||||
* Копит скользящее окно последних байт, чтобы маркер `finish_reason`/`[DONE]`,
|
||||
* разрезанный границей чтения, всё равно был распознан, и запоминает два факта:
|
||||
* видели ли непустой `finish_reason` и видели ли `[DONE]`.
|
||||
*/
|
||||
internal class StreamEndDetector(private val windowSize: Int = STREAM_SCAN_WINDOW) {
|
||||
var sawFinishReason = false
|
||||
private set
|
||||
|
||||
var sawDone = false
|
||||
private set
|
||||
|
||||
private var window: String = ""
|
||||
|
||||
fun feed(bytes: ByteArray, from: Int = 0, length: Int = bytes.size) {
|
||||
if (length <= 0) return
|
||||
val scan = window + bytes.decodeToString(from, from + length)
|
||||
if (!sawDone && scan.contains("[DONE]")) sawDone = true
|
||||
if (!sawFinishReason && hasNonNullFinishReasonText(scan)) sawFinishReason = true
|
||||
window = if (scan.length > windowSize) scan.substring(scan.length - windowSize) else scan
|
||||
}
|
||||
|
||||
/**
|
||||
* Маркер нужен, только если апстрим закрыл поток после штатного `finish_reason`,
|
||||
* но сам `[DONE]` не прислал. При обрыве БЕЗ `finish_reason` маркер НЕ дописываем.
|
||||
*/
|
||||
val missingDoneMarker: Boolean get() = sawFinishReason && !sawDone
|
||||
}
|
||||
|
||||
/**
|
||||
* Сырой байтовый passthrough стрима (`think_tags: off`) с выравниванием контракта
|
||||
* конца потока: байты уходят клиенту как есть, а если апстрим закрыл поток, не
|
||||
* прислав `data: [DONE]`, но `finish_reason` в потоке был — дописываем маркер.
|
||||
*/
|
||||
internal suspend fun ByteWriteChannel.streamRawWithDoneContract(source: ByteReadChannel) {
|
||||
val detector = StreamEndDetector()
|
||||
val buf = ByteArray(8192)
|
||||
while (true) {
|
||||
val n = source.readAvailable(buf)
|
||||
if (n == -1) break
|
||||
if (n > 0) {
|
||||
writeFully(buf, 0, n)
|
||||
flush()
|
||||
detector.feed(buf, 0, n)
|
||||
}
|
||||
}
|
||||
if (detector.missingDoneMarker) {
|
||||
emitUtf8(SSE_DONE_MARKER)
|
||||
flush()
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Импорты под новые типы (`ByteReadChannel`, `ByteWriteChannel`, `readAvailable`, `writeFully`, `flush`,
|
||||
`jsonArray`, `jsonObject`, `JsonPrimitive`) — проверить, что они уже есть в файле; добавлять только недостающие.
|
||||
|
||||
### 3.2. Правка ветки `think_tags: off` в `handleChat`
|
||||
|
||||
Было (внутри `call.respondBytesWriter(...)`):
|
||||
|
||||
```kotlin
|
||||
if (thinkMode == "off") {
|
||||
// Флажок не выставлен — сырой байтовый passthrough как раньше.
|
||||
val buf = ByteArray(8192)
|
||||
while (true) {
|
||||
val n = ch.readAvailable(buf)
|
||||
if (n == -1) break
|
||||
if (n > 0) {
|
||||
writeFully(buf, 0, n)
|
||||
flush()
|
||||
}
|
||||
}
|
||||
} else {
|
||||
streamSseWithThinkTags(ch, thinkMode)
|
||||
}
|
||||
```
|
||||
|
||||
Стало:
|
||||
|
||||
```kotlin
|
||||
if (thinkMode == "off") {
|
||||
// Флажок не выставлен — сырой байтовый passthrough + выравнивание конца потока.
|
||||
ch.streamRawWithDoneContract(this)
|
||||
} else {
|
||||
streamSseWithThinkTags(ch, thinkMode)
|
||||
}
|
||||
```
|
||||
(если `this` в этом контексте — не `ByteWriteChannel`, использовать тот ресивер, который уже использовался
|
||||
существующими вызовами `writeFully`/`flush` в этой лямбде; компилятор — источник истины).
|
||||
|
||||
### 3.3. Правка `streamSseWithThinkTags`
|
||||
|
||||
Добавить два флага и запись фактов:
|
||||
|
||||
```kotlin
|
||||
internal suspend fun ByteWriteChannel.streamSseWithThinkTags(source: ByteReadChannel, thinkMode: String) {
|
||||
val splitters = mutableMapOf<Int, ThinkTagSplitter>()
|
||||
val addReasoning = thinkMode == "split"
|
||||
var sawDone = false
|
||||
var sawFinishReason = false
|
||||
while (true) {
|
||||
val line = source.readLine(LineEnding.Lenient) ?: break
|
||||
when {
|
||||
line.startsWith("data:") -> {
|
||||
val payload = line.removePrefix("data:").trim()
|
||||
if (payload == "[DONE]") {
|
||||
sawDone = true
|
||||
emitUtf8(SSE_DONE_MARKER)
|
||||
} else {
|
||||
val obj = runCatching { json.parseToJsonElement(payload).jsonObject }.getOrNull()
|
||||
if (obj != null && hasFinishReason(obj)) sawFinishReason = true
|
||||
val out = obj?.let { transformThinkChunk(it, splitters, thinkMode, addReasoning) }
|
||||
if (out == null) emitUtf8("$line\n\n") else emitUtf8("data: $out\n\n")
|
||||
}
|
||||
}
|
||||
else -> emitUtf8("$line\n")
|
||||
}
|
||||
flush()
|
||||
}
|
||||
// ... существующий блок сброса удержанных хвостов сплиттеров — БЕЗ изменений ...
|
||||
// НОВОЕ: после хвостов дописать маркер, если апстрим его не прислал, а finish_reason был.
|
||||
if (sawFinishReason && !sawDone) {
|
||||
emitUtf8(SSE_DONE_MARKER)
|
||||
flush()
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Порядок важен: маркер дописывается **после** блока сброса хвостов (сначала текст, потом терминатор).
|
||||
Остальную логику функции (`transformThinkChunk`, `emitUtf8("$line\n")` для служебных строк) не менять.
|
||||
|
||||
## 4. Тесты (новый файл)
|
||||
|
||||
`src/commonTest/kotlin/pw/binom/llmproxy/StreamDoneContractTest.kt` — по образцу существующего
|
||||
`ThinkTagStreamTest` (тот же приём: `ByteChannel(autoFlush = true)`, запись входа, `close(null)`,
|
||||
чтение выхода в буфер; для сырого потока вызывать `streamRawWithDoneContract`).
|
||||
|
||||
Обязательные случаи (имена — как ниже, чтобы можно было ссылаться в отчёте):
|
||||
|
||||
| Тест | Вход | Ожидание |
|
||||
| --- | --- | --- |
|
||||
| `textHelperSeesFinishReason` | строка `{"choices":[{"finish_reason":"stop"}]}` | `hasNonNullFinishReasonText` = true |
|
||||
| `textHelperIgnoresNullFinishReason` | `{"choices":[{"finish_reason":null}]}` | false |
|
||||
| `textHelperIgnoresEscapedOccurrenceInContent` | `{"delta":{"content":"\"finish_reason\":\"stop\""}}` | false |
|
||||
| `detectorSeesFinishReasonSplitAcrossReads` | два `feed`: `...finish_rea` + `son":"stop"...` | `sawFinishReason` = true, `missingDoneMarker` = true |
|
||||
| `detectorSeesDoneSplitAcrossReads` | два `feed`: `data: [DO` + `NE]` | `sawDone` = true, `missingDoneMarker` = false |
|
||||
| `detectorSeesDoneAndFinishReason` | `finish_reason` + `[DONE]` в одном feed | `missingDoneMarker` = false |
|
||||
| `detectorTruncatedStreamNeedsNoMarker` | `feed` с контентом без `finish_reason` | `missingDoneMarker` = false |
|
||||
| `rawStreamIsByteExactAndAppendsDone` | SSE с `finish_reason`, без `[DONE]` | выход = вход + `"data: [DONE]\n\n"` (байт-в-байт префикс), маркер ровно один |
|
||||
| `rawStreamTruncatedPassesThroughUntouched` | SSE без `finish_reason` и без `[DONE]` | выход = вход (ни одного `[DONE]`) |
|
||||
| `rawStreamKeepsUpstreamDoneMarker` | SSE с `finish_reason` и своим `[DONE]` | `[DONE]` ровно один (не дублируется) |
|
||||
| `thinkStreamAppendsDoneWhenUpstreamClosedWithoutMarker` | `streamSseWithThinkTags(..., "split")`, чанк с `finish_reason:"stop"`, без `[DONE]` | выход заканчивается `data: [DONE]\n\n`, ровно один маркер |
|
||||
| `thinkStreamTruncatedNeedsNoMarker` | mode `split`, чанк с контентом, без `finish_reason` | в выходе нет `[DONE]` |
|
||||
| `thinkStreamKeepsSingleDone` | mode `split`, `finish_reason` + `data: [DONE]` | ровно один `[DONE]` |
|
||||
|
||||
## 5. Мутационная приёмка (по TESTING.md)
|
||||
|
||||
После того как тесты зелёные, проверить их силу — по одному изменению, каждый раз с
|
||||
`./gradlew cleanJvmTest jvmTest`, затем `git checkout --` на файл:
|
||||
|
||||
1. `missingDoneMarker` → всегда `false` (детектор сырого пути) — должны упасть тесты сырого пути
|
||||
(`rawStreamIsByteExactAndAppendsDone`) и тест разрезанного маркера;
|
||||
2. `if (sawFinishReason && !sawDone)` → `if (!sawDone)` в `streamSseWithThinkTags` — должен упасть
|
||||
`thinkStreamTruncatedNeedsNoMarker` (это защита от «обрыв выглядит как успех»);
|
||||
3. `if (sawFinishReason && !sawDone)` → `if (false)` — должен упасть
|
||||
`thinkStreamAppendsDoneWhenUpstreamClosedWithoutMarker`.
|
||||
|
||||
Тест, который при такой поломке остаётся зелёным, считается пустым.
|
||||
|
||||
## 6. Команды проверки
|
||||
|
||||
```bash
|
||||
./gradlew jvmTest # все тесты зелёные; сейчас в проекте 66 тестов, стало больше
|
||||
./gradlew cleanJvmTest jvmTest # прогон с нуля (обязателен для мутационной приёмки)
|
||||
./gradlew clean jvmTest fatJar # полная сборка
|
||||
```
|
||||
Числа тестов смотреть в `build/test-results/jvmTest/*.xml` (атрибуты `tests` / `failures` / `errors`),
|
||||
а не по строкам в консоли.
|
||||
|
||||
## 7. Границы работ
|
||||
|
||||
- Трогать **только**: `src/commonMain/kotlin/pw/binom/llmproxy/Main.kt`,
|
||||
`src/commonTest/kotlin/pw/binom/llmproxy/StreamDoneContractTest.kt`, `TESTING.md` (добавить строку про
|
||||
новый тест-файл и пункты мутационной приёмки).
|
||||
- **НЕ** менять `ThinkTagSplitter.kt`, non-stream путь, `transformThinkMessage`, `transformThinkChunk`,
|
||||
фрейминг существующих чанков, `README.md`/`CONFIG.md`.
|
||||
- **НЕ** добавлять зависимости, не менять версии в `build.gradle.kts`, не менять `config.yaml`.
|
||||
- **НЕ** коммитить и не пушить.
|
||||
- Внешние сети/апстримы не дёргать — только локальные тесты.
|
||||
|
||||
## 8. Вне области (и почему)
|
||||
|
||||
- **Обрыв без `finish_reason` остаётся без маркера** — так усечённый стрим не выглядит успешным
|
||||
(см. п. 2). Да, в этом случае Bifrost снова будет ждать закрытия сокета; это осознанный выбор
|
||||
в пользу честности контракта, а не «ускорения любой ценой».
|
||||
- **Non-stream путь** не трогаем: там тула сама собирает ответ из чанков (`rebuildFromChunks`) и
|
||||
маркер не нужен.
|
||||
- **Bifrost и сам MiniMax** не трогаем: фикс на нашем хопе лечит всех клиентов сразу.
|
||||
@@ -19,6 +19,7 @@ kotlinx-serialization-json = { module = "org.jetbrains.kotlinx:kotlinx-serializa
|
||||
yamlkt = { module = "net.mamoe.yamlkt:yamlkt", version.ref = "yamlkt" }
|
||||
kotlinx-coroutines-core = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-core", version.ref = "coroutines" }
|
||||
kotlinx-datetime = { module = "org.jetbrains.kotlinx:kotlinx-datetime", version.ref = "datetime" }
|
||||
kotlinx-coroutines-test = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-test", version.ref = "coroutines" }
|
||||
kotlinx-io-core = { module = "org.jetbrains.kotlinx:kotlinx-io-core", version.ref = "kotlinxIo" }
|
||||
logback-classic = { module = "ch.qos.logback:logback-classic", version.ref = "logback" }
|
||||
kotlin-test = { module = "org.jetbrains.kotlin:kotlin-test" }
|
||||
|
||||
@@ -242,16 +242,8 @@ private suspend fun handleChat(
|
||||
call.respondBytesWriter(ContentType.parse(ct), status) {
|
||||
val ch = resp.body<ByteReadChannel>()
|
||||
if (thinkMode == "off") {
|
||||
// Флажок не выставлен — сырой байтовый passthrough как раньше.
|
||||
val buf = ByteArray(8192)
|
||||
while (true) {
|
||||
val n = ch.readAvailable(buf)
|
||||
if (n == -1) break
|
||||
if (n > 0) {
|
||||
writeFully(buf, 0, n)
|
||||
flush()
|
||||
}
|
||||
}
|
||||
// Флажок не выставлен — сырой байтовый passthrough + выравнивание конца потока.
|
||||
streamRawWithDoneContract(ch)
|
||||
} else {
|
||||
streamSseWithThinkTags(ch, thinkMode)
|
||||
}
|
||||
@@ -644,6 +636,93 @@ internal fun transformThinkMessage(obj: JsonObject, thinkMode: String): String?
|
||||
return JsonObject(obj.toMutableMap().apply { this["choices"] = JsonArray(newChoices) }).toString()
|
||||
}
|
||||
|
||||
/** Терминальный маркер SSE, который клиенты (в т.ч. Bifrost) ждут как конец потока. */
|
||||
private const val SSE_DONE_MARKER = "data: [DONE]\n\n"
|
||||
|
||||
/** Размер скользящего окна детектора: маркер может разрезаться границей чтения. */
|
||||
private const val STREAM_SCAN_WINDOW = 128
|
||||
|
||||
/**
|
||||
* Есть ли в тексте `finish_reason` со значением, отличным от `null`. Экранированные
|
||||
* вхождения (`\"finish_reason\\\":\\\"stop\\\"` внутри содержимого ответа) не считаются:
|
||||
* смотрим только на неэкранированную кавычку.
|
||||
*/
|
||||
internal fun hasNonNullFinishReasonText(text: String): Boolean {
|
||||
var from = 0
|
||||
while (true) {
|
||||
val i = text.indexOf("finish_reason", from)
|
||||
if (i < 0) return false
|
||||
val quote = i - 1
|
||||
val escaped = quote >= 0 && text[quote] == '"' && quote > 0 && text[quote - 1] == '\\'
|
||||
if (!escaped) {
|
||||
val rest = text.substring(i + "finish_reason".length)
|
||||
.dropWhile { it == ' ' || it == ':' || it == '"' }
|
||||
if (!rest.startsWith("null")) return true
|
||||
}
|
||||
from = i + 1
|
||||
}
|
||||
}
|
||||
|
||||
/** Структурная проверка чанка: в `choices[*].finish_reason` есть непустая строка. */
|
||||
internal fun hasFinishReason(obj: JsonObject): Boolean =
|
||||
obj["choices"]?.jsonArray?.any { el ->
|
||||
val fr = (el.jsonObject["finish_reason"] as? JsonPrimitive)?.takeIf { it.isString }?.content
|
||||
!fr.isNullOrEmpty()
|
||||
} ?: false
|
||||
|
||||
/**
|
||||
* Детектор контракта конца SSE для сырого passthrough-потока (`think_tags: off`).
|
||||
* Копит скользящее окно последних байт, чтобы маркер `finish_reason`/`[DONE]`,
|
||||
* разрезанный границей чтения, всё равно был распознан, и запоминает два факта:
|
||||
* видели ли непустой `finish_reason` и видели ли `[DONE]`.
|
||||
*/
|
||||
internal class StreamEndDetector(private val windowSize: Int = STREAM_SCAN_WINDOW) {
|
||||
var sawFinishReason = false
|
||||
private set
|
||||
|
||||
var sawDone = false
|
||||
private set
|
||||
|
||||
private var window: String = ""
|
||||
|
||||
fun feed(bytes: ByteArray, from: Int = 0, length: Int = bytes.size) {
|
||||
if (length <= 0) return
|
||||
val scan = window + bytes.decodeToString(from, from + length)
|
||||
if (!sawDone && scan.contains("[DONE]")) sawDone = true
|
||||
if (!sawFinishReason && hasNonNullFinishReasonText(scan)) sawFinishReason = true
|
||||
window = if (scan.length > windowSize) scan.substring(scan.length - windowSize) else scan
|
||||
}
|
||||
|
||||
/**
|
||||
* Маркер нужен, только если апстрим закрыл поток после штатного `finish_reason`,
|
||||
* но сам `[DONE]` не прислал. При обрыве БЕЗ `finish_reason` маркер НЕ дописываем.
|
||||
*/
|
||||
val missingDoneMarker: Boolean get() = sawFinishReason && !sawDone
|
||||
}
|
||||
|
||||
/**
|
||||
* Сырой байтовый passthrough стрима (`think_tags: off`) с выравниванием контракта
|
||||
* конца потока: байты уходят клиенту как есть, а если апстрим закрыл поток, не
|
||||
* прислав `data: [DONE]`, но `finish_reason` в потоке был — дописываем маркер.
|
||||
*/
|
||||
internal suspend fun ByteWriteChannel.streamRawWithDoneContract(source: ByteReadChannel) {
|
||||
val detector = StreamEndDetector()
|
||||
val buf = ByteArray(8192)
|
||||
while (true) {
|
||||
val n = source.readAvailable(buf)
|
||||
if (n == -1) break
|
||||
if (n > 0) {
|
||||
writeFully(buf, 0, n)
|
||||
flush()
|
||||
detector.feed(buf, 0, n)
|
||||
}
|
||||
}
|
||||
if (detector.missingDoneMarker) {
|
||||
emitUtf8(SSE_DONE_MARKER)
|
||||
flush()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Построчный разбор SSE-стрима с рассечением think-тегов. Строки, не начинающиеся
|
||||
* с `data:`, и `data: [DONE]` уходят клиенту без изменений (с `\n`). Прочие
|
||||
@@ -652,18 +731,22 @@ internal fun transformThinkMessage(obj: JsonObject, thinkMode: String): String?
|
||||
* исходная строка. Каждую строку сразу `flush()`, чтобы стрим не «залипал» в
|
||||
* буфере. В конце потока накопленные хвосты сплиттеров сбрасываются финиш-чанком.
|
||||
*/
|
||||
private suspend fun ByteWriteChannel.streamSseWithThinkTags(source: ByteReadChannel, thinkMode: String) {
|
||||
internal suspend fun ByteWriteChannel.streamSseWithThinkTags(source: ByteReadChannel, thinkMode: String) {
|
||||
val splitters = mutableMapOf<Int, ThinkTagSplitter>()
|
||||
val addReasoning = thinkMode == "split"
|
||||
var sawDone = false
|
||||
var sawFinishReason = false
|
||||
while (true) {
|
||||
val line = source.readLine(LineEnding.Lenient) ?: break
|
||||
when {
|
||||
line.startsWith("data:") -> {
|
||||
val payload = line.removePrefix("data:").trim()
|
||||
if (payload == "[DONE]") {
|
||||
emitUtf8("data: [DONE]\n\n")
|
||||
sawDone = true
|
||||
emitUtf8(SSE_DONE_MARKER)
|
||||
} else {
|
||||
val obj = runCatching { json.parseToJsonElement(payload).jsonObject }.getOrNull()
|
||||
if (obj != null && hasFinishReason(obj)) sawFinishReason = true
|
||||
val out = obj?.let { transformThinkChunk(it, splitters, thinkMode, addReasoning) }
|
||||
if (out == null) emitUtf8("$line\n\n") else emitUtf8("data: $out\n\n")
|
||||
}
|
||||
@@ -699,6 +782,12 @@ private suspend fun ByteWriteChannel.streamSseWithThinkTags(source: ByteReadChan
|
||||
flush()
|
||||
}
|
||||
}
|
||||
// После сброса хвостов дописываем маркер, только если апстрим его не прислал,
|
||||
// а `finish_reason` в потоке был (при обрыве без него маркер НЕ дописываем).
|
||||
if (sawFinishReason && !sawDone) {
|
||||
emitUtf8(SSE_DONE_MARKER)
|
||||
flush()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -710,7 +799,7 @@ private suspend fun ByteWriteChannel.streamSseWithThinkTags(source: ByteReadChan
|
||||
* Чанк без `choices` или без строкового `delta.content` не меняется —
|
||||
* возвращается null (отдать исходную строку как есть).
|
||||
*/
|
||||
private fun transformThinkChunk(
|
||||
internal fun transformThinkChunk(
|
||||
obj: JsonObject,
|
||||
splitters: MutableMap<Int, ThinkTagSplitter>,
|
||||
thinkMode: String,
|
||||
|
||||
@@ -533,7 +533,7 @@ class ConfigLogicTest {
|
||||
|
||||
// значение у апстрима — берётся оно, провайдер игнорируется
|
||||
assertEquals("split", effectiveThinkTags(upSplit, null))
|
||||
assertEquals("split", effectiveThinkTags(upSplit, prov("true")))
|
||||
assertEquals("split", effectiveThinkTags(upSplit, prov("strip")))
|
||||
|
||||
// апстрим null — берётся провайдерский
|
||||
assertEquals("split", effectiveThinkTags(upNull, prov("split")))
|
||||
|
||||
@@ -0,0 +1,176 @@
|
||||
package pw.binom.llmproxy
|
||||
|
||||
import io.ktor.utils.io.ByteChannel
|
||||
import io.ktor.utils.io.close
|
||||
import io.ktor.utils.io.readAvailable
|
||||
import io.ktor.utils.io.writeFully
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFalse
|
||||
import kotlin.test.assertTrue
|
||||
import kotlinx.coroutines.test.runTest
|
||||
|
||||
class StreamDoneContractTest {
|
||||
|
||||
/** Терминальный маркер, который должен оказаться (или не оказаться) в конце вывода. */
|
||||
private val done = "data: [DONE]\n\n"
|
||||
|
||||
/** Прогнать сырой passthrough-поток через боевую обвязку и вернуть записанные байты как строку. */
|
||||
private suspend fun runRaw(input: String): String {
|
||||
val src = ByteChannel(autoFlush = true)
|
||||
src.writeFully(input.encodeToByteArray())
|
||||
src.close(null)
|
||||
val out = ByteChannel(autoFlush = true)
|
||||
out.streamRawWithDoneContract(src)
|
||||
out.close(null)
|
||||
return readAll(out)
|
||||
}
|
||||
|
||||
/** Прогнать SSE через think-обвязку и вернуть записанные байты как строку. */
|
||||
private suspend fun runThink(input: String, mode: String): String {
|
||||
val src = ByteChannel(autoFlush = true)
|
||||
src.writeFully(input.encodeToByteArray())
|
||||
src.close(null)
|
||||
val out = ByteChannel(autoFlush = true)
|
||||
out.streamSseWithThinkTags(src, mode)
|
||||
out.close(null)
|
||||
return readAll(out)
|
||||
}
|
||||
|
||||
private suspend fun readAll(ch: ByteChannel): String {
|
||||
val sb = StringBuilder()
|
||||
val buf = ByteArray(512)
|
||||
while (true) {
|
||||
val n = ch.readAvailable(buf)
|
||||
if (n == -1) break
|
||||
if (n > 0) sb.append(buf.decodeToString(0, n))
|
||||
}
|
||||
return sb.toString()
|
||||
}
|
||||
|
||||
private fun countOccurrences(text: String, needle: String): Int {
|
||||
var count = 0
|
||||
var from = 0
|
||||
while (true) {
|
||||
val i = text.indexOf(needle, from)
|
||||
if (i < 0) return count
|
||||
count++
|
||||
from = i + needle.length
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun textHelperSeesFinishReason() {
|
||||
assertTrue(hasNonNullFinishReasonText("""{"choices":[{"finish_reason":"stop"}]}"""))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun textHelperIgnoresNullFinishReason() {
|
||||
assertFalse(hasNonNullFinishReasonText("""{"choices":[{"finish_reason":null}]}"""))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun textHelperIgnoresEscapedOccurrenceInContent() {
|
||||
// Экранированное вхождение внутри content — это не поле finish_reason.
|
||||
assertFalse(hasNonNullFinishReasonText("""{"delta":{"content":"\"finish_reason\":\"stop\""}}"""))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun detectorSeesFinishReasonSplitAcrossReads() {
|
||||
// Маркер finish_reason разрезан границей чтения: детектор обязан его собрать.
|
||||
val detector = StreamEndDetector()
|
||||
detector.feed("""{"choices":[{"finish_rea""".encodeToByteArray())
|
||||
detector.feed("""son":"stop"}]}""".encodeToByteArray())
|
||||
assertTrue(detector.sawFinishReason)
|
||||
assertTrue(detector.missingDoneMarker)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun detectorSeesDoneSplitAcrossReads() {
|
||||
// [DONE] разрезан границей чтения: детектор видит его и маркер не нужен.
|
||||
val detector = StreamEndDetector()
|
||||
detector.feed("data: [DO".encodeToByteArray())
|
||||
detector.feed("NE]".encodeToByteArray())
|
||||
assertTrue(detector.sawDone)
|
||||
assertFalse(detector.missingDoneMarker)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun detectorSeesDoneAndFinishReason() {
|
||||
// В одном feed и finish_reason, и [DONE] — дописывать ничего не нужно.
|
||||
val detector = StreamEndDetector()
|
||||
detector.feed("""{"choices":[{"finish_reason":"stop"}]} data: [DONE]""".encodeToByteArray())
|
||||
assertFalse(detector.missingDoneMarker)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun detectorTruncatedStreamNeedsNoMarker() {
|
||||
// Обрыв без finish_reason: маркер не дописываем (усечённый стрим честно остаётся усечённым).
|
||||
val detector = StreamEndDetector()
|
||||
detector.feed("""{"choices":[{"delta":{"content":"hi"}}]}""".encodeToByteArray())
|
||||
assertFalse(detector.missingDoneMarker)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun rawStreamIsByteExactAndAppendsDone() = runTest {
|
||||
// Сырой поток с finish_reason и без [DONE]: байты входа не меняются, маркер дописывается ровно один.
|
||||
val input = "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hi\"},\"finish_reason\":\"stop\"}]}\n\n"
|
||||
|
||||
val out = runRaw(input)
|
||||
|
||||
assertEquals(input + done, out, "выход должен быть вход + один маркер")
|
||||
assertEquals(1, countOccurrences(out, done), "маркер должен быть ровно один")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun rawStreamTruncatedPassesThroughUntouched() = runTest {
|
||||
// Обрыв без finish_reason: ни одного [DONE], вход не меняется.
|
||||
val input = "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hi\"}}]}\n\n"
|
||||
|
||||
val out = runRaw(input)
|
||||
|
||||
assertEquals(input, out, "усечённый поток должен уйти как есть")
|
||||
assertEquals(0, countOccurrences(out, "[DONE]"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun rawStreamKeepsUpstreamDoneMarker() = runTest {
|
||||
// Апстрим прислал свой [DONE]: второй маркер не дописывается.
|
||||
val input = "data: {\"choices\":[{\"finish_reason\":\"stop\"}]}\n\n" + done
|
||||
|
||||
val out = runRaw(input)
|
||||
|
||||
assertEquals(1, countOccurrences(out, done), "маркер не должен дублироваться")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun thinkStreamAppendsDoneWhenUpstreamClosedWithoutMarker() = runTest {
|
||||
// think-обвязка: finish_reason есть, [DONE] нет — маркер дописывается ровно один.
|
||||
val input = "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hi\"},\"finish_reason\":\"stop\"}]}\n\n"
|
||||
|
||||
val out = runThink(input, "split")
|
||||
|
||||
assertTrue(out.endsWith(done), "выход должен заканчиваться маркером: $out")
|
||||
assertEquals(1, countOccurrences(out, done), "маркер должен быть ровно один")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun thinkStreamTruncatedNeedsNoMarker() = runTest {
|
||||
// think-обвязка: обрыв без finish_reason — маркер не дописываем.
|
||||
val input = "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hi\"}}]}\n\n"
|
||||
|
||||
val out = runThink(input, "split")
|
||||
|
||||
assertEquals(0, countOccurrences(out, "[DONE]"), "при обрыве маркера быть не должно: $out")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun thinkStreamKeepsSingleDone() = runTest {
|
||||
// think-обвязка: finish_reason и собственный [DONE] — маркер остаётся ровно один.
|
||||
val input = "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hi\"},\"finish_reason\":\"stop\"}]}\n\n" + done
|
||||
|
||||
val out = runThink(input, "split")
|
||||
|
||||
assertEquals(1, countOccurrences(out, done), "маркер не должен дублироваться")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,179 @@
|
||||
package pw.binom.llmproxy
|
||||
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFalse
|
||||
import kotlin.test.assertNotNull
|
||||
import kotlin.test.assertNull
|
||||
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
|
||||
|
||||
class ThinkTagChunkTest {
|
||||
|
||||
// Литералы тегов OPEN/CLOSE из ThinkTagSplitter собираем из символьных
|
||||
// кусочков, чтобы не записывать тег единой строкой в исходнике.
|
||||
private val openTag = '<' + "think" + '>'
|
||||
private val closeTag = '<' + "/think" + '>'
|
||||
// Маркер рассуждения (5 букв) собираем из Unicode-кодов, без единой строки.
|
||||
private val reasoning = listOf(0x0420, 0x0410, 0x0417, 0x0423, 0x041C)
|
||||
.map { Char(it) }
|
||||
.joinToString("")
|
||||
|
||||
private fun deltaOf(result: String, index: Int): JsonObject =
|
||||
Json.parseToJsonElement(result).jsonObject
|
||||
.get("choices")!!.jsonArray[index].jsonObject
|
||||
.get("delta")!!.jsonObject
|
||||
|
||||
@Test
|
||||
fun chunkSplitHoldsTailAcrossChunks() {
|
||||
// Хвост открывающего тега, разрезанный на границе чанков, переживает
|
||||
// два отдельных вызова через общий splitters (getOrPut по index).
|
||||
val splitters = mutableMapOf<Int, ThinkTagSplitter>()
|
||||
|
||||
// Чанк 1: текст + обрезанный префикс открывающего тега (не завершает тег).
|
||||
val prefix = openTag.dropLast(1)
|
||||
val obj1 = Json.parseToJsonElement(
|
||||
"""{"choices":[{"index":0,"delta":{"content":"текст$prefix"}}]}""",
|
||||
).jsonObject
|
||||
val r1 = transformThinkChunk(obj1, splitters, "split", true)
|
||||
assertNotNull(r1)
|
||||
val d1 = deltaOf(r1, 0)
|
||||
assertEquals("текст", d1["content"]!!.jsonPrimitive.content)
|
||||
assertFalse(d1.containsKey("reasoning_content"))
|
||||
|
||||
// Чанк 2 (тот же splitters): остаток открывающего тега + маркер.
|
||||
val rest = openTag.last()
|
||||
val obj2 = Json.parseToJsonElement(
|
||||
"""{"choices":[{"index":0,"delta":{"content":"$rest$reasoning"}}]}""",
|
||||
).jsonObject
|
||||
val r2 = transformThinkChunk(obj2, splitters, "split", true)
|
||||
assertNotNull(r2)
|
||||
val d2 = deltaOf(r2, 0)
|
||||
assertFalse(d2.containsKey("content"))
|
||||
assertEquals(reasoning, d2["reasoning_content"]!!.jsonPrimitive.content)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun chunkStripRemovesReasoning() {
|
||||
// Режим strip: блок целиком (теги + рассуждения) вырезается,
|
||||
// content склеивается в «AB», ключа reasoning_content нет.
|
||||
val splitters = mutableMapOf<Int, ThinkTagSplitter>()
|
||||
val c = "A" + openTag + reasoning + closeTag + "B"
|
||||
val obj = Json.parseToJsonElement(
|
||||
"""{"choices":[{"index":0,"delta":{"content":"$c"}}]}""",
|
||||
).jsonObject
|
||||
val r = transformThinkChunk(obj, splitters, "strip", false)
|
||||
assertNotNull(r)
|
||||
val d = deltaOf(r, 0)
|
||||
assertEquals("AB", d["content"]!!.jsonPrimitive.content)
|
||||
assertFalse(d.containsKey("reasoning_content"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun chunkDropsEmptyContentFieldWhenAllGoesToReasoning() {
|
||||
// Режим split: весь контент чанка — think-блок, поэтому content пуст
|
||||
// и ключ убирается из delta, а рассуждения уходят в reasoning_content.
|
||||
val splitters = mutableMapOf<Int, ThinkTagSplitter>()
|
||||
val c = openTag + reasoning + closeTag
|
||||
val obj = Json.parseToJsonElement(
|
||||
"""{"choices":[{"index":0,"delta":{"content":"$c"}}]}""",
|
||||
).jsonObject
|
||||
val r = transformThinkChunk(obj, splitters, "split", true)
|
||||
assertNotNull(r)
|
||||
val d = deltaOf(r, 0)
|
||||
assertFalse(d.containsKey("content"))
|
||||
assertEquals(reasoning, d["reasoning_content"]!!.jsonPrimitive.content)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun chunkReturnsNullWhenNoChoices() {
|
||||
// Нет ключа choices — чанк не трогаем, отдаём исходную строку (null).
|
||||
val splitters = mutableMapOf<Int, ThinkTagSplitter>()
|
||||
val obj = Json.parseToJsonElement("""{"model":"m"}""").jsonObject
|
||||
assertNull(transformThinkChunk(obj, splitters, "split", true))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun chunkReturnsNullWhenDeltaContentNotAString() {
|
||||
// content = null (JsonNull) и content = массив частей — не строка,
|
||||
// значит choice не трогаем, весь чанк не меняется -> null.
|
||||
val s1 = mutableMapOf<Int, ThinkTagSplitter>()
|
||||
val o1 = Json.parseToJsonElement(
|
||||
"""{"choices":[{"index":0,"delta":{"content":null}}]}""",
|
||||
).jsonObject
|
||||
assertNull(transformThinkChunk(o1, s1, "split", true))
|
||||
|
||||
val s2 = mutableMapOf<Int, ThinkTagSplitter>()
|
||||
val o2 = Json.parseToJsonElement(
|
||||
"""{"choices":[{"index":0,"delta":{"content":[{"type":"text","text":"hi"}]}}]}""",
|
||||
).jsonObject
|
||||
assertNull(transformThinkChunk(o2, s2, "split", true))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun chunkSeparateSplittersPerIndex() {
|
||||
// Один объект splitters на все три вызова: по каждому index держим
|
||||
// своего сплиттера, поэтому обрезанные хвосты по index не путаются.
|
||||
val splitters = mutableMapOf<Int, ThinkTagSplitter>()
|
||||
val prefix = openTag.dropLast(1)
|
||||
val rest = openTag.last()
|
||||
|
||||
// Вызов 1: два choice — index 0 «A»+префикс тега, index 1 «B»+тот же префикс.
|
||||
val c1 = """{"choices":[{"index":0,"delta":{"content":"A$prefix"}},{"index":1,"delta":{"content":"B$prefix"}}]}"""
|
||||
val r1 = transformThinkChunk(Json.parseToJsonElement(c1).jsonObject, splitters, "split", true)
|
||||
assertNotNull(r1)
|
||||
|
||||
// Вызов 2: index 1 получает остаток тега + маркер → рассуждения у index 1.
|
||||
val c2 = """{"choices":[{"index":1,"delta":{"content":"$rest${reasoning}1"}}]}"""
|
||||
val r2 = transformThinkChunk(Json.parseToJsonElement(c2).jsonObject, splitters, "split", true)
|
||||
assertNotNull(r2)
|
||||
val d2 = deltaOf(r2, 0)
|
||||
assertEquals(reasoning + "1", d2["reasoning_content"]!!.jsonPrimitive.content)
|
||||
|
||||
// Вызов 3: index 0 получает остаток тега + маркер → рассуждения у index 0.
|
||||
val c3 = """{"choices":[{"index":0,"delta":{"content":"$rest${reasoning}0"}}]}"""
|
||||
val r3 = transformThinkChunk(Json.parseToJsonElement(c3).jsonObject, splitters, "split", true)
|
||||
assertNotNull(r3)
|
||||
val d3 = deltaOf(r3, 0)
|
||||
assertEquals(reasoning + "0", d3["reasoning_content"]!!.jsonPrimitive.content)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun chunkKeepsUntouchedChoiceAndFinishReason() {
|
||||
// Первый choice целиком идёт через split; второй без ключа content
|
||||
// (в нём только finish_reason) — остаётся нетронутым, как и был.
|
||||
val splitters = mutableMapOf<Int, ThinkTagSplitter>()
|
||||
val c = "A" + openTag + reasoning + closeTag + "B"
|
||||
val c1 = """{"choices":[{"index":0,"delta":{"content":"$c"}},{"index":1,"delta":{"finish_reason":"stop"}}]}"""
|
||||
val r = transformThinkChunk(Json.parseToJsonElement(c1).jsonObject, splitters, "split", true)
|
||||
assertNotNull(r)
|
||||
val d0 = deltaOf(r, 0)
|
||||
assertEquals("AB", d0["content"]!!.jsonPrimitive.content)
|
||||
assertEquals(reasoning, d0["reasoning_content"]!!.jsonPrimitive.content)
|
||||
val d1 = deltaOf(r, 1)
|
||||
assertEquals("stop", d1["finish_reason"]!!.jsonPrimitive.content)
|
||||
assertFalse(d1.containsKey("content"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun chunkIndexDefaultsToZeroWhenAbsent() {
|
||||
// Ключ index отсутствует — оба раза используем сплиттер под индекс 0,
|
||||
// поэтому хвост первого чанка доживал до рассуждений второго.
|
||||
val splitters = mutableMapOf<Int, ThinkTagSplitter>()
|
||||
val prefix = openTag.dropLast(1)
|
||||
val rest = openTag.last()
|
||||
|
||||
val c1 = """{"choices":[{"delta":{"content":"A$prefix"}}]}"""
|
||||
val r1 = transformThinkChunk(Json.parseToJsonElement(c1).jsonObject, splitters, "split", true)
|
||||
assertNotNull(r1)
|
||||
|
||||
val c2 = """{"choices":[{"delta":{"content":"$rest$reasoning"}}]}"""
|
||||
val r2 = transformThinkChunk(Json.parseToJsonElement(c2).jsonObject, splitters, "split", true)
|
||||
assertNotNull(r2)
|
||||
val d2 = deltaOf(r2, 0)
|
||||
assertEquals(reasoning, d2["reasoning_content"]!!.jsonPrimitive.content)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,179 @@
|
||||
package pw.binom.llmproxy
|
||||
|
||||
import io.ktor.utils.io.ByteChannel
|
||||
import io.ktor.utils.io.close
|
||||
import io.ktor.utils.io.readAvailable
|
||||
import io.ktor.utils.io.writeFully
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertTrue
|
||||
import kotlinx.coroutines.test.runTest
|
||||
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
|
||||
|
||||
class ThinkTagStreamTest {
|
||||
|
||||
/** Прогнать SSE-текст через боевую обвязку и вернуть то, что она записала. */
|
||||
private suspend fun runStream(input: String, mode: String): String {
|
||||
val src = ByteChannel(autoFlush = true)
|
||||
src.writeFully(input.encodeToByteArray())
|
||||
src.close(null)
|
||||
val out = ByteChannel(autoFlush = true)
|
||||
out.streamSseWithThinkTags(src, mode)
|
||||
out.close(null)
|
||||
val sb = StringBuilder()
|
||||
val buf = ByteArray(512)
|
||||
while (true) {
|
||||
val n = out.readAvailable(buf)
|
||||
if (n == -1) break
|
||||
if (n > 0) sb.append(buf.decodeToString(0, n))
|
||||
}
|
||||
return sb.toString()
|
||||
}
|
||||
|
||||
// Литералы тегов OPEN/CLOSE из ThinkTagSplitter собираем из символьных
|
||||
// кусочков, чтобы не записывать тег единой строкой в исходнике.
|
||||
private val openTag = '<' + "think" + '>'
|
||||
private val closeTag = '<' + "/think" + '>'
|
||||
// Маркер рассуждения (5 букв) собираем из Unicode-кодов, без единой строки.
|
||||
private val reasoning = listOf(0x0420, 0x0410, 0x0417, 0x0423, 0x041C)
|
||||
.map { Char(it) }
|
||||
.joinToString("")
|
||||
|
||||
/** JSON-нагрузки из data-строк вывода (строки, начинающиеся с «data:»), без служебного [DONE]. */
|
||||
private fun dataPayloads(sse: String): List<String> =
|
||||
sse.lines()
|
||||
.filter { it.startsWith("data:") && it.removePrefix("data:").trim() != "[DONE]" }
|
||||
.map { it.removePrefix("data:").trim() }
|
||||
|
||||
private fun deltaOf(result: String, index: Int): JsonObject =
|
||||
Json.parseToJsonElement(result).jsonObject
|
||||
.get("choices")!!.jsonArray[index].jsonObject
|
||||
.get("delta")!!.jsonObject
|
||||
|
||||
@Test
|
||||
fun streamDoneAndServiceLinesPassThrough() = runTest {
|
||||
// Служебные строки SSE и маркер конца потока должны дойти до клиента без изменений.
|
||||
val input = buildString {
|
||||
append("data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"hi\"}}]}\n")
|
||||
append("\n")
|
||||
append("data: [DONE]\n")
|
||||
append("\n")
|
||||
append("event: ping\n")
|
||||
append("\n")
|
||||
append(": keep-alive\n")
|
||||
append("\n")
|
||||
}
|
||||
|
||||
val out = runStream(input, "split")
|
||||
|
||||
assertTrue(out.contains("data: [DONE]"), "маркер конца потока потерян: $out")
|
||||
assertTrue(out.contains("event: ping"), "служебная строка event потеряна: $out")
|
||||
assertTrue(out.contains(": keep-alive"), "строка-комментарий потеряна: $out")
|
||||
assertTrue(out.contains("hi"), "текстовый чанк потерян: $out")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun streamSplitsReasoningAcrossDataChunks() = runTest {
|
||||
// Хвост открывающего тега разрезан на границе двух data-событий: первое
|
||||
// отдаёт content «A» и хвост держит, второе завершает тег и блок.
|
||||
val prefix = openTag.take(3)
|
||||
val rest = openTag.drop(3)
|
||||
val content2 = rest + reasoning + closeTag + "B"
|
||||
val input = buildString {
|
||||
append("data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"A$prefix\"}}]}\n")
|
||||
append("\n")
|
||||
append("data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"$content2\"}}]}\n")
|
||||
append("\n")
|
||||
}
|
||||
|
||||
val out = runStream(input, "split")
|
||||
val payloads = dataPayloads(out)
|
||||
|
||||
val last = deltaOf(payloads.last(), 0)
|
||||
assertEquals("B", last["content"]!!.jsonPrimitive.content)
|
||||
assertEquals(reasoning, last["reasoning_content"]!!.jsonPrimitive.content)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun streamUnclosedBlockCarriesReasoningInSameChunk() = runTest {
|
||||
// Незакрытый think-блок в конце потока: reasoning отдаётся сразу в том же
|
||||
// чанке, где пришёл, финиш-чанка не появляется — удержанного хвоста нет.
|
||||
val content = "A" + openTag + "МЫСЛИ"
|
||||
val input = buildString {
|
||||
append("data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"$content\"}}]}\n")
|
||||
append("\n")
|
||||
}
|
||||
|
||||
val payloads = dataPayloads(runStream(input, "split"))
|
||||
|
||||
assertEquals(1, payloads.size, "финиш-чанка быть не должно: $payloads")
|
||||
val delta = deltaOf(payloads.single(), 0)
|
||||
assertEquals("A", delta["content"]!!.jsonPrimitive.content)
|
||||
assertEquals("МЫСЛИ", delta["reasoning_content"]!!.jsonPrimitive.content)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun streamFlushesHeldTailOnFinish() = runTest {
|
||||
// Поток обрывается на неполном префиксе открывающего тега: удержанный
|
||||
// хвост не теряется и сбрасывается финиш-чанком в конце.
|
||||
val tail = openTag.take(3)
|
||||
val input = buildString {
|
||||
append("data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"A$tail\"}}]}\n")
|
||||
append("\n")
|
||||
}
|
||||
|
||||
val payloads = dataPayloads(runStream(input, "split"))
|
||||
|
||||
assertEquals(2, payloads.size, "финиш-чанк с удержанным хвостом потерян: $payloads")
|
||||
val flushed = deltaOf(payloads.last(), 0)
|
||||
assertEquals(tail, flushed["content"]!!.jsonPrimitive.content)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun streamInvalidJsonPassesThroughVerbatim() = runTest {
|
||||
// data-строка, которая не является JSON, доходит до клиента без изменений.
|
||||
val input = buildString {
|
||||
append("data: {это не json\n")
|
||||
append("\n")
|
||||
}
|
||||
|
||||
val out = runStream(input, "split")
|
||||
|
||||
assertTrue(out.contains("data: {это не json"), "битая строка не дошла как есть: $out")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun streamChunkWithoutChoicesPassesThrough() = runTest {
|
||||
// Чанк с пустыми choices не меняется и уходит клиенту исходной строкой.
|
||||
val input = buildString {
|
||||
append("data: {\"usage\":{\"total_tokens\":5},\"choices\":[]}\n")
|
||||
append("\n")
|
||||
}
|
||||
|
||||
val out = runStream(input, "split")
|
||||
|
||||
assertTrue(out.contains("\"total_tokens\":5"), "чанк с usage изменился: $out")
|
||||
}
|
||||
|
||||
@Test
|
||||
fun streamModeStripEmitsNoReasoning() = runTest {
|
||||
// Режим strip: текст внутри think-тегов выбрасывается, поля
|
||||
// reasoning_content нет, обычный текст остаётся.
|
||||
val content = "A" + openTag + reasoning + closeTag + "B"
|
||||
val input = buildString {
|
||||
append("data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"$content\"}}]}\n")
|
||||
append("\n")
|
||||
}
|
||||
|
||||
val out = runStream(input, "strip")
|
||||
val payloads = dataPayloads(out)
|
||||
|
||||
assertTrue(!out.contains("reasoning_content"), "в режиме strip не должно быть reasoning_content: $out")
|
||||
assertEquals(1, payloads.size, "ожидался ровно один чанк: $payloads")
|
||||
assertEquals("AB", deltaOf(payloads.single(), 0)["content"]!!.jsonPrimitive.content, "теги и рассуждение должны быть вырезаны: $payloads")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,174 @@
|
||||
package pw.binom.llmproxy
|
||||
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFalse
|
||||
import kotlin.test.assertNotNull
|
||||
import kotlin.test.assertNull
|
||||
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
|
||||
|
||||
class ThinkTagTransformTest {
|
||||
|
||||
// Точные литералы тегов из ThinkTagSplitter (OPEN/CLOSE): собираем из
|
||||
// отдельных символов, чтобы не писать тег единой строкой в исходнике.
|
||||
private val openTag = '<' + "think" + '>'
|
||||
private val closeTag = '<' + "/think" + '>'
|
||||
|
||||
private fun messageOf(result: String, index: Int): JsonObject =
|
||||
Json.parseToJsonElement(result).jsonObject
|
||||
.get("choices")!!.jsonArray[index].jsonObject
|
||||
.get("message")!!.jsonObject
|
||||
|
||||
@Test
|
||||
fun splitSingleBlockMovesReasoningAndKeepsOtherFields() {
|
||||
// Проверяем: в режиме split содержимое блока уходит в reasoning_content,
|
||||
// в content остаётся «AB», а служебные поля (model, usage, …) не тронуты.
|
||||
val c = "A" + openTag + "РАЗУМ" + closeTag + "B"
|
||||
val obj = Json.parseToJsonElement(
|
||||
"""
|
||||
{"id":"c1","object":"chat.completion","created":123,"model":"m1",
|
||||
"usage":{"prompt_tokens":1,"completion_tokens":2,"total_tokens":3},
|
||||
"system_fingerprint":"sf1",
|
||||
"choices":[{"index":0,"message":{"role":"assistant","content":"$c"},"finish_reason":"stop"}]}
|
||||
""".trimIndent(),
|
||||
).jsonObject
|
||||
|
||||
val result = transformThinkMessage(obj, "split")
|
||||
assertNotNull(result)
|
||||
val out = Json.parseToJsonElement(result).jsonObject
|
||||
val msg = out.get("choices")!!.jsonArray[0].jsonObject.get("message")!!.jsonObject
|
||||
assertEquals("AB", msg["content"]!!.jsonPrimitive.content)
|
||||
assertEquals("РАЗУМ", msg["reasoning_content"]!!.jsonPrimitive.content)
|
||||
assertEquals("m1", out["model"]!!.jsonPrimitive.content)
|
||||
assertEquals(3, out["usage"]!!.jsonObject["total_tokens"]!!.jsonPrimitive.content.toInt())
|
||||
}
|
||||
|
||||
@Test
|
||||
fun splitTwoBlocksConcatenateInOrder() {
|
||||
// Проверяем порядок конкатенации reasoning, когда два think-блока идут подряд.
|
||||
val c = openTag + "ПЕРВЫЙ" + closeTag + "X" + openTag + "ВТОРОЙ" + closeTag
|
||||
val obj = Json.parseToJsonElement(
|
||||
"""{"choices":[{"index":0,"message":{"content":"$c"}}]}""",
|
||||
).jsonObject
|
||||
|
||||
val result = transformThinkMessage(obj, "split")
|
||||
assertNotNull(result)
|
||||
val msg = messageOf(result, 0)
|
||||
assertEquals("X", msg["content"]!!.jsonPrimitive.content)
|
||||
assertEquals("ПЕРВЫЙВТОРОЙ", msg["reasoning_content"]!!.jsonPrimitive.content)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun splitAppendsToExistingReasoningContent() {
|
||||
// Проверяем, что прежнее reasoning_content не теряется, а новое дописывается.
|
||||
val c = openTag + "NEW" + closeTag
|
||||
val obj = Json.parseToJsonElement(
|
||||
"""{"choices":[{"message":{"content":"$c","reasoning_content":"OLD"}}]}""",
|
||||
).jsonObject
|
||||
|
||||
val result = transformThinkMessage(obj, "split")
|
||||
assertNotNull(result)
|
||||
val msg = messageOf(result, 0)
|
||||
assertEquals("OLDNEW", msg["reasoning_content"]!!.jsonPrimitive.content)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun splitUnclosedTagGoesToReasoning() {
|
||||
// Проверяем, что незакрытый открывающий тег уводит хвост целиком в reasoning.
|
||||
val c = "текст" + openTag + "мысли"
|
||||
val obj = Json.parseToJsonElement(
|
||||
"""{"choices":[{"message":{"content":"$c"}}]}""",
|
||||
).jsonObject
|
||||
|
||||
val result = transformThinkMessage(obj, "split")
|
||||
assertNotNull(result)
|
||||
val msg = messageOf(result, 0)
|
||||
assertEquals("текст", msg["content"]!!.jsonPrimitive.content)
|
||||
assertEquals("мысли", msg["reasoning_content"]!!.jsonPrimitive.content)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun stripRemovesTagsAndLeavesNoReasoningKey() {
|
||||
// Проверяем: в режиме strip теги и рассуждения вырезаются,
|
||||
// а ключ reasoning_content в message отсутствует вовсе.
|
||||
val c = "A" + openTag + "РАЗУМ" + closeTag + "B"
|
||||
val obj = Json.parseToJsonElement(
|
||||
"""{"choices":[{"message":{"content":"$c"}}]}""",
|
||||
).jsonObject
|
||||
|
||||
val result = transformThinkMessage(obj, "strip")
|
||||
assertNotNull(result)
|
||||
val msg = messageOf(result, 0)
|
||||
assertEquals("AB", msg["content"]!!.jsonPrimitive.content)
|
||||
assertFalse(msg.containsKey("reasoning_content"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun missingChoicesReturnsNull() {
|
||||
// Проверяем: без ключа choices функция не вносит изменений (null).
|
||||
val obj = Json.parseToJsonElement(
|
||||
"""{"id":"c1","model":"m1"}""",
|
||||
).jsonObject
|
||||
assertNull(transformThinkMessage(obj, "split"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun nonStringContentReturnsNull() {
|
||||
// content = null (JsonNull) — менять нечего, функция возвращает null.
|
||||
val nullContent = Json.parseToJsonElement(
|
||||
"""{"choices":[{"message":{"content":null}}]}""",
|
||||
).jsonObject
|
||||
assertNull(transformThinkMessage(nullContent, "split"))
|
||||
|
||||
// content = массив частей (JsonArray) — тоже не трогаем, null.
|
||||
val arrayContent = Json.parseToJsonElement(
|
||||
"""{"choices":[{"message":{"content":[{"type":"text","text":"hi"}]}}]}""",
|
||||
).jsonObject
|
||||
assertNull(transformThinkMessage(arrayContent, "split"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun twoChoicesEachKeepOwnThinkBlock() {
|
||||
// Проверяем, что каждый choice обрабатывается независимо: у каждого свой
|
||||
// вырезанный content и свой reasoning_content.
|
||||
val c0 = "А" + openTag + "А-раз" + closeTag + "Б"
|
||||
val c1 = "В" + openTag + "В-раз" + closeTag + "Г"
|
||||
val obj = Json.parseToJsonElement(
|
||||
"""
|
||||
{"choices":[
|
||||
{"index":0,"message":{"content":"$c0"},"finish_reason":"stop"},
|
||||
{"index":1,"message":{"content":"$c1"},"finish_reason":"stop"}]}
|
||||
""".trimIndent(),
|
||||
).jsonObject
|
||||
|
||||
val result = transformThinkMessage(obj, "split")
|
||||
assertNotNull(result)
|
||||
val out = Json.parseToJsonElement(result).jsonObject.get("choices")!!.jsonArray
|
||||
val m0 = out[0].jsonObject.get("message")!!.jsonObject
|
||||
val m1 = out[1].jsonObject.get("message")!!.jsonObject
|
||||
assertEquals("АБ", m0["content"]!!.jsonPrimitive.content)
|
||||
assertEquals("А-раз", m0["reasoning_content"]!!.jsonPrimitive.content)
|
||||
assertEquals("ВГ", m1["content"]!!.jsonPrimitive.content)
|
||||
assertEquals("В-раз", m1["reasoning_content"]!!.jsonPrimitive.content)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun offModePassthroughKeepsTags() {
|
||||
// Проверяем passthrough: в режиме off теги остаются в content,
|
||||
// reasoning_content не добавляется, а результат — не null.
|
||||
val c = "A" + openTag + "РАЗУМ" + closeTag + "B"
|
||||
val obj = Json.parseToJsonElement(
|
||||
"""{"choices":[{"message":{"content":"$c"}}]}""",
|
||||
).jsonObject
|
||||
|
||||
val result = transformThinkMessage(obj, "off")
|
||||
assertNotNull(result)
|
||||
val msg = messageOf(result, 0)
|
||||
assertEquals("A" + openTag + "РАЗУМ" + closeTag + "B", msg["content"]!!.jsonPrimitive.content)
|
||||
assertFalse(msg.containsKey("reasoning_content"))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user