Feature/lang graph (#2319)

* add langgraph * datasource: initial commit * datasource: datasource details and chunks * datasource: Document Store Node * more changes * Document Store - Base functionality * Document Store Loader Component * Document Store Loader Component * before merging the modularity PR * after merging the modularity PR * preview mode * initial draft PR * fixes * minor updates and fixes * preview with loader and splitter * preview with credential * show stored chunks * preview update... * edit config * save, preview and other changes * save, preview and other changes * save, process and other changes * save, process and other changes * alpha1 - for internal testing * rerouting urls * bug fix on new leader create * pagination support for chunks * delete document store * Update pnpm-lock.yaml * doc store card view * Update store files to use updated storage functions, Document Store Table View and other changes * ui changes * add expanded chunk dialog, improve ui * change throw Error to InternalError * Bug Fixes and removal of subFolder, adding of view chunks for store * lint fixes * merge changes * DocumentStoreStatus component * ui changes for doc store * add remove metadata key field, add custom document loader * add chatflows used doc store chips * add types/interfaces to DocumentStore Services * document loader list dialog title bar color change * update interfaces * Whereused Chatflow Name and Added chunkNo to retain order of created chunks. * use typeorm order chunkNo, ui changes * update tabler icons react * cleanup agents * add pysandbox tool * add abort functionality, loading next agent * add empty view svg * update chatflow tool with chatId * rename to agentflows * update worker for prompt input values * update dashboard to agentflows, agentcanvas * fix marketplace use template * add agentflow templates * resolve merge conflict * update baseURL --------- Co-authored-by: vinodkiran <vinodkiran@usa.net> Co-authored-by: Vinod Paidimarry <vinodkiran@outlook.in>
2026-06-28 17:01:00 +03:00 · 2024-05-21 16:36:42 +01:00
parent 95f1090bed
commit 8ebc4dcfd5
92 changed files with 7216 additions and 701 deletions
@@ -0,0 +1,345 @@
+import {
+    ICommonObject,
+    IMultiAgentNode,
+    IAgentReasoning,
+    ITeamState,
+    ConsoleCallbackHandler,
+    additionalCallbacks
+} from 'flowise-components'
+import { IChatFlow, IComponentNodes, IDepthQueue, IReactFlowNode, IReactFlowObject } from '../Interface'
+import { Server } from 'socket.io'
+import { buildFlow, getStartingNodes, getEndingNodes, constructGraphs, databaseEntities } from '../utils'
+import { getRunningExpressApp } from './getRunningExpressApp'
+import logger from './logger'
+import { StateGraph, END } from '@langchain/langgraph'
+import { BaseMessage, HumanMessage } from '@langchain/core/messages'
+import { cloneDeep, flatten } from 'lodash'
+import { replaceInputsWithConfig, resolveVariables } from '.'
+import { StatusCodes } from 'http-status-codes'
+import { InternalFlowiseError } from '../errors/internalFlowiseError'
+import { getErrorMessage } from '../errors/utils'
+
+/**
+ * Build Agent Graph
+ * @param {IChatFlow} chatflow
+ * @param {string} chatId
+ * @param {string} sessionId
+ * @param {ICommonObject} incomingInput
+ * @param {string} baseURL
+ * @param {Server} socketIO
+ */
+export const buildAgentGraph = async (
+    chatflow: IChatFlow,
+    chatId: string,
+    sessionId: string,
+    incomingInput: ICommonObject,
+    baseURL?: string,
+    socketIO?: Server
+): Promise<any> => {
+    try {
+        const appServer = getRunningExpressApp()
+        const chatflowid = chatflow.id
+
+        /*** Get chatflows and prepare data  ***/
+        const flowData = chatflow.flowData
+        const parsedFlowData: IReactFlowObject = JSON.parse(flowData)
+        const nodes = parsedFlowData.nodes
+        const edges = parsedFlowData.edges
+
+        /*** Get Ending Node with Directed Graph  ***/
+        const { graph, nodeDependencies } = constructGraphs(nodes, edges)
+        const directedGraph = graph
+
+        const endingNodes = getEndingNodes(nodeDependencies, directedGraph, nodes)
+
+        /*** Get Starting Nodes with Reversed Graph ***/
+        const constructedObj = constructGraphs(nodes, edges, { isReversed: true })
+        const nonDirectedGraph = constructedObj.graph
+        let startingNodeIds: string[] = []
+        let depthQueue: IDepthQueue = {}
+        const endingNodeIds = endingNodes.map((n) => n.id)
+        for (const endingNodeId of endingNodeIds) {
+            const resx = getStartingNodes(nonDirectedGraph, endingNodeId)
+            startingNodeIds.push(...resx.startingNodeIds)
+            depthQueue = Object.assign(depthQueue, resx.depthQueue)
+        }
+        startingNodeIds = [...new Set(startingNodeIds)]
+
+        // Initialize nodes like ChatModels, Tools, etc.
+        const reactFlowNodes = await buildFlow(
+            startingNodeIds,
+            nodes,
+            edges,
+            graph,
+            depthQueue,
+            appServer.nodesPool.componentNodes,
+            incomingInput.question,
+            [],
+            chatId,
+            sessionId,
+            chatflowid,
+            appServer.AppDataSource,
+            incomingInput?.overrideConfig,
+            appServer.cachePool,
+            false,
+            undefined,
+            incomingInput.uploads,
+            baseURL
+        )
+
+        const options = {
+            chatId,
+            sessionId,
+            chatflowid,
+            logger,
+            analytic: chatflow.analytic,
+            appDataSource: appServer.AppDataSource,
+            databaseEntities: databaseEntities,
+            cachePool: appServer.cachePool,
+            uploads: incomingInput.uploads,
+            baseURL,
+            signal: new AbortController()
+        }
+
+        let streamResults
+        let finalResult = ''
+        let agentReasoning: IAgentReasoning[] = []
+
+        const workerNodes: IReactFlowNode[] = reactFlowNodes.filter((node: IReactFlowNode) => node.data.name === 'worker')
+        const supervisorNodes: IReactFlowNode[] = reactFlowNodes.filter((node: IReactFlowNode) => node.data.name === 'supervisor')
+
+        const mapNameToLabel: Record<string, string> = {}
+
+        for (const node of [...workerNodes, ...supervisorNodes]) {
+            mapNameToLabel[node.data.instance.name] = node.data.instance.label
+        }
+
+        try {
+            streamResults = await compileGraph(
+                chatflow,
+                mapNameToLabel,
+                reactFlowNodes,
+                endingNodeIds,
+                appServer.nodesPool.componentNodes,
+                options,
+                startingNodeIds,
+                incomingInput.question,
+                incomingInput?.overrideConfig
+            )
+
+            if (streamResults) {
+                let isStreamingStarted = false
+                for await (const output of await streamResults) {
+                    if (!output?.__end__) {
+                        const agentName = Object.keys(output)[0]
+                        const usedTools = output[agentName]?.messages
+                            ? output[agentName].messages.map((msg: any) => msg.additional_kwargs?.usedTools)
+                            : []
+                        const sourceDocuments = output[agentName]?.messages
+                            ? output[agentName].messages.map((msg: any) => msg.additional_kwargs?.sourceDocuments)
+                            : []
+                        const messages = output[agentName]?.messages ? output[agentName].messages.map((msg: any) => msg.content) : []
+                        const reasoning = {
+                            agentName: mapNameToLabel[agentName],
+                            messages,
+                            next: output[agentName]?.next,
+                            instructions: output[agentName]?.instructions,
+                            usedTools: flatten(usedTools),
+                            sourceDocuments: flatten(sourceDocuments)
+                        }
+                        agentReasoning.push(reasoning)
+                        if (socketIO && incomingInput.socketIOClientId) {
+                            if (!isStreamingStarted) {
+                                isStreamingStarted = true
+                                socketIO.to(incomingInput.socketIOClientId).emit('start', JSON.stringify(agentReasoning))
+                            }
+
+                            socketIO.to(incomingInput.socketIOClientId).emit('agentReasoning', JSON.stringify(agentReasoning))
+
+                            // Send loading next agent indicator
+                            if (reasoning.next && reasoning.next !== 'FINISH' && reasoning.next !== 'END') {
+                                socketIO
+                                    .to(incomingInput.socketIOClientId)
+                                    .emit('nextAgent', mapNameToLabel[reasoning.next] || reasoning.next)
+                            }
+                        }
+                    } else {
+                        finalResult = output.__end__.messages.length ? output.__end__.messages.pop()?.content : ''
+                        if (Array.isArray(finalResult)) finalResult = output.__end__.instructions
+
+                        if (finalResult === incomingInput.question) {
+                            const supervisorNode = reactFlowNodes.find((node: IReactFlowNode) => node.data.name === 'supervisor')
+                            const llm = supervisorNode?.data?.instance?.llm
+                            if (llm) {
+                                const res = await llm.invoke(incomingInput.question)
+                                finalResult = res?.content
+                            }
+                        }
+
+                        if (socketIO && incomingInput.socketIOClientId) {
+                            socketIO.to(incomingInput.socketIOClientId).emit('token', finalResult)
+                        }
+                    }
+                }
+
+                return { finalResult, agentReasoning }
+            }
+        } catch (e) {
+            if (socketIO && incomingInput.socketIOClientId) {
+                socketIO.to(incomingInput.socketIOClientId).emit('abort')
+            }
+            return { finalResult, agentReasoning }
+        }
+        return streamResults
+    } catch (e) {
+        logger.error('[server]: Error:', e)
+        throw new InternalFlowiseError(StatusCodes.INTERNAL_SERVER_ERROR, `Error buildAgentGraph - ${getErrorMessage(e)}`)
+    }
+}
+
+/**
+ * Compile Graph
+ * @param {IChatFlow} chatflow
+ * @param {Record<string, string>} mapNameToLabel
+ * @param {IReactFlowNode[]} reactflowNodes
+ * @param {string[]} workerNodeIds
+ * @param {IComponentNodes} componentNodes
+ * @param {ICommonObject} options
+ * @param {string[]} startingNodeIds
+ * @param {string} question
+ * @param {ICommonObject} overrideConfig
+ */
+const compileGraph = async (
+    chatflow: IChatFlow,
+    mapNameToLabel: Record<string, string>,
+    reactflowNodes: IReactFlowNode[] = [],
+    workerNodeIds: string[],
+    componentNodes: IComponentNodes,
+    options: ICommonObject,
+    startingNodeIds: string[],
+    question: string,
+    overrideConfig?: ICommonObject
+) => {
+    const appServer = getRunningExpressApp()
+    const channels: ITeamState = {
+        messages: {
+            value: (x: BaseMessage[], y: BaseMessage[]) => x.concat(y),
+            default: () => []
+        },
+        next: 'initialState',
+        instructions: "Solve the user's request.",
+        team_members: []
+    }
+
+    const workflowGraph = new StateGraph<ITeamState>({
+        //@ts-ignore
+        channels
+    })
+
+    const workerNodes = reactflowNodes.filter((node) => workerNodeIds.includes(node.data.id))
+
+    let supervisorWorkers: { [key: string]: IMultiAgentNode[] } = {}
+
+    // Init worker nodes
+    for (const workerNode of workerNodes) {
+        const nodeInstanceFilePath = componentNodes[workerNode.data.name].filePath as string
+        const nodeModule = await import(nodeInstanceFilePath)
+        const newNodeInstance = new nodeModule.nodeClass()
+
+        let flowNodeData = cloneDeep(workerNode.data)
+        if (overrideConfig) flowNodeData = replaceInputsWithConfig(flowNodeData, overrideConfig)
+        flowNodeData = resolveVariables(flowNodeData, reactflowNodes, question, [])
+
+        try {
+            const workerResult: IMultiAgentNode = await newNodeInstance.init(flowNodeData, question, options)
+            const parentSupervisor = workerResult.parentSupervisorName
+            if (!parentSupervisor || workerResult.type !== 'worker') continue
+            if (Object.prototype.hasOwnProperty.call(supervisorWorkers, parentSupervisor)) {
+                supervisorWorkers[parentSupervisor].push(workerResult)
+            } else {
+                supervisorWorkers[parentSupervisor] = [workerResult]
+            }
+
+            workflowGraph.addNode(workerResult.name, workerResult.node)
+        } catch (e) {
+            throw new InternalFlowiseError(StatusCodes.INTERNAL_SERVER_ERROR, `Error initialize worker nodes - ${getErrorMessage(e)}`)
+        }
+    }
+
+    // Init supervisor nodes
+    for (const supervisor in supervisorWorkers) {
+        const supervisorInputLabel = mapNameToLabel[supervisor]
+        const supervisorNode = reactflowNodes.find((node) => supervisorInputLabel === node.data.inputs?.supervisorName)
+        if (!supervisorNode) continue
+
+        const nodeInstanceFilePath = componentNodes[supervisorNode.data.name].filePath as string
+        const nodeModule = await import(nodeInstanceFilePath)
+        const newNodeInstance = new nodeModule.nodeClass()
+
+        let flowNodeData = cloneDeep(supervisorNode.data)
+
+        if (overrideConfig) flowNodeData = replaceInputsWithConfig(flowNodeData, overrideConfig)
+        flowNodeData = resolveVariables(flowNodeData, reactflowNodes, question, [])
+
+        if (flowNodeData.inputs) flowNodeData.inputs.workerNodes = supervisorWorkers[supervisor]
+
+        try {
+            const supervisorResult: IMultiAgentNode = await newNodeInstance.init(flowNodeData, question, options)
+            if (!supervisorResult.workers?.length) continue
+
+            if (supervisorResult.moderations && supervisorResult.moderations.length > 0) {
+                try {
+                    for (const moderation of supervisorResult.moderations) {
+                        question = await moderation.checkForViolations(question)
+                    }
+                } catch (e) {
+                    throw new InternalFlowiseError(StatusCodes.INTERNAL_SERVER_ERROR, getErrorMessage(e))
+                }
+            }
+
+            workflowGraph.addNode(supervisorResult.name, supervisorResult.node)
+
+            for (const worker of supervisorResult.workers) {
+                workflowGraph.addEdge(worker, supervisorResult.name)
+            }
+
+            let conditionalEdges: { [key: string]: string } = {}
+            for (let i = 0; i < supervisorResult.workers.length; i++) {
+                conditionalEdges[supervisorResult.workers[i]] = supervisorResult.workers[i]
+            }
+
+            workflowGraph.addConditionalEdges(supervisorResult.name, (x: ITeamState) => x.next, {
+                ...conditionalEdges,
+                FINISH: END
+            })
+
+            workflowGraph.setEntryPoint(supervisorResult.name)
+
+            // Add agentflow to pool
+            ;(workflowGraph as any).signal = options.signal
+            appServer.chatflowPool.add(
+                `${chatflow.id}_${options.chatId}`,
+                workflowGraph as any,
+                reactflowNodes.filter((node) => startingNodeIds.includes(node.id)),
+                overrideConfig
+            )
+
+            // TODO: add persistence
+            // const memory = new MemorySaver()
+            const graph = workflowGraph.compile()
+
+            const loggerHandler = new ConsoleCallbackHandler(logger)
+            const callbacks = await additionalCallbacks(flowNodeData, options)
+
+            // Return stream result as we should only have 1 supervisor
+            return await graph.stream(
+                {
+                    messages: [new HumanMessage({ content: question })]
+                },
+                { recursionLimit: supervisorResult?.recursionLimit ?? 100, callbacks: [loggerHandler, ...callbacks] }
+            )
+        } catch (e) {
+            throw new InternalFlowiseError(StatusCodes.INTERNAL_SERVER_ERROR, `Error initialize supervisor nodes - ${getErrorMessage(e)}`)
+        }
+    }
+}
@@ -1,7 +1,18 @@
 import { Request } from 'express'
 import { IFileUpload, convertSpeechToText, ICommonObject, addSingleFileToStorage, addArrayFilesToStorage } from 'flowise-components'
 import { StatusCodes } from 'http-status-codes'
-import { IncomingInput, IMessage, INodeData, IReactFlowObject, IReactFlowNode, IDepthQueue, chatType, IChatMessage } from '../Interface'
+import {
+    IncomingInput,
+    IMessage,
+    INodeData,
+    IReactFlowObject,
+    IReactFlowNode,
+    IDepthQueue,
+    chatType,
+    IChatMessage,
+    IChatFlow,
+    IReactFlowEdge
+} from '../Interface'
 import { InternalFlowiseError } from '../errors/internalFlowiseError'
 import { ChatFlow } from '../database/entities/ChatFlow'
 import { Server } from 'socket.io'
@@ -30,6 +41,8 @@ import { omit } from 'lodash'
 import * as fs from 'fs'
 import logger from './logger'
 import { utilAddChatMessage } from './addChatMesage'
+import { buildAgentGraph } from './buildAgentGraph'
+import { getErrorMessage } from '../errors/utils'

 /**
 * Build Chatflow
@@ -41,6 +54,8 @@ export const utilBuildChatflow = async (req: Request, socketIO?: Server, isInter
    try {
        const appServer = getRunningExpressApp()
        const chatflowid = req.params.id
+        const baseURL = `${req.protocol}://${req.get('host')}`
+
        let incomingInput: IncomingInput = req.body
        let nodeToExecuteData: INodeData
        const chatflow = await appServer.AppDataSource.getRepository(ChatFlow).findOneBy({
@@ -140,11 +155,34 @@ export const utilBuildChatflow = async (req: Request, socketIO?: Server, isInter
        const nodes = parsedFlowData.nodes
        const edges = parsedFlowData.edges

-        // Get session ID
+        /*** Get session ID ***/
        const memoryNode = findMemoryNode(nodes, edges)
        const memoryType = memoryNode?.data.label
        let sessionId = getMemorySessionId(memoryNode, incomingInput, chatId, isInternal)

+        /*** Get Ending Node with Directed Graph  ***/
+        const { graph, nodeDependencies } = constructGraphs(nodes, edges)
+        const directedGraph = graph
+        const endingNodes = getEndingNodes(nodeDependencies, directedGraph, nodes)
+
+        /*** If the graph is an agent graph, build the agent response ***/
+        if (endingNodes.filter((node) => node.data.category === 'Multi Agents').length) {
+            return await utilBuildAgentResponse(
+                chatflow,
+                isInternal,
+                chatId,
+                memoryType ?? '',
+                sessionId,
+                userMessageDateTime,
+                fileUploads,
+                incomingInput,
+                nodes,
+                edges,
+                socketIO,
+                baseURL
+            )
+        }
+
        // Get prepend messages
        const prependMessages = incomingInput.history

@@ -153,7 +191,6 @@ export const utilBuildChatflow = async (req: Request, socketIO?: Server, isInter
         * - Still in sync (i.e the flow has not been modified since)
         * - Existing overrideConfig and new overrideConfig are the same
         * - Flow doesn't start with/contain nodes that depend on incomingInput.question
-         * TODO: convert overrideConfig to hash when we no longer store base64 string but filepath
         ***/
        const isFlowReusable = () => {
            return (
@@ -176,13 +213,7 @@ export const utilBuildChatflow = async (req: Request, socketIO?: Server, isInter
                `[server]: Reuse existing chatflow ${chatflowid} with ending node ${nodeToExecuteData.label} (${nodeToExecuteData.id})`
            )
        } else {
-            /*** Get Ending Node with Directed Graph  ***/
-            const { graph, nodeDependencies } = constructGraphs(nodes, edges)
-            const directedGraph = graph
-
-            const endingNodes = getEndingNodes(nodeDependencies, directedGraph, nodes)
-
-            let isCustomFunctionEndingNode = endingNodes.some((node) => node.data?.outputs?.output === 'EndingNode')
+            const isCustomFunctionEndingNode = endingNodes.some((node) => node.data?.outputs?.output === 'EndingNode')

            for (const endingNode of endingNodes) {
                const endingNodeData = endingNode.data
@@ -268,7 +299,10 @@ export const utilBuildChatflow = async (req: Request, socketIO?: Server, isInter
                appServer.cachePool,
                false,
                undefined,
-                incomingInput.uploads
+                incomingInput.uploads,
+                baseURL,
+                socketIO,
+                incomingInput.socketIOClientId
            )

            const nodeToExecute =
@@ -372,13 +406,92 @@ export const utilBuildChatflow = async (req: Request, socketIO?: Server, isInter
        // this is used when input text is empty but question is in audio format
        result.question = incomingInput.question
        result.chatId = chatId
-        result.chatMessageId = chatMessage.id
+        result.chatMessageId = chatMessage?.id
        if (sessionId) result.sessionId = sessionId
        if (memoryType) result.memoryType = memoryType

        return result
-    } catch (e: any) {
+    } catch (e) {
        logger.error('[server]: Error:', e)
-        throw new InternalFlowiseError(StatusCodes.INTERNAL_SERVER_ERROR, e.message)
+        throw new InternalFlowiseError(StatusCodes.INTERNAL_SERVER_ERROR, getErrorMessage(e))
+    }
+}
+
+const utilBuildAgentResponse = async (
+    chatflow: IChatFlow,
+    isInternal: boolean,
+    chatId: string,
+    memoryType: string,
+    sessionId: string,
+    userMessageDateTime: Date,
+    fileUploads: IFileUpload[],
+    incomingInput: ICommonObject,
+    nodes: IReactFlowNode[],
+    edges: IReactFlowEdge[],
+    socketIO?: Server,
+    baseURL?: string
+) => {
+    try {
+        const appServer = getRunningExpressApp()
+        const streamResults = await buildAgentGraph(chatflow, chatId, sessionId, incomingInput, baseURL, socketIO)
+        if (streamResults) {
+            const { finalResult, agentReasoning } = streamResults
+            const userMessage: Omit<IChatMessage, 'id'> = {
+                role: 'userMessage',
+                content: incomingInput.question,
+                chatflowid: chatflow.id,
+                chatType: isInternal ? chatType.INTERNAL : chatType.EXTERNAL,
+                chatId,
+                memoryType,
+                sessionId,
+                createdDate: userMessageDateTime,
+                fileUploads: incomingInput.uploads ? JSON.stringify(fileUploads) : undefined,
+                leadEmail: incomingInput.leadEmail
+            }
+            await utilAddChatMessage(userMessage)
+
+            const apiMessage: Omit<IChatMessage, 'id' | 'createdDate'> = {
+                role: 'apiMessage',
+                content: finalResult,
+                chatflowid: chatflow.id,
+                chatType: isInternal ? chatType.INTERNAL : chatType.EXTERNAL,
+                chatId,
+                memoryType,
+                sessionId
+            }
+            if (agentReasoning.length) apiMessage.agentReasoning = JSON.stringify(agentReasoning)
+            const chatMessage = await utilAddChatMessage(apiMessage)
+
+            await appServer.telemetry.sendTelemetry('prediction_sent', {
+                version: await getAppVersion(),
+                chatlowId: chatflow.id,
+                chatId,
+                type: isInternal ? chatType.INTERNAL : chatType.EXTERNAL,
+                flowGraph: getTelemetryFlowObj(nodes, edges)
+            })
+
+            // Prepare response
+            let result: ICommonObject = {}
+            result.text = finalResult
+            result.question = incomingInput.question
+            result.chatId = chatId
+            result.chatMessageId = chatMessage?.id
+            if (sessionId) result.sessionId = sessionId
+            if (memoryType) result.memoryType = memoryType
+            if (agentReasoning.length) result.agentReasoning = agentReasoning
+
+            await appServer.telemetry.sendTelemetry('graph_compiled', {
+                version: await getAppVersion(),
+                graphId: chatflow.id,
+                type: isInternal ? chatType.INTERNAL : chatType.EXTERNAL,
+                flowGraph: getTelemetryFlowObj(nodes, edges)
+            })
+
+            return result
+        }
+        return undefined
+    } catch (e) {
+        logger.error('[server]: Error:', e)
+        throw new InternalFlowiseError(StatusCodes.INTERNAL_SERVER_ERROR, getErrorMessage(e))
    }
 }
@@ -18,7 +18,7 @@ export const utilGetUploadsConfig = async (chatflowid: string): Promise<any> =>
        throw new InternalFlowiseError(StatusCodes.NOT_FOUND, `Chatflow ${chatflowid} not found`)
    }

-    const uploadAllowedNodes = ['llmChain', 'conversationChain', 'reactAgentChat', 'conversationalAgent', 'toolAgent']
+    const uploadAllowedNodes = ['llmChain', 'conversationChain', 'reactAgentChat', 'conversationalAgent', 'toolAgent', 'supervisor']
    const uploadProcessingNodes = ['chatOpenAI', 'chatAnthropic', 'awsChatBedrock', 'azureChatOpenAI', 'chatGoogleGenerativeAI']

    const flowObj = JSON.parse(chatflow.flowData)
@@ -1,6 +1,7 @@
 import path from 'path'
 import fs from 'fs'
 import logger from './logger'
+import { Server } from 'socket.io'
 import {
    IComponentCredentials,
    IComponentNodes,
@@ -267,9 +268,10 @@ export const getEndingNodes = (
                endingNodeData &&
                endingNodeData.category !== 'Chains' &&
                endingNodeData.category !== 'Agents' &&
-                endingNodeData.category !== 'Engine'
+                endingNodeData.category !== 'Engine' &&
+                endingNodeData.category !== 'Multi Agents'
            ) {
-                error = new InternalFlowiseError(StatusCodes.INTERNAL_SERVER_ERROR, `Ending node must be either a Chain or Agent`)
+                error = new InternalFlowiseError(StatusCodes.INTERNAL_SERVER_ERROR, `Ending node must be either a Chain or Agent or Engine`)
                continue
            }
        }
@@ -443,7 +445,10 @@ export const buildFlow = async (
    cachePool?: CachePool,
    isUpsert?: boolean,
    stopNodeId?: string,
-    uploads?: IFileUpload[]
+    uploads?: IFileUpload[],
+    baseURL?: string,
+    socketIO?: Server,
+    socketIOClientId?: string
 ) => {
    const flowNodes = cloneDeep(reactFlowNodes)

@@ -496,7 +501,10 @@ export const buildFlow = async (
                    databaseEntities,
                    cachePool,
                    dynamicVariables,
-                    uploads
+                    uploads,
+                    baseURL,
+                    socketIO,
+                    socketIOClientId
                })
                if (indexResult) upsertHistory['result'] = indexResult
                logger.debug(`[server]: Finished upserting ${reactFlowNode.data.label} (${reactFlowNode.data.id})`)
@@ -520,7 +528,10 @@ export const buildFlow = async (
                    cachePool,
                    isUpsert,
                    dynamicVariables,
-                    uploads
+                    uploads,
+                    baseURL,
+                    socketIO,
+                    socketIOClientId
                })

                // Save dynamic variables
@@ -1048,7 +1059,6 @@ export const findAvailableConfigs = (reactFlowNodes: IReactFlowNode[], component
            }
        }
    }
-
    return configs
 }