Akış (streaming) desenlerini doğru kur

İki Moonborn uç noktası Server-Sent Events (SSE / Sunucu Gönderimli Olaylar) ile akış (streaming) yapar:

İkisi de aynı SSE zarfını (envelope) kullanır, sadece olay tipleri farklıdır. Bu rehber olay şekillerini, ham okuma döngüsünü, iptal akışını ve neden yerleşik yeniden bağlanma olmadığını anlatır.

Bu rehberi bitirdiğinde

• Üretim ve sohbet akış olaylarının şekillerini bileceksin.
• Ham SSE'yi çözümleyen bir okuma döngüsü yazabileceksin (hata ayıklama + alt seviye entegrasyon için).
• SDK'ların (client.chat.messages.stream(), client.personas.createStream()) aynı akışı nasıl sardığını (wrap) bileceksin.
• AbortController ile akışı temiz kesebilecek, sunucu tarafı faturalamayı durdurabileceksin.
• SSE'nin yerleşik yeniden bağlanma (reconnect) taşımadığını ve düşen akışların nasıl kurtarılacağını (recovery) bileceksin.

Ön koşul: API anahtarı + temel fetch / SSE bilgisi. Üst seviye akışı SDK ile yapıyorsan Bir persona ile sohbet oturumu kur eğitiminde token token akış adımı var.

Üretim hattı olayları

event: step.started        data: { step: "soul_draft" }
event: step.completed      data: { step: "soul_draft", durationMs: 4500, model: "claude-opus-4-7" }
event: step.started        data: { step: "self_enrich" }
event: step.completed      data: { step: "self_enrich", durationMs: 3200, model: "claude-sonnet-4-6" }
...
event: pipeline.completed  data: { personaId: "per_01H...", durationMs: 67000 }

Altı adım sırayla step.started + step.completed yayınlar:

1. intent_parse
2. soul_draft
3. self_enrich
4. mask_build
5. surface_ground
6. audit

pipeline.completed sonlandırıcı olaydır — son persona kimliğini ve toplam duvar saati süresini (wall time) taşır.

Sohbet yanıt olayları

event: token       data: { "delta": "Beni" }
event: token       data: { "delta": " tetikleyen" }
event: token       data: { "delta": " şey" }
event: token       data: { "delta": " aradaki" }
event: token       data: { "delta": " boşluk." }
event: completed   data: { "messageId": "msg_01H...", "driftScore": 0.14, "driftThreshold": 0.30, "driftAlert": false }

Token'lar sırayla akar; completed olayı yanıt biterken bir kez gelir ve drift zarfını (envelope) taşır.

Ham SSE okuma döngüsü

SDK olmadan SSE'yi çözümlemek için klasik desen:

const controller = new AbortController();
 
const res = await fetch('https://api.moonborn.co/v1/chat/sessions/ses_.../messages?stream=true', {
  method: 'POST',
  headers: {
    Authorization: `Bearer ${process.env.MOONBORN_API_KEY}`,
    'Content-Type': 'application/json',
    Accept: 'text/event-stream',
  },
  body: JSON.stringify({ content: 'Selam.' }),
  signal: controller.signal,
});
 
const reader = res.body!.getReader();
const decoder = new TextDecoder();
let buf = '';
 
while (true) {
  const { value, done } = await reader.read();
  if (done) break;
 
  buf += decoder.decode(value, { stream: true });
 
  let idx: number;
  while ((idx = buf.indexOf('\n\n')) !== -1) {
    const frame = buf.slice(0, idx);
    buf = buf.slice(idx + 2);
 
    const lines = frame.split('\n');
    const event = lines.find((l) => l.startsWith('event:'))?.slice(6).trim();
    const dataRaw = lines.find((l) => l.startsWith('data:'))?.slice(5).trim();
    if (!event || !dataRaw) continue;
 
    const data = JSON.parse(dataRaw);
 
    if (event === 'token') {
      process.stdout.write(data.delta);
    } else if (event === 'completed') {
      console.log('\n\ndrift:', data.driftScore, data.driftAlert);
    }
  }
}

İptal — akışı temiz kes

fetch çağrısına bir AbortController.signal geçir. Kullanıcı akışı iptal ederse:

const controller = new AbortController();
 
// Akışı başlat
const streamPromise = startStream(controller.signal);
 
// 2 saniye sonra iptal et
setTimeout(() => controller.abort(), 2000);
 
await streamPromise;

Yeniden bağlanma — neden yok, nasıl kurtarılır

SSE'nin yerleşik sürdürmesi (resume) yoktur. İstemci akış ortasında düşerse üretim veya yanıt sunucu tarafında zaten çalışıyordur.

Kurtarma deseni

Akış koptuğunda akışı yeniden deneme — sadece sonuçtan oku.

try {
  await streamMessages(/* ... */);
} catch (err) {
  // Akış koptu — sunucu tarafı çoktan çalışıyor olabilir
  // Son mesajı ve drift zarfını yokla (poll)
  const messages = await client.chat.messages.list({
    sessionId: 'ses_...',
    limit: 1,
  });
  const latest = messages.data[0];
  console.log('Recovery:', latest.content, latest.driftScore);
}

Akışı kullanıcı deneyimine bağlama desenleri

Plan gereksinimi

Her plan — akışta sınırlama yoktur.

İlgili