additional internal end point for streaming and associated changes

packages/components/nodes/agents/ConversationalRetrievalToolAgent/ConversationalRetrievalToolAgent.ts

@@ -7,7 +7,16 @@ import { ChatPromptTemplate, MessagesPlaceholder, HumanMessagePromptTemplate, Pr
 import { formatToOpenAIToolMessages } from 'langchain/agents/format_scratchpad/openai_tools'
 import { getBaseClasses } from '../../../src/utils'
 import { type ToolsAgentStep } from 'langchain/agents/openai/output_parser'
-import { FlowiseMemory, ICommonObject, INode, INodeData, INodeParams, IUsedTool, IVisionChatModal } from '../../../src/Interface'
+import {
+    FlowiseMemory,
+    ICommonObject,
+    INode,
+    INodeData,
+    INodeParams,
+    IServerSideEventStreamer,
+    IUsedTool,
+    IVisionChatModal
+} from "../../../src/Interface";
 import { ConsoleCallbackHandler, CustomChainHandler, additionalCallbacks } from '../../../src/handler'
 import { AgentExecutor, ToolCallingAgentOutputParser } from '../../../src/agents'
 import { Moderation, checkInputs, streamResponse } from '../../moderation/Moderation'
@@ -104,16 +113,19 @@ class ConversationalRetrievalToolAgent_Agents implements INode {
         const memory = nodeData.inputs?.memory as FlowiseMemory
         const moderations = nodeData.inputs?.inputModeration as Moderation[]
 
-        const isStreamable = options.socketIO && options.socketIOClientId
+        const shouldStreamResponse = options.shouldStreamResponse
+        const sseStreamer: IServerSideEventStreamer = options.sseStreamer as IServerSideEventStreamer
+        const chatId = options.chatId
 
         if (moderations && moderations.length > 0) {
             try {
                 // Use the output of the moderation chain as input for the OpenAI Function Agent
                 input = await checkInputs(moderations, input)
             } catch (e) {
                 await new Promise((resolve) => setTimeout(resolve, 500))
-                if (isStreamable)
-                    streamResponse(options.socketIO && options.socketIOClientId, e.message, options.socketIO, options.socketIOClientId)
+                if (shouldStreamResponse) {
+                    streamResponse(sseStreamer, chatId, e.message)
+                }
                 return formatResponse(e.message)
             }
         }
@@ -127,15 +139,15 @@ class ConversationalRetrievalToolAgent_Agents implements INode {
         let sourceDocuments: ICommonObject[] = []
         let usedTools: IUsedTool[] = []
 
-        if (isStreamable) {
-            const handler = new CustomChainHandler(options.socketIO, options.socketIOClientId)
+        if (shouldStreamResponse) {
+            const handler = new CustomChainHandler(sseStreamer, chatId)
             res = await executor.invoke({ input }, { callbacks: [loggerHandler, handler, ...callbacks] })
             if (res.sourceDocuments) {
-                options.socketIO.to(options.socketIOClientId).emit('sourceDocuments', flatten(res.sourceDocuments))
+                sseStreamer.streamSourceDocumentsEvent(chatId, JSON.stringify(flatten(res.sourceDocuments)))
                 sourceDocuments = res.sourceDocuments
             }
             if (res.usedTools) {
-                options.socketIO.to(options.socketIOClientId).emit('usedTools', res.usedTools)
+                sseStreamer.streamUsedToolsEvent(chatId, JSON.stringify(res.usedTools))
                 usedTools = res.usedTools
             }
         } else {

packages/server/src/controllers/internal-predictions/index.ts

@@ -5,35 +5,47 @@ import { getRunningExpressApp } from '../../utils/getRunningExpressApp'
 // Send input message and get prediction result (Internal)
 const createInternalPrediction = async (req: Request, res: Response, next: NextFunction) => {
     try {
-        const chatId = req.params.chatId
-        //getRunningExpressApp().sseStreamer.addClient(chatId, res)
+        const apiResponse = await utilBuildChatflow(req, true)
+        return res.json(apiResponse)
+    } catch (error) {
+        next(error)
+    }
+}
+
+// Send input message and stream prediction result using SSE (Internal)
+const createAndStreamInternalPrediction = async (req: Request, res: Response, next: NextFunction) => {
+    try {
+        const chatId = req.body.chatId
+        getRunningExpressApp().sseStreamer.addClient(chatId, res)
+        res.setHeader('Content-Type', 'text/event-stream')
+        res.setHeader('Cache-Control', 'no-cache')
+        res.setHeader('Connection', 'keep-alive')
+        res.flushHeaders()
 
         const apiResponse = await utilBuildChatflow(req, true)
-        if (apiResponse.isStreamValid) {
-            const sseStreamer = getRunningExpressApp().sseStreamer
-            if (apiResponse.chatId) {
-                sseStreamer.streamCustomEvent(apiResponse.chatId, 'chatId', apiResponse.chatId)
-            }
-            if (apiResponse.chatMessageId) {
-                sseStreamer.streamCustomEvent(apiResponse.chatId, 'chatMessageId', apiResponse.chatMessageId)
-            }
-            if (apiResponse.question) {
-                sseStreamer.streamCustomEvent(apiResponse.chatId, 'question', apiResponse.question)
-            }
-            if (apiResponse.sessionId) {
-                sseStreamer.streamCustomEvent(apiResponse.chatId, 'sessionId', apiResponse.sessionId)
-            }
-            if (apiResponse.memoryType) {
-                sseStreamer.streamCustomEvent(apiResponse.chatId, 'memoryType', apiResponse.memoryType)
-            }
-            sseStreamer.removeClient(apiResponse.chatId)
+        const sseStreamer = getRunningExpressApp().sseStreamer
+        if (apiResponse.chatId) {
+            sseStreamer.streamCustomEvent(apiResponse.chatId, 'chatId', apiResponse.chatId)
         }
-
-        return res.json(apiResponse)
+        if (apiResponse.chatMessageId) {
+            sseStreamer.streamCustomEvent(apiResponse.chatId, 'chatMessageId', apiResponse.chatMessageId)
+        }
+        if (apiResponse.question) {
+            sseStreamer.streamCustomEvent(apiResponse.chatId, 'question', apiResponse.question)
+        }
+        if (apiResponse.sessionId) {
+            sseStreamer.streamCustomEvent(apiResponse.chatId, 'sessionId', apiResponse.sessionId)
+        }
+        if (apiResponse.memoryType) {
+            sseStreamer.streamCustomEvent(apiResponse.chatId, 'memoryType', apiResponse.memoryType)
+        }
+        sseStreamer.removeClient(apiResponse.chatId)
+        return
     } catch (error) {
         next(error)
     }
 }
 export default {
-    createInternalPrediction
+    createInternalPrediction,
+    createAndStreamInternalPrediction
 }

packages/server/src/controllers/predictions/index.ts

@@ -61,28 +61,32 @@ const createPrediction = async (req: Request, res: Response, next: NextFunction)
             const apiResponse = await predictionsServices.buildChatflow(req)
             if (streamable?.isStreaming && req.body.streaming === 'true') {
                 const sseStreamer = getRunningExpressApp().sseStreamer
+                const metadataJson: any = {}
                 if (apiResponse.chatId) {
-                    sseStreamer.streamCustomEvent(apiResponse.chatId, 'chatId', apiResponse.chatId)
+                    metadataJson['chatId'] = apiResponse.chatId
+                    //sseStreamer.streamCustomEvent(apiResponse.chatId, 'chatId', apiResponse.chatId)
                 }
                 if (apiResponse.chatMessageId) {
-                    sseStreamer.streamCustomEvent(apiResponse.chatId, 'chatMessageId', apiResponse.chatMessageId)
+                    metadataJson['chatMessageId'] = apiResponse.chatMessageId
+                    //sseStreamer.streamCustomEvent(apiResponse.chatId, 'chatMessageId', apiResponse.chatMessageId)
                 }
                 if (apiResponse.question) {
-                    sseStreamer.streamCustomEvent(apiResponse.chatId, 'question', apiResponse.question)
+                    metadataJson['question'] = apiResponse.question
+                    //sseStreamer.streamCustomEvent(apiResponse.chatId, 'question', apiResponse.question)
                 }
                 if (apiResponse.sessionId) {
-                    sseStreamer.streamCustomEvent(apiResponse.chatId, 'sessionId', apiResponse.sessionId)
+                    metadataJson['sessionId'] = apiResponse.sessionId
+                    //sseStreamer.streamCustomEvent(apiResponse.chatId, 'sessionId', apiResponse.sessionId)
                 }
                 if (apiResponse.memoryType) {
-                    sseStreamer.streamCustomEvent(apiResponse.chatId, 'memoryType', apiResponse.memoryType)
+                    metadataJson['memoryType'] = apiResponse.memoryType
+                    //sseStreamer.streamCustomEvent(apiResponse.chatId, 'memoryType', apiResponse.memoryType)
                 }
+                sseStreamer.streamCustomEvent(apiResponse.chatId, 'metadata', JSON.stringify(metadataJson))
                 sseStreamer.removeClient(apiResponse.chatId)
-            }
-            if (req.body.streaming === 'true') {
                 return
-            } else {
-                return res.json(apiResponse)
             }
+            return res.json(apiResponse)
         } else {
             throw new InternalFlowiseError(StatusCodes.UNAUTHORIZED, `This site is not allowed to access this chatbot`)
         }

packages/server/src/index.ts

@@ -135,9 +135,9 @@ export class App {
             '/api/v1/leads',
             '/api/v1/get-upload-file',
             '/api/v1/ip',
-            '/api/v1/ping',
-                '/api/v1/events'
-            ]const URL_CASE_INSENSITIVE_REGEX: RegExp = /\/api\/v1\//i
+            '/api/v1/ping'
+        ]
+        const URL_CASE_INSENSITIVE_REGEX: RegExp = /\/api\/v1\//i
         const URL_CASE_SENSITIVE_REGEX: RegExp = /\/api\/v1\//
 
         if (process.env.FLOWISE_USERNAME && process.env.FLOWISE_PASSWORD) {
@@ -203,7 +203,6 @@ export class App {
 
         this.app.use('/api/v1', flowiseApiV1Router)
         this.sseStreamer = new SSEStreamer(this.app)
-        this.sseStreamer.setupSSEEndpoint()
 
         // ----------------------------------------
         // Configure number of proxies in Host Environment

packages/server/src/routes/internal-predictions/index.ts

@@ -4,5 +4,6 @@ const router = express.Router()
 
 // CREATE
 router.post(['/', '/:id'], internalPredictionsController.createInternalPrediction)
+router.post(['/', '/stream/:id'], internalPredictionsController.createAndStreamInternalPrediction)
 
 export default router