Feat: Receive reply messages of different event types from the agent #3221 (#8100)

### What problem does this PR solve?
Feat: Receive reply messages of different event types from the agent
#3221

### Type of change


- [x] New Feature (non-breaking change which adds functionality)

web/src/hooks/use-send-message.ts

+import { Authorization } from '@/constants/authorization';
+import api from '@/utils/api';
+import { getAuthorization } from '@/utils/authorization-util';
+import { EventSourceParserStream } from 'eventsource-parser/stream';
+import { useCallback, useRef, useState } from 'react';
+
+export enum MessageEventType {
+  WorkflowStarted = 'workflow_started',
+  NodeStarted = 'node_started',
+  NodeFinished = 'node_finished',
+  Message = 'message',
+  MessageEnd = 'message_end',
+  WorkflowFinished = 'workflow_finished',
+}
+
+export interface IAnswerEvent<T> {
+  event: MessageEventType;
+  message_id: string;
+  created_at: number;
+  task_id: string;
+  data: T;
+}
+
+export interface INodeData {
+  inputs: Record<string, any>;
+  outputs: Record<string, any>;
+  component_id: string;
+  error: null | string;
+  elapsed_time: number;
+  created_at: number;
+}
+
+export interface IMessageData {
+  content: string;
+}
+
+export type INodeEvent = IAnswerEvent<INodeData>;
+
+export type IMessageEvent = IAnswerEvent<IMessageData>;
+
+export type IEventList = Array<INodeEvent | IMessageEvent>;
+
+export const useSendMessageBySSE = (url: string = api.completeConversation) => {
+  const [answerList, setAnswerList] = useState<IEventList>([]);
+  const [done, setDone] = useState(true);
+  const timer = useRef<any>();
+  const sseRef = useRef<AbortController>();
+
+  const initializeSseRef = useCallback(() => {
+    sseRef.current = new AbortController();
+  }, []);
+
+  const resetAnswerList = useCallback(() => {
+    if (timer.current) {
+      clearTimeout(timer.current);
+    }
+    timer.current = setTimeout(() => {
+      setAnswerList([]);
+      clearTimeout(timer.current);
+    }, 1000);
+  }, []);
+
+  const send = useCallback(
+    async (
+      body: any,
+      controller?: AbortController,
+    ): Promise<{ response: Response; data: ResponseType } | undefined> => {
+      initializeSseRef();
+      try {
+        setDone(false);
+        const response = await fetch(url, {
+          method: 'POST',
+          headers: {
+            [Authorization]: getAuthorization(),
+            'Content-Type': 'application/json',
+          },
+          body: JSON.stringify(body),
+          signal: controller?.signal || sseRef.current?.signal,
+        });
+
+        const res = response.clone().json();
+
+        const reader = response?.body
+          ?.pipeThrough(new TextDecoderStream())
+          .pipeThrough(new EventSourceParserStream())
+          .getReader();
+
+        while (true) {
+          const x = await reader?.read();
+          if (x) {
+            const { done, value } = x;
+            if (done) {
+              console.info('done');
+              resetAnswerList();
+              break;
+            }
+            try {
+              const val = JSON.parse(value?.data || '');
+
+              console.info('data:', val);
+
+              setAnswerList((list) => {
+                const nextList = [...list];
+                nextList.push(val);
+                return nextList;
+              });
+            } catch (e) {
+              console.warn(e);
+            }
+          }
+        }
+        console.info('done?');
+        setDone(true);
+        resetAnswerList();
+        return { data: await res, response };
+      } catch (e) {
+        setDone(true);
+        resetAnswerList();
+
+        console.warn(e);
+      }
+    },
+    [initializeSseRef, url, resetAnswerList],
+  );
+
+  const stopOutputMessage = useCallback(() => {
+    sseRef.current?.abort();
+  }, []);
+
+  return {
+    send,
+    answerList,
+    done,
+    setDone,
+    resetAnswerList,
+    stopOutputMessage,
+  };
+};

web/src/pages/agent/chat/box.tsx

+import MessageItem from '@/components/message-item';
+import { MessageType } from '@/constants/chat';
+import { useGetFileIcon } from '@/pages/chat/hooks';
+import { buildMessageItemReference } from '@/pages/chat/utils';
+import { Spin } from 'antd';
+
+import { useSendNextMessage } from './hooks';
+
+import MessageInput from '@/components/message-input';
+import PdfDrawer from '@/components/pdf-drawer';
+import { useClickDrawer } from '@/components/pdf-drawer/hooks';
+import { useFetchFlow } from '@/hooks/flow-hooks';
+import { useFetchUserInfo } from '@/hooks/user-setting-hooks';
+import { buildMessageUuidWithRole } from '@/utils/chat';
+
+const AgentChatBox = () => {
+  const {
+    sendLoading,
+    handleInputChange,
+    handlePressEnter,
+    value,
+    loading,
+    ref,
+    derivedMessages,
+    reference,
+    stopOutputMessage,
+  } = useSendNextMessage();
+
+  const { visible, hideModal, documentId, selectedChunk, clickDocumentButton } =
+    useClickDrawer();
+  useGetFileIcon();
+  const { data: userInfo } = useFetchUserInfo();
+  const { data: canvasInfo } = useFetchFlow();
+
+  return (
+    <>
+      <section className="flex flex-1 flex-col pl-5 h-[90vh]">
+        <div className="flex-1 ">
+          <div>
+            <Spin spinning={loading}>
+              {derivedMessages?.map((message, i) => {
+                return (
+                  <MessageItem
+                    loading={
+                      message.role === MessageType.Assistant &&
+                      sendLoading &&
+                      derivedMessages.length - 1 === i
+                    }
+                    key={buildMessageUuidWithRole(message)}
+                    nickname={userInfo.nickname}
+                    avatar={userInfo.avatar}
+                    avatarDialog={canvasInfo.avatar}
+                    item={message}
+                    reference={buildMessageItemReference(
+                      { message: derivedMessages, reference },
+                      message,
+                    )}
+                    clickDocumentButton={clickDocumentButton}
+                    index={i}
+                    showLikeButton={false}
+                    sendLoading={sendLoading}
+                  ></MessageItem>
+                );
+              })}
+            </Spin>
+          </div>
+          <div ref={ref} />
+        </div>
+        <MessageInput
+          showUploadIcon={false}
+          value={value}
+          sendLoading={sendLoading}
+          disabled={false}
+          sendDisabled={sendLoading}
+          conversationId=""
+          onPressEnter={handlePressEnter}
+          onInputChange={handleInputChange}
+          stopOutputMessage={stopOutputMessage}
+        />
+      </section>
+      <PdfDrawer
+        visible={visible}
+        hideModal={hideModal}
+        documentId={documentId}
+        chunk={selectedChunk}
+      ></PdfDrawer>
+    </>
+  );
+};
+
+export default AgentChatBox;

web/src/pages/agent/chat/chat-sheet.tsx

 import {
   Sheet,
   SheetContent,
-  SheetDescription,
   SheetHeader,
   SheetTitle,
-  SheetTrigger,
 } from '@/components/ui/sheet';
 import { IModalProps } from '@/interfaces/common';
+import { cn } from '@/lib/utils';
+import AgentChatBox from './box';
 
-export function ChatSheet({ visible }: IModalProps<any>) {
+export function ChatSheet({ visible, hideModal }: IModalProps<any>) {
   return (
-    <Sheet open={visible} modal={false}>
-      <SheetTrigger>Open</SheetTrigger>
-      <SheetContent>
+    <Sheet open={visible} modal={false} onOpenChange={hideModal}>
+      <SheetTitle className="hidden"></SheetTitle>
+      <SheetContent className={cn('top-20 p-0')}>
         <SheetHeader>
           <SheetTitle>Are you absolutely sure?</SheetTitle>
-          <SheetDescription>
-            This action cannot be undone. This will permanently delete your
-            account and remove your data from our servers.
-          </SheetDescription>
         </SheetHeader>
+        <AgentChatBox></AgentChatBox>
       </SheetContent>
     </Sheet>
   );

web/src/pages/agent/chat/hooks.ts

+import { MessageType } from '@/constants/chat';
+import { useFetchFlow } from '@/hooks/flow-hooks';
+import {
+  useHandleMessageInputChange,
+  useSelectDerivedMessages,
+} from '@/hooks/logic-hooks';
+import {
+  IEventList,
+  IMessageEvent,
+  MessageEventType,
+  useSendMessageBySSE,
+} from '@/hooks/use-send-message';
+import { Message } from '@/interfaces/database/chat';
+import i18n from '@/locales/config';
+import api from '@/utils/api';
+import { message } from 'antd';
+import trim from 'lodash/trim';
+import { useCallback, useEffect } from 'react';
+import { useParams } from 'umi';
+import { v4 as uuid } from 'uuid';
+import { receiveMessageError } from '../utils';
+
+const antMessage = message;
+
+export const useSelectNextMessages = () => {
+  const { data: flowDetail, loading } = useFetchFlow();
+  const reference = flowDetail.dsl.reference;
+  const {
+    derivedMessages,
+    ref,
+    addNewestQuestion,
+    addNewestAnswer,
+    removeLatestMessage,
+    removeMessageById,
+    removeMessagesAfterCurrentMessage,
+  } = useSelectDerivedMessages();
+
+  return {
+    reference,
+    loading,
+    derivedMessages,
+    ref,
+    addNewestQuestion,
+    addNewestAnswer,
+    removeLatestMessage,
+    removeMessageById,
+    removeMessagesAfterCurrentMessage,
+  };
+};
+
+function findMessageFromList(eventList: IEventList) {
+  const event = eventList.find((x) => x.event === MessageEventType.Message) as
+    | IMessageEvent
+    | undefined;
+
+  return event?.data?.content;
+}
+
+export const useSendNextMessage = () => {
+  const {
+    reference,
+    loading,
+    derivedMessages,
+    ref,
+    addNewestQuestion,
+    addNewestAnswer,
+    removeLatestMessage,
+    removeMessageById,
+  } = useSelectNextMessages();
+  const { id: agentId } = useParams();
+  const { handleInputChange, value, setValue } = useHandleMessageInputChange();
+  const { refetch } = useFetchFlow();
+
+  const { send, answerList, done, stopOutputMessage } = useSendMessageBySSE(
+    api.runCanvas,
+  );
+
+  const sendMessage = useCallback(
+    async ({ message }: { message: Message; messages?: Message[] }) => {
+      const params: Record<string, unknown> = {
+        id: agentId,
+      };
+      params.running_hint_text = i18n.t('flow.runningHintText', {
+        defaultValue: 'is running...🕞',
+      });
+      if (message.content) {
+        params.query = message.content;
+        // params.message_id = message.id;
+        params.inputs = {}; // begin operator inputs
+      }
+      const res = await send(params);
+
+      if (receiveMessageError(res)) {
+        antMessage.error(res?.data?.message);
+
+        // cancel loading
+        setValue(message.content);
+        removeLatestMessage();
+      } else {
+        refetch(); // pull the message list after sending the message successfully
+      }
+    },
+    [agentId, send, setValue, removeLatestMessage, refetch],
+  );
+
+  const handleSendMessage = useCallback(
+    async (message: Message) => {
+      sendMessage({ message });
+    },
+    [sendMessage],
+  );
+
+  useEffect(() => {
+    const message = findMessageFromList(answerList);
+    if (message) {
+      addNewestAnswer({
+        answer: message,
+        reference: {
+          chunks: [],
+          doc_aggs: [],
+          total: 0,
+        },
+      });
+    }
+  }, [answerList, addNewestAnswer]);
+
+  const handlePressEnter = useCallback(() => {
+    if (trim(value) === '') return;
+    const id = uuid();
+    if (done) {
+      setValue('');
+      handleSendMessage({ id, content: value.trim(), role: MessageType.User });
+    }
+    addNewestQuestion({
+      content: value,
+      id,
+      role: MessageType.User,
+    });
+  }, [addNewestQuestion, handleSendMessage, done, setValue, value]);
+
+  const fetchPrologue = useCallback(async () => {
+    // fetch prologue
+    const sendRet = await send({ id: agentId });
+    if (receiveMessageError(sendRet)) {
+      message.error(sendRet?.data?.message);
+    } else {
+      refetch();
+    }
+  }, [agentId, refetch, send]);
+
+  useEffect(() => {
+    fetchPrologue();
+  }, [fetchPrologue]);
+
+  return {
+    handlePressEnter,
+    handleInputChange,
+    value,
+    sendLoading: !done,
+    reference,
+    loading,
+    derivedMessages,
+    ref,
+    removeMessageById,
+    stopOutputMessage,
+  };
+};

web/src/pages/agent/form-sheet/next.tsx

 
   return (
     <Sheet open={visible} modal={false}>
-      <SheetTitle className="hidden"></SheetTitle>
       <SheetContent className={cn('top-20 p-0')} closeIcon={false}>
         <SheetHeader>
+          <SheetTitle className="hidden"></SheetTitle>
           <section className="flex-col border-b py-2 px-5">
             <div className="flex items-center gap-2 pb-3">
               <OperatorIcon