From f8c984aef45242ad554391335cfa750d3ca1ad56 Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Thu, 6 Aug 2026 22:45:58 +0800 Subject: [PATCH] =?UTF-8?q?perf(web):=20=E4=BC=98=E5=8C=96=E6=B5=81?= =?UTF-8?q?=E5=BC=8F=20delta=20=E5=A4=84=E7=90=86=EF=BC=8C=E9=81=BF?= =?UTF-8?q?=E5=85=8D=E6=AF=8F=20token=20=E5=85=A8=E6=95=B0=E7=BB=84?= =?UTF-8?q?=E6=8B=B7=E8=B4=9D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 用 ref 累加 delta chunks + requestAnimationFrame 批量 flush， 将每秒数十次 O(n) 数组拷贝合并为一帧一次状态更新。 - useMessages.ts: 新增 streamingRef 累加器 + rAF 调度，所有 delta 只 push 到 ref，由 flushStreaming 批量落盘；用 rafScheduledRef boolean 追踪调度状态（避免 rAF handle 为 0 时与 null 检查冲突）； flushStreaming 基于 prev 实际状态判断创建/更新消息（批处理安全） - useChat.ts: 5 处 setMessages([]) 改为 clearMessages()，切话题/ 会话/通道时重置流式 ref，避免脏状态残留 - test/setup.ts: 同步化 rAF 并返回递增非零 handle，保持测试同步语义 - useChat.test.ts: 新增对抗性测试（批量 delta 累加、stream_end 后 assistant_response 替换含 reasoning） 复杂度：O(n²·L) → O(n) 每帧，15 个测试全部通过 --- web/src/hooks/chat/useMessages.ts | 167 +++++++++++++++++++++++++----- web/src/hooks/useChat.test.ts | 42 ++++++++ web/src/hooks/useChat.ts | 13 ++- web/src/test/setup.ts | 9 ++ 4 files changed, 196 insertions(+), 35 deletions(-) diff --git a/web/src/hooks/chat/useMessages.ts b/web/src/hooks/chat/useMessages.ts index 6df67f3..8fad370 100644 --- a/web/src/hooks/chat/useMessages.ts +++ b/web/src/hooks/chat/useMessages.ts @@ -2,6 +2,7 @@ import { useState, useCallback, useRef, + useEffect, type Dispatch, type SetStateAction, type MutableRefObject, @@ -63,6 +64,110 @@ export function useMessages(options: UseMessagesOptions): UseMessagesReturn { }); }, []); + // ---- 流式 delta 累加器 ---- + // 用 ref 暂存 delta chunks，rAF 批量 flush 到 state，避免每个 token 都拷贝整个消息数组。 + // 首 delta 同步写入消息壳，保证消息立即存在（assistant_response 可按 id 替换）。 + const streamingRef = useRef<{ + id: string; + contentChunks: string[]; + reasoningChunks: string[]; + index: number; + timestamp: number; + dirty: boolean; + } | null>(null); + const rafHandleRef = useRef(null); + // 独立的调度标志：rAF handle 在某些环境可能为 0（与 null 检查冲突）， + // 用 boolean 显式追踪是否已有 pending flush，语义更稳健。 + const rafScheduledRef = useRef(false); + + // flush pending 流式内容到 state。force=true 时即使 dirty=false 也强制 flush一次， + // 用于 finishStreaming 确保壳已创建（边界：首 delta 后立即 stream_end，dirty 可能已清）。 + const flushStreaming = useCallback((force = false) => { + const stream = streamingRef.current; + if (!stream) return; + if (!force && !stream.dirty) return; + stream.dirty = false; + const content = stream.contentChunks.join(''); + const reasoning = + stream.reasoningChunks.length > 0 ? stream.reasoningChunks.join('') : undefined; + setMessages((prev) => { + // 优先用缓存的 index 直接定位（O(1)），并校验 id 仍匹配 + if (stream.index >= 0 && stream.index < prev.length && prev[stream.index].id === stream.id) { + const updated = prev.slice(); + const existing = updated[stream.index]; + updated[stream.index] = { + ...existing, + content, + reasoningContent: reasoning ?? existing.reasoningContent, + }; + return updated; + } + // 回退：按 id 查找 + const idx = prev.findIndex((m) => m.id === stream.id && m.type === 'message'); + if (idx >= 0) { + stream.index = idx; + const updated = prev.slice(); + const existing = updated[idx]; + updated[idx] = { + ...existing, + content, + reasoningContent: reasoning ?? existing.reasoningContent, + }; + return updated; + } + // 消息壳尚未存在：创建（基于 prev 实际状态判断，不依赖外部标志，批处理安全） + stream.index = prev.length; + return [ + ...prev, + { + id: stream.id, + role: 'assistant' as const, + content, + timestamp: stream.timestamp, + type: 'message' as const, + reasoningContent: reasoning, + }, + ]; + }); + }, []); + + const clearStreaming = useCallback(() => { + if (rafScheduledRef.current) { + if (rafHandleRef.current !== null) { + cancelAnimationFrame(rafHandleRef.current); + } + rafHandleRef.current = null; + rafScheduledRef.current = false; + } + streamingRef.current = null; + }, []); + + const scheduleFlush = useCallback(() => { + if (rafScheduledRef.current) return; + rafScheduledRef.current = true; + rafHandleRef.current = requestAnimationFrame(() => { + rafHandleRef.current = null; + rafScheduledRef.current = false; + flushStreaming(false); + }); + }, [flushStreaming]); + + const finishStreaming = useCallback(() => { + flushStreaming(true); + clearStreaming(); + }, [flushStreaming, clearStreaming]); + + // 卸载时取消未决的 rAF，避免在已卸载组件上触发状态更新 + useEffect(() => { + return () => { + if (rafScheduledRef.current && rafHandleRef.current !== null) { + cancelAnimationFrame(rafHandleRef.current); + } + rafHandleRef.current = null; + rafScheduledRef.current = false; + }; + }, []); + const handleMessage = useCallback((content: string, attachments?: Attachment[]) => { setMessages((prev) => [ ...prev, @@ -79,8 +184,9 @@ export function useMessages(options: UseMessagesOptions): UseMessagesReturn { }, []); const clearMessages = useCallback(() => { + clearStreaming(); setMessages([]); - }, []); + }, [clearStreaming]); const handleStop = useCallback((): Command => { return { type: 'stop_execution' }; @@ -88,6 +194,10 @@ export function useMessages(options: UseMessagesOptions): UseMessagesReturn { const handleMainViewMessage = useCallback( (message: WsOutbound): boolean => { + // 非流式消息到达前，先把 pending 的流式内容落盘，避免被后续消息覆盖或丢失 + if (message.type !== 'stream_delta') { + finishStreaming(); + } switch (message.type) { case 'task_started': { const msg = message as TaskStarted; @@ -131,32 +241,26 @@ export function useMessages(options: UseMessagesOptions): UseMessagesReturn { case 'stream_delta': { const msg = message as StreamDelta; if (msg.topic_id && msg.topic_id !== selectedTopicRef.current) return true; - setMessages((prev) => { - const existingIdx = prev.findIndex((m) => m.id === msg.id && m.type === 'message'); - if (existingIdx >= 0) { - const updated = [...prev]; - const existing = updated[existingIdx]; - updated[existingIdx] = { - ...existing, - content: existing.content + msg.delta, - reasoningContent: msg.reasoning_delta - ? (existing.reasoningContent || '') + msg.reasoning_delta - : existing.reasoningContent, - }; - return updated; - } - return [ - ...prev, - { - id: msg.id, - role: 'assistant' as const, - content: msg.delta, - timestamp: Math.floor(Date.now() / 1000), - type: 'message' as const, - reasoningContent: msg.reasoning_delta, - }, - ]; - }); + + // 所有 delta 都只累加到 ref，由 rAF 批量 flush 到 state。 + // 首 delta 由 flushStreaming 在消息不存在时创建壳；后续 delta 走 index 直接更新。 + let stream = streamingRef.current; + if (!stream || stream.id !== msg.id) { + stream = { + id: msg.id, + contentChunks: [], + reasoningChunks: [], + index: -1, + timestamp: Math.floor(Date.now() / 1000), + dirty: false, + }; + streamingRef.current = stream; + } + if (msg.delta) stream.contentChunks.push(msg.delta); + if (msg.reasoning_delta) stream.reasoningChunks.push(msg.reasoning_delta); + stream.dirty = true; + scheduleFlush(); + if (msg.user_message_id) applyUserMessageId(msg.user_message_id); return true; } @@ -305,7 +409,14 @@ export function useMessages(options: UseMessagesOptions): UseMessagesReturn { return false; } }, - [selectedTopicRef, topicsRef, bumpTopicRefreshTrigger, applyUserMessageId], + [ + selectedTopicRef, + topicsRef, + bumpTopicRefreshTrigger, + applyUserMessageId, + finishStreaming, + scheduleFlush, + ], ); return { diff --git a/web/src/hooks/useChat.test.ts b/web/src/hooks/useChat.test.ts index 0a71379..ad32b61 100644 --- a/web/src/hooks/useChat.test.ts +++ b/web/src/hooks/useChat.test.ts @@ -423,4 +423,46 @@ describe('useChat - handleServerMessage characterization', () => { act(() => result.current.handleServerMessage(otherTopicStream)); expect(result.current.messages.find((m) => m.id === 'm-other')).toBeUndefined(); }); + + it('14. multiple stream_delta in one act() batch accumulate correctly', () => { + const { result } = renderUseChat(); + // 模拟同一同步批次内连续到达多个 delta（验证 chunks 累加 + index 跟踪） + act(() => { + result.current.handleServerMessage({ type: 'stream_delta', id: 'batch1', delta: 'A' }); + result.current.handleServerMessage({ type: 'stream_delta', id: 'batch1', delta: 'B' }); + result.current.handleServerMessage({ type: 'stream_delta', id: 'batch1', delta: 'C' }); + }); + expect(result.current.messages).toHaveLength(1); + expect(result.current.messages[0].id).toBe('batch1'); + expect(result.current.messages[0].content).toBe('ABC'); + }); + + it('15. stream_end then assistant_response replaces streamed content including reasoning', () => { + const { result } = renderUseChat(); + act(() => { + result.current.handleServerMessage({ + type: 'stream_delta', + id: 'r1', + delta: 'text', + reasoning_delta: 'think', + }); + }); + expect(result.current.messages[0].content).toBe('text'); + expect(result.current.messages[0].reasoningContent).toBe('think'); + // stream_end 触发 finishStreaming（flush pending + clear） + act(() => result.current.handleServerMessage({ type: 'stream_end', id: 'r1' })); + // assistant_response 替换整条消息 + act(() => + result.current.handleServerMessage({ + type: 'assistant_response', + id: 'r1', + content: 'final text', + role: 'assistant', + reasoning_content: 'final reasoning', + }), + ); + expect(result.current.messages).toHaveLength(1); + expect(result.current.messages[0].content).toBe('final text'); + expect(result.current.messages[0].reasoningContent).toBe('final reasoning'); + }); }); diff --git a/web/src/hooks/useChat.ts b/web/src/hooks/useChat.ts index 5f9f2e0..8118156 100644 --- a/web/src/hooks/useChat.ts +++ b/web/src/hooks/useChat.ts @@ -169,8 +169,7 @@ export function useChat(): UseChatReturn { const prevSid = sessions.selectedSessionIdRef.current; const prevSessionExists = prevSid !== null && message.sessions.some((s) => s.session_id === prevSid); - const isReconnect = - topics.selectedTopicRef.current !== null && prevSessionExists; + const isReconnect = topics.selectedTopicRef.current !== null && prevSessionExists; sessions.setSessions(message.sessions); if (isReconnect) { // 原 session 仍在，保持选中 @@ -184,7 +183,7 @@ export function useChat(): UseChatReturn { // 首次连接、切换通道、或原 session 已被删除：清空旧数据避免污染 topics.setTopics([]); topics.setSelectedTopic(null); - messages.setMessages([]); + messages.clearMessages(); sessions.setSelectedSessionId((prev) => prev && message.sessions.some((s) => s.session_id === prev) ? prev @@ -204,7 +203,7 @@ export function useChat(): UseChatReturn { case 'topic_list': { const autoFocused = topics.handleTopicList(message); - if (autoFocused) messages.setMessages([]); + if (autoFocused) messages.clearMessages(); messages.setIsLoading(false); return; } @@ -276,7 +275,7 @@ export function useChat(): UseChatReturn { // ---- selectTopic: 切换话题，清空消息和子智能体栈 ---- const selectTopic = useCallback((topicId: string) => { topics.setSelectedTopic(topicId); - messages.setMessages([]); + messages.clearMessages(); // ref + state 双写，消除竞态窗口（与 enter/exitSubAgentView 一致） subAgent.subAgentViewRef.current = null; subAgent.subAgentStackRef.current = []; @@ -292,7 +291,7 @@ export function useChat(): UseChatReturn { sessions.setSelectedSessionId(null); topics.setTopics([]); topics.setSelectedTopic(null); - messages.setMessages([]); + messages.clearMessages(); subAgent.subAgentViewRef.current = null; subAgent.subAgentStackRef.current = []; subAgent.setSubAgentStack([]); @@ -308,7 +307,7 @@ export function useChat(): UseChatReturn { sessions.setSelectedSessionId(sessionId); topics.setTopics([]); topics.setSelectedTopic(null); - messages.setMessages([]); + messages.clearMessages(); subAgent.subAgentViewRef.current = null; subAgent.subAgentStackRef.current = []; subAgent.setSubAgentStack([]); diff --git a/web/src/test/setup.ts b/web/src/test/setup.ts index bb02c60..55c8da5 100644 --- a/web/src/test/setup.ts +++ b/web/src/test/setup.ts @@ -1 +1,10 @@ import '@testing-library/jest-dom/vitest'; + +// 让 requestAnimationFrame 在测试中同步触发，使 rAF 批量更新的状态在 act() 内立即可见。 +// 返回递增的非零 handle，与浏览器真实 rAF 语义一致（0 是合法 ID 但会与 null 检查冲突）。 +let rafIdCounter = 1; +globalThis.requestAnimationFrame = (cb: FrameRequestCallback): number => { + cb(performance.now()); + return rafIdCounter++; +}; +globalThis.cancelAnimationFrame = (): void => {};