React 与 ChatGPT API 集成的流式响应工程化实践：从单次请求到 Server-Sent Events 的架构演进 - 简希博客

问题背景

ChatGPT API 的流式响应（stream: true）返回的是 Server-Sent Events 格式的 text/event-stream，而非标准 JSON。直接用 fetch().then(res => res.json()) 会当场崩溃——浏览器没有内置 SSE 流解析器。

反模式：把流当一次性请求

// ❌ 流式接口被当成普通 JSON 消费，解析失败
async function askChatGPT(prompt: string) {
  const res = await fetch('https://api.openai.com/v1/chat/completions', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Authorization': `Bearer ${API_KEY}`,
    },
    body: JSON.stringify({
      model: 'gpt-4',
      messages: [{ role: 'user', content: prompt }],
      stream: true,
    }),
  });
  return res.json(); // 💥 SyntaxError: Unexpected token 'd', "data: {"id"..." is not valid JSON
}

核心矛盾：服务端返回的是 SSE 字节流 data: {...}\n\n，而 res.json() 期望的是完整 JSON body。

第一步：手动解析 SSE 流

ReadableStream 是浏览器原生的流处理接口。拆开 response.body，逐行消费。

async function* streamChatGPT(prompt: string): AsyncGenerator<string> {
  const res = await fetch('https://api.openai.com/v1/chat/completions', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      'Authorization': `Bearer ${API_KEY}`,
    },
    body: JSON.stringify({
      model: 'gpt-4',
      messages: [{ role: 'user', content: prompt }],
      stream: true,
    }),
  });

  const reader = res.body!.getReader();
  const decoder = new TextDecoder();
  let buffer = '';

  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: ')) continue;
      const data = line.slice(6);
      if (data === '[DONE]') return;

      try {
        const parsed = JSON.parse(data);
        const delta = parsed.choices?.[0]?.delta?.content;
        if (delta) yield delta;
      } catch { /* 跳过非 JSON 行 */ }
    }
  }
}

关键细节：用 buffer 应对 TCP 分帧——一次 read() 得到的 chunk 可能在 data: 行中间断开。

第二步：封装自定义 Hook

把流解析逻辑收拢到 useChatStream，对外暴露 messages、send、isStreaming 三个原语。

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

interface Message {
  role: 'user' | 'assistant';
  content: string;
}

function useChatStream() {
  const [messages, setMessages] = useState<Message[]>([]);
  const [isStreaming, setIsStreaming] = useState(false);
  const abortRef = useRef<AbortController | null>(null);

  const send = useCallback(async (userInput: string) => {
    const userMsg: Message = { role: 'user', content: userInput };
    const assistantMsg: Message = { role: 'assistant', content: '' };
    setMessages(prev => [...prev, userMsg, assistantMsg]);
    setIsStreaming(true);

    abortRef.current = new AbortController();

    try {
      const res = await fetch('https://api.openai.com/v1/chat/completions', {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          'Authorization': `Bearer ${API_KEY}`,
        },
        body: JSON.stringify({
          model: 'gpt-4',
          messages: [...messages, userMsg].map(({ role, content }) => ({ role, content })),
          stream: true,
        }),
        signal: abortRef.current.signal,
      });

      const reader = res.body!.getReader();
      const decoder = new TextDecoder();
      let buffer = '';

      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: ')) continue;
          const data = line.slice(6);
          if (data === '[DONE]') break;
          try {
            const parsed = JSON.parse(data);
            const delta = parsed.choices?.[0]?.delta?.content;
            if (delta) {
              assistantMsg.content += delta;
              setMessages(prev => {
                const copy = [...prev];
                copy[copy.length - 1] = { ...assistantMsg };
                return copy;
              });
            }
          } catch { /* skip */ }
        }
      }
    } catch (err: unknown) {
      if (err instanceof DOMException && err.name === 'AbortError') {
        setMessages(prev => [...prev.slice(0, -1)]); // 移除半截的 assistant 消息
      }
    } finally {
      setIsStreaming(false);
    }
  }, [messages]);

  const abort = useCallback(() => abortRef.current?.abort(), []);

  return { messages, send, isStreaming, abort };
}

这里有一个容易被忽略的性能陷阱：用 prev => [...prev, ...] 每次都复制整个数组。消息列表长到几百条时开销可观。优化方向：

// ✅ 只更新最后一条消息的 content，避免全量拷贝
setMessages(prev =>
  prev.map((msg, i) =>
    i === prev.length - 1
      ? { ...msg, content: msg.content + delta }
      : msg
  )
);

第三步：竞态治理

用户快速连续提问时，前一个请求的响应还在流式返回就发起新请求，两个 setMessages 互相覆盖——典型的竞态问题。

// ❌ 没有竞态控制：两次 send 的 setMessages 交错执行
// send("什么是闭包？")
// send("用 Rust 实现") // assistantMsg 被两个请求同时修改

const send = useCallback(async (userInput: string) => {
  abortRef.current?.abort(); // ✅ 发起新请求前中止上一个
  // ...
}, []);

AbortController 一行解决：新请求触发时 abort() 旧的 fetch，signal 让 Reader 的 read() 抛出 AbortError，catch 块清理掉半截消息。

组件集成

function ChatBox() {
  const { messages, send, isStreaming, abort } = useChatStream();
  const [input, setInput] = useState('');

  return (
    <div className="chat-container">
      <div className="messages">
        {messages.map((msg, i) => (
          <div key={i} className={`bubble ${msg.role}`}>
            {msg.content}
          </div>
        ))}
      </div>
      <form onSubmit={e => { e.preventDefault(); send(input); setInput(''); }}>
        <input value={input} onChange={e => setInput(e.target.value)} />
        <button type="submit" disabled={isStreaming}>发送</button>
        <button type="button" onClick={abort}>停止</button>
      </form>
    </div>
  );
}

类型安全增强

把 OpenAI 的响应结构收拢到类型定义里，拒绝 any 裸奔：

type ChatGPTChunk = {
  id: string;
  object: 'chat.completion.chunk';
  created: number;
  model: string;
  choices: Array<{
    index: number;
    delta: { role?: string; content?: string };
    finish_reason: 'stop' | 'length' | null;
  }>;
};

// parse 处改用类型断言
const parsed: ChatGPTChunk = JSON.parse(data);

总结

React 中集成 ChatGPT 流式响应的核心路径：fetch + ReadableStream → SSE 手动解析 → useChatStream Hook → AbortController 竞态治理。这条路没有魔法，本质上是对 ReadableStreamDefaultReader 的一次工程化封装。理解了这层，无论是接入 Claude API 还是自建流式后端，套路完全复用。