# How the Stream Utility Manages Real-Time Token Streaming in Lemon AI

> Discover how Lemon AI's stream utility manages real-time token streaming directly from LLMs. Learn about SSE and chunked formats for efficient client delivery.

- Repository: [hexdocom/lemonai](https://github.com/hexdocom/lemonai)
- Tags: internals
- Published: 2026-03-03

---

**Lemon AI's stream utility creates a Node.js `PassThrough` stream, returns an `onTokenStream` callback that encodes tokens into SSE or chunked formats, and pipes them to the client in real-time as the LLM generates them.**

The `hexdocom/lemonai` repository implements a sophisticated three-layer architecture to deliver low-latency token streaming from large language models. At the heart of this system lies the stream utility in [`src/utils/stream.util.js`](https://github.com/hexdocom/lemonai/blob/main/src/utils/stream.util.js), which abstracts protocol-specific encoding and stream management away from the LLM and runtime layers.

## The Three-Layer Streaming Architecture

Lemon AI decouples stream creation, LLM communication, and runtime orchestration into distinct layers that communicate via the `onTokenStream` callback.

### Layer 1: Stream Utility and PassThrough Creation

The `handleStream` function in [`src/utils/stream.util.js`](https://github.com/hexdocom/lemonai/blob/main/src/utils/stream.util.js) initializes a Node.js `PassThrough` stream and configures HTTP headers based on the requested protocol. It supports three output formats: **OpenAI-style SSE**, **plain SSE** (Base64-encoded), and **raw chunked transfer**.

```javascript
// src/utils/stream.util.js (excerpt)
const { PassThrough } = require("stream");

const handleStream = (responseType = 'sse', response, debug = true) => {
  const stream = new PassThrough();
  let onTokenStream = new Function();

  if (responseType === "openai-sse") {
    response.type = "text/event-stream";
    response.set("Cache-Control", "no-cache");
    response.set("Connection", "keep-alive");
    onTokenStream = (token, model = "gpt") => {
      const encoded = JSON.stringify({
        id: uuidv4(),
        object: "chat.completion.chunk",
        created: Math.floor(Date.now() / 1000),
        model,
        choices: [{ index: 0, delta: { role: "assistant", content: token }, finish_reason: null }]
      });
      stream.write(`data: ${encoded}\n\n`);
    };
  }
  // ... additional protocols (sse, stream) omitted for brevity
  return { stream, onTokenStream };
};

```

### Layer 2: LLM Request Configuration

Every LLM provider in Lemon AI extends the base class defined in [`src/completion/llm.base.js`](https://github.com/hexdocom/lemonai/blob/main/src/completion/llm.base.js). When initiating a chat completion, the base class sets `stream: true` and `responseType: "stream"` in the Axios configuration, ensuring the underlying HTTP client returns a Node.js stream rather than a buffered response.

```javascript
// src/completion/llm.base.js (excerpt)
async request(messages = [], options = {}) {
  const body = {
    model: options.model || this.model,
    messages,
    stream: true,                 // Enable token streaming
  };
  const config = {
    url: this.CHAT_COMPLETION_URL,
    method: "post",
    headers: { "Content-Type": "application/json" },
    data: body,
    responseType: "stream",       // Axios returns a Node stream
  };
  const response = await axios.request(config);
  return response;
}

```

### Layer 3: Runtime Token Forwarding

The runtime layer, exemplified by [`src/runtime/LocalRuntime.js`](https://github.com/hexdocom/lemonai/blob/main/src/runtime/LocalRuntime.js), bridges the LLM's raw HTTP stream with the utility's `onTokenStream` callback. It listens for `data` events on the Axios response stream, extracts token text from chunks, and invokes `onTokenStream` to push the token through the `PassThrough` stream to the client.

```javascript
// src/runtime/LocalRuntime.js (excerpt)
async callLLM(prompt, context = {}, options = {}) {
  const { onTokenStream } = context;
  const response = await llm.request([ { role: "user", content: prompt } ], options);
  
  response.data.on('data', chunk => {
    const token = chunk.toString();   // Extract token from LLM chunk
    if (onTokenStream) onTokenStream(token); // Push to client stream
  });
}

```

## Wiring the Stream Through the Router

API endpoints in [`src/routers/agent/run.js`](https://github.com/hexdocom/lemonai/blob/main/src/routers/agent/run.js) and [`src/routers/agent/chat.js`](https://github.com/hexdocom/lemonai/blob/main/src/routers/agent/chat.js) orchestrate the initialization. They invoke `handleStream` to obtain the `stream` and `onTokenStream` pair, inject `onTokenStream` into the runtime context, and assign the `PassThrough` stream to the Koa response body.

```javascript
// src/routers/agent/run.js (excerpt)
const handleStream = require("@src/utils/stream.util");

// Initialize streaming infrastructure
body.responseType = body.responseType || "sse";
const { stream, onTokenStream } = handleStream(body.responseType, response);

// Propagate callback through the agent execution chain
const context = {
  onTokenStream,
  conversation_id,
  user_id: ctx.state.user.id,
};
await runAgentTask(context);   // Triggers LLM calls that use the callback

// Expose stream to HTTP client
ctx.body = stream;

```

## Summary

- **Stream Utility ([`src/utils/stream.util.js`](https://github.com/hexdocom/lemonai/blob/main/src/utils/stream.util.js))**: Creates a Node.js `PassThrough` stream, configures protocol-specific HTTP headers (SSE or chunked), and returns an `onTokenStream` callback that encodes and writes tokens to the stream in real time.
- **LLM Base ([`src/completion/llm.base.js`](https://github.com/hexdocom/lemonai/blob/main/src/completion/llm.base.js))**: Enables streaming at the transport layer by setting `stream: true` and `responseType: "stream"` in Axios requests, returning a raw Node HTTP stream.
- **Runtime ([`src/runtime/LocalRuntime.js`](https://github.com/hexdocom/lemonai/blob/main/src/runtime/LocalRuntime.js))**: Listens for `data` events on the LLM stream and forwards each chunk to `onTokenStream`, bridging the LLM output with the client-facing SSE connection.
- **Router ([`src/routers/agent/run.js`](https://github.com/hexdocom/lemonai/blob/main/src/routers/agent/run.js))**: Initializes the streaming pipeline by invoking `handleStream`, injecting the callback into the execution context, and assigning the `PassThrough` stream to the HTTP response body.

## Frequently Asked Questions

### What is the role of the PassThrough stream in Lemon AI?

The `PassThrough` stream acts as an intermediary buffer that decouples the LLM's token generation from the HTTP response. Created in [`src/utils/stream.util.js`](https://github.com/hexdocom/lemonai/blob/main/src/utils/stream.util.js), it allows the `onTokenStream` callback to write encoded tokens immediately while the Koa server streams them to the client via Server-Sent Events or chunked transfer encoding.

### How does Lemon AI handle different streaming protocols?

The `handleStream` function in [`src/utils/stream.util.js`](https://github.com/hexdocom/lemonai/blob/main/src/utils/stream.util.js) detects the requested protocol via the `responseType` parameter. For `openai-sse`, it formats tokens as JSON chunks matching OpenAI's chat completion schema; for plain `sse`, it Base64-encodes tokens; for `stream`, it uses raw chunked transfer encoding without SSE framing.

### Where is the stream utility defined in the codebase?

The core streaming logic resides in [`src/utils/stream.util.js`](https://github.com/hexdocom/lemonai/blob/main/src/utils/stream.util.js). This module exports the `handleStream` function, which is imported by API routers such as [`src/routers/agent/run.js`](https://github.com/hexdocom/lemonai/blob/main/src/routers/agent/run.js) and [`src/routers/agent/chat.js`](https://github.com/hexdocom/lemonai/blob/main/src/routers/agent/chat.js) to initialize real-time token delivery.

### How does the runtime know when to send tokens to the client?

The runtime receives an `onTokenStream` callback through its execution context, which originates from the router's call to `handleStream`. In [`src/runtime/LocalRuntime.js`](https://github.com/hexdocom/lemonai/blob/main/src/runtime/LocalRuntime.js), the runtime attaches a `data` event listener to the LLM's Axios response stream; each time a chunk arrives, it invokes `onTokenStream(chunk.toString())`, pushing the token to the client immediately.