Jump to content

Connect SuperML | Leeroopedia MCP: Equip your AI agents with best practices, code verification, and debugging knowledge. Powered by Leeroo — building Organizational Superintelligence. Contact us at founders@leeroo.com.

Implementation:Openai Openai node Stream

From Leeroopedia
Revision as of 13:36, 16 February 2026 by Admin (talk | contribs) (Auto-imported from implementations/Openai_Openai_node_Stream.md)
(diff) ← Older revision | Latest revision (diff) | Newer revision → (diff)
Knowledge Sources
Domains SDK, Streaming
Last Updated 2026-02-15 12:00 GMT

Overview

The Stream class is the foundational generic wrapper for consuming Server-Sent Events (SSE) responses from the OpenAI API as async iterables.

Description

Stream<Item> implements AsyncIterable<Item>, allowing consumers to use for await...of loops to process streamed data. It is the core primitive upon which higher-level streaming abstractions such as ChatCompletionStream, AssistantStream, and ResponseStream are built.

The class provides two static factory methods for constructing streams: fromSSEResponse parses a raw HTTP Response with an SSE body by decoding chunks, splitting on double newlines, and yielding parsed JSON objects (stopping when the [DONE] sentinel is received). fromReadableStream takes a newline-delimited ReadableStream of JSON values, typically one produced by toReadableStream(), and yields each parsed line as an item.

The stream tracks an AbortController that is used to cancel the underlying HTTP request. If the consumer breaks out of the iteration early, the controller is automatically aborted. The tee() method splits a stream into two independent streams that can be consumed at different speeds, and toReadableStream() serializes the stream back into a newline-delimited ReadableStream of JSON strings for cross-boundary transport (e.g., server to browser).

Usage

This class is used internally whenever a streaming API call is made (e.g., client.chat.completions.create({ stream: true })). It is also used directly when bridging streams between server and client via Stream.fromReadableStream() and .toReadableStream().

Code Reference

Source Location

Signature

export class Stream<Item> implements AsyncIterable<Item> {
  controller: AbortController;

  constructor(
    iterator: () => AsyncIterator<Item>,
    controller: AbortController,
    client?: OpenAI,
  );

  static fromSSEResponse<Item>(
    response: Response,
    controller: AbortController,
    client?: OpenAI,
  ): Stream<Item>;

  static fromReadableStream<Item>(
    readableStream: ReadableStream,
    controller: AbortController,
    client?: OpenAI,
  ): Stream<Item>;

  [Symbol.asyncIterator](): AsyncIterator<Item>;
  tee(): [Stream<Item>, Stream<Item>];
  toReadableStream(): ReadableStream;
}

Import

import { Stream } from 'openai/streaming';

I/O Contract

Inputs

Name Type Required Description
iterator () => AsyncIterator<Item> Yes Factory function that returns the async iterator producing stream items
controller AbortController Yes Controller used to abort the underlying HTTP request
client OpenAI No Client instance used for logging

Outputs

Name Type Description
Stream<Item> AsyncIterable<Item> An async iterable that yields parsed items from the SSE stream
tee() [Stream<Item>, Stream<Item>] Two independent streams reading from the same source
toReadableStream() ReadableStream A web-standard ReadableStream of newline-delimited JSON

Usage Examples

Basic Usage

import OpenAI from 'openai';

const client = new OpenAI();
const stream = await client.chat.completions.create({
  model: 'gpt-4o',
  messages: [{ role: 'user', content: 'Hello' }],
  stream: true,
});

for await (const chunk of stream) {
  process.stdout.write(chunk.choices[0]?.delta?.content || '');
}

Splitting a Stream

const [stream1, stream2] = stream.tee();

// Consume independently
for await (const chunk of stream1) { /* logging */ }
for await (const chunk of stream2) { /* processing */ }

Server-to-Client Transport

// Server side: convert to ReadableStream for HTTP response
const readableStream = stream.toReadableStream();

// Client side: reconstruct from ReadableStream
const clientStream = Stream.fromReadableStream<ChatCompletionChunk>(readableStream, new AbortController());
for await (const chunk of clientStream) {
  // process chunk
}

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment