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

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, 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 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.

// 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. 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.

// 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, 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.

// 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 and 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.

// 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): 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): 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): 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): 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, 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 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. This module exports the handleStream function, which is imported by API routers such as src/routers/agent/run.js and 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, 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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}