LangChain.js streaming responses in an Express API

Streaming LLM responses in an Express API isn’t just about calling stream() and piping to the response. You need proper error handling, connection lifecycle management, and a way to surface token usage without buffering the entire response. This guide walks through a complete langchain.js streaming express api implementation that handles backpressure, provider failures, and client disconnections gracefully.

Step 1: Set up the project and dependencies

Create a new Node project with TypeScript and install the minimal dependencies. We’ll use the OpenAI provider for this example, but the streaming pattern works identically across Anthropic, Google, and other providers supported by LangChain.js.

mkdir langchain-streaming-api && cd langchain-streaming-api
npm init -y
npm install express langchain @langchain/openai zod
npm install -D typescript tsx @types/express @types/node

Create a tsconfig.json targeting modern Node:

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "lib": ["ES2022"],
    "outDir": "./dist",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "resolveJsonModule": true,
    "declaration": true
  },
  "include": ["src/**/*"],
  "exclude": ["node_modules", "dist"]
}

Add a dev script to package.json:

"scripts": {
  "dev": "tsx watch src/index.ts",
  "build": "tsc",
  "start": "node dist/index.js"
}

Step 2: Define the request schema and types

Validate incoming requests with Zod. This prevents malformed payloads from reaching the model and gives you a single source of truth for the API contract.

// src/types.ts
import { z } from "zod";

export const ChatRequestSchema = z.object({
  messages: z.array(
    z.object({
      role: z.enum(["system", "user", "assistant"]),
      content: z.string().min(1).max(100_000),
    })
  ).min(1).max(100),
  model: z.string().optional(),
  temperature: z.number().min(0).max(2).optional(),
  maxTokens: z.number().int().positive().max(8192).optional(),
  stream: z.boolean().default(true),
});

export type ChatRequest = z.infer<typeof ChatRequestSchema>;

export interface StreamChunk {
  content: string;
  usage?: {
    promptTokens: number;
    completionTokens: number;
    totalTokens: number;
  };
  finishReason?: "stop" | "length" | "content_filter" | "tool_calls" | "error";
}

Step 3: Create the LangChain streaming service

Extract the model invocation into a service class. This keeps your route handlers thin and makes the streaming logic testable in isolation.

// src/services/llm.ts
import { ChatOpenAI } from "@langchain/openai";
import { HumanMessage, SystemMessage, AIMessage } from "@langchain/core/messages";
import { ChatRequest, StreamChunk } from "../types";

export class LLMService {
  private model: ChatOpenAI;

  constructor(apiKey: string, defaultModel = "gpt-4o-mini") {
    this.model = new ChatOpenAI({
      apiKey,
      model: defaultModel,
      streaming: true,
      // These callbacks fire for each token; we'll use them to build our stream
      callbacks: [],
    });
  }

  async *streamChat(request: ChatRequest): AsyncGenerator<StreamChunk> {
    const messages = request.messages.map((m) => {
      switch (m.role) {
        case "system":
          return new SystemMessage(m.content);
        case "user":
          return new HumanMessage(m.content);
        case "assistant":
          return new AIMessage(m.content);
      }
    });

    // Bind request-specific params without mutating the shared model instance
    const model = request.model || request.temperature || request.maxTokens
      ? this.model.bind({
          ...(request.model && { model: request.model }),
          ...(request.temperature && { temperature: request.temperature }),
          ...(request.maxTokens && { maxTokens: request.maxTokens }),
        })
      : this.model;

    let accumulatedContent = "";
    let usage: StreamChunk["usage"] = undefined;

    const stream = await model.stream(messages);

    for await (const chunk of stream) {
      const content = chunk.content as string;
      if (content) {
        accumulatedContent += content;
        yield { content };
      }

      // Usage metadata arrives in the final chunk's response_metadata
      if (chunk.response_metadata?.tokenUsage) {
        usage = {
          promptTokens: chunk.response_metadata.tokenUsage.promptTokens,
          completionTokens: chunk.response_metadata.tokenUsage.completionTokens,
          totalTokens: chunk.response_metadata.tokenUsage.totalTokens,
        };
      }

      if (chunk.response_metadata?.finishReason) {
        yield {
          content: "",
          usage,
          finishReason: chunk.response_metadata.finishReason,
        };
      }
    }

    // Fallback if finishReason never arrived
    if (!usage) {
      yield { content: "", finishReason: "stop" };
    }
  }
}

Key details: the generator yields each token chunk immediately, and we surface usage metadata when the provider includes it in response_metadata. The bind() call creates a per-request model instance so concurrent requests with different parameters don’t interfere.

Step 4: Build the Express route with proper SSE formatting

Server-Sent Events (SSE) is the standard way to stream text to browser clients. It works over plain HTTP, handles reconnection automatically, and doesn’t require WebSocket infrastructure.

// src/routes/chat.ts
import { Request, Response } from "express";
import { ChatRequestSchema } from "../types";
import { LLMService } from "../services/llm";

export function createChatRouter(llmService: LLMService) {
  const router = require("express").Router();

  router.post("/chat", async (req: Request, res: Response) => {
    // Validate request
    const parseResult = ChatRequestSchema.safeParse(req.body);
    if (!parseResult.success) {
      return res.status(400).json({
        error: "Invalid request",
        details: parseResult.error.flatten(),
      });
    }

    const request = parseResult.data;

    // Client explicitly asked for non-streaming? Handle that separately.
    if (!request.stream) {
      return handleNonStreaming(req, res, llmService, request);
    }

    // SSE headers
    res.setHeader("Content-Type", "text/event-stream");
    res.setHeader("Cache-Control", "no-cache, no-transform");
    res.setHeader("Connection", "keep-alive");
    res.setHeader("X-Accel-Buffering", "no"); // Disable nginx buffering
    res.flushHeaders();

    // Track client connection
    const clientId = Date.now().toString(36);
    console.log(`[${clientId}] Stream started`);

    const cleanup = () => {
      console.log(`[${clientId}] Stream ended`);
    };

    req.on("close", cleanup);
    req.on("error", cleanup);

    try {
      for await (const chunk of llmService.streamChat(request)) {
        // Check if client disconnected
        if (req.destroyed || req.socket.destroyed) {
          break;
        }

        // Format as SSE: each yield becomes one event
        const payload = JSON.stringify(chunk);
        res.write(`data: ${payload}\n\n`);

        // Explicit flush ensures chunks don't buffer in Node's internal queue
        // This is critical for low-latency token delivery
        if (!res.flushHeaders()) {
          // flushHeaders returns false if kernel buffer is full
          await new Promise((resolve) => res.once("drain", resolve));
        }
      }

      // Signal completion
      res.write("data: [DONE]\n\n");
      res.end();
    } catch (error) {
      console.error(`[${clientId}] Stream error:`, error);
      if (!res.destroyed) {
        res.write(`data: ${JSON.stringify({ error: "Stream failed" })}\n\n`);
        res.end();
      }
    }
  });

  return router;
}

async function handleNonStreaming(
  req: Request,
  res: Response,
  llmService: LLMService,
  request: ReturnType<typeof ChatRequestSchema.parse>
) {
  // Collect the full stream into a single response
  let fullContent = "";
  let usage: any = undefined;
  let finishReason: any = undefined;

  for await (const chunk of llmService.streamChat(request)) {
    fullContent += chunk.content;
    if (chunk.usage) usage = chunk.usage;
    if (chunk.finishReason) finishReason = chunk.finishReason;
  }

  res.json({
    content: fullContent,
    usage,
    finishReason,
  });
}

The flushHeaders() call after each write is the critical piece most tutorials miss. Without it, Node’s TCP stack may buffer multiple tokens before sending, defeating the purpose of streaming. The drain event handles backpressure when the client can’t keep up.

Step 5: Wire it all together in the entry point

// src/index.ts
import express from "express";
import { createChatRouter } from "./routes/chat";
import { LLMService } from "./services/llm";

const app = express();
app.use(express.json({ limit: "1mb" }));

// Health check for load balancers
app.get("/health", (_req, res) => {
  res.json({ status: "ok", timestamp: new Date().toISOString() });
});

// Initialize LLM service
const apiKey = process.env.OPENAI_API_KEY;
if (!apiKey) {
  console.error("OPENAI_API_KEY not set");
  process.exit(1);
}

const llmService = new LLMService(apiKey);
app.use("/api", createChatRouter(llmService));

// Global error handler
app.use((err: Error, _req: express.Request, res: express.Response, _next: express.NextFunction) => {
  console.error("Unhandled error:", err);
  res.status(500).json({ error: "Internal server error" });
});

const port = process.env.PORT || 3000;
app.listen(port, () => {
  console.log(`Server listening on http://localhost:${port}`);
});

Create a .env file with your API key:

OPENAI_API_KEY=sk-...
PORT=3000

Step 6: Verify the streaming endpoint works

Start the server and test with curl. The -N flag disables curl’s output buffering so you see tokens as they arrive.

npm run dev

In another terminal:

curl -N -X POST http://localhost:3000/api/chat \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [
      {"role": "user", "content": "Count to 10 slowly, one number per token"}
    ],
    "stream": true
  }'

You should see output like:

data: {"content":"1"}

data: {"content":"2"}

data: {"content":"3"}

...

data: {"content":"","usage":{"promptTokens":12,"completionTokens":15,"totalTokens":27},"finishReason":"stop"}

data: [DONE]

Each data: line is a valid JSON object you can parse incrementally on the client. The final [DONE] sentinel matches the OpenAI streaming convention.

Step 7: Handle provider failures with fallback

Production systems need resilience when a provider degrades. If you’re routing through a gateway that exposes multiple providers (like n4n.ai does with 240+ models behind one OpenAI-compatible endpoint), you can implement fallback at the service layer without changing your route handlers.

// src/services/llm.ts (extended)
import { ChatOpenAI } from "@langchain/openai";
// ... existing imports

export class LLMService {
  private primary: ChatOpenAI;
  private fallback: ChatOpenAI | null = null;

  constructor(
    primaryApiKey: string,
    primaryModel = "gpt-4o-mini",
    fallbackConfig?: { apiKey: string; model: string; baseUrl?: string }
  ) {
    this.primary = new ChatOpenAI({
      apiKey: primaryApiKey,
      model: primaryModel,
      streaming: true,
    });

    if (fallbackConfig) {
      this.fallback = new ChatOpenAI({
        apiKey: fallbackConfig.apiKey,
        model: fallbackConfig.model,
        streaming: true,
        configuration: fallbackConfig.baseUrl ? { baseURL: fallbackConfig.baseUrl } : undefined,
      });
    }
  }

  async *streamChat(request: ChatRequest): AsyncGenerator<StreamChunk> {
    const models = [this.primary, this.fallback].filter(Boolean) as ChatOpenAI[];

    for (const model of models) {
      try {
        yield* this.streamWithModel(model, request);
        return; // Success, exit the generator
      } catch (error) {
        console.warn(`Model ${model.model} failed, trying fallback:`, error);
        // Continue to next model
      }
    }

    throw new Error("All models exhausted");
  }

  private async *streamWithModel(model: ChatOpenAI, request: ChatRequest): AsyncGenerator<StreamChunk> {
    // ... same streaming logic as before, but using the passed model
    const messages = request.messages.map((m) => {
      switch (m.role) {
        case "system": return new SystemMessage(m.content);
        case "user": return new HumanMessage(m.content);
        case "assistant": return new AIMessage(m.content);
      }
    });

    const boundModel = request.model || request.temperature || request.maxTokens
      ? model.bind({
          ...(request.model && { model: request.model }),
          ...(request.temperature && { temperature: request.temperature }),
          ...(request.maxTokens && { maxTokens: request.maxTokens }),
        })
      : model;

    const stream = await boundModel.stream(messages);

    for await (const chunk of stream) {
      const content = chunk.content as string;
      if (content) yield { content };

      if (chunk.response_metadata?.tokenUsage) {
        yield {
          content: "",
          usage: {
            promptTokens: chunk.response_metadata.tokenUsage.promptTokens,
            completionTokens: chunk.response_metadata.tokenUsage.completionTokens,
            totalTokens: chunk.response_metadata.tokenUsage.totalTokens,
          },
        };
      }

      if (chunk.response_metadata?.finishReason) {
        yield { content: "", finishReason: chunk.response_metadata.finishReason };
      }
    }
  }
}

Update the entry point to pass fallback config:

// src/index.ts (partial)
const llmService = new LLMService(
  process.env.OPENAI_API_KEY!,
  "gpt-4o-mini",
  process.env.FALLBACK_API_KEY
    ? {
        apiKey: process.env.FALLBACK_API_KEY,
        model: process.env.FALLBACK_MODEL || "claude-3-haiku-20240307",
        baseUrl: process.env.FALLBACK_BASE_URL, // e.g., "https://api.n4n.ai/v1"
      }
    : undefined
);

This pattern keeps your Express routes clean while the service handles provider diversity. The gateway forwards cache-control hints from providers, so you can also respect upstream caching directives if needed.

Step 8: Build a minimal client for verification

Create a simple HTML file to verify end-to-end streaming in a browser. This also demonstrates the correct SSE parsing pattern.

<!-- public/index.html -->
<!DOCTYPE html>
<html>
<head>
  <meta charset="utf-8">
  <title>Streaming Test</title>
  <style>
    body { font-family: system-ui; max-width: 800px; margin: 2rem auto; padding: 0 1rem; }
    #output { white-space: pre-wrap; border: 1px solid #ddd; padding: 1rem; min-height: 200px; }
    .usage { color: #666; font-size: 0.875rem; margin-top: 1rem; }
    button { padding: 0.5rem 1rem; font-size: 1rem; }
  </style>
</head>
<body>
  <h1>LangChain.js Streaming Demo</h1>
  <textarea id="prompt" rows="4" style="width:100%">Explain streaming in three sentences.</textarea>
  <br><br>
  <button id="send">Stream Response</button>
  <div id="output"></div>
  <div class="usage" id="usage"></div>

  <script>
    const output = document.getElementById('output');
    const usageEl = document.getElementById('usage');
    const send = document.getElementById('send');

    send.addEventListener('click', async () => {
      output.textContent = '';
      usageEl.textContent = '';
      send.disabled = true;

      const response = await fetch('/api/chat', {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({
          messages: [{ role: 'user', content: document.getElementById('prompt').value }],
          stream: true
        })
      });

      const reader = response.body?.getReader();
      const decoder = new TextDecoder();
      let buffer = '';

      try {
        while (true) {
          const { done, value } = await reader!.read();
          if (done) break;

          buffer += decoder.decode(value, { stream: true });
          const lines = buffer.split('\n');
          buffer = lines.pop() || '';

          for (const line of lines) {
            if (line.startsWith('data: ')) {
              const data = line.slice(6);
              if (data === '[DONE]') continue;

              try {
                const chunk = JSON.parse(data);
                if (chunk.content) output.textContent += chunk.content;
                if (chunk.usage) {
                  usageEl.textContent = `Tokens: ${chunk.usage.promptTokens} in, ${chunk.usage.completionTokens} out, ${chunk.usage.totalTokens} total`;
                }
                if (chunk.finishReason) {
                  usageEl.textContent += ` | Finish: ${chunk.finishReason}`;
                }
              } catch (e) {
                console.warn('Parse error:', data);
              }
            }
          }
        }
      } finally {
        send.disabled = false;
      }
    });
  </script>
</body>
</html>

Serve it by adding static middleware in src/index.ts:

import path from "path";
import { fileURLToPath } from "url";

const __dirname = path.dirname(fileURLToPath(import.meta.url));
app.use(express.static(path.join(__dirname, "../public")));

Visit http://localhost:3000 and click “Stream Response” to watch tokens arrive in real time.

Step 9: Add request logging and observability

You can’t debug what you can’t see. Add a lightweight request logger that captures latency, token counts, and error rates without external dependencies.

// src/middleware/logging.ts
import { Request, Response, NextFunction } from "express";

export function requestLogger(req: Request, res: Response, next: NextFunction) {
  const start = process.hrtime.bigint();
  const requestId = crypto.randomUUID().slice(0, 8);

  // Attach request ID for correlation
  (req as any).requestId = requestId;

  res.on("finish", () => {
    const durationMs = Number(process.hrtime.bigint() - start) / 1_000_000;
    const log = {
      requestId,
      method: req.method,
      path: req.path,
      status: res.statusCode,
      durationMs: Math.round(durationMs),
      ip: req.ip,
      userAgent: req.get("user-agent"),
    };

    // Structured JSON for log aggregation
    console.log(JSON.stringify(log));
  });

  next();
}

Apply it in src/index.ts:

import { requestLogger } from "./middleware/logging";
app.use(requestLogger);

Sample log output:

{"requestId":"a1b2c3d4","method":"POST","path":"/api/chat","status":200,"durationMs":1247,"ip":"::1","userAgent":"curl/8.5.0"}

Step 10: Deploy considerations

When moving to production, address these operational concerns:

Process management: Run with PM2 or a container orchestrator. Set NODE_ENV=production to enable Express optimizations.

# ecosystem.config.js for PM2
module.exports = {
  apps: [{
    name: "langchain-api",
    script: "dist/index.js",
    instances: "max",
    exec_mode: "cluster",
    env: { NODE_ENV: "production", PORT: 3000 },
  }]
};

Reverse proxy: Put nginx or a cloud load balancer in front. Critical nginx settings for streaming:

location /api/chat {
    proxy_pass http://localhost:3000;
    proxy_http_version 1.1;
    proxy_set_header Connection "";
    proxy_cache off;
    proxy_buffering off;        # Critical: disable buffering
    proxy_read_timeout 300s;    # Long timeout for slow models
    proxy_send_timeout 300s;
}

Rate limiting: Add per-IP or per-key limits before the route handler. express-rate-limit works well:

import rateLimit from "express-rate-limit";

const limiter = rateLimit({
  windowMs: 60_000,
  max: 60,
  keyGenerator: (req) => req.ip || "unknown",
  handler: (req, res) => res.status(429).json({ error: "Rate limit exceeded" }),
});

app.use("/api/chat", limiter);

Secrets: Never commit API keys. Use your platform’s secret manager (AWS Secrets Manager, GCP Secret Manager, Vault, etc.) and inject at runtime.

Verification checklist

Before considering the implementation complete, verify each of these:

• curl -N shows tokens arriving incrementally, not all at once
• Client disconnect mid-stream cleans up server resources (check logs)
• Non-streaming mode ("stream": false) returns a single JSON response
• Invalid request bodies return 400 with Zod validation details
• Provider failure triggers fallback and completes successfully
• Token usage appears in the final chunk for both streaming and non-streaming
• Load test with 50 concurrent streams doesn’t OOM or stall
• nginx proxy_buffering off confirmed in staging

What to extend next

This foundation handles the hard parts: backpressure, connection lifecycle, provider fallback, and observability. From here you can add:

• Tool calling: Parse tool_calls from chunks and execute functions before resuming the stream
• Conversation persistence: Store messages in Redis or Postgres with the requestId for debugging
• Authentication: Validate API keys in middleware, attach tier limits to the request object
• Metrics: Export Prometheus counters for stream_started, stream_completed, stream_failed, tokens_generated

The streaming pattern stays the same regardless of model provider or feature additions. Master the generator + SSE + backpressure triangle and the rest becomes straightforward composition.