How to Cut Streaming LLM Latency to Sub-500ms with Next.js & Edge Workers

[User Browser]
      │
      │ (1) DNS / TLS + Request Ingress
      ▼
[Global Edge CDN (Cloudflare / Vercel Edge)]
      │
      ├─► (Cache Hit: Prompt Prefix Cache in Redis Edge) ──► Instant Prefill Response
      │
      ▼ (Cache Miss: Forward to Origin / Provider)
[Anthropic Claude 3.5 Sonnet / OpenAI GPT-4o Ingress]
      │
      │ (2) Streaming SSE Tokens Chunk-by-Chunk
      ▼
[TransformStream Pipeline (Sanitization + Markdown Token Parsing)]
      │
      ▼ (3) HTTP/2 or HTTP/3 Server-Sent Events
[Client React 19 UI: useActionState + RSC Stream Consumer]

import { NextRequest, NextResponse } from "next/server";
import { Anthropic } from "@anthropic-ai/sdk";

export const runtime = "edge"; // Enforce edge runtime execution (V8 isolate)
export const dynamic = "force-dynamic";

const anthropic = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY || "",
});

export async function POST(req: NextRequest) {
  try {
    const { messages, tone, systemContext } = await req.json();

    if (!messages || !Array.isArray(messages)) {
      return NextResponse.json(
        { error: "Invalid payload: messages array required." },
        { status: 400 }
      );
    }

    // 1. Establish an SSE readable stream with TransformStream
    const { readable, writable } = new TransformStream();
    const writer = writable.getWriter();
    const textEncoder = new TextEncoder();

    // 2. Initiate the upstream LLM streaming request asynchronously
    const stream = await anthropic.messages.stream({
      model: "claude-3-5-sonnet-20241022",
      max_tokens: 2048,
      temperature: 0.7,
      system: [
        {
          type: "text",
          text: `You are an expert technical advisor. Profile context: ${systemContext}`,
          // Prompt Caching breakpoint enabled for prompt reuse
          cache_control: { type: "ephemeral" },
        },
      ],
      messages,
    });

    // 3. Pipe incoming chunks to client without blocking
    (async () => {
      try {
        for await (const chunk of stream) {
          if (
            chunk.type === "content_block_delta" &&
            chunk.delta.type === "text_delta"
          ) {
            const rawToken = chunk.delta.text;
            const sseEvent = `data: ${JSON.stringify({ text: rawToken })}\n\n`;
            await writer.write(textEncoder.encode(sseEvent));
          }
        }
        // Signal stream termination
        await writer.write(textEncoder.encode("data: [DONE]\n\n"));
      } catch (streamError) {
        console.error("Stream pipe interrupted:", streamError);
        const errorEvent = `data: ${JSON.stringify({ error: "Stream interrupted" })}\n\n`;
        await writer.write(textEncoder.encode(errorEvent));
      } finally {
        await writer.close();
      }
    })();

    // 4. Return the response immediately with appropriate SSE streaming headers
    return new Response(readable, {
      status: 200,
      headers: {
        "Content-Type": "text/event-stream; charset=utf-8",
        "Cache-Control": "no-cache, no-transform",
        Connection: "keep-alive",
        "X-Accel-Buffering": "no", // Disables NGINX proxy buffering
      },
    });
  } catch (err: unknown) {
    const errorMessage =
      err instanceof Error ? err.message : "Internal Server Error";
    return NextResponse.json({ error: errorMessage }, { status: 500 });
  }
}

// Architecture of a Cache-Optimized System Prompt Construct
const systemPrompts = [
  {
    type: "text" as const,
    text: BASE_SYSTEM_PROMPT_CONSTANTS, // ~1500 tokens of rules, style guide, JSON schemas
    cache_control: { type: "ephemeral" }, // First cache barrier
  },
  {
    type: "text" as const,
    text: CURRENT_SESSION_VECTOR_CONTEXT, // User-specific dynamic memory
  },
];

"use client";

import { useState, useCallback, useRef } from "react";

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

export function useEdgeLLMStream() {
  const [content, setContent] = useState<string>("");
  const [isStreaming, setIsStreaming] = useState<boolean>(false);
  const bufferRef = useRef<string>("");
  const rafIdRef = useRef<number | null>(null);

  const startStream = useCallback(
    async (endpoint: string, payload: unknown, options?: StreamOptions) => {
      setIsStreaming(true);
      setContent("");
      bufferRef.current = "";

      try {
        const response = await fetch(endpoint, {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify(payload),
        });

        if (!response.ok || !response.body) {
          throw new Error(`Failed stream initiation: ${response.statusText}`);
        }

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

        // Micro-batch state updates using requestAnimationFrame
        const flushBuffer = () => {
          setContent(bufferRef.current);
          rafIdRef.current = null;
        };

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

          const chunk = decoder.decode(value, { stream: true });
          const lines = chunk.split("\n\n");

          for (const line of lines) {
            if (!line.startsWith("data: ")) continue;
            const dataStr = line.replace("data: ", "").trim();

            if (dataStr === "[DONE]") {
              streamDone = true;
              break;
            }

            try {
              const parsed = JSON.parse(dataStr);
              if (parsed.text) {
                bufferRef.current += parsed.text;
                options?.onToken?.(parsed.text);

                // Throttle React renders to 60fps monitor refresh
                if (rafIdRef.current === null) {
                  rafIdRef.current = requestAnimationFrame(flushBuffer);
                }
              }
            } catch {
              // Partial line fragmented across TCP packet; handle in production ring buffer
            }
          }
        }

        // Final flush
        setContent(bufferRef.current);
        options?.onComplete?.(bufferRef.current);
      } catch (err: unknown) {
        const error = err instanceof Error ? err : new Error(String(err));
        options?.onError?.(error);
      } finally {
        setIsStreaming(false);
        if (rafIdRef.current) cancelAnimationFrame(rafIdRef.current);
      }
    },
    []
  );

  return { content, isStreaming, startStream };
}