LLM SSE оқиғаларын біріздендіру стримингті қалай ретте ұстайды

LLM-нен келетін ағынды for chunk: print(chunk.text) цикліне дейін қысқартуға болмайды. Мұндай код демо кезінде жұмыс істейді, ал продакшнда құрал аргументтерін жоғалтады, бас тартуды желі қатесімен шатастырады және сокет жабылғанын жауап аяқталды деп қабылдайды.

Стримингке арналған дұрыс ортақ қабат барлық провайдерді бір-біріне ұқсатуға тырыспайды. Ол өзінің шағын оқиғалар тізбегін бекітеді, блоктардың ретін сақтайды және дұрыс финалды апатты үзілістен анық ажыратады. Соның арқасында клиенттер бір контракт алады, ал адаптерлер жүйе шекарасында бөтен ерекшеліктерді еркін талдай алады.

SSE оқиғалардың шекарасын береді, бірақ мағынасын анықтамайды

SSE хабарламаларды HTTP арқылы Content-Type: text/event-stream түрінде жеткізудің мәтіндік тәсілін анықтайды. HTML Standard бойынша әр хабарлама event:, data: сияқты жолдардан және жазбаны аяқтайтын бос жолдан тұрады. Стандарт бір хабарламада бірнеше data: жолын және браузер қайта қосылғанда қайтара алатын id идентификаторын да қолдайды.

Бұдан SSE модель жауабын стримингтеуді біледі деген қорытынды шықпайды. Ол дельтаның не екенін, құрал шақыруы қашан ашылатынын, фрагменттерді орындарымен ауыстыруға болатынын немесе сәтті аяқталу деп нені санау керегін айтпайды. Бұл провайдер API-інің семантикасы.

Іс жүзінде кемінде төрт түрлі модель кездеседі:

• OpenAI-үйлесімді Chat Completions ағыны JSON-ды көбіне data: ішінде жібереді және оны [DONE] жолымен аяқтайды.
• OpenAI Responses API response.created, дельта қосу оқиғалары және response.completed немесе response.failed сияқты типтелген оқиғаларды қайтарады.
• Claude Messages API атауы бар SSE оқиғаларын жібереді және жауапты content block өмірлік циклі арқылы құрады.
• Gemini streamGenerateContent бірінен соң бірі келетін GenerateContentResponse объектілерін қайтарады. Бір объект кандидаттың кезекті бөлігін немесе финал ақпаратын қамтуы мүмкін.

Қате атаудан басталады. Команда «бізде SSE бар» дейді де, Claude үшін choices[0].delta.content өрісіне есептелген өңдегішті қолданады немесе Gemini-ден [DONE] күтеді. Сым арқылы бәрі SSE болып көрінуі мүмкін, бірақ олардың жауап протоколдары әртүрлі.

Жалпы формат токендерді емес, блоктарды сипаттауы керек

Біртұтас мәтін дельтасы қолданба тек чатты басып шығарған кезде ыңғайлы. Модель функция шақырып, құрылымдалған объект қайтарып, бас тарту жіберіп немесе жасырын reasoning бөлігін шығарған сәтте «токен» интеграцияның негізгі бірлігі болудан қалады.

Негізге контент блогын алыңыз. Блокта тұрақты block_id, реттік index және түр болады. Оның ішіне дельталар келеді. Мәтін блогы жолдар алады, құрал шақыруы блогы атауды, шақыру идентификаторын және JSON фрагменттерін алады, ал reasoning блогы мәтін немесе тұтастықты тексеруге арналған жеке белгі алуы мүмкін.

Шлюз бен қолданба арасында қолданатын ең аз контракт мынадай:

{
  "stream_id": "st_01J...",
  "seq": 17,
  "type": "block.delta",
  "block": {
    "id": "b_1",
    "index": 1,
    "kind": "tool_call"
  },
  "delta": {
    "kind": "json_text",
    "text": "{\"city\":\"Alma\"
  }
}

Бұл контракт үшін алты міндетті тип қажет:

1. stream.start шлюздің ағынды қабылдап, оған идентификатор бергенін хабарлайды.
2. block.start нақты блокты ашады және оның түрін жариялайды.
3. block.delta ашық блокқа өзгермейтін фрагмент қосады.
4. block.stop блокты жабады және оған жаңа дельталар жіберуге тыйым салады.
5. stream.done сәтті аяқталғанын растайды және финал метадеректерін қамтиды.
6. stream.error ағынды код, санат және қауіпсіз хабарламасы бар қатемен аяқтайды.

stream.start оқиғасы «провайдер генерацияны бастады» дегенді білдірмейді. Ол тек сіздің жария контрактіңіздің басталғанын көрсетеді. stream.done оқиғасы «клиент HTTP body-ді толық оқып шықты» дегенге тең емес. Ол адаптердің сәтті нәтиже туралы жеткілікті растау алғанын білдіреді.

stream.heartbeat және stream.warning оқиғаларын қосуға болады, бірақ оларды клиент үшін міндетті етпеңіз. Ping, SSE түсіндірмесі және кезекті мәтін дельтасы жауап күйін өзгертпеуі керек.

Дельталардың реті фрагмент өлшемінен маңызды

Провайдер бөліктердің ыңғайлы өлшеміне кепілдік бермейді. Бір дельтада сөз, бірнеше сөйлем, бос жол, байт деңгейінде оқылған Unicode таңбасының жартысы немесе жабылатын жақшасы жоқ JSON бөлігі болуы мүмкін. Интерфейсіңіз токен шекарасы туралы болжам бермеуі керек.

Сенімді ереже қарапайым: адаптер бір логикалық блок үшін алған дельталарын сол ретімен береді және UI ыңғайлы болсын деп әртүрлі блоктарды біріктірмейді. Блоктарда индекстер болса, index финал контент массивіндегі орынды, ал seq клиентке оқиғалардың жеткізілу ретін анықтайды.

Модель алдымен мәтін жазып, кейін функция шақырып, функция нәтижесінен соң жалғасын жазатын жауапты қарастырайық. Дұрыс тізбек мынадай:

stream.start
block.start text index=0
block.delta text="Кестені тексеріп жатырмын."
block.stop text index=0
block.start tool_call index=1
block.delta json_text="{\"date\":\"2026-07-23\""
block.delta json_text="}"
block.stop tool_call index=1
block.start text index=2
block.delta text="Бүгін қолжетімді..."
block.stop text index=2
stream.done

Қолданба шақыруды орындамай тұрып, екінші мәтін блогын пайдаланушыға көрсетуге болмайды. Тіпті нақты провайдер жауап бөліктерін параллель генерациялай алса да, бұл ереже сақталады. index=0 және index=2 блоктарын бір буферге біріктіріп, мағынасын кейін қалпына келтіремін деуге болмайды. Аудитке, қайта іске қосуға және құралдар оркестрациясына бастапқы құрылым қажет болады.

Reasoning тағдырын бөлек шешіңіз. Өнім саясаты пайдаланушыға оны көрсетуге тыйым салса, reasoning-ді кәдімгі мәтінмен алмастырмаңыз және үнсіз алып тастамаңыз. visibility: "internal" блогын қорғалған контурға беріңіз немесе сұрау деңгейінде көрсетуді өшіріңіз. Әртүрлі модельдерде жауаптың бұл бөлігіне қол жеткізу және тексеру ережелері өзгеше, сондықтан әмбебап UI жалаушасы қауіпті.

Ағынның аяқталуы провайдердің дәлелін талап етеді

Жабық байланыс тек байланыс жабық екенін білдіреді. Сервер жауапты штаттық түрде аяқтауы мүмкін, прокси ұзақ сұрауды үзуі мүмкін, пайдаланушы желіден ажырауы мүмкін, ал upstream JSON аргументтерінің жартысын жіберген соң істен шығуы мүмкін. Бұл жағдайлардың бәрін бір done күйіне біріктіруге болмайды.

Әр адаптерде финал сигналдарының нақты кестесі болуы керек. Мысалы, OpenAI-үйлесімді Chat Completions үшін бұл валидті JSON чанктарынан кейінгі [DONE] болуы мүмкін. OpenAI Responses API жағдайында HTTP ағынының соңына ғана емес, терминал response.completed, response.failed, response.incomplete немесе response.cancelled оқиғаларына сүйену керек. OpenAI құжаттамасында оқиғалар sequence_number мәнін қамтиды. Бұл диагностикаға пайдалы сыртқы дерек, бірақ жария seq мәнін бәрібір шлюз тағайындауы керек.

Claude құжаттамасы өмірлік циклді ерекше анық сипаттайды: message_start, одан кейін content_block_start, content_block_delta және content_block_stop арқылы бір немесе бірнеше блок, содан соң message_delta және финал message_stop. Claude жаңа оқиға түрлері пайда болуы мүмкін екенін де ескертеді, сондықтан клиент белгісіз типтерді тыныш өңдеуі керек. Бұл ереже тек Claude үшін емес, кез келген адаптерге жарайды.

Gemini streamGenerateContent режимінде [DONE] сияқты міндетті финал маркерін емес, жауап объектілерінің тізбегін жібереді. Адаптер кандидаттың аяқталғанын көрсететін құжатталған өрістерге және ағынмен келген финал деректеріне қарауы керек. Соңғы объектіден кейін байланыстың жабылуын, оның алдында күйді анықтайтын дерек болмаса, жеткілікті растау деп санауға болмайды.

Ішкі жүйеде үш терминал күй ұстаңыз:

• completed провайдер штаттық нәтижені растағанын білдіреді.
• failed провайдер қате жібергенін немесе адаптер протоколдың анық қатесін алғанын білдіреді.
• interrupted тасымалдау үзілгенін немесе клиент расталған финалға дейін сұрауды тоқтатқанын білдіреді.

interrupted үшін алынған мәтін дельталарын қаралама ретінде сақтауға болады, бірақ мұндай жауапты агенттің дайын нәтижесі ретінде жазуға болмайды. Әсіресе соңғы ашық блок tool_call немесе схема бойынша JSON жауабы болса.

Құрал аргументтерін блок жабылғанға дейін жинау керек

Толық емес JSON JSON болып саналмайды. Бұл қарапайым көрінеді, бірақ көптеген оркестратор дәл осы жерде қысқартылған аргументпен функция шақырады немесе модель шығарған мәтінді регуляр өрнектермен «жөндей» бастайды.

Claude input_json_delta ішінде partial_json жол фрагменттерін жібереді және оларды талдаудан бұрын жинауды ұсынады. Құрылымдалған нәтиже жөніндегі Gemini құжаттамасы да толық объект алынғанша валидті жартылай JSON жолдарын біріктіру керегін айтады. Инженерлік қорытынды біреу: ағынды біртіндеп көрсетуге болады, бірақ орындалатын объект блок аяқталғаннан кейін ғана пайда болады.

Адаптерде бүкіл жауапқа бір буфер емес, әр block_id үшін жеке буфер болуы керек. Құрал шақырулары бірнеше блоктан тұруы мүмкін, ал кейбір API параллель шақыруларға рұқсат береді.

type ToolBuffer = {
  name?: string;
  callId?: string;
  rawArguments: string;
  closed: boolean;
};

function appendToolDelta(buf: ToolBuffer, part: string) {
  if (buf.closed) throw new Error("delta after block.stop");
  buf.rawArguments += part;
}

function closeToolBlock(buf: ToolBuffer) {
  buf.closed = true;
  const args = JSON.parse(buf.rawArguments);
  return { name: buf.name, callId: buf.callId, arguments: args };
}

Бұл код rawArguments мәнін әр дельта сайын талдауға әдейі тырыспайды. Егер UI шақырудың қалай қалыптасып жатқанын көрсеткісі келсе, шикі мәтінді техникалық режимде шығара алады. Құрал орындаушысы block.stop оқиғасын күтіп, содан кейін JSON мен аргументтер схемасын тексеруі керек.

JSON.parse қатесін құрал қатесі ретінде жасыруға болмайды. Бұл жауап протоколының немесе модельдің structured output режимімен үйлеспеуінің қатесі. Журналда провайдер ID-і, модель ID-і, бастапқы оқиғалар және блок позициясы қалуы керек, бірақ пайдаланушы құпиялары мен толық промпт әдепкіде жазылмауы тиіс.

SSE ішіндегі қате клиентке қате ретінде жетуі керек

Жауап басындағы HTTP 200 генерация сәтті өтті дегенді білдірмейді. Тақырыптар жіберілгеннен кейін сервер жауапты HTTP 429, 500 немесе 529 күйіне ауыстыра алмайды. Сондықтан провайдерлер қатені ашылған ағынның ортасында жеке оқиға ретінде жіберуі мүмкін.

Claude қате объектісі, мысалы overloaded_error, бар event: error оқиғасын нақты сипаттайды. OpenAI Responses API-де сәтсіздіктің жеке терминал оқиғалары бар. Басқа API-де қате диагностиканы SDK-ге қалдырып, тасымалдаудың үзілуі ретінде көрінуі мүмкін. Ортақ қабат осы нұсқалардың бәрін бір stream.error оқиғасына айналдыруы керек.

Пайдалы қате құрылымы барлық қате бірдей сияқты көрінбеуі тиіс:

{
  "type": "stream.error",
  "stream_id": "st_01J...",
  "seq": 24,
  "error": {
    "category": "upstream_overloaded",
    "retryable": true,
    "provider_code": "overloaded_error",
    "message": "Провайдер уақытша шамадан тыс жүктелген"
  }
}

stream.error оқиғасынан кейін барлық жабылмаған блоктарды тек ішкі күйде жабыңыз, бірақ клиентке жалған block.stop жібермеңіз. Әйтпесе клиент құрал аргументтері толық, ал мәтіндік жауап мағыналы түрде аяқталды деп ойлайды.

Сұрауды автоматты қайталауға сырттан байқалатын әсер пайда болғанға дейін ғана болады. Қарапайым чат үшін бұл пайдаланушы көрген алғашқы дельтаға дейінгі кезең. Құралдары бар агентте шекара қатаңырақ: идемпотенттік кілт болмаса, құрал ақша есептен шығаруы, хат жіберуі немесе жазбаны өзгертуі мүмкін болғаннан кейін іске қосуды қайталауға болмайды.

Белгісіз оқиғалар API жаңартуларынан кейін де өңделуі керек

Провайдерлер стриминг протоколдарын кеңейтеді. Reasoning, дәйексөз, аудио немесе серверлік құрал туралы жаңа оқиға switch өңдегішін exception арқылы құлатып, пайдаланушыға дайын мәтінді жеткізуді тоқтатпауы керек.

Өңдеуді екі қабатқа бөліңіз. Бірінші қабат SSE-ні декодтап, бастапқы оқиғаны диагностика ізінде сақтайды. Екінші қабат белгілі типтерді сіздің контрактыңызбен сәйкестендіреді. Белгісіз типті ignored деп белгілеп, метриканы арттырады және провайдер оны терминал қате деп атамаған жағдайда ағынды жалғастырады.

Бұл өзгерістерді мәңгі үнсіз елемеуге шақыру емес. Белгісіз тип алерттер мен тест транскрипттерінде көрінуі керек. Бірақ жаңа міндетті емес өріс немесе ping оқиғасы үшін жұмыс істеп тұрған ағынның құлауы бақылауы бар басқарылатын өткізіп жіберуден нашар.

Heartbeat оқиғалары бизнес таймаутын шексіз ұзартпауы керек. Тірі байланысты көрсететін тасымалдау таймаутын және мазмұнды дельтаның немесе құрал орындалуының құжатталған оқиғасының пайда болуын талап ететін прогресс таймаутын бөлек ұстаңыз. Әйтпесе upstream ping жіберіп байланысты ұстап тұрады, ал пайдаланушыңыз шексіз күтеді.

Адаптер соңғы күй автоматы болуы керек

Стримингтегі шарттар тез көбейіп, қарапайым if жиынына сыймай қалады. Шағын соңғы күй автоматы тыйым салынған өтулерді көзге түсіреді және дұрыс тест жазуға мүмкіндік береді.

Ағын күйін былай сипаттауға болады:

idle -> open -> receiving -> terminal
                     |          |
                     v          v
                  interrupted  completed | failed

receiving ішінде әр блоктың күйін сақтаңыз: new, open, closed. block.delta оқиғасына тек open блок үшін, ал stream.done оқиғасына барлық блок жабылған кезде ғана рұқсат беріңіз. Егер провайдер ашық блокпен бірге терминал оқиғасын жіберсе, адаптер жетіспейтін JSON-ды ойдан шығармай, ағынды үйлесімділік қатесімен аяқтауы керек.

Әр жазылған ағын үшін мына ережелерді тексеріңіз:

1. Бірінші жария элемент әрқашан stream.start болады.
2. Әр block.delta үшін бұрын ашылған және әлі жабылмаған блок болуы керек.
3. Бір stream_id аясында seq нөмірі қатаң өсуі тиіс.
4. stream.done немесе stream.error оқиғасынан кейін жаңа жария оқиғалар болмауы керек.
5. Ашық блок бар кезде stream.done жіберілмеуі тиіс.

Бұл ережелер көзге бірден көрінбейтін қателерді ұстайды: жабылғаннан кейінгі қайталама дельта, параллель tool call кезіндегі шатасқан индекс, usage келмей тұрып жіберілген финал және reconnect-тен кейін чанкты қате қайталап қолдану.

Тек тірі сұрауларды емес, транскрипттерді тестілеу керек

Модельге жасалған тірі тест қажет, бірақ ол сирек ақауларды нашар қайталайды. Адаптер нақты протокол оқиғаларының сақталған транскрипттерінен өткен кезде ғана сенімді қорғаныс пайда болады.

Әр провайдер үшін кемінде бес анонимдендірілген тізбек жинаңыз: қарапайым мәтін, бірнеше блок, фрагменттелген JSON бар құрал шақыруы, мәтіннің бір бөлігінен кейінгі терминал қате және терминал сигналы жоқ үзілу. Белгісіз оқиғамен жағдайды да қосыңыз. Claude үшін ping қосыңыз, өйткені ол кез келген жерде пайда болуы мүмкін. OpenAI Responses үшін сәтсіз терминал күйін, ал Gemini үшін мәтін бірнеше жауап объектісімен келетін ағынды қосыңыз.

Тек күтілетін финал мәтінді тексермеңіз. Нормаланған оқиғалардың толық тізбегін тексеріңіз. Мұндай тест мынадай көрінеді:

expect(normalize(transcript)).toEqual([
  { type: "stream.start", seq: 1 },
  { type: "block.start", seq: 2, block: { index: 0, kind: "text" } },
  { type: "block.delta", seq: 3, delta: { text: "Сәлем" } },
  { type: "block.stop", seq: 4 },
  { type: "stream.done", seq: 5, status: "completed" }
]);

Транскрипттерге property test те қосыңыз. Генератор бір мәтін дельтасын кездейсоқ фрагменттерге бөле алады, heartbeat кірістіре алады және белгісіз міндетті емес оқиғаларды қайталай алады. Соңында жиналған мәтін сол күйінде қалуы керек, ал автомат инварианттарды бұзбауы тиіс.

AI Router командаларға бірнеше жеткізуші үшін OpenAI-үйлесімді кіріс береді, бірақ endpoint үйлесімділігі стриминг семантикасындағы айырмашылықтарды жоймайды. Осындай кіріс үстіне шлюз немесе клиент құрып жатсаңыз, нақты бағытыңыз қандай оқиғаларды сақтап, нормализациялайтынын, әсіресе құралдар мен финал күйлері үшін тексеріңіз.

Клиентке қарапайым контракт, ал шлюзге толық шындық керек

Фронтендке әдетте мәтін, күй және прогресс жеткілікті. Оркестраторға блок индекстері, tool call идентификаторлары, тоқтау себептері, usage және провайдердің бастапқы күйі қажет. Браузерді бұл күрделілікті ұстауға мәжбүрлемеңіз, бірақ оны шлюзде жоғалтпаңыз.

Міндеттерді дұрыс бөлу мынадай: адаптер бөтен ағынды оқып, қатаң ішкі оқиғалар жасайды; оркестратор блоктар мен құралдар бойынша шешім қабылдайды; клиент көрсетуге болатын деректердің қауіпсіз проекциясын алады. Бұл ретте бір нормаланған журнал жауаптың неге аяқталғанын, JSON қай жерде үзілгенін және модельден нақты не келгенін түсіндіруге мүмкіндік беруі керек.

Егер қазіргі контрактыңыз delta, [DONE] және catch мәндерінен тұрса, барлық интеграцияны бірден қайта жазудан бастамаңыз. Tool call бар бір ағынды алып, оны транскрипт ретінде бекітіңіз, block.start пен block.stop қосыңыз, содан кейін расталған терминал оқиғасы жоқ done күйіне тыйым салыңыз. Осыдан соң LLM API арасындағы жасырын айырмашылықтардың көбі пайдаланушы интерфейсінде көрінбей қалады.