Skip to content
React AI 应用架构：组件设计、状态管理与流式交互
概述
LLM 应用前端的核心问题是：模型响应不是一次性返回，而是持续到达的增量数据。React 应用需要把一段不断增长的文本（或结构化内容）渲染到页面上，同时支持用户中断、重新生成和错误恢复。这类需求无法直接套用传统请求-响应模型的写法，需要从流式数据的消费、状态模型的设计和组件边界的划分三个层面来组织代码。
基本概念
前端分层
AI 对话前端可以按职责分为三层：
text
┌───────────────────────────────────────────────┐
│ UI 层 │
│ MessageList / ChatInput / StreamingText │
├───────────────────────────────────────────────┤
│ 状态层 │
│ messages / status / error / usage │
├───────────────────────────────────────────────┤
│ 数据访问层 │
│ fetch / ReadableStream / SSE parser / Abort │
└───────────────────────────────────────────────┘- 数据访问层负责向 LLM API 发起请求、读取流式响应、解析 SSE 事件、处理中断信号。它不关心 UI。
- 状态层是 UI 与数据访问之间的接口。数据访问层产生增量事件（
delta），状态层把增量合并进消息数组，并维护请求状态。 - UI 层通过 props 或 context 读取状态，不直接调用 fetch，也不解析
ReadableStream。
分层的收益在于：UI 组件可以独立测试；把 fetch 换成 WebSocket 或 mock 数据时只需替换数据访问层；状态层可以从 useReducer 换成 Zustand，UI 不需要改动。
ReadableStream 与异步迭代
fetch 返回的 response.body 是 ReadableStream 实例。在支持异步迭代的环境中可以逐块读取响应内容。
javascript
const response = await fetch('/api/chat', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ messages }),
signal,
});
if (!response.ok) {
throw new Error(`HTTP ${response.status}`);
}
for await (const chunk of response.body) {
// chunk 是 Uint8Array
}chunk 是二进制数据，需要解码为文本：
javascript
const decoder = new TextDecoder();
for await (const chunk of response.body) {
const text = decoder.decode(chunk, { stream: true });
process(text);
}TextDecoder 的 stream: true 选项会在多字节字符被拆分到多个 chunk 时保留片段，等后续字节到达后继续解码。如果不使用这个选项，一个被拆开的 UTF-8 字符会产生乱码。
浏览器兼容性：所有现代浏览器都支持 ReadableStream 的异步迭代。如果遇到不支持的环境，可以改用 response.body.getReader() 配合 reader.read() 手动读取。
SSE 解析
SSE（Server-Sent Events）是一种基于文本的流式传输协议。LLM API 返回的典型 SSE 格式如下：
text
data: {"id":"chatcmpl-xxx","choices":[{"index":0,"delta":{"content":"你"},"finish_reason":null}]}
data: {"id":"chatcmpl-xxx","choices":[{"index":0,"delta":{"content":"好"},"finish_reason":null}]}
data: [DONE]每行以 data: 开头，后面是 JSON 字符串，行之间以空行分隔。OpenAI Chat Completions 的流式响应由多个 chat.completion.chunk 组成，choices[0].delta 是模型输出的增量，可能包含 content、tool_calls、function_call、refusal 等字段。
浏览器原生支持的 EventSource 无法设置 Authorization 请求头，因此许多 LLM 服务无法直接使用。更通用的做法是用 fetch 发起请求，再手动解析响应体。
下面是一个解析 SSE 的异步生成器：
javascript
async function* parseSSE(response) {
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) {
const trimmed = line.trim();
if (!trimmed.startsWith('data:')) {
// 忽略 event:、ping 等非 data 行
continue;
}
const data = trimmed.slice(5).trim();
if (data === '[DONE]') return;
yield data;
}
}
} finally {
reader.releaseLock();
}
}注意点：
buffer是解析器实例的局部变量。每次调用parseSSE都会创建独立作用域，多个并发流各自持有自己的 buffer，不会互相写入。- 非
data:行、空行、注释行会被忽略。部分服务会插入ping事件，过滤后不会影响业务数据。 - 如果服务在 HTTP 层返回了错误状态，应该在进入解析前处理，而不是把错误页面的 HTML 当作 SSE 解析。
response.ok检查是必需的。
调用方式：
javascript
const response = await fetch('/api/chat', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ messages }),
signal,
});
if (!response.ok) {
const error = await response.json().catch(() => null);
throw new Error(error?.error?.message || `HTTP ${response.status}`);
}
for await (const raw of parseSSE(response)) {
const event = JSON.parse(raw);
const delta = event.choices?.[0]?.delta?.content;
if (delta) {
applyDelta(delta);
}
}也可以使用社区维护的 parse-sse 库。它提供 parseServerSentEvents(response)，返回的流支持 for await...of、pipeTo、pipeThrough 和 tee，并允许自定义 fetch 请求头。
工作原理
消息模型设计
消息模型需要同时覆盖三种情况：
- 用户输入和模型输出的纯文本
- 图片、附件等多模态内容
- 工具调用、函数调用、引用等结构化数据
一种可行的结构是让 content 从字符串升级为内容块（content part）数组。每个 part 是一个可辨识联合：
typescript
type Role = 'system' | 'user' | 'assistant' | 'tool';
interface TextContentPart {
type: 'text';
text: string;
}
interface ImageContentPart {
type: 'image';
image: string; // base64 或 URL
mimeType?: string;
}
interface ToolCallContentPart {
type: 'tool-call';
toolCallId: string;
toolName: string;
args: Record<string, unknown>;
}
interface ToolResultContentPart {
type: 'tool-result';
toolCallId: string;
result: unknown;
}
interface CitationContentPart {
type: 'citation';
citationId: string;
title?: string;
url?: string;
}
type ContentPart =
| TextContentPart
| ImageContentPart
| ToolCallContentPart
| ToolResultContentPart
| CitationContentPart;消息本体：
typescript
interface Message {
id: string;
role: Role;
content: string | ContentPart[];
createdAt: number;
status: 'pending' | 'streaming' | 'done' | 'error';
error?: string;
meta?: {
model?: string;
usage?: {
promptTokens: number;
completionTokens: number;
};
finishReason?: string;
};
}content 保留 string 是为了兼容历史消息：只包含文本的消息仍然是 string；新产生的多模态消息使用 ContentPart[]。流式更新时，TextContentPart 可以原地追加文本；工具调用 part 则等 tool_calls 参数完全到达后一次性更新。
需要注意，流式 chunk 的 delta.role 不一定总是 assistant。工具调用场景下，role 可能是 tool。前端消息模型需要按 role 路由 delta，而不是假设所有增量都追加到同一条 assistant 消息。
组件划分与数据流
UI 组件可以按容器/展示划分：
text
ChatContainer
├── MessageList（展示组件）
│ ├── MessageItem
│ │ └── StreamingText
│ └── AutoScrollAnchor
├── ChatInput（展示组件）
└── StatusBar（展示组件）ChatContainer从状态层读取数据，把messages、status、回调函数传给子组件。MessageList只负责布局和滚动，不关心消息内容来自哪里。StreamingText只接收text字符串，并把光标移到末尾。ChatInput通过onSubmit上抛字符串，由容器层调用状态层的发送逻辑。
数据流方向：
text
用户输入
→ ChatInput.onSubmit(text)
→ ChatContainer.handleSubmit(text)
→ 数据访问层 fetch + parseSSE
→ 状态层 dispatch(STREAM_DELTA)
→ ChatContainer 从 context 读取最新状态
→ MessageList 重新渲染展示组件不应该直接接收流式增量。如果将 delta 传入 MessageItem，每次增量都会触发从容器到叶子组件的全链路更新，而且子组件需要知道何时追加、何时替换，流式逻辑会散布在多个组件中。让展示组件接收完整的 Message 对象，流式合并逻辑集中在状态层，渲染层可以保持简单。
消息状态与请求状态的管理
消息状态和请求状态应当分开。消息状态描述对话内容，请求状态描述一次生成过程。
typescript
interface ChatState {
messages: Message[];
requestId: string | null;
status: 'idle' | 'loading' | 'streaming' | 'done' | 'error';
error: string | null;
}状态定义：
idle：没有进行中的请求。loading：请求已发出，尚未收到第一个数据块。streaming：已经收到第一个数据块，正在累积内容。done：正常结束或用户中断。error：请求失败。
requestId 用于区分不同轮次的请求，避免多个并发请求互相覆盖。
useReducer 与 Context
使用 useReducer 管理消息状态，通过 Context 提供给组件树。
typescript
type ChatAction =
| { type: 'ADD_MESSAGE'; message: Message }
| { type: 'REQUEST_START'; requestId: string }
| { type: 'STREAM_START'; messageId: string }
| { type: 'STREAM_DELTA'; messageId: string; delta: string }
| { type: 'STREAM_DONE'; messageId: string; finishReason?: string; usage?: Usage }
| { type: 'STREAM_ERROR'; messageId: string; error: string }
| { type: 'RESET' };Reducer：
typescript
function chatReducer(state: ChatState, action: ChatAction): ChatState {
switch (action.type) {
case 'ADD_MESSAGE':
return { ...state, messages: [...state.messages, action.message] };
case 'REQUEST_START':
return { ...state, requestId: action.requestId, status: 'loading' };
case 'STREAM_START':
return {
...state,
status: 'streaming',
messages: state.messages.map((message) =>
message.id === action.messageId
? { ...message, status: 'streaming' }
: message
),
};
case 'STREAM_DELTA':
return {
...state,
messages: state.messages.map((message) => {
if (message.id !== action.messageId) return message;
if (typeof message.content === 'string') {
return { ...message, content: message.content + action.delta };
}
// ContentPart 数组：追加到最后一段文本 part
const partIndex = message.content.length - 1;
const lastPart = message.content[partIndex];
if (lastPart?.type === 'text') {
const nextContent = [...message.content];
nextContent[partIndex] = {
...lastPart,
text: lastPart.text + action.delta,
};
return { ...message, content: nextContent };
}
return message;
}),
};
case 'STREAM_DONE':
return {
...state,
status: 'done',
messages: state.messages.map((message) =>
message.id === action.messageId
? {
...message,
status: 'done',
meta: {
...message.meta,
finishReason: action.finishReason,
usage: action.usage,
},
}
: message
),
};
case 'STREAM_ERROR':
return {
...state,
status: 'error',
error: action.error,
messages: state.messages.map((message) =>
message.id === action.messageId
? { ...message, status: 'error', error: action.error }
: message
),
};
default:
return state;
}
}REQUEST_START 与 STREAM_START 分开，是为了让 loading 状态真实可达。请求发出后先进入 loading，收到第一个 SSE 事件时才切换为 streaming。
Context：
typescript
const ChatContext = createContext<{
state: ChatState;
dispatch: React.Dispatch<ChatAction>;
} | null>(null);
export function ChatProvider({ children }: { children: React.ReactNode }) {
const [state, dispatch] = useReducer(chatReducer, initialState);
return (
<ChatContext.Provider value={{ state, dispatch }}>
{children}
</ChatContext.Provider>
);
}
export function useChatContext() {
const ctx = useContext(ChatContext);
if (!ctx) throw new Error('useChatContext 必须在 ChatProvider 内使用');
return ctx;
}示例
流式交互的 React 实现
将流式读取放在容器层的异步函数中。每收到一个增量就 dispatch(STREAM_DELTA)。
typescript
async function handleSubmit(input: string) {
const userMessage: Message = {
id: crypto.randomUUID(),
role: 'user',
content: input,
createdAt: Date.now(),
status: 'done',
};
const assistantMessage: Message = {
id: crypto.randomUUID(),
role: 'assistant',
content: '',
createdAt: Date.now(),
status: 'pending',
};
const previousMessages = stateRef.current.messages;
dispatch({ type: 'ADD_MESSAGE', message: userMessage });
dispatch({ type: 'ADD_MESSAGE', message: assistantMessage });
dispatch({ type: 'REQUEST_START', requestId: assistantMessage.id });
const controller = new AbortController();
abortRef.current = controller;
try {
const requestMessages = [...previousMessages, userMessage];
const response = await fetch('/api/chat', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ messages: requestMessages }),
signal: controller.signal,
});
if (!response.ok) {
const errBody = await response.json().catch(() => null);
throw new Error(errBody?.error?.message || `HTTP ${response.status}`);
}
let receivedFirstChunk = false;
for await (const raw of parseSSE(response)) {
if (!receivedFirstChunk) {
dispatch({ type: 'STREAM_START', messageId: assistantMessage.id });
receivedFirstChunk = true;
}
const event = JSON.parse(raw);
const delta = event.choices?.[0]?.delta?.content;
if (delta) {
dispatch({
type: 'STREAM_DELTA',
messageId: assistantMessage.id,
delta,
});
}
if (event.choices?.[0]?.finish_reason) {
dispatch({
type: 'STREAM_DONE',
messageId: assistantMessage.id,
finishReason: event.choices[0].finish_reason,
usage: event.usage,
});
}
}
dispatch({ type: 'STREAM_DONE', messageId: assistantMessage.id });
} catch (error) {
if (controller.signal.aborted) {
dispatch({
type: 'STREAM_DONE',
messageId: assistantMessage.id,
finishReason: 'aborted',
});
} else {
dispatch({
type: 'STREAM_ERROR',
messageId: assistantMessage.id,
error: String(error),
});
}
}
}两个容易被忽略的设计：
stateRef的存在。handleSubmit是异步函数，闭包内捕获的state是发起请求那一刻的快照。流式读取过程中，state已经更新，闭包内的state不会变。因此需要通过 ref 始终保存最新的消息列表：
typescript
const stateRef = useRef(state);
useEffect(() => {
stateRef.current = state;
}, [state]);- 并发流的隔离。
parseSSE的 buffer 是局部的，不同请求之间天然隔离。状态更新以messageId为索引，只修改对应的消息。abortRef保存最新一次的AbortController：发送新请求时，如果上一次请求仍在进行，会先被中断，避免两个流同时写入同一个消息列表。
如果业务上需要允许并发请求，就不能在发送新请求时直接中断旧请求，而应该按 requestId 收集 AbortController，由上层决定何时取消。
API
上面的逻辑可以抽取成 hooks，供不同组件复用。
useAbort
useAbort 管理当前激活的请求：
typescript
function useAbort() {
const abortRef = useRef<AbortController | null>(null);
const abort = useCallback(() => {
abortRef.current?.abort();
}, []);
const createSignal = useCallback(() => {
abortRef.current?.abort();
const controller = new AbortController();
abortRef.current = controller;
return controller.signal;
}, []);
useEffect(() => {
return () => abortRef.current?.abort();
}, []);
return { abort, createSignal };
}createSignal 会先中断上一个请求，因此一个组件内同一时刻最多只有一个激活的流式请求。
useChat
useChat 管理消息数组、发送、停止和重新生成：
typescript
function useChat(options?: {
api?: string;
initialMessages?: Message[];
onFinish?: (message: Message) => void;
onError?: (error: Error) => void;
}) {
const { api = '/api/chat', initialMessages = [] } = options ?? {};
const [messages, setMessages] = useState<Message[]>(initialMessages);
const [status, setStatus] = useState<ChatState['status']>('idle');
const [error, setError] = useState<Error | null>(null);
const { abort, createSignal } = useAbort();
const messagesRef = useRef(messages);
useEffect(() => {
messagesRef.current = messages;
}, [messages]);
const sendRequest = useCallback(
async (requestMessages: Message[], assistantId: string) => {
const signal = createSignal();
try {
const response = await fetch(api, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ messages: requestMessages }),
signal,
});
if (!response.ok) {
const body = await response.json().catch(() => null);
throw new Error(body?.error?.message || `HTTP ${response.status}`);
}
let started = false;
for await (const raw of parseSSE(response)) {
if (!started) {
setStatus('streaming');
setMessages((prev) =>
prev.map((m) =>
m.id === assistantId ? { ...m, status: 'streaming' } : m
)
);
started = true;
}
const event = JSON.parse(raw);
const delta = event.choices?.[0]?.delta?.content;
if (delta) {
setMessages((prev) =>
prev.map((m) => {
if (m.id !== assistantId) return m;
if (typeof m.content !== 'string') return m;
return { ...m, content: m.content + delta };
})
);
}
}
setMessages((prev) =>
prev.map((m) =>
m.id === assistantId ? { ...m, status: 'done' } : m
)
);
setStatus('done');
} catch (err) {
if (signal.aborted) {
setMessages((prev) =>
prev.map((m) =>
m.id === assistantId ? { ...m, status: 'done' } : m
)
);
setStatus('done');
} else {
setError(err as Error);
setStatus('error');
setMessages((prev) =>
prev.map((m) =>
m.id === assistantId ? { ...m, status: 'error', error: String(err) } : m
)
);
}
}
},
[api, createSignal]
);
const sendMessage = useCallback(
async (input: string) => {
const userMessage: Message = {
id: crypto.randomUUID(),
role: 'user',
content: input,
createdAt: Date.now(),
status: 'done',
};
const assistantMessage: Message = {
id: crypto.randomUUID(),
role: 'assistant',
content: '',
createdAt: Date.now(),
status: 'pending',
};
const previousMessages = messagesRef.current;
setMessages([...previousMessages, userMessage, assistantMessage]);
setStatus('loading');
setError(null);
await sendRequest([...previousMessages, userMessage], assistantMessage.id);
},
[sendRequest]
);
const stop = useCallback(() => abort(), [abort]);
const reload = useCallback(async () => {
const lastUserIndex = messagesRef.current
.map((m) => m.role)
.lastIndexOf('user');
if (lastUserIndex === -1) return;
const previousMessages = messagesRef.current.slice(0, lastUserIndex + 1);
const assistantMessage: Message = {
id: crypto.randomUUID(),
role: 'assistant',
content: '',
createdAt: Date.now(),
status: 'pending',
};
setMessages([...previousMessages, assistantMessage]);
setStatus('loading');
setError(null);
await sendRequest(previousMessages, assistantMessage.id);
}, [sendRequest]);
return { messages, sendMessage, stop, reload, status, error };
}reload 的语义是重发最后一轮请求。它找到最后一条 user 消息，移除该消息之后的所有 assistant 消息，再创建一个新的空 assistant 消息，并把请求体设为 previousMessages。用户消息不会重复追加。
useCompletion
useCompletion 面向补全场景，不维护对话历史：
typescript
function useCompletion(options?: { api?: string }) {
const { api = '/api/completion' } = options ?? {};
const [completion, setCompletion] = useState('');
const [status, setStatus] = useState('idle');
const [error, setError] = useState<Error | null>(null);
const { abort, createSignal } = useAbort();
const complete = useCallback(
async (prompt: string) => {
setStatus('loading');
setCompletion('');
setError(null);
const signal = createSignal();
try {
const response = await fetch(api, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ prompt }),
signal,
});
if (!response.ok) {
throw new Error(`HTTP ${response.status}`);
}
let started = false;
let text = '';
for await (const raw of parseSSE(response)) {
if (!started) {
setStatus('streaming');
started = true;
}
const event = JSON.parse(raw);
const delta =
event.choices?.[0]?.text ?? event.choices?.[0]?.delta?.content;
if (delta) {
text += delta;
setCompletion(text);
}
}
setStatus('done');
} catch (err) {
if (signal.aborted) {
setStatus('done');
} else {
setError(err as Error);
setStatus('error');
}
}
},
[api, createSignal]
);
return { completion, complete, status, error, stop: abort };
}Vercel AI SDK 的 useChat 和 useCompletion 提供了类似但更完整的封装，内部处理了消息状态、SSE 解析、请求取消和错误回调，并支持与 Next.js App Router 的 Route Handler 配合。
注意点
中断、重试与错误处理
中断依赖 AbortController。将 signal 传入 fetch，调用 abort() 后：
- fetch 的 Promise 被拒绝，抛出
AbortError。 - 如果流正在读取，
reader.read()也会被拒绝。 - 读取循环应检查
signal.aborted，主动抛出signal.reason。
错误处理需要区分以下类型：
- HTTP 错误：状态码非 2xx。处理方式是在进入流读取前读取响应体，抛出包含状态码和错误信息的异常。
- 网络错误：fetch 拒绝，通常是
TypeError，例如断网、CORS、DNS 失败。 - SSE 解析错误：
JSON.parse遇到非法数据。 - 中断错误：
AbortError，不是真正的错误，UI 应恢复到完成状态。
重试的语义是重新发送最后一轮请求。reload 的流程是：移除正在流式更新的 assistant 消息，然后重新调用 sendMessage。用户消息保留在消息列表中，避免重复输入。
如果上一次请求已经成功写入消息数组，重试前需要移除对应的 assistant 消息，否则会出现两条 assistant 回复。
渲染节流与虚拟列表
每收到一个 chunk 就 setMessages，会让 React 在短时间内执行大量渲染。可以控制渲染频率来降低开销。
一个带尾随触发的 useThrottledValue 实现：
typescript
function useThrottledValue<T>(value: T, delayMs = 50): T {
const [throttled, setThrottled] = useState(value);
const latestValue = useRef(value);
const lastTime = useRef(0);
const timer = useRef<ReturnType<typeof setTimeout> | null>(null);
useEffect(() => {
latestValue.current = value;
}, [value]);
useEffect(() => {
const tick = () => {
timer.current = null;
lastTime.current = Date.now();
setThrottled(latestValue.current);
};
const remaining = delayMs - (Date.now() - lastTime.current);
if (remaining <= 0) {
tick();
} else if (!timer.current) {
timer.current = setTimeout(tick, remaining);
}
}, [value, delayMs]);
useEffect(() => {
return () => {
if (timer.current) clearTimeout(timer.current);
};
}, []);
return throttled;
}这个实现会合并 delayMs 内的多次更新，并始终以最新值作为最终输出。它适用于流式文本：中间帧可以跳过，最后一帧必须显示。
也可以使用 startTransition 标记流式更新为低优先级，避免阻塞用户输入：
typescript
import { startTransition } from 'react';
// 收到 delta 时
startTransition(() => {
dispatch({ type: 'STREAM_DELTA', messageId, delta });
});startTransition 只是给 React 一个提示：这个更新可以被中断。它会跳过不必要的中间渲染，不会减少状态更新本身的次数。如果 delta 频率很高，仍然建议配合节流。
虚拟列表适用于消息数量很大的对话。react-window 的 FixedSizeList 是常见实现：
tsx
import { FixedSizeList } from 'react-window';
function MessageList({ messages }: { messages: Message[] }) {
return (
<FixedSizeList
height={600}
itemCount={messages.length}
itemSize={80}
itemData={messages}
>
{({ index, style, data }) => (
<div style={style}>
<MessageItem message={data[index]} />
</div>
)}
</FixedSizeList>
);
}流式场景中，如果列表底部持续有新内容，需要自动滚动跟随：
typescript
function useAutoScroll(messages: Message[], isStreaming: boolean) {
const listRef = useRef<HTMLDivElement>(null);
const stickToBottom = useRef(true);
const onScroll = () => {
const el = listRef.current;
if (!el) return;
stickToBottom.current =
el.scrollHeight - el.scrollTop - el.clientHeight < 40;
};
useEffect(() => {
const el = listRef.current;
if (el && stickToBottom.current && isStreaming) {
el.scrollTop = el.scrollHeight;
}
}, [messages, isStreaming]);
return { listRef, onScroll };
}如果用户向上滚动查看历史消息，stickToBottom 会变为 false，流式更新时列表不再自动滚动。用户滚回底部后恢复跟随。
应用
状态库集成：Zustand、Redux Toolkit 与 React Query
useReducer + Context 是内置方案，适合中小型应用。当状态复杂度增加，或需要跨多个页面共享会话状态时，可以使用外部状态库。
Zustand
Zustand 的 store 通过不可变更新来改变状态：
typescript
import { create } from 'zustand';
interface ChatStore {
messages: Message[];
status: ChatStatus;
error: string | null;
addMessage: (message: Message) => void;
updateMessage: (id: string, patch: Partial<Message>) => void;
setStatus: (status: ChatStatus) => void;
}
const useChatStore = create<ChatStore>((set) => ({
messages: [],
status: 'idle',
error: null,
addMessage: (message) =>
set((state) => ({ messages: [...state.messages, message] })),
updateMessage: (id, patch) =>
set((state) => ({
messages: state.messages.map((m) =>
m.id === id ? { ...m, ...patch } : m
),
})),
setStatus: (status) => set({ status }),
}));在异步逻辑中读取 store 的最新值：
typescript
const messages = useChatStore.getState().messages;流式循环中调用 action：
typescript
useChatStore.getState().updateMessage(assistantId, { content: nextContent });Redux Toolkit
Redux Toolkit 的 createSlice 可以简化 action 和 reducer 的定义：
typescript
import { createSlice } from '@reduxjs/toolkit';
const chatSlice = createSlice({
name: 'chat',
initialState,
reducers: {
messageAdded(state, action) {
state.messages.push(action.payload);
},
deltaReceived(state, action) {
const message = state.messages.find(
(m) => m.id === action.payload.messageId
);
if (message && typeof message.content === 'string') {
message.content += action.payload.delta;
}
},
},
});
export const { messageAdded, deltaReceived } = chatSlice.actions;Redux Toolkit 基于 Immer，reducer 内部可以直接修改嵌套状态。
React Query / SWR 的边界
React Query 和 SWR 面向服务端状态，擅长缓存、重试和重新验证。AI 对话状态有两个来源：服务端持久化数据（例如历史会话列表）和客户端临时数据（正在生成的流式文本）。前者适合放进 React Query：
typescript
const { data: sessions } = useQuery({
queryKey: ['sessions'],
queryFn: fetchSessions,
});后者不应该放进 React Query 的缓存中。流式消息是持续变化的临时状态，用 query 缓存会导致频繁失效和重新验证，而且 Query 的请求状态无法表达“正在流式接收”这个中间状态。
流式请求可以封装成 mutation。useMutation 的请求生命周期与流式过程需要单独设计：
typescript
const mutation = useMutation({
mutationFn: async (input: string) => {
let finalText = '';
for await (const event of streamChat({ messages: buildMessages(input) })) {
const delta = event.choices?.[0]?.delta?.content;
if (delta) {
finalText += delta;
useChatStore.getState().updateMessage(assistantId, {
content: finalText,
});
}
}
return finalText;
},
onSuccess: () => {
queryClient.invalidateQueries({ queryKey: ['sessions', sessionId] });
},
});这样，React Query 处理请求生命周期，Zustand 处理流式状态。职责需要明确：React Query 不缓存流式中间结果，流式状态始终由本地 store 持有。
React 18 还提供了 useSyncExternalStore，用于让外部 store 与 React 并发渲染机制协作。状态库内部通过它来订阅 store 变化，组件无需手动 forceUpdate。
生态与趋势：Vercel AI SDK、Next.js App Router 与流式 UI
Vercel AI SDK 提供 useChat、useCompletion 等客户端 hooks，也提供服务端工具把模型返回值包装成 SSE 响应。
typescript
import { useChat } from '@ai-sdk/react';
const { messages, input, handleInputChange, handleSubmit, isLoading, stop } =
useChat({ api: '/api/chat' });在 Next.js App Router 的 Route Handler 中，可以把模型流直接作为 SSE 返回：
typescript
// app/api/chat/route.ts
import { streamText } from 'ai';
import { openai } from '@ai-sdk/openai';
export async function POST(req: Request) {
const { messages } = await req.json();
const result = streamText({
model: openai('gpt-4o-mini'),
messages,
});
return result.toDataStreamResponse();
}toDataStreamResponse() 返回 SSE 格式的响应，客户端 useChat 可以直接消费。
Next.js App Router 还允许 React Server Components（RSC）在服务端异步渲染，并将渲染结果流式传输到客户端。这个机制与 LLM SSE 流的区别在于：
- SSE 流传输的是数据，客户端解析后自己更新组件。
- RSC 流传输的是服务端渲染好的 UI 片段，React 在客户端渐进式填充 Suspense 边界。
例如，一个 RSC 页面可以先输出布局和 loading 状态，再异步加载耗时内容：
tsx
export default async function Page() {
return (
<Suspense fallback={<p>加载中…</p>}>
<AssistantInfo />
</Suspense>
);
}
async function AssistantInfo() {
const info = await fetchAssistantInfo();
return <p>{info.name}</p>;
}React 会先把 <p>加载中…</p> 发送到客户端，等 fetchAssistantInfo() 完成后，再把最终内容以流式更新的方式传输过去。
React 19 的 use 可以直接在组件中读取 Promise，并结合 Suspense：
tsx
import { use } from 'react';
function CommentSection({
commentsPromise,
}: {
commentsPromise: Promise<Comment[]>;
}) {
const comments = use(commentsPromise);
return <CommentList comments={comments} />;
}use 的语义是读取 Promise：未完成时挂起，完成后重新渲染。它不能直接消费 ReadableStream，因此逐 token 的 LLM 输出仍然需要在客户端读取 ReadableStream。RSC 流式 UI 更适合“加载完成后一次性展示”的异步片段，而不是每个 token 的增量文本。
streamText 返回的结果不仅包含文本，也可以包含工具调用。对于工具调用，前端需要按 tool_call_id 将用户侧的工具结果回传给模型，而不是简单地把所有增量追加到同一段文本中。这与前面消息模型中预留 ToolCallContentPart 和 ToolResultContentPart 的原因一致。
流式 UI 是当前生态中的一个探索方向：服务端不仅返回文本，还能返回组件树。该领域的 API 仍在快速演化，架构上值得保留的是数据访问层与状态层的独立性，以便将来替换具体的流式协议。