llm-streaming-response-handler

LLM 流式响应处理器

擅长构建生产级 LLM 响应流式界面，让响应感觉即时且灵敏。

适用场景

✅ 适用于：

• 带有打字动画的聊天界面
• 实时 AI 助手
• 带实时预览的代码生成
• 渐进式显示的文档摘要
• 任何用户期待 LLM 立即反馈的 UI

❌ 不适用于：

• 批量文档处理（没有用户实时查看）
• 不支持流式传输的 API
• 基于 WebSocket 的双向聊天（使用 Socket.IO）
• 简单的请求/响应（使用 fetch 即可）

快速决策树

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

---

技术选型

Server-Sent Events (SSE) - 推荐

LLM 流式传输为何优先选择 SSE 而不是 WebSockets：

• 简单：基于 HTTP，可与现有基础设施配合使用
• 自动重连：内置重连逻辑
• 防火墙友好：通过代理时比 WebSockets 更简单
• 非常适合单向传输：LLM 只会从服务端向客户端流式传输

时间线：

• 2015-2020：WebSockets 用于所有场景
• 2020：SSE 开始用于流式 API
• 2023+：SSE 成为 LLM 流式传输的标准（OpenAI、Anthropic）
• 2024：Vercel AI SDK 推广 SSE 模式

流式 API

提供商	流式传输方式	响应格式
OpenAI	SSE	data: {"choices":[{"delta":{"content":"token"}}]}
Anthropic	SSE	data: {"type":"content_block_delta","delta":{"text":"token"}}
Claude (API)	SSE	data: {"delta":{"text":"token"}}
Vercel AI SDK	SSE	在各提供商之间进行规范化

---

常见反模式

反模式 1：显示前先缓冲

新手想法：“收集所有 token，然后显示完整响应”

问题：这完全违背了流式传输的目的。

错误做法：

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

正确做法：

// ✅ 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
    }
  }
}

时间线：

• 2023 年前：许多应用会缓冲整个响应
• 2023 年起：逐 token 显示已成为预期做法

---

反模式 2：不取消流

问题：用户无法停止生成，浪费 token 和金钱。

表现：“停止”按钮不起作用或根本不存在。

正确做法：

// ✅ 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>
);

---

反模式 3：没有错误恢复

问题：流在响应中途失败，用户只能看到部分文本，却不知道发生了故障。

正确做法：

// ✅ 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>
)}

---

反模式 4：未关闭流导致内存泄漏

问题：流未清理，导致内存泄漏。

表现：多次请求后浏览器运行缓慢。

正确做法：

// ✅ 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]);

---

反模式 5：Token 之间没有输入指示器

问题：Token 生成较慢时，UI 会显得卡住。

正确做法：

// ✅ 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; }
}

---

实现模式

模式 1：基本 SSE 流处理器

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);
}

模式 2：用于流式传输的 React Hook

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>
  );
}

模式 3：服务端流式传输（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'
    }
  });
}

---

生产环境检查清单

□ 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

---

适用与不适用场景

场景	使用流式传输？
聊天界面	✅ 是
长篇内容生成	✅ 是
带预览的代码生成	✅ 是
短文本补全（<50 个单词）	❌ 否 - 使用常规 fetch
后台任务	❌ 否 - 使用任务队列
双向聊天	⚠️ 改用 WebSockets

---

技术对比

特性	SSE	WebSockets	Long Polling
复杂度	低	中	高
自动重连	✅	❌	❌
双向传输	❌	✅	❌
防火墙友好	✅	⚠️	✅
浏览器支持	✅ 所有现代浏览器	✅ 所有现代浏览器	✅ 通用
LLM API 支持	✅ 标准	❌ 少见	❌ 不使用

---

参考资料

• /references/sse-protocol.md - Server-Sent Events 规范详情
• /references/vercel-ai-sdk.md - Vercel AI SDK 集成模式
• /references/error-recovery.md - 流错误处理策略

脚本

• scripts/stream_tester.ts - 在本地测试 SSE 端点
• scripts/token_counter.ts - 在生成前估算成本

---

此技能适用于：大模型流式传输实现 | SSE 协议 | 实时 UI 更新 | 取消 | 错误恢复 | 逐个 Token 显示