update memory by SSE in preview

web/app/components/workflow/panel/debug-and-preview/hooks.ts

@@ -13,7 +13,7 @@ import {
   useWorkflowRun,
 } from '../../hooks'
 import { NodeRunningStatus, WorkflowRunningStatus } from '../../types'
-import { useWorkflowStore } from '../../store'
+import { useStore, useWorkflowStore } from '../../store'
 import { DEFAULT_ITER_TIMES, DEFAULT_LOOP_TIMES } from '../../constants'
 import type {
   ChatItem,
@@ -63,6 +63,8 @@ export const useChat = (
   const { fetchInspectVars } = useSetWorkflowVarsWithValue()
   const [suggestedQuestions, setSuggestQuestions] = useState<string[]>([])
   const suggestedQuestionsAbortControllerRef = useRef<AbortController | null>(null)
+  const conversationVariables = useStore(s => s.conversationVariables)
+  const setConversationVariables = useStore(s => s.setConversationVariables)
   const {
     setIterTimes,
     setLoopTimes,
@@ -499,6 +501,18 @@ export const useChat = (
             })
           }
         },
+        onMemoryUpdate: ({ data }) => {
+          const currentMemoryIndex = conversationVariables.findIndex(item => item.id === data.memory_id)
+          const newList = produce(conversationVariables, (draft) => {
+            if (currentMemoryIndex > -1) {
+              draft[currentMemoryIndex] = {
+                ...draft[currentMemoryIndex],
+                ...data,
+              }
+            }
+          })
+          setConversationVariables(newList)
+        },
       },
     )
   }, [threadMessages, chatTree.length, updateCurrentQAOnTree, handleResponding, formSettings?.inputsForm, handleRun, notify, t, config?.suggested_questions_after_answer?.enabled, fetchInspectVars, invalidAllLastRun])

web/service/base.ts

@@ -12,6 +12,7 @@ import type {
   LoopFinishedResponse,
   LoopNextResponse,
   LoopStartedResponse,
+  MemoryUpdateResponse,
   NodeFinishedResponse,
   NodeStartedResponse,
   ParallelBranchFinishedResponse,
@@ -62,6 +63,7 @@ export type IOnLoopStarted = (workflowStarted: LoopStartedResponse) => void
 export type IOnLoopNext = (workflowStarted: LoopNextResponse) => void
 export type IOnLoopFinished = (workflowFinished: LoopFinishedResponse) => void
 export type IOnAgentLog = (agentLog: AgentLogResponse) => void
+export type IOnMemoryUpdate = (memory: MemoryUpdateResponse) => void
 
 export type IOtherOptions = {
   isPublicAPI?: boolean
@@ -97,6 +99,7 @@ export type IOtherOptions = {
   onLoopNext?: IOnLoopNext
   onLoopFinish?: IOnLoopFinished
   onAgentLog?: IOnAgentLog
+  onMemoryUpdate?: IOnMemoryUpdate
 }
 
 function unicodeToChar(text: string) {
@@ -152,6 +155,7 @@ const handleStream = (
   onTTSEnd?: IOnTTSEnd,
   onTextReplace?: IOnTextReplace,
   onAgentLog?: IOnAgentLog,
+  onMemoryUpdate?: IOnMemoryUpdate,
 ) => {
   if (!response.ok)
     throw new Error('Network response was not ok')
@@ -270,6 +274,9 @@ const handleStream = (
             else if (bufferObj.event === 'tts_message_end') {
               onTTSEnd?.(bufferObj.message_id, bufferObj.audio)
             }
+            else if (bufferObj.event === 'memory_update') {
+              onMemoryUpdate?.(bufferObj as MemoryUpdateResponse)
+            }
           }
         })
         buffer = lines[lines.length - 1]
@@ -363,6 +370,7 @@ export const ssePost = async (
     onLoopStart,
     onLoopNext,
     onLoopFinish,
+    onMemoryUpdate,
   } = otherOptions
   const abortController = new AbortController()
 
@@ -465,6 +473,7 @@ export const ssePost = async (
       onTTSEnd,
       onTextReplace,
       onAgentLog,
+      onMemoryUpdate,
       )
     }).catch((e) => {
       if (e.toString() !== 'AbortError: The user aborted a request.' && !e.toString().includes('TypeError: Cannot assign to read only property'))

web/types/workflow.ts

@@ -290,6 +290,15 @@ export type AgentLogResponse = {
   data: AgentLogItemWithChildren
 }
 
+export type MemoryUpdateResponse = {
+  task_id: string
+  event: string
+  data: {
+    memory_id: string
+    memory: any // TODO
+  }
+}
+
 export type WorkflowRunHistory = {
   id: string
   version: string