Llm Streaming Response Handler

LLM Streaming Response Handler

Expert in building production-grade streaming interfaces for LLM responses that feel instant and responsive.

When to Use

✅ Use for:

• Chat interfaces with typing animation
• Real-time AI assistants
• Code generation with live preview
• Document summarization with progressive display
• Any UI where users expect immediate feedback from LLMs

❌ NOT for:

• Batch document processing (no user watching)
• APIs that don't support streaming
• WebSocket-based bidirectional chat (use Socket.IO)
• Simple request/response (fetch is fine)

Quick Decision Tree

Does your LLM interaction:
├── Need immediate visual feedback? → Streaming
├── Display long-form content (>100 words)? → Streaming
├── User expects typewriter effect? → Streaming
├── Short response (<50 words)? → Regular fetch
└── Background processing? → Regular fetch

---

Technology Selection

Server-Sent Events (SSE) - Recommended

Why SSE over WebSockets for LLM streaming:

• Simplicity: HTTP-based, works with existing infrastructure
• Auto-reconnect: Built-in reconnection logic
• Firewall-friendly: Easier than WebSockets through proxies
• One-way perfect: LLMs only stream server → client

Timeline:

• 2015-2020: WebSockets for everything
• 2020: SSE adoption for streaming APIs
• 2023+: SSE standard for LLM streaming (OpenAI, Anthropic)
• 2024: Vercel AI SDK popularizes SSE patterns

Streaming APIs

---

Common Anti-Patterns

Anti-Pattern 1: Buffering Before Display

Novice thinking: "Collect all tokens, then show complete response"

Problem: Defeats the entire purpose of streaming.

Wrong approach:

// ❌ Waits for entire response before showing anything
const response = await fetch('/api/chat', { method: 'POST', body: prompt });
const fullText = await response.text();
setMessage(fullText); // User sees nothing until done

Correct approach:

// ✅ Display tokens as they arrive
const response = await fetch('/api/chat', {
  method: 'POST',
  body: JSON.stringify({ prompt })
});

const reader = response.body.getReader();
const decoder = new TextDecoder();

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

  const chunk = decoder.decode(value);
  const lines = chunk.split('\n').filter(line => line.trim());

  for (const line of lines) {
    if (line.startsWith('data: ')) {
      const data = JSON.parse(line.slice(6));
      setMessage(prev => prev + data.content); // Update immediately
    }
  }
}

Timeline:

• Pre-2023: Many apps buffered entire response
• 2023+: Token-by-token display expected

---

Anti-Pattern 2: No Stream Cancellation

Problem: User can't stop generation, wasting tokens and money.

Symptom: "Stop" button doesn't work or doesn't exist.

Correct approach:

// ✅ AbortController for cancellation
const [abortController, setAbortController] = useState<AbortController | null>(null);

const streamResponse = async () => {
  const controller = new AbortController();
  setAbortController(controller);

  try {
    const response = await fetch('/api/chat', {
      signal: controller.signal,
      method: 'POST',
      body: JSON.stringify({ prompt })
    });

    // Stream handling...
  } catch (error) {
    if (error.name === 'AbortError') {
      console.log('Stream cancelled by user');
    }
  } finally {
    setAbortController(null);
  }
};

const cancelStream = () => {
  abortController?.abort();
};

return (
  <button onClick={cancelStream} disabled={!abortController}>
    Stop Generating
  </button>
);

---

Anti-Pattern 3: No Error Recovery

Problem: Stream fails mid-response, user sees partial text with no indication of failure.

Correct approach:

// ✅ Error states and recovery
const [streamState, setStreamState] = useState<'idle' | 'streaming' | 'error' | 'complete'>('idle');
const [errorMessage, setErrorMessage] = useState<string | null>(null);

try {
  setStreamState('streaming');

  // Streaming logic...

  setStreamState('complete');
} catch (error) {
  setStreamState('error');

  if (error.name === 'AbortError') {
    setErrorMessage('Generation stopped');
  } else if (error.message.includes('429')) {
    setErrorMessage('Rate limit exceeded. Try again in a moment.');
  } else {
    setErrorMessage('Something went wrong. Please retry.');
  }
}

// UI feedback
{streamState === 'error' && (
  <div className="error-banner">
    {errorMessage}
    <button onClick={retryStream}>Retry</button>
  </div>
)}

---

Anti-Pattern 4: Memory Leaks from Unclosed Streams

Problem: Streams not cleaned up, causing memory leaks.

Symptom: Browser slows down after multiple requests.

Correct approach:

// ✅ Cleanup with useEffect
useEffect(() => {
  let reader: ReadableStreamDefaultReader | null = null;

  const streamResponse = async () => {
    const response = await fetch('/api/chat', { ... });
    reader = response.body.getReader();

    // Streaming...
  };

  streamResponse();

  // Cleanup on unmount
  return () => {
    reader?.cancel();
  };
}, [prompt]);

---

Anti-Pattern 5: No Typing Indicator Between Tokens

Problem: UI feels frozen between slow tokens.

Correct approach:

// ✅ Animated cursor during generation
<div className="message">
  {content}
  {isStreaming && <span className="typing-cursor">▊</span>}
</div>

.typing-cursor {
  animation: blink 1s step-end infinite;
}

@keyframes blink {
  50% { opacity: 0; }
}

---

Implementation Patterns

Pattern 1: Basic SSE Stream Handler

async function* streamCompletion(prompt: string) {
  const response = await fetch('/api/chat', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ prompt })
  });

  const reader = response.body!.getReader();
  const decoder = new TextDecoder();

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

    const chunk = decoder.decode(value);
    const lines = chunk.split('\n');

    for (const line of lines) {
      if (line.startsWith('data: ')) {
        const data = JSON.parse(line.slice(6));

        if (data.content) {
          yield data.content;
        }

        if (data.done) {
          return;
        }
      }
    }
  }
}

// Usage
for await (const token of streamCompletion('Hello')) {
  console.log(token);
}

Pattern 2: React Hook for Streaming

import { useState, useCallback } from 'react';

interface UseStreamingOptions {
  onToken?: (token: string) => void;
  onComplete?: (fullText: string) => void;
  onError?: (error: Error) => void;
}

export function useStreaming(options: UseStreamingOptions = {}) {
  const [content, setContent] = useState('');
  const [isStreaming, setIsStreaming] = useState(false);
  const [error, setError] = useState<Error | null>(null);
  const [abortController, setAbortController] = useState<AbortController | null>(null);

  const stream = useCallback(async (prompt: string) => {
    const controller = new AbortController();
    setAbortController(controller);
    setIsStreaming(true);
    setError(null);
    setContent('');

    try {
      const response = await fetch('/api/chat', {
        method: 'POST',
        signal: controller.signal,
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify({ prompt })
      });

      const reader = response.body!.getReader();
      const decoder = new TextDecoder();

      let accumulated = '';

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

        const chunk = decoder.decode(value);
        const lines = chunk.split('\n').filter(line => line.trim());

        for (const line of lines) {
          if (line.startsWith('data: ')) {
            const data = JSON.parse(line.slice(6));

            if (data.content) {
              accumulated += data.content;
              setContent(accumulated);
              options.onToken?.(data.content);
            }
          }
        }
      }

      options.onComplete?.(accumulated);
    } catch (err) {
      if (err.name !== 'AbortError') {
        setError(err as Error);
        options.onError?.(err as Error);
      }
    } finally {
      setIsStreaming(false);
      setAbortController(null);
    }
  }, [options]);

  const cancel = useCallback(() => {
    abortController?.abort();
  }, [abortController]);

  return { content, isStreaming, error, stream, cancel };
}

// Usage in component
function ChatInterface() {
  const { content, isStreaming, stream, cancel } = useStreaming({
    onToken: (token) => console.log('New token:', token),
    onComplete: (text) => console.log('Done:', text)
  });

  return (
    <div>
      <div className="message">
        {content}
        {isStreaming && <span className="cursor">▊</span>}
      </div>

      <button onClick={() => stream('Tell me a story')} disabled={isStreaming}>
        Generate
      </button>

      {isStreaming && <button onClick={cancel}>Stop</button>}
    </div>
  );
}

Pattern 3: Server-Side Streaming (Next.js)

// app/api/chat/route.ts
import { OpenAI } from 'openai';

export const runtime = 'edge'; // Required for streaming

export async function POST(req: Request) {
  const { prompt } = await req.json();

  const openai = new OpenAI({
    apiKey: process.env.OPENAI_API_KEY
  });

  const stream = await openai.chat.completions.create({
    model: 'gpt-4',
    messages: [{ role: 'user', content: prompt }],
    stream: true
  });

  // Convert OpenAI stream to SSE format
  const encoder = new TextEncoder();

  const readable = new ReadableStream({
    async start(controller) {
      try {
        for await (const chunk of stream) {
          const content = chunk.choices[0]?.delta?.content;

          if (content) {
            const sseMessage = `data: ${JSON.stringify({ content })}\n\n`;
            controller.enqueue(encoder.encode(sseMessage));
          }
        }

        // Send completion signal
        controller.enqueue(encoder.encode('data: {"done":true}\n\n'));
        controller.close();
      } catch (error) {
        controller.error(error);
      }
    }
  });

  return new Response(readable, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive'
    }
  });
}

---

Production Checklist

□ AbortController for cancellation
□ Error states with retry capability
□ Typing indicator during generation
□ Cleanup on component unmount
□ Rate limiting on API route
□ Token usage tracking
□ Streaming fallback (if API fails)
□ Accessibility (screen reader announces updates)
□ Mobile-friendly (touch targets for stop button)
□ Network error recovery (auto-retry on disconnect)
□ Max response length enforcement
□ Cost estimation before generation

---

When to Use vs Avoid

---

Technology Comparison

---

References

• /references/sse-protocol.md - Server-Sent Events specification details
• /references/vercel-ai-sdk.md - Vercel AI SDK integration patterns
• /references/error-recovery.md - Stream error handling strategies

Scripts

• scripts/stream_tester.ts - Test SSE endpoints locally
• scripts/token_counter.ts - Estimate costs before generation

---

This skill guides: LLM streaming implementation | SSE protocol | Real-time UI updates | Cancellation | Error recovery | Token-by-token display

Stream Error Recovery Strategies

Production patterns for handling errors during LLM streaming and providing graceful recovery.

Error Categories

1. Network Errors

• Connection timeouts
• DNS failures
• Lost connectivity mid-stream

2. API Errors

• Rate limits (429)
• Authentication (401, 403)
• Server errors (500, 503)
• Invalid requests (400)

3. Stream-Specific Errors

• Malformed SSE data
• Unexpected stream termination
• Backpressure overload
• Client-side cancellation

---

Pattern 1: Retry with Exponential Backoff

async function streamWithRetry(
  url: string,
  options: RequestInit,
  maxRetries = 3
): Promise<Response> {
  let attempt = 0;

  while (attempt < maxRetries) {
    try {
      const response = await fetch(url, options);

      // Don't retry on client errors (4xx except rate limit)
      if (response.status >= 400 && response.status < 500 && response.status !== 429) {
        throw new Error(`Client error: ${response.status}`);
      }

      // Don't retry on success
      if (response.ok) {
        return response;
      }

      // Retry on server errors and rate limits
      throw new Error(`Server error: ${response.status}`);

    } catch (error) {
      attempt++;

      if (attempt >= maxRetries) {
        throw error;
      }

      // Exponential backoff: 1s, 2s, 4s
      const delay = Math.pow(2, attempt) * 1000;
      console.log(`Retry ${attempt}/${maxRetries} after ${delay}ms`);

      await new Promise(resolve => setTimeout(resolve, delay));
    }
  }

  throw new Error('Max retries exceeded');
}

// Usage
const response = await streamWithRetry('/api/chat', {
  method: 'POST',
  body: JSON.stringify({ prompt })
});

---

Pattern 2: Rate Limit Handling

async function handleRateLimit(response: Response): Promise<void> {
  if (response.status === 429) {
    const retryAfter = response.headers.get('Retry-After');

    if (retryAfter) {
      const delay = parseInt(retryAfter) * 1000;
      console.log(`Rate limited. Retrying after ${delay}ms`);
      await new Promise(resolve => setTimeout(resolve, delay));
    } else {
      // No Retry-After header, use default backoff
      await new Promise(resolve => setTimeout(resolve, 60000)); // 1 minute
    }
  }
}

// Usage
const response = await fetch('/api/chat', { ... });

if (response.status === 429) {
  await handleRateLimit(response);
  // Retry request
  return fetch('/api/chat', { ... });
}

---

Pattern 3: Graceful Degradation

Show partial response if stream fails mid-way.

function useStreamingWithFallback() {
  const [content, setContent] = useState('');
  const [error, setError] = useState<Error | null>(null);
  const [isComplete, setIsComplete] = useState(false);

  const stream = async (prompt: string) => {
    setError(null);
    setIsComplete(false);

    try {
      const response = await fetch('/api/chat', {
        method: 'POST',
        body: JSON.stringify({ prompt })
      });

      const reader = response.body!.getReader();
      const decoder = new TextDecoder();

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

        if (done) {
          setIsComplete(true);
          break;
        }

        const chunk = decoder.decode(value);
        const lines = chunk.split('\n');

        for (const line of lines) {
          if (line.startsWith('data: ')) {
            const data = JSON.parse(line.slice(6));
            if (data.content) {
              setContent(prev => prev + data.content);
            }
          }
        }
      }
    } catch (err) {
      setError(err as Error);
      // Keep partial content visible
      console.error('Stream failed, showing partial response:', content);
    }
  };

  return { content, error, isComplete, stream };
}

// UI shows partial content + error
{content && (
  <div>
    {content}
    {error && !isComplete && (
      <div className="error-banner">
        ⚠️ Connection lost. Showing partial response.
        <button onClick={() => stream(lastPrompt)}>Resume</button>
      </div>
    )}
  </div>
)}

---

Pattern 4: Resume from Last Token

For very long streams, save checkpoints.

interface StreamCheckpoint {
  prompt: string;
  content: string;
  lastTokenIndex: number;
}

async function streamWithCheckpoints(prompt: string) {
  let checkpoint: StreamCheckpoint | null = loadCheckpoint(prompt);

  if (checkpoint) {
    console.log('Resuming from checkpoint:', checkpoint.lastTokenIndex);
  }

  const response = await fetch('/api/chat', {
    method: 'POST',
    body: JSON.stringify({
      prompt,
      resumeFrom: checkpoint?.lastTokenIndex || 0
    })
  });

  const reader = response.body!.getReader();
  const decoder = new TextDecoder();

  let tokenIndex = checkpoint?.lastTokenIndex || 0;
  let content = checkpoint?.content || '';

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

      const chunk = decoder.decode(value);
      // Process chunk...

      content += newToken;
      tokenIndex++;

      // Save checkpoint every 100 tokens
      if (tokenIndex % 100 === 0) {
        saveCheckpoint({ prompt, content, lastTokenIndex: tokenIndex });
      }
    }
  } catch (error) {
    // Checkpoint saved, can resume later
    console.error('Stream interrupted at token', tokenIndex);
    throw error;
  }
}

---

Pattern 5: Client-Side Timeout

Prevent infinite hanging.

async function streamWithTimeout(
  url: string,
  options: RequestInit,
  timeoutMs = 30000
): Promise<void> {
  const controller = new AbortController();

  // Set timeout
  const timeoutId = setTimeout(() => {
    controller.abort();
  }, timeoutMs);

  try {
    const response = await fetch(url, {
      ...options,
      signal: controller.signal
    });

    // Reset timeout on each chunk
    const reader = response.body!.getReader();
    const decoder = new TextDecoder();

    while (true) {
      // Clear and reset timeout
      clearTimeout(timeoutId);
      const chunkTimeoutId = setTimeout(() => controller.abort(), timeoutMs);

      const { done, value } = await reader.read();

      clearTimeout(chunkTimeoutId);

      if (done) break;

      // Process chunk...
    }
  } catch (error) {
    if (error.name === 'AbortError') {
      throw new Error('Request timed out');
    }
    throw error;
  } finally {
    clearTimeout(timeoutId);
  }
}

---

Pattern 6: User-Friendly Error Messages

function getErrorMessage(error: unknown): string {
  if (error instanceof Error) {
    // Network errors
    if (error.message.includes('Failed to fetch')) {
      return 'Network error. Please check your connection.';
    }

    // Timeout
    if (error.name === 'AbortError') {
      return 'Request timed out. Please try again.';
    }

    // Generic
    return error.message;
  }

  // HTTP errors
  if (typeof error === 'object' && error !== null && 'status' in error) {
    const status = (error as any).status;

    switch (status) {
      case 400:
        return 'Invalid request. Please try rephrasing.';
      case 401:
        return 'Please log in to continue.';
      case 403:
        return 'You don\'t have permission to do this.';
      case 429:
        return 'Too many requests. Please wait a moment.';
      case 500:
        return 'Server error. Please try again later.';
      case 503:
        return 'Service temporarily unavailable.';
      default:
        return `Error ${status}: Something went wrong.`;
    }
  }

  return 'An unexpected error occurred.';
}

// UI
{error && (
  <div className="error-message">
    <span>{getErrorMessage(error)}</span>
    <button onClick={retry}>Try Again</button>
  </div>
)}

---

Pattern 7: Fallback to Non-Streaming

If streaming fails repeatedly, fall back to standard request.

async function adaptiveStream(prompt: string): Promise<string> {
  try {
    // Try streaming first
    return await streamResponse(prompt);
  } catch (error) {
    console.log('Streaming failed, falling back to standard request');

    // Fallback to non-streaming
    const response = await fetch('/api/chat', {
      method: 'POST',
      body: JSON.stringify({ prompt, stream: false })
    });

    const data = await response.json();
    return data.content;
  }
}

---

Pattern 8: Error Logging & Monitoring

async function streamWithMonitoring(prompt: string) {
  const startTime = Date.now();

  try {
    await streamResponse(prompt);

    // Log success
    await logMetric({
      event: 'stream_success',
      duration: Date.now() - startTime,
      prompt
    });

  } catch (error) {
    const duration = Date.now() - startTime;

    // Log error with context
    await logError({
      event: 'stream_error',
      error: error.message,
      duration,
      prompt,
      userAgent: navigator.userAgent,
      timestamp: new Date().toISOString()
    });

    // Send to error tracking (Sentry, etc.)
    if (window.Sentry) {
      window.Sentry.captureException(error, {
        tags: {
          component: 'chat-stream',
          duration
        }
      });
    }

    throw error;
  }
}

---

Pattern 9: Circuit Breaker

Stop trying after repeated failures.

class CircuitBreaker {
  private failures = 0;
  private lastFailureTime = 0;
  private state: 'closed' | 'open' | 'half-open' = 'closed';

  async execute<T>(fn: () => Promise<T>): Promise<T> {
    // Circuit open: reject immediately
    if (this.state === 'open') {
      const timeSinceFailure = Date.now() - this.lastFailureTime;

      if (timeSinceFailure < 60000) { // 1 minute cooldown
        throw new Error('Circuit breaker open. Try again later.');
      }

      // Try half-open
      this.state = 'half-open';
    }

    try {
      const result = await fn();

      // Success: reset
      this.failures = 0;
      this.state = 'closed';

      return result;
    } catch (error) {
      this.failures++;
      this.lastFailureTime = Date.now();

      // Trip circuit after 3 failures
      if (this.failures >= 3) {
        this.state = 'open';
      }

      throw error;
    }
  }
}

const breaker = new CircuitBreaker();

// Usage
try {
  await breaker.execute(() => streamResponse(prompt));
} catch (error) {
  // Handle error or show cached response
}

---

Production Checklist

□ Retry logic with exponential backoff
□ Rate limit handling (Retry-After header)
□ Graceful degradation (show partial)
□ User-friendly error messages
□ Timeout enforcement
□ Error logging to monitoring service
□ Circuit breaker for repeated failures
□ Fallback to non-streaming
□ Checkpoint saving for long streams
□ Network status detection
□ Offline mode (show cached)
□ Cancel button always works

---

Testing Error Scenarios

Simulate Network Failure

// Test: Network interruption
if (process.env.NODE_ENV === 'development') {
  // Randomly fail 10% of requests
  if (Math.random() < 0.1) {
    throw new Error('Simulated network failure');
  }
}

Simulate Rate Limit

// Test: Rate limit response
return new Response('Too many requests', {
  status: 429,
  headers: {
    'Retry-After': '5' // 5 seconds
  }
});

Simulate Partial Stream

// Test: Stream fails halfway
controller.enqueue(encoder.encode('data: First half\n\n'));
await new Promise(resolve => setTimeout(resolve, 1000));
controller.error(new Error('Connection lost'));

---

Resources

• MDN: Error Handling
• Circuit Breaker Pattern
• Retry Best Practices

Server-Sent Events (SSE) Protocol

Deep dive into the SSE specification and implementation details for LLM streaming.

Protocol Basics

SSE is a simple HTTP-based protocol for server-to-client streaming.

Request:

GET /api/stream HTTP/1.1
Accept: text/event-stream
Cache-Control: no-cache

Response:

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

data: First message

data: Second message

data: {"type":"completion","text":"Token"}

Message Format

Data Field

The only required field. Can span multiple lines.

data: Simple message

data: Multi-line
data: message
data: here

data: {"json":"works too"}

Event Field

Custom event types (default is "message").

event: status
data: {"state":"processing"}

event: completion
data: {"content":"Hello"}

ID Field

Event identifier for reconnection.

id: 1
data: First event

id: 2
data: Second event

Client can reconnect with Last-Event-ID header:

GET /api/stream HTTP/1.1
Last-Event-ID: 2

Retry Field

Reconnection delay in milliseconds.

retry: 3000
data: Reconnect after 3 seconds if disconnected

Client Implementation

EventSource API (Browser)

const eventSource = new EventSource('/api/stream');

eventSource.onmessage = (event) => {
  console.log('Data:', event.data);
};

eventSource.addEventListener('status', (event) => {
  console.log('Status:', JSON.parse(event.data));
});

eventSource.onerror = (error) => {
  console.error('SSE error:', error);
  eventSource.close();
};

// Close connection
eventSource.close();

Limitations:

• No custom headers (can't send Authorization)
• No POST requests (GET only)
• Limited error handling

Fetch API (More Control)

const response = await fetch('/api/stream', {
  method: 'POST',
  headers: {
    'Content-Type': 'application/json',
    'Authorization': `Bearer ${token}`
  },
  body: JSON.stringify({ prompt: 'Hello' })
});

const reader = response.body!.getReader();
const decoder = new TextDecoder();

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

  const chunk = decoder.decode(value);
  console.log(chunk);
}

Advantages:

• Custom headers
• POST requests
• AbortController support
• Better error handling

Server Implementation

Node.js (Express)

import express from 'express';

app.get('/api/stream', (req, res) => {
  // Set SSE headers
  res.setHeader('Content-Type', 'text/event-stream');
  res.setHeader('Cache-Control', 'no-cache');
  res.setHeader('Connection', 'keep-alive');

  // Send initial message
  res.write('data: Connected\n\n');

  // Send periodic updates
  const interval = setInterval(() => {
    res.write(`data: ${Date.now()}\n\n`);
  }, 1000);

  // Cleanup on disconnect
  req.on('close', () => {
    clearInterval(interval);
    res.end();
  });
});

Next.js (App Router)

// app/api/stream/route.ts
export async function GET() {
  const encoder = new TextEncoder();

  const stream = new ReadableStream({
    async start(controller) {
      // Send messages
      controller.enqueue(encoder.encode('data: Hello\n\n'));
      controller.enqueue(encoder.encode('data: World\n\n'));

      // Close stream
      controller.close();
    }
  });

  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache',
      'Connection': 'keep-alive'
    }
  });
}

Edge Runtime (Vercel/Cloudflare)

export const runtime = 'edge';

export async function POST(req: Request) {
  const { prompt } = await req.json();

  const stream = new ReadableStream({
    async start(controller) {
      const encoder = new TextEncoder();

      // Stream from LLM
      for await (const chunk of llmStream(prompt)) {
        const message = `data: ${JSON.stringify({ content: chunk })}\n\n`;
        controller.enqueue(encoder.encode(message));
      }

      controller.close();
    }
  });

  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream',
      'Cache-Control': 'no-cache'
    }
  });
}

Reconnection Logic

SSE has built-in reconnection, but Fetch API requires manual implementation.

class SSEClient {
  private reconnectDelay = 1000;
  private maxReconnectDelay = 30000;
  private reconnectAttempts = 0;

  async connect(url: string, onMessage: (data: string) => void) {
    try {
      const response = await fetch(url);
      const reader = response.body!.getReader();
      const decoder = new TextDecoder();

      // Reset reconnect state on successful connection
      this.reconnectAttempts = 0;
      this.reconnectDelay = 1000;

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

        const chunk = decoder.decode(value);
        const lines = chunk.split('\n');

        for (const line of lines) {
          if (line.startsWith('data: ')) {
            onMessage(line.slice(6));
          }
        }
      }
    } catch (error) {
      // Exponential backoff
      this.reconnectAttempts++;
      this.reconnectDelay = Math.min(
        this.reconnectDelay * 2,
        this.maxReconnectDelay
      );

      console.log(`Reconnecting in ${this.reconnectDelay}ms...`);

      setTimeout(() => {
        this.connect(url, onMessage);
      }, this.reconnectDelay);
    }
  }
}

CORS Configuration

SSE requires proper CORS headers.

// Server-side
app.use((req, res, next) => {
  res.setHeader('Access-Control-Allow-Origin', '*');
  res.setHeader('Access-Control-Allow-Methods', 'GET, POST');
  res.setHeader('Access-Control-Allow-Headers', 'Content-Type');
  next();
});

Heartbeat Pattern

Send periodic pings to keep connection alive.

// Server
const heartbeat = setInterval(() => {
  res.write(': heartbeat\n\n'); // Comment, ignored by client
}, 15000);

req.on('close', () => {
  clearInterval(heartbeat);
});

Why: Some proxies close idle connections after 30 seconds.

Error Handling

Client-Side Errors

try {
  const response = await fetch('/api/stream');

  if (!response.ok) {
    throw new Error(`HTTP ${response.status}`);
  }

  if (!response.body) {
    throw new Error('No response body');
  }

  // Stream handling...
} catch (error) {
  if (error.name === 'AbortError') {
    console.log('Stream cancelled');
  } else if (error.message.includes('Failed to fetch')) {
    console.error('Network error');
  } else {
    console.error('Unknown error:', error);
  }
}

Server-Side Errors

// Send error as SSE event
async start(controller) {
  try {
    // Streaming logic...
  } catch (error) {
    const errorMessage = `event: error\ndata: ${JSON.stringify({
      message: error.message
    })}\n\n`;
    controller.enqueue(encoder.encode(errorMessage));
    controller.close();
  }
}

Performance Considerations

Backpressure

Slow clients can overwhelm server. Handle backpressure:

const stream = new ReadableStream({
  async start(controller) {
    for await (const chunk of dataSource) {
      // Wait if client is slow
      if (controller.desiredSize !== null && controller.desiredSize <= 0) {
        await new Promise(resolve => setTimeout(resolve, 100));
      }

      controller.enqueue(encoder.encode(`data: ${chunk}\n\n`));
    }
  }
});

Memory Management

Close connections when done:

// Client
useEffect(() => {
  const controller = new AbortController();

  fetch('/api/stream', { signal: controller.signal })
    .then(/* ... */);

  return () => {
    controller.abort();
  };
}, []);

Security

Rate Limiting

import rateLimit from 'express-rate-limit';

const limiter = rateLimit({
  windowMs: 60 * 1000, // 1 minute
  max: 10, // 10 requests per minute
  message: 'Too many requests'
});

app.get('/api/stream', limiter, (req, res) => {
  // Stream handler...
});

Authentication

// Server
const token = req.headers.authorization?.split(' ')[1];
if (!verifyToken(token)) {
  return res.status(401).send('Unauthorized');
}

Production Checklist

□ Content-Type: text/event-stream header set
□ Cache-Control: no-cache header set
□ CORS configured correctly
□ Heartbeat implemented (for proxies)
□ Reconnection logic with exponential backoff
□ Error events sent to client
□ Rate limiting enabled
□ Authentication required
□ Memory cleanup on disconnect
□ Timeout for long-running streams

Resources

• MDN: Server-Sent Events
• SSE Specification
• EventSource API

Vercel AI SDK Integration Patterns

Production patterns for using Vercel AI SDK to handle LLM streaming across providers.

Why Vercel AI SDK?

Unified API across OpenAI, Anthropic, Google, Cohere, Hugging Face:

• Same streaming interface
• Automatic retries
• Built-in token counting
• React hooks

Timeline:

• 2023: Released as ai package
• 2024: Became de facto standard for Next.js AI apps
• 2024+: Supports 15+ LLM providers

---

Installation

npm install ai @ai-sdk/openai @ai-sdk/anthropic

---

Pattern 1: Basic Streaming (useChat Hook)

'use client';

import { useChat } from 'ai/react';

export function ChatInterface() {
  const { messages, input, handleInputChange, handleSubmit, isLoading } = useChat({
    api: '/api/chat'
  });

  return (
    <div>
      <div className="messages">
        {messages.map((message) => (
          <div key={message.id} className={`message ${message.role}`}>
            {message.content}
          </div>
        ))}
      </div>

      <form onSubmit={handleSubmit}>
        <input
          value={input}
          onChange={handleInputChange}
          disabled={isLoading}
          placeholder="Type a message..."
        />
        <button type="submit" disabled={isLoading}>
          Send
        </button>
      </form>
    </div>
  );
}

Server Route (app/api/chat/route.ts):

import { openai } from '@ai-sdk/openai';
import { streamText } from 'ai';

export const runtime = 'edge';

export async function POST(req: Request) {
  const { messages } = await req.json();

  const result = await streamText({
    model: openai('gpt-4-turbo'),
    messages
  });

  return result.toAIStreamResponse();
}

What you get:

• ✅ Streaming automatically handled
• ✅ Message history managed
• ✅ Loading states
• ✅ Error handling
• ✅ Optimistic updates

---

Pattern 2: Streaming with Tools (Function Calling)

import { openai } from '@ai-sdk/openai';
import { streamText, tool } from 'ai';
import { z } from 'zod';

export async function POST(req: Request) {
  const { messages } = await req.json();

  const result = await streamText({
    model: openai('gpt-4-turbo'),
    messages,
    tools: {
      getWeather: tool({
        description: 'Get weather for a location',
        parameters: z.object({
          location: z.string().describe('City name')
        }),
        execute: async ({ location }) => {
          const weather = await fetchWeather(location);
          return weather;
        }
      }),
      createReminder: tool({
        description: 'Create a reminder',
        parameters: z.object({
          text: z.string(),
          time: z.string()
        }),
        execute: async ({ text, time }) => {
          await db.reminders.create({ text, time });
          return { success: true };
        }
      })
    }
  });

  return result.toAIStreamResponse();
}

Client (automatic tool execution):

const { messages } = useChat({
  api: '/api/chat',
  onToolCall: ({ toolCall }) => {
    console.log('Tool called:', toolCall.toolName, toolCall.args);
  }
});

---

Pattern 3: Multi-Provider Fallback

import { openai } from '@ai-sdk/openai';
import { anthropic } from '@ai-sdk/anthropic';
import { streamText } from 'ai';

export async function POST(req: Request) {
  const { messages } = await req.json();

  try {
    // Try OpenAI first
    const result = await streamText({
      model: openai('gpt-4-turbo'),
      messages
    });

    return result.toAIStreamResponse();
  } catch (error) {
    // Fallback to Anthropic
    console.log('OpenAI failed, falling back to Claude');

    const result = await streamText({
      model: anthropic('claude-3-sonnet-20240229'),
      messages
    });

    return result.toAIStreamResponse();
  }
}

---

Pattern 4: Custom useChat with Abort

import { useChat } from 'ai/react';

export function ChatWithAbort() {
  const {
    messages,
    input,
    handleInputChange,
    handleSubmit,
    isLoading,
    stop
  } = useChat({
    api: '/api/chat'
  });

  return (
    <div>
      {/* Messages */}
      {messages.map((m) => (
        <div key={m.id}>{m.content}</div>
      ))}

      {/* Form */}
      <form onSubmit={handleSubmit}>
        <input value={input} onChange={handleInputChange} />
        <button type="submit" disabled={isLoading}>
          Send
        </button>
        {isLoading && (
          <button type="button" onClick={stop}>
            Stop
          </button>
        )}
      </form>
    </div>
  );
}

---

Pattern 5: Streaming Object (Structured Output)

For structured data (not just text).

import { openai } from '@ai-sdk/openai';
import { streamObject } from 'ai';
import { z } from 'zod';

const recipeSchema = z.object({
  name: z.string(),
  ingredients: z.array(z.object({
    name: z.string(),
    amount: z.string()
  })),
  instructions: z.array(z.string())
});

export async function POST(req: Request) {
  const { prompt } = await req.json();

  const result = await streamObject({
    model: openai('gpt-4-turbo'),
    schema: recipeSchema,
    prompt: `Generate a recipe for: ${prompt}`
  });

  return result.toTextStreamResponse();
}

Client:

'use client';

import { experimental_useObject as useObject } from 'ai/react';

export function RecipeGenerator() {
  const { object, submit, isLoading } = useObject({
    api: '/api/generate-recipe',
    schema: recipeSchema
  });

  return (
    <div>
      <button onClick={() => submit('chocolate cake')}>
        Generate Recipe
      </button>

      {object && (
        <div>
          <h2>{object.name}</h2>
          <h3>Ingredients:</h3>
          <ul>
            {object.ingredients?.map((ing, i) => (
              <li key={i}>{ing.amount} {ing.name}</li>
            ))}
          </ul>
          <h3>Instructions:</h3>
          <ol>
            {object.instructions?.map((step, i) => (
              <li key={i}>{step}</li>
            ))}
          </ol>
        </div>
      )}
    </div>
  );
}

---

Pattern 6: Token Counting & Cost Tracking

import { openai } from '@ai-sdk/openai';
import { streamText } from 'ai';

export async function POST(req: Request) {
  const { messages } = await req.json();

  const result = await streamText({
    model: openai('gpt-4-turbo'),
    messages,
    onFinish: ({ usage }) => {
      console.log('Token usage:', usage);
      // { promptTokens: 50, completionTokens: 100, totalTokens: 150 }

      const cost = calculateCost(usage, 'gpt-4-turbo');
      await db.usage.create({ cost, tokens: usage.totalTokens });
    }
  });

  return result.toAIStreamResponse();
}

function calculateCost(usage: any, model: string): number {
  const pricing = {
    'gpt-4-turbo': { input: 0.00001, output: 0.00003 } // per token
  };

  const rates = pricing[model];
  return (
    usage.promptTokens * rates.input +
    usage.completionTokens * rates.output
  );
}

---

Pattern 7: Middleware (Logging, Auth)

import { openai } from '@ai-sdk/openai';
import { streamText } from 'ai';

export async function POST(req: Request) {
  // Authentication
  const session = await getSession(req);
  if (!session) {
    return new Response('Unauthorized', { status: 401 });
  }

  // Rate limiting
  const { allowed } = await rateLimit(session.user.id);
  if (!allowed) {
    return new Response('Rate limit exceeded', { status: 429 });
  }

  const { messages } = await req.json();

  // Log request
  await db.chatLog.create({
    userId: session.user.id,
    prompt: messages[messages.length - 1].content,
    timestamp: new Date()
  });

  const result = await streamText({
    model: openai('gpt-4-turbo'),
    messages,
    onFinish: async ({ text, usage }) => {
      // Log response
      await db.chatLog.update({
        response: text,
        tokens: usage.totalTokens
      });
    }
  });

  return result.toAIStreamResponse();
}

---

Pattern 8: Custom System Prompts

const result = await streamText({
  model: openai('gpt-4-turbo'),
  system: `You are a helpful assistant for a cooking app.
           Only provide recipes and cooking advice.
           Keep responses under 200 words.`,
  messages
});

---

Pattern 9: Temperature & Model Parameters

const result = await streamText({
  model: openai('gpt-4-turbo'),
  messages,
  temperature: 0.7,  // Creativity (0-2)
  maxTokens: 500,    // Response length limit
  topP: 0.9,         // Nucleus sampling
  frequencyPenalty: 0.5,  // Reduce repetition
  presencePenalty: 0.5    // Encourage new topics
});

---

Production Checklist

□ Rate limiting per user
□ Token usage tracking
□ Cost monitoring
□ Error logging
□ Authentication required
□ System prompt defined
□ Max tokens set
□ onFinish handler for analytics
□ Multi-provider fallback
□ Tool calling validated (if used)

---

Comparison: Vercel AI SDK vs Raw API

Use Vercel AI SDK when:

• Building with Next.js
• Need React hooks
• Want provider flexibility
• Prefer TypeScript safety

Use raw API when:

• Non-React framework
• Bundle size critical
• Need custom streaming logic
• Provider-specific features

---

Resources

• Vercel AI SDK Docs
• Examples
• API Reference