From b08eb009bb6c1203bc0db13143db4c7dd4b5be50 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E5=AD=90=E9=BB=98?= <925456043@qq.com> Date: Mon, 27 Jul 2026 19:41:11 +0800 Subject: [PATCH] =?UTF-8?q?perf:=20=E4=BC=98=E5=8C=96=20Agent=20=E6=B5=81?= =?UTF-8?q?=E5=BC=8F=E4=BC=9A=E8=AF=9D=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 合并流式通知与持久化写入，限制浏览器会话缓存 - 增量投影轮次事件并稳定关联异步工具任务 --- .../agentChatRuntimeManager.test.ts | 78 ++++++++- .../ai/agent-chat/agentChatRuntimeManager.ts | 114 ++++++++++++- .../ai/agents/components/AgentTryoutPanel.vue | 6 +- .../useAgentTryoutRawRounds.test.ts | 154 ++++++++++++++++++ .../composables/useAgentTryoutRawRounds.ts | 72 +++++++- .../composables/useAgentTryoutStream.ts | 22 ++- .../components/chat-timeline/ChatTimeline.vue | 42 +++-- .../chat-timeline/__tests__/builder.test.ts | 63 +++++++ .../src/components/chat-timeline/builder.ts | 34 +++- .../src/components/chat-timeline/types.ts | 1 + 10 files changed, 539 insertions(+), 47 deletions(-) diff --git a/easyflow-ui-admin/app/src/views/ai/agent-chat/agentChatRuntimeManager.test.ts b/easyflow-ui-admin/app/src/views/ai/agent-chat/agentChatRuntimeManager.test.ts index 066e43c9..6ae1d1e3 100644 --- a/easyflow-ui-admin/app/src/views/ai/agent-chat/agentChatRuntimeManager.test.ts +++ b/easyflow-ui-admin/app/src/views/ai/agent-chat/agentChatRuntimeManager.test.ts @@ -1,7 +1,7 @@ // @vitest-environment happy-dom import { createPinia, setActivePinia } from 'pinia'; -import { beforeEach, describe, expect, it, vi } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { useUserStore } from '@easyflow/stores'; @@ -23,6 +23,10 @@ describe('agentChatRuntimeManager', () => { vi.clearAllMocks(); }); + afterEach(() => { + vi.useRealTimers(); + }); + it('replaces draft image URLs and isolates snapshots by account', async () => { let callbacks: any; vi.mocked(sendAgentChat).mockImplementation((_data, options) => { @@ -95,4 +99,76 @@ describe('agentChatRuntimeManager', () => { userStore.setUserInfo(firstAccount); expect(agentChatRuntimeManager.getSnapshot('101')).toBeUndefined(); }); + + it('coalesces streaming notifications and persists terminal state immediately', async () => { + vi.useFakeTimers(); + let callbacks: any; + vi.mocked(sendAgentChat).mockImplementation((_data, options) => { + callbacks = options; + return Promise.resolve() as any; + }); + const account = { + avatar: '', + id: 'stream-user', + loginName: 'stream-user', + nickname: '流式用户', + tenantId: 'tenant-1', + }; + useUserStore().setUserInfo(account); + + await agentChatRuntimeManager.start({ + agentId: 'agent-1', + prompt: '流式测试', + sessionId: 'stream-session', + }); + const listener = vi.fn(); + const unsubscribe = agentChatRuntimeManager.subscribe(listener); + const storageSpy = vi.spyOn(sessionStorage, 'setItem'); + + for (const delta of ['A', 'B', 'C']) { + callbacks.onMessage({ + data: JSON.stringify({ + domain: 'LLM', + payload: { delta }, + type: 'MESSAGE', + }), + }); + } + + expect(listener).not.toHaveBeenCalled(); + expect(storageSpy).not.toHaveBeenCalled(); + expect( + JSON.stringify( + agentChatRuntimeManager.getSnapshot('stream-session')?.items, + ), + ).toContain('ABC'); + + await vi.advanceTimersByTimeAsync(50); + expect(listener).toHaveBeenCalledTimes(1); + expect(storageSpy).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(250); + expect(storageSpy).toHaveBeenCalledTimes(2); + + callbacks.onMessage({ + data: JSON.stringify({ + domain: 'LLM', + payload: { delta: 'D' }, + type: 'MESSAGE', + }), + }); + listener.mockClear(); + storageSpy.mockClear(); + callbacks.onFinished(); + + expect(listener).toHaveBeenCalledTimes(1); + expect(storageSpy).toHaveBeenCalledTimes(2); + expect(agentChatRuntimeManager.getSnapshot('stream-session')).toEqual( + expect.objectContaining({ completed: true, sending: false }), + ); + + unsubscribe(); + storageSpy.mockRestore(); + clearAgentChatBrowserCache(account); + }); }); diff --git a/easyflow-ui-admin/app/src/views/ai/agent-chat/agentChatRuntimeManager.ts b/easyflow-ui-admin/app/src/views/ai/agent-chat/agentChatRuntimeManager.ts index ad201358..9f6394f0 100644 --- a/easyflow-ui-admin/app/src/views/ai/agent-chat/agentChatRuntimeManager.ts +++ b/easyflow-ui-admin/app/src/views/ai/agent-chat/agentChatRuntimeManager.ts @@ -64,10 +64,15 @@ interface StartOptions { } const STORAGE_VERSION = 2; +const STREAM_NOTIFY_INTERVAL_MS = 50; +const STREAM_PERSIST_INTERVAL_MS = 300; +const MAX_RUNTIME_SESSIONS_PER_IDENTITY = 10; const sessions = new Map(); const listeners = new Set<() => void>(); const latestSessionIds = new Map(); +const persistTimers = new Map>(); +let notifyTimer: ReturnType | undefined; function clone(value: T): T { const serialized = JSON.stringify(value); @@ -108,6 +113,24 @@ function notify() { } } +function notifyNow() { + if (notifyTimer) { + clearTimeout(notifyTimer); + notifyTimer = undefined; + } + notify(); +} + +function scheduleNotify() { + if (notifyTimer || listeners.size === 0) { + return; + } + notifyTimer = setTimeout(() => { + notifyTimer = undefined; + notify(); + }, STREAM_NOTIFY_INTERVAL_MS); +} + function persistSession(state: RuntimeSessionState) { const storage = safeSessionStorage(); if (!storage) { @@ -118,7 +141,7 @@ function persistSession(state: RuntimeSessionState) { agentName: state.agentName, completed: state.completed, error: state.error, - items: clone(state.items), + items: state.items, prompt: state.prompt, roundId: state.roundId, sessionId: state.sessionId, @@ -136,6 +159,82 @@ function persistSession(state: RuntimeSessionState) { } } +function cancelPersistTimer(scopedSessionKey: string) { + const timer = persistTimers.get(scopedSessionKey); + if (!timer) { + return; + } + clearTimeout(timer); + persistTimers.delete(scopedSessionKey); +} + +function removeSessionCache( + scopedSessionKey: string, + state: RuntimeSessionState, +) { + cancelPersistTimer(scopedSessionKey); + sessions.delete(scopedSessionKey); + try { + safeSessionStorage()?.removeItem( + storageKey(state.identity, state.sessionId), + ); + } catch { + // 本地缓存清理失败不影响服务端会话。 + } +} + +function pruneRuntimeSessions(identity: string, currentSessionId: string) { + const scopedSessions = [...sessions.entries()] + .filter(([, session]) => session.identity === identity) + .sort((first, second) => second[1].updatedAt - first[1].updatedAt); + if (scopedSessions.length <= MAX_RUNTIME_SESSIONS_PER_IDENTITY) { + return; + } + const protectedKeys = new Set( + scopedSessions + .filter( + ([, session]) => + session.sending || session.sessionId === currentSessionId, + ) + .map(([key]) => key), + ); + let retainedCount = protectedKeys.size; + for (const [key, session] of scopedSessions) { + if (protectedKeys.has(key)) { + continue; + } + if (retainedCount < MAX_RUNTIME_SESSIONS_PER_IDENTITY) { + retainedCount += 1; + continue; + } + removeSessionCache(key, session); + } +} + +function touchState(state: RuntimeSessionState) { + state.updatedAt = Date.now(); + latestSessionIds.set(state.identity, state.sessionId); + sessions.set(sessionKey(state.identity, state.sessionId), state); +} + +function scheduleStateUpdate(state: RuntimeSessionState) { + touchState(state); + const scopedSessionKey = sessionKey(state.identity, state.sessionId); + if (!persistTimers.has(scopedSessionKey)) { + persistTimers.set( + scopedSessionKey, + setTimeout(() => { + persistTimers.delete(scopedSessionKey); + const current = sessions.get(scopedSessionKey); + if (current) { + persistSession(current); + } + }, STREAM_PERSIST_INTERVAL_MS), + ); + } + scheduleNotify(); +} + function restoreSession(identity: string, sessionId: string) { if (!identity) { return undefined; @@ -179,11 +278,11 @@ function restoreSession(identity: string, sessionId: string) { } function upsertState(state: RuntimeSessionState) { - state.updatedAt = Date.now(); - latestSessionIds.set(state.identity, state.sessionId); - sessions.set(sessionKey(state.identity, state.sessionId), state); + touchState(state); + cancelPersistTimer(sessionKey(state.identity, state.sessionId)); persistSession(state); - notify(); + pruneRuntimeSessions(state.identity, state.sessionId); + notifyNow(); } function runningSession(identity = identityScope()) { @@ -272,11 +371,12 @@ function replaceAcceptedImages( onAgentChatCacheClear((identity) => { for (const [key, session] of sessions) { if (session.identity === identity) { + cancelPersistTimer(key); sessions.delete(key); } } latestSessionIds.delete(identity); - notify(); + notifyNow(); }); export const agentChatRuntimeManager = { @@ -383,7 +483,7 @@ export const agentChatRuntimeManager = { void options.onInputAccepted?.(); } applyAgentSseEnvelope(current.items, envelope, { roundId }); - upsertState(current); + scheduleStateUpdate(current); }, }, ); diff --git a/easyflow-ui-admin/app/src/views/ai/agents/components/AgentTryoutPanel.vue b/easyflow-ui-admin/app/src/views/ai/agents/components/AgentTryoutPanel.vue index 9e86aad1..9d8e192c 100644 --- a/easyflow-ui-admin/app/src/views/ai/agents/components/AgentTryoutPanel.vue +++ b/easyflow-ui-admin/app/src/views/ai/agents/components/AgentTryoutPanel.vue @@ -10,7 +10,7 @@ import type { AgentToolBinding, } from '../types'; -import { computed, onMounted, ref, watch } from 'vue'; +import { computed, onBeforeUnmount, onMounted, ref, watch } from 'vue'; import { ChatTimeline } from '@easyflow/common-ui'; import { BrushCleaning } from '@easyflow/icons'; @@ -39,6 +39,7 @@ const emit = defineEmits<{ close: [] }>(); const { loading, clearDraftSession, + dispose, markToolApproving, markToolRejected, copyMessageText, @@ -80,6 +81,9 @@ async function activateComposer() { } onMounted(() => void activateComposer()); +onBeforeUnmount(() => { + dispose(); +}); watch( () => [props.agent.id, props.agent.localId], diff --git a/easyflow-ui-admin/app/src/views/ai/agents/composables/useAgentTryoutRawRounds.test.ts b/easyflow-ui-admin/app/src/views/ai/agents/composables/useAgentTryoutRawRounds.test.ts index 33f67d5a..ebd9203e 100644 --- a/easyflow-ui-admin/app/src/views/ai/agents/composables/useAgentTryoutRawRounds.test.ts +++ b/easyflow-ui-admin/app/src/views/ai/agents/composables/useAgentTryoutRawRounds.test.ts @@ -241,4 +241,158 @@ describe('useAgentTryoutRawRounds', () => { toolCallId: 'call-approval', }); }); + + it('异步工作流轮询事件始终归并到首张审批卡', () => { + const store = useAgentTryoutRawRounds({ + mode: 'draft', + sessionId: 'session-raw-async-tool', + }); + const roundId = store.createRound('生成文档'); + store.recordEvent(roundId, { + domain: 'TOOL', + payload: { + input: { user_input: '写一篇小作文' }, + requestId: 'req-async', + resumeToken: 'resume-async', + toolCallId: 'submit-call-1', + toolName: '文档生成', + }, + type: 'FORM_REQUEST', + }); + store.recordEvent(roundId, { + domain: 'TOOL', + payload: { + asyncTool: true, + phase: 'submit', + sourceToolCallId: 'submit-call-1', + status: 'RUNNING', + taskId: 'task-1', + toolCallId: 'task-1', + toolName: '文档生成', + }, + type: 'TOOL_RESULT', + }); + for (const sourceToolCallId of ['observe-call-1', 'observe-call-2']) { + store.recordEvent(roundId, { + domain: 'TOOL', + payload: { + asyncTool: true, + input: { taskId: 'task-1' }, + phase: 'observe', + sourceToolCallId, + status: 'RUNNING', + taskId: 'task-1', + toolCallId: 'task-1', + toolName: '文档生成', + }, + type: 'TOOL_CALL', + }); + } + store.recordEvent(roundId, { + domain: 'TOOL', + payload: { + asyncTool: true, + phase: 'result', + sourceToolCallId: 'result-call-1', + status: 'SUCCEEDED', + taskId: 'task-1', + toolCallId: 'task-1', + toolName: '文档生成', + }, + type: 'TOOL_RESULT', + }); + + const tools = store + .buildTimelineItems() + .filter((item) => item.type === 'tool'); + + expect(tools).toHaveLength(1); + expect(tools[0]).toMatchObject({ + mode: 'approval', + status: 'success', + taskId: 'task-1', + toolCallId: 'task-1', + toolName: '文档生成', + }); + }); + + it('连续文本增量压缩后刷新内容保持一致', () => { + const sessionId = 'stream-compaction'; + const store = useAgentTryoutRawRounds({ + mode: 'draft', + sessionId, + }); + const roundId = store.createRound('你好'); + const liveItems = store.buildTimelineItems(); + const assistantText = ( + items: ReturnType, + ) => + items + .flatMap((item) => + item.type === 'message' && item.role === 'assistant' + ? item.parts + .filter((part) => part.type === 'text') + .map((part) => part.content) + : [], + ) + .join(''); + + for (const delta of ['你', '好', '，', '世界']) { + const event = store.recordEvent(roundId, { + domain: 'LLM', + payload: { delta }, + type: 'MESSAGE', + }); + if (!event) { + throw new Error('流式事件记录失败'); + } + store.projectEvent(liveItems, roundId, event); + } + + expect(store.currentVariant(roundId)?.runtimeEvents).toHaveLength(1); + expect(store.currentVariant(roundId)?.runtimeEvents[0]?.payload.delta).toBe( + '你好，世界', + ); + expect(assistantText(liveItems)).toBe('你好，世界'); + + store.flush(); + const restored = useAgentTryoutRawRounds({ + mode: 'draft', + sessionId, + }); + expect(assistantText(restored.buildTimelineItems())).toBe('你好，世界'); + }); + + it('存在待持久化增量时结束事件仍立即落盘', () => { + const sessionId = 'terminal-persist'; + const store = useAgentTryoutRawRounds({ + mode: 'draft', + sessionId, + }); + const roundId = store.createRound('结束测试'); + + store.recordEvent(roundId, { + domain: 'LLM', + payload: { reasoning: '思考中' }, + type: 'THINKING', + }); + store.recordEvent(roundId, { + domain: 'SYSTEM', + payload: {}, + type: 'DONE', + }); + + const restored = useAgentTryoutRawRounds({ + mode: 'draft', + sessionId, + }); + expect(restored.currentVariant(roundId)?.status).toBe('completed'); + expect( + restored + .currentVariant(roundId) + ?.runtimeEvents.some( + (event) => event.domain === 'SYSTEM' && event.type === 'DONE', + ), + ).toBe(true); + }); }); diff --git a/easyflow-ui-admin/app/src/views/ai/agents/composables/useAgentTryoutRawRounds.ts b/easyflow-ui-admin/app/src/views/ai/agents/composables/useAgentTryoutRawRounds.ts index 31f254dc..96d5f56d 100644 --- a/easyflow-ui-admin/app/src/views/ai/agents/composables/useAgentTryoutRawRounds.ts +++ b/easyflow-ui-admin/app/src/views/ai/agents/composables/useAgentTryoutRawRounds.ts @@ -45,7 +45,7 @@ const STORAGE_VERSION = 1; const MAX_ROUNDS = 50; const MAX_VARIANTS = 10; const STORAGE_PREFIX = 'easyflow:agent-tryout-raw-rounds'; -const PERSIST_DEBOUNCE_MS = 300; +const PERSIST_DEBOUNCE_MS = 500; const memorySessions = new Map(); function createRoundId() { @@ -275,6 +275,45 @@ function selectedVariant(round: AgentTryoutRawRound) { ); } +function streamingPayloadKey(event: AgentTryoutRuntimeEvent) { + if (event.domain !== 'LLM') { + return undefined; + } + let candidates: string[] = []; + if (event.type === 'MESSAGE') { + candidates = ['delta']; + } else if (event.type === 'THINKING') { + candidates = ['reasoning', 'delta', 'text']; + } + return candidates.find((key) => typeof event.payload[key] === 'string'); +} + +function appendRuntimeEvent( + variant: AgentTryoutRawVariant, + event: AgentTryoutRuntimeEvent, +) { + const previous = variant.runtimeEvents[variant.runtimeEvents.length - 1]; + const currentKey = streamingPayloadKey(event); + const previousKey = previous ? streamingPayloadKey(previous) : undefined; + if ( + previous && + previous.domain === event.domain && + previous.type === event.type && + currentKey && + currentKey === previousKey + ) { + previous.payload = { + ...previous.payload, + ...event.payload, + [currentKey]: + asText(previous.payload[currentKey]) + + asText(event.payload[currentKey]), + }; + return; + } + variant.runtimeEvents.push(event); +} + function visibleText(item: ChatTimelineMessageItem) { return item.parts .filter((part) => part.type === 'text') @@ -529,6 +568,7 @@ function projectEventToTimeline( payload.toolDisplayName ?? rawToolName ?? '工具', ); const asyncTool = payload.asyncTool === true; + const taskInput = asRecord(payload.input ?? payload.toolInput); ChatTimelineBuilder.upsertToolCall(items, { input: payload.input ?? payload.toolInput, output: asyncTool @@ -549,6 +589,12 @@ function projectEventToTimeline( variantIndex, 'knowledge-retrieval', ), + sourceToolCallId: asyncTool + ? asText(payload.sourceToolCallId ?? payload.source_tool_call_id) + : undefined, + taskId: asyncTool + ? asText(payload.taskId ?? taskInput.taskId ?? taskInput.task_id) + : undefined, toolCallId: asText( payload.toolCallId ?? payload.taskId ?? @@ -650,11 +696,6 @@ export function useAgentTryoutRawRounds(options: { } function schedulePersist() { - const key = storageKey(options.mode, options.sessionId); - memorySessions.set( - key, - [...rounds.values()].map((item) => clone(item)), - ); if (persistTimer) { return; } @@ -737,7 +778,7 @@ export function useAgentTryoutRawRounds(options: { payload: event.payload || {}, type: event.type.toUpperCase(), }; - variant.runtimeEvents.push(runtimeEvent); + appendRuntimeEvent(variant, runtimeEvent); if (runtimeEvent.domain === 'SYSTEM' && runtimeEvent.type === 'DONE') { variant.status = 'completed'; round.status = 'completed'; @@ -754,9 +795,23 @@ export function useAgentTryoutRawRounds(options: { runtimeEvent.domain === 'ERROR' ) { persistNow(); - return; + return runtimeEvent; } schedulePersist(); + return runtimeEvent; + } + + function projectEvent( + items: ChatTimelineItem[], + roundId: string, + event: AgentTryoutRuntimeEvent, + ) { + const round = rounds.get(roundId); + const variant = round && selectedVariant(round); + if (!round || !variant) { + return; + } + projectEventToTimeline(items, event, round.roundId, variant.variantIndex); } function completeRound(roundId: string) { @@ -849,6 +904,7 @@ export function useAgentTryoutRawRounds(options: { createRound, currentVariant, getPrompt, + projectEvent, recordEvent, regenerateRound, selectVariant, diff --git a/easyflow-ui-admin/app/src/views/ai/agents/composables/useAgentTryoutStream.ts b/easyflow-ui-admin/app/src/views/ai/agents/composables/useAgentTryoutStream.ts index 8f854e6f..7284861c 100644 --- a/easyflow-ui-admin/app/src/views/ai/agents/composables/useAgentTryoutStream.ts +++ b/easyflow-ui-admin/app/src/views/ai/agents/composables/useAgentTryoutStream.ts @@ -114,6 +114,7 @@ export function useAgentTryoutStream() { const sessionChanged = activeSessionId !== sessionId; activeSessionId = sessionId; if (!rawRounds || sessionChanged) { + rawRounds?.flush(); rawRounds = useAgentTryoutRawRounds({ mode: 'draft', sessionId, @@ -174,12 +175,18 @@ export function useAgentTryoutStream() { const payload = envelope.payload || {}; if (activeRoundId) { - rawRounds?.recordEvent(activeRoundId, { + const runtimeEvent = rawRounds?.recordEvent(activeRoundId, { domain, payload, type, }); - rebuildTimeline(); + if (runtimeEvent) { + rawRounds?.projectEvent( + timelineItems.value, + activeRoundId, + runtimeEvent, + ); + } } if (domain === 'LLM' && type === 'MESSAGE') { @@ -340,12 +347,23 @@ export function useAgentTryoutStream() { finishStoppedRun(); } + function dispose() { + if (loading.value) { + userStopped = true; + sseClient.abort(); + finishStoppedRun(); + return; + } + rawRounds?.flush(); + } + return { loading, clearDraftSession, markToolApproving, markToolRejected, copyMessageText, + dispose, regenerateDraft, selectVariant, syncDraftContext, diff --git a/easyflow-ui-admin/packages/effects/common-ui/src/components/chat-timeline/ChatTimeline.vue b/easyflow-ui-admin/packages/effects/common-ui/src/components/chat-timeline/ChatTimeline.vue index 8666b05d..432ae1c5 100644 --- a/easyflow-ui-admin/packages/effects/common-ui/src/components/chat-timeline/ChatTimeline.vue +++ b/easyflow-ui-admin/packages/effects/common-ui/src/components/chat-timeline/ChatTimeline.vue @@ -6,7 +6,7 @@ import type { ChatTimelineToolApprovalPayload, } from './types'; -import { nextTick, onBeforeUnmount, ref, watch } from 'vue'; +import { computed, nextTick, onBeforeUnmount, ref, watch } from 'vue'; import ChatTimelineItem from './ChatTimelineItem.vue'; @@ -38,6 +38,15 @@ let preservedScrollTop: number | undefined; const bottomThreshold = 24; let scrollFrame = 0; +const assistantActionAnchorIds = computed(() => { + const latestAssistantByRound = new Map(); + for (const item of props.items) { + if (item.type === 'message' && item.role === 'assistant' && item.roundId) { + latestAssistantByRound.set(item.roundId, item.id); + } + } + return new Set(latestAssistantByRound.values()); +}); function isNearBottom(container: HTMLElement) { return ( @@ -85,26 +94,13 @@ function canRegenerateMessage(item: ChatTimelineItemType) { return item.type === 'message' && (props.regenerable?.(item) ?? false); } -function isAssistantActionAnchor( - item: ChatTimelineItemType, - index: number, - items: ChatTimelineItemType[], -) { - if (item.type !== 'message' || item.role !== 'assistant' || !item.roundId) { - return false; - } - for (let cursor = items.length - 1; cursor >= 0; cursor -= 1) { - const current = items[cursor]; - if ( - current && - current.type === 'message' && - current.role === 'assistant' && - current.roundId === item.roundId - ) { - return cursor === index; - } - } - return false; +function isAssistantActionAnchor(item: ChatTimelineItemType) { + return ( + item.type === 'message' && + item.role === 'assistant' && + Boolean(item.roundId) && + assistantActionAnchorIds.value.has(item.id) + ); } function isVariantLoading(item: ChatTimelineItemType) { @@ -152,9 +148,9 @@ watch(