ranuts/stream — پاسخ‌های استریمی مدل

تحلیل Server-Sent Events، واژگانی مستقل از فراهم‌کننده برای یک پاسخ استریم‌شده مدل، و تاکردن آن واژگان به بلوک‌های قابل رسم.

import { parseEventStream, mapEventStream, createStreamAccumulator } from 'ranuts/stream';

نقطه ورودی خودش را دارد. هیچ‌چیز اینجا به DOM دست نمی‌زند، پس یک پاسخ را می‌شود در یک آزمون یا روی سرور تا کرد؛ وارد کردنش از ranuts/utils ماژول‌های رو به DOM را هم با خود می‌کشد.

هیچ فراهم‌کننده‌ای اینجا ساکن نیست. هر API متعارف تکمیل گفت‌وگو همان چهار چیز را استریم می‌کند (متن دستیار، متن استدلالِ جداگانه محاسبه‌شده، فراخوانی ابزار، شمار توکن)، اما هرکدام آن‌ها را جور دیگری نام می‌گذارد و درهم می‌بافد. نگاشتنِ رویداد یک فراهم‌کننده به StreamChunk تنها گامِ وابسته به فراهم‌کننده است و همان‌جا پیش شما می‌ماند: پختنِ یک قالب سیم درون کتابخانه، دو لایه دیگر را برای هرکس دیگری بی‌مصرف می‌کرد.

سه لایه

لایه	چه می‌کند
parseEventStream(source)	بایت‌ها ← ServerSentEvent. فقط لایه انتقال.
StreamChunk	واژگانی که یک پاسخ در آن می‌رسد.
createStreamAccumulator()	تکه‌ها را به بلوک‌هایی تا می‌کند که یک نما می‌تواند رسمشان کند.

mapEventStream(source, map) دو تای نخست را به هم می‌دوزد: رویدادها را می‌پیماید و می‌گذارد نگاشت شما برای هرکدام صفر یا چند تکه برگرداند. برگرداندن [] همان راهِ دور انداختن یک keep-alive یا نگهبان [DONE] است.

واژگان

type StreamChunk =
  | { type: 'block-start'; index: number; blockType: ContentBlockType }
  | { type: 'text-delta'; index: number; text: string }
  | { type: 'reasoning-delta'; index: number; text: string }
  | { type: 'tool-call-delta'; index: number; id: string; name?: string; argumentsDelta: string }
  | { type: 'block-end'; index: number; block: ContentBlock }
  | { type: 'usage'; usage: TokenUsage }
  | { type: 'finish'; reason: FinishReason };

• index دلتاهای درهم‌بافته را به هم مربوط می‌کند. استدلال و متن درهم می‌رسند و چند فراخوانی ابزار هم‌زمان باز می‌شوند، پس ترتیب رسیدن، گروه‌بندی نیست.
• block-end بلوکِ سرهم‌شده را می‌آورد و بر آنچه دلتاها ساخته‌اند می‌چربد. مصرف‌کننده‌ای که تنها بلوک‌های تمام‌شده را می‌خواهد می‌تواند هر دلتا را نادیده بگیرد.
• آرگومان‌های ابزار همان متن خام JSON می‌مانند. نیمی از یک سند JSON یک مقدار نیست. arguments را یک بار و پس از finish تحلیل کنید: تحلیلِ argumentsDelta در میانه جریان، همان‌جایی است که فراخوانی ابزارهای استریمی معمولاً می‌شکنند.
• block-start اختیاری است. چند فراهم‌کننده بلوک را با نخستین دلتایش باز می‌کنند، پس انباشتگر هم در صورت نیاز یکی باز می‌کند. در نگاشت خودتان هم آن را الزامی نکنید.
• finish پایان می‌دهد. usage پیش از آن می‌رسد؛ پس از آن چیزی نمی‌آید.

تا کردن یک پاسخ

const accumulator = createStreamAccumulator();

for await (const chunk of mapEventStream(response.body, toStreamChunks)) {
  accumulator.push(chunk);
  render(accumulator.snapshot());
}

const { blocks, usage, finishReason } = accumulator.snapshot();
const calls = accumulator.toolCalls(); // آرگومان‌ها هنوز متن‌اند — همین‌جا تحلیلشان کنید

snapshot() تغییرناپذیر است: عکس لحظه‌ای‌ای که در میانه جریان گرفته شود مقدارهای همان لحظه را نگه می‌دارد، پس یک نما می‌تواند یکی را در دست بگیرد بی‌آنکه push بعدی زیر دستش عوضش کند. text() و reasoning() بلوک‌های خودشان را به ترتیب اندیس به هم می‌چسبانند و reset() نمونه را برای پاسخ بعدی خالی می‌کند.

تحلیل‌گر SSE از پس چه چیزهایی برمی‌آید

قواعد قاب‌بندی کم‌شمارند و تقریباً هرگز کامل پیاده نمی‌شوند. parseEventStream این‌ها را پوشش می‌دهد:

• مرز یک تکه در هر جایی، از جمله وسط یک نویسه چندبایتی و میان دو نیمهٔ یک \r\n
• فیلدهای تکراری data: که با \n به هم می‌پیوندند
• برداشتن دقیقاً یک فاصله پس از دونقطه
• خط‌های توضیح :، همان راهی که سرورها با آن اتصال را گرم نگه می‌دارند
• BOM آغازین
• بلوک پایانی‌ای که سرور هرگز با خط خالی تمامش نکرده
• ReadableStreamی که Symbol.asyncIterator ندارد

جز ReadableStream، هر AsyncIterable<Uint8Array>ی را هم می‌پذیرد، پس یک آزمون می‌تواند بدون شبکه برش‌های بایتی به آن بدهد.

یک نگاشت واقعی

packages/im در همین مخزن یک مصرف‌کننده کارآمد است: مسیر SSE سازگار با OpenAI، نگاشت به StreamChunk، و نمایی که به‌جای چسباندن دلتاها به دست خودش، یک عکس لحظه‌ای را نگه می‌دارد. آزمون رفت‌وبرگشتش بایت‌های سرور واقعی را با چند اندازه تکه از میان کلاینت واقعی می‌گذراند، پس این دو نیمه نمی‌توانند از هم فاصله بگیرند.

همچنین ببینید

• ranuts/conversation: رویدادهای حاصل را به گره‌های قابل رسم می‌نگارد
• <r-conversation>: آن گره‌ها را رسم می‌کند