← Back to the summary

A Producer-Consumer EventStream Channel That Decouples LLM Streaming from Agent Loops

This article is the third installment in the "Developing a Coding Agent from Scratch" series, primarily introducing the Agent's event stream mechanism and the implementation of an EventStream channel.

We all know that when calling a large model API, streaming responses are typically enabled. The large model continuously returns data chunks via the SSE (Server-Sent Events) protocol. Each SSE message carries a JSON string as a payload in its data field. The JSON structures returned by different providers (such as field names delta or content) vary. Taking text generation as an example, when you ask "What's the weather like today?", the large model continuously returns data chunks via the SSE (Server-Sent Events) protocol, in a format similar to:

// First SSE message
data: {
  "id": "chatcmpl-123",
  "object": "chat.completion.chunk",
  "created": 1694268190,
  "model": "gpt-4",
  "choices": [
    {
      "index": 0,
      "delta": {
        "content": "Today"
      },
      "finish_reason": null
    }
  ]
}
// Second SSE message
data: {
  "id": "chatcmpl-123",
  "object": "chat.completion.chunk",
  "created": 1694268190,
  "model": "gpt-4",
  "choices": [
    {
      "index": 0,
      "delta": {
        "content": "the weather"
      },
      "finish_reason": null
    }
  ]
}
// Third SSE message
data: {
  "id": "chatcmpl-123",
  "object": "chat.completion.chunk",
  "created": 1694268190,
  "model": "gpt-4",
  "choices": [
    {
      "index": 0,
      "delta": {
        "content": "is sunny"
      },
      "finish_reason": null
    }
  ]
}
// and so on

"Note: For ease of understanding, this section only uses text delta events as an example. In an actual Coding Agent project, the stream will also include events like tool_call (tool invocation) and done (end flag), which we will handle uniformly in the subsequent adapter implementation chapter."

Later, we will add a unified adapter to adapt the event streams from different providers into a unified event stream format, for example:

{
    "type": "text_delta",
    "contentIndex": 0,
    "delta": "Today"
}
{
    "type": "text_delta",
    "contentIndex": 0,
    "delta": "the weather"
}
{
    "type": "text_delta",
    "contentIndex": 0,
    "delta": "is sunny"
}
// and so on

After obtaining the results, our Agent can process the results according to the event stream format, such as rendering to the terminal, calling tools, etc.

Producers and Consumers

Having understood the above, you've probably already figured out that the EventStream channel is essentially a producer-consumer model. In this model, the producer is the adapter code responsible for calling push() to inject events into the channel (it fetches raw data from the large model API), and the consumer is the Agent's for await...of loop. The producer continuously produces events; for example, when we request the large model API, it continuously returns "Today", "the weather", "is sunny", etc. The consumer continuously consumes events, taking each event and continuously rendering it on the terminal interface. Next, we will start implementing an EventStream channel.

Implementing the EventStream Channel

First, let's briefly understand what we are ultimately implementing:

EventStream is a real-time data channel. It seamlessly connects the two asynchronous activities of the large model pushing data and the Agent consuming data through the same in-memory object, allowing the Agent to use a simple for await...of loop to read the real-time stream as if reading an array.

Before starting the implementation, let's clarify a key concept: In order for the EventStream class to be traversable by a for await...of loop, it must be an AsyncIterable. We need to implement the [Symbol.asyncIterator] method on the class, which returns an asynchronous iterator object (containing a next(): Promise<...> method). When the for await...of loop traverses an EventStream instance, it calls the [Symbol.asyncIterator] method, returning an asynchronous iterator object. The asynchronous iterator object's next() method returns a Promise that eventually resolves an event object. When the event object's type is done, the loop ends.

Next, we will implement the EventStream class step by step. First, create an event-stream.ts file in the packages/ai/src/utils directory.

Defining Option Types

First, we define the EventStreamOptions type for the options passed when instantiating this class:

interface EventStreamOptions<TEvent,TResult> {
    validate(event: TEvent): void;
    isTerminal(event: TEvent): boolean;
    getResult(event: TEvent): TResult;
}

This way, when instantiating the EventStream class, we can pass in an options object to define the behavior of the event stream. For example, we can validate whether an event conforms to the expected format, determine if an event is the end, or get the final result when the stream ends.

Then, we continue by writing the Waiter consumer type to handle the resolve and reject operations of the asynchronous iterator, as well as the normalizeError function to uniformly convert unknown error objects to the Error type.

interface Waiter<TEvent> {
  resolve(result: IteratorResult<TEvent>): void;
  reject(cause: unknown): void;
}

function normalizeError(cause: unknown): Error {
  return cause instanceof Error ? cause : new Error(String(cause));
}

Adding Class Fields and Constructor

Next, we start defining the EventStream class. Because we will later support event types (TEvent) from multiple APIs and the final return result (TResult) (including parameters like total token count, end reason, elapsed time, etc.), we define them here in generic form. As mentioned above, this class is an AsyncIterable, so we implement it based on the AsyncIterable interface.

export class EventStream<TEvent, TResult> implements AsyncIterable<TEvent> {
    // Event stream options
    private readonly options: EventStreamOptions<TEvent, TResult>;
    // Event queue
    private readonly queue: TEvent[] = [];
    // Consumer wait queue
    private readonly waiters: Waiter<TEvent>[] = [];
    // Final result Promise
    private readonly finalResultPromise: Promise<TResult>;
    // Resolve final result
    private resolveFinalResult!: (result: TResult) => void;
    // Reject final result
    private rejectFinalResult!: (cause: unknown) => void;
    // Whether the stream has ended
    private terminal = false;
    // Error object
    private failure?: Error;
    // Whether the iterator has been claimed (only one iterator can run per instance)
    private iteratorClaimed = false;

    constructor(options: EventStreamOptions<TEvent, TResult>) {
        this.options = options;
        // When the result method is called, it returns the final result Promise. When the stream ends, the final result is returned when this.resolveFinalResult is executed.
        this.finalResultPromise = new Promise<TResult>((resolve, reject) => {
            this.resolveFinalResult = resolve;
            this.rejectFinalResult = reject;
        });

        void this.finalResultPromise.catch(() => undefined);
    }

    [Symbol.asyncIterator](): AsyncIterator<any> {
        return {
            async next() {
                return { value: '', done: false };
            }
        };
    }
    // Get the final result
    result(): Promise<TResult> {
		return this.finalResultPromise;
	}
}

Note: Here we temporarily add [Symbol.asyncIterator]() to prevent errors; the specific implementation will be added later.

void this.finalResultPromise.catch(() => undefined); This line of code only marks the internal Promise as having a rejection handler, preventing unhandled rejection warnings when the caller only iterates and does not call result(). result() still returns the original Promise, so failures will still reject and will not be swallowed.

Adding the `push` Method to the Class

In the EventStream class, we need a method to add events to the event queue. This method is the push method. It's simple to understand: when requesting the large model API, we call this method to add the returned events one by one to the event queue. But before that, we need to add a series of checks. First, add a function assertCanPush to determine if an event can be added. Throw an error directly when there is a failure reason or the stream has already ended.

private assertCanPush(): void {
		if (this.failure) {
			throw this.failure;
		}
		if (this.terminal) {
			throw new Error("EventStream is already settled");
		}
	}

Add another fail method for handling error situations. When the large model API returns an error code, the network is interrupted, or other unrecoverable exceptions occur, we call the fail method to handle them. This method stores the error object in the dedicated failure field and immediately rejects all consumers (waiters) that are waiting for data. At the same time, it also calls rejectFinalResult, putting the final result Promise into a rejected state.

	fail(cause: unknown): void {
		if (this.failure) {
			return;
		}
		if (this.terminal) {
			throw new Error("EventStream is already settled");
		}

		const error = normalizeError(cause);
		this.failure = error;
		this.rejectFinalResult(error);
		while (this.waiters.length > 0) {
			const waiter = this.waiters.shift();
			waiter?.reject(error);
		}
	}

Next, we can implement the push method:

  push(event: TEvent): void {
    // Throws an error if it doesn't pass
    this.assertCanPush();
    // Final result when the stream ends
    let terminalResult: { value: TResult } | undefined;
    try {
      // Validate if the event is legal
      this.options.validate(event);
      // If the stream event ends, get the final result
      if (this.options.isTerminal(event)) {
        terminalResult = { value: this.options.getResult(event) };
      }
    } catch (cause) {
      const error = normalizeError(cause);
      // If an error occurs, call the fail method to reject the waiting waiters and the Promise for the final result
      this.fail(error);
      // Throw the error
      throw error;
    }

    if (terminalResult) {
      this.terminal = true;
      // Resolve the final result Promise
      this.resolveFinalResult(terminalResult.value);
    }

    const waiter = this.waiters.shift();
    if (waiter) {
      // If there is a waiting consumer, resolve the event directly to it
      waiter.resolve({ value: event, done: false });
    } else {
      // If there is no waiting consumer, add the event to the queue
      this.queue.push(event);
    }
    // This while loop will not execute in a standard single-consumer scenario; it is used to defend against extreme concurrency scenarios, ensuring no waiters are missed. This is a typical practice of defensive programming.
    if (terminalResult) {
      while (this.waiters.length > 0) {
        const pendingWaiter = this.waiters.shift();
        pendingWaiter?.resolve({ value: undefined, done: true });
      }
    }
  }

Here, waiter is the waiting voucher that the Agent is suspended on when calling nextEvent(). It contains the resolve and reject methods needed to wake up the Agent. For example:

// Assuming options (including validate, isTerminal, getResult) have been defined
const stream = new EventStream<MyEvent, MyResult>(options);
for await (const event of stream) {
    console.log(event);
}

This will call the [Symbol.asyncIterator]() method. Next, we will start implementing the [Symbol.asyncIterator]() method.

At this point, the push method is responsible for putting data in, while the nextEvent method is responsible for taking data out. They are linked through the two internal arrays queue and waiters: when the consumer calls nextEvent() and there is data, it returns directly; when there is no data, the resolve method is stored in waiters, waiting for push to wake it up. This completely decouples the two ends in time.

Implementing the `[Symbol.asyncIterator]()` Method

When we for await...of an instantiated EventStream, the [Symbol.asyncIterator]() method is called. This method returns an asynchronous iterator object containing a next function. The next function returns a Promise object that resolves to an object containing value and done properties. value is the event, and done is a boolean. If the event is the last event, done is true, and the loop stops.

  [Symbol.asyncIterator](): AsyncIterator<TEvent> {
    if (this.iteratorClaimed) {
      throw new Error("EventStream supports exactly one async iterator");
    }
    // Set the iterator as claimed to prevent duplicate calls
    this.iteratorClaimed = true;

    return {
      next: () => this.nextEvent(),
    };
  }

Implement the nextEvent method:

  private nextEvent(): Promise<IteratorResult<TEvent>> {
    // If there are events in the queue, directly take the first event and resolve
    if (this.queue.length > 0) {
      const queuedEvent = this.queue.shift() as TEvent;
      return Promise.resolve({ value: queuedEvent, done: false });
    }
    // If there is an error, reject the Promise
    if (this.failure) {
      return Promise.reject(this.failure);
    }
    // If the stream event has ended, resolve as undefined, done is true, ending the loop
    if (this.terminal) {
      return Promise.resolve({ value: undefined, done: true });
    }
    // If the event queue is empty, save the waiter's resolve and reject methods. When an event is pushed, calling resolve will continue the loop.
    return new Promise<IteratorResult<TEvent>>((resolve, reject) => {
      this.waiters.push({ resolve, reject });
    });
  }

At this point, EventStream is implemented. Finally, import and export EventStream in ai/src/index.ts:

export { EventStream } from "./utils/event-stream.ts";

Summary

So far, we have fully implemented an EventStream event stream channel. Let's review its core design:

What problem does it solve?

In large model streaming responses, the speed at which the producer (API adapter) pushes data and the speed at which the consumer (Agent loop) consumes data often do not match. EventStream completely decouples the two ends in time through an in-memory buffer channel:

The producer only needs to call push(event), without caring whether the consumer is ready.
The consumer only needs to write for await...of, without caring whether the data "arrives immediately" or "arrives later."

How does it work internally?

Component	Role	Corresponding Variable
Event Queue	Event buffer. When the producer is faster than the consumer, events are temporarily stored here and taken out one by one when the consumer is idle.	`queue`
Consumer Wait Queue	When the consumer is faster than the producer, the `resolve` of `nextEvent()` is suspended here, waiting to be awakened after the producer `push`es.	`waiters`
Stream End Flag	Set to `true` when a terminal event (`done`) is received, ensuring the next `next()` returns `done: true`, allowing the loop to exit gracefully.	`terminal`
Failure State	Records an error when the API reports an error or the network is interrupted. All subsequent `next()` calls are directly rejected, implementing "fail-fast."	`failure`
Final Result Promise	After the stream ends, statistical information extracted via `getResult` (such as token consumption, end reason) is returned by the `result()` method, without interfering with the loop body.	`finalResultPromise`

The Git branch for this chapter is EventStream