Feature/add ability to upload file from chat (#3059)

add ability to upload file from chat

packages/server/src/ChatflowPool.ts

@@ -16,13 +16,21 @@ export class ChatflowPool {
      * @param {IReactFlowNode[]} startingNodes
      * @param {ICommonObject} overrideConfig
      */
-    add(chatflowid: string, endingNodeData: INodeData | undefined, startingNodes: IReactFlowNode[], overrideConfig?: ICommonObject) {
+    add(
+        chatflowid: string,
+        endingNodeData: INodeData | undefined,
+        startingNodes: IReactFlowNode[],
+        overrideConfig?: ICommonObject,
+        chatId?: string
+    ) {
         this.activeChatflows[chatflowid] = {
             startingNodes,
             endingNodeData,
             inSync: true
         }
         if (overrideConfig) this.activeChatflows[chatflowid].overrideConfig = overrideConfig
+        if (chatId) this.activeChatflows[chatflowid].chatId = chatId
+
         logger.info(`[server]: Chatflow ${chatflowid} added into ChatflowPool`)
     }
 

packages/server/src/Interface.ts

@@ -231,6 +231,7 @@ export interface IActiveChatflows {
         endingNodeData?: INodeData
         inSync: boolean
         overrideConfig?: ICommonObject
+        chatId?: string
     }
 }
 

packages/server/src/routes/vectors/index.ts

@@ -14,6 +14,6 @@ router.post(
     vectorsController.getRateLimiterMiddleware,
     vectorsController.upsertVectorMiddleware
 )
-router.post(['/internal-upsert/', '/internal-upsert/:id'], vectorsController.createInternalUpsert)
+router.post(['/internal-upsert/', '/internal-upsert/:id'], upload.array('files'), vectorsController.createInternalUpsert)
 
 export default router

packages/server/src/utils/buildChatflow.ts

@@ -1,5 +1,12 @@
 import { Request } from 'express'
-import { IFileUpload, convertSpeechToText, ICommonObject, addSingleFileToStorage, addArrayFilesToStorage } from 'flowise-components'
+import {
+    IFileUpload,
+    convertSpeechToText,
+    ICommonObject,
+    addSingleFileToStorage,
+    addArrayFilesToStorage,
+    mapMimeTypeToInputField
+} from 'flowise-components'
 import { StatusCodes } from 'http-status-codes'
 import {
     IncomingInput,
@@ -18,7 +25,6 @@ import { ChatFlow } from '../database/entities/ChatFlow'
 import { Server } from 'socket.io'
 import { getRunningExpressApp } from '../utils/getRunningExpressApp'
 import {
-    mapMimeTypeToInputField,
     isFlowValidForStream,
     buildFlow,
     getTelemetryFlowObj,
@@ -32,7 +38,8 @@ import {
     getMemorySessionId,
     isSameOverrideConfig,
     getEndingNodes,
-    constructGraphs
+    constructGraphs,
+    isSameChatId
 } from '../utils'
 import { validateChatflowAPIKey } from './validateKey'
 import { databaseEntities } from '.'
@@ -201,6 +208,7 @@ export const utilBuildChatflow = async (req: Request, socketIO?: Server, isInter
          * - Node Data already exists in pool
          * - Still in sync (i.e the flow has not been modified since)
          * - Existing overrideConfig and new overrideConfig are the same
+         * - Existing chatId and new chatId is the same
          * - Flow doesn't start with/contain nodes that depend on incomingInput.question
          ***/
         const isFlowReusable = () => {
@@ -209,6 +217,7 @@ export const utilBuildChatflow = async (req: Request, socketIO?: Server, isInter
                 Object.prototype.hasOwnProperty.call(appServer.chatflowPool.activeChatflows, chatflowid) &&
                 appServer.chatflowPool.activeChatflows[chatflowid].inSync &&
                 appServer.chatflowPool.activeChatflows[chatflowid].endingNodeData &&
+                isSameChatId(appServer.chatflowPool.activeChatflows[chatflowid].chatId, chatId) &&
                 isSameOverrideConfig(
                     isInternal,
                     appServer.chatflowPool.activeChatflows[chatflowid].overrideConfig,
@@ -338,7 +347,7 @@ export const utilBuildChatflow = async (req: Request, socketIO?: Server, isInter
             )
             nodeToExecuteData = reactFlowNodeData
 
-            appServer.chatflowPool.add(chatflowid, nodeToExecuteData, startingNodes, incomingInput?.overrideConfig)
+            appServer.chatflowPool.add(chatflowid, nodeToExecuteData, startingNodes, incomingInput?.overrideConfig, chatId)
         }
 
         logger.debug(`[server]: Running ${nodeToExecuteData.label} (${nodeToExecuteData.id})`)

packages/server/src/utils/getUploadsConfig.ts

@@ -2,14 +2,22 @@ import { StatusCodes } from 'http-status-codes'
 import { INodeParams } from 'flowise-components'
 import { ChatFlow } from '../database/entities/ChatFlow'
 import { getRunningExpressApp } from '../utils/getRunningExpressApp'
-import { IUploadFileSizeAndTypes, IReactFlowNode } from '../Interface'
+import { IUploadFileSizeAndTypes, IReactFlowNode, IReactFlowEdge } from '../Interface'
 import { InternalFlowiseError } from '../errors/internalFlowiseError'
 
+type IUploadConfig = {
+    isSpeechToTextEnabled: boolean
+    isImageUploadAllowed: boolean
+    isFileUploadAllowed: boolean
+    imgUploadSizeAndTypes: IUploadFileSizeAndTypes[]
+    fileUploadSizeAndTypes: IUploadFileSizeAndTypes[]
+}
+
 /**
  * Method that checks if uploads are enabled in the chatflow
  * @param {string} chatflowid
  */
-export const utilGetUploadsConfig = async (chatflowid: string): Promise<any> => {
+export const utilGetUploadsConfig = async (chatflowid: string): Promise<IUploadConfig> => {
     const appServer = getRunningExpressApp()
     const chatflow = await appServer.AppDataSource.getRepository(ChatFlow).findOneBy({
         id: chatflowid
@@ -18,21 +26,17 @@ 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',
-        'supervisor',
-        'seqStart'
-    ]
-    const uploadProcessingNodes = ['chatOpenAI', 'chatAnthropic', 'awsChatBedrock', 'azureChatOpenAI', 'chatGoogleGenerativeAI']
-
     const flowObj = JSON.parse(chatflow.flowData)
-    const imgUploadSizeAndTypes: IUploadFileSizeAndTypes[] = []
+    const nodes: IReactFlowNode[] = flowObj.nodes
+    const edges: IReactFlowEdge[] = flowObj.edges
 
     let isSpeechToTextEnabled = false
+    let isImageUploadAllowed = false
+    let isFileUploadAllowed = false
+
+    /*
+     * Check for STT
+     */
     if (chatflow.speechToText) {
         const speechToTextProviders = JSON.parse(chatflow.speechToText)
         for (const provider in speechToTextProviders) {
@@ -46,39 +50,72 @@ export const utilGetUploadsConfig = async (chatflowid: string): Promise<any> =>
         }
     }
 
-    let isImageUploadAllowed = false
-    const nodes: IReactFlowNode[] = flowObj.nodes
+    /*
+     * Condition for isFileUploadAllowed
+     * 1.) vector store with fileUpload = true && connected to a document loader with fileType
+     */
+    const fileUploadSizeAndTypes: IUploadFileSizeAndTypes[] = []
+    for (const node of nodes) {
+        if (node.data.category === 'Vector Stores' && node.data.inputs?.fileUpload) {
+            // Get the connected document loader node fileTypes
+            const sourceDocumentEdges = edges.filter(
+                (edge) => edge.target === node.id && edge.targetHandle === `${node.id}-input-document-Document`
+            )
+            for (const edge of sourceDocumentEdges) {
+                const sourceNode = nodes.find((node) => node.id === edge.source)
+                if (!sourceNode) continue
+                const fileType = sourceNode.data.inputParams.find((param) => param.type === 'file' && param.fileType)?.fileType
+                if (fileType) {
+                    fileUploadSizeAndTypes.push({
+                        fileTypes: fileType.split(', '),
+                        maxUploadSize: 500
+                    })
+                    isFileUploadAllowed = true
+                }
+            }
+            break
+        }
+    }
 
     /*
      * Condition for isImageUploadAllowed
-     * 1.) one of the uploadAllowedNodes exists
-     * 2.) one of the uploadProcessingNodes exists + allowImageUploads is ON
+     * 1.) one of the imgUploadAllowedNodes exists
+     * 2.) one of the imgUploadLLMNodes exists + allowImageUploads is ON
      */
-    if (!nodes.some((node) => uploadAllowedNodes.includes(node.data.name))) {
-        return {
-            isSpeechToTextEnabled,
-            isImageUploadAllowed: false,
-            imgUploadSizeAndTypes
-        }
+    const imgUploadSizeAndTypes: IUploadFileSizeAndTypes[] = []
+    const imgUploadAllowedNodes = [
+        'llmChain',
+        'conversationChain',
+        'reactAgentChat',
+        'conversationalAgent',
+        'toolAgent',
+        'supervisor',
+        'seqStart'
+    ]
+    const imgUploadLLMNodes = ['chatOpenAI', 'chatAnthropic', 'awsChatBedrock', 'azureChatOpenAI', 'chatGoogleGenerativeAI']
+
+    if (nodes.some((node) => imgUploadAllowedNodes.includes(node.data.name))) {
+        nodes.forEach((node: IReactFlowNode) => {
+            if (imgUploadLLMNodes.indexOf(node.data.name) > -1) {
+                // TODO: for now the maxUploadSize is hardcoded to 5MB, we need to add it to the node properties
+                node.data.inputParams.map((param: INodeParams) => {
+                    if (param.name === 'allowImageUploads' && node.data.inputs?.['allowImageUploads']) {
+                        imgUploadSizeAndTypes.push({
+                            fileTypes: 'image/gif;image/jpeg;image/png;image/webp;'.split(';'),
+                            maxUploadSize: 5
+                        })
+                        isImageUploadAllowed = true
+                    }
+                })
+            }
+        })
     }
 
-    nodes.forEach((node: IReactFlowNode) => {
-        if (uploadProcessingNodes.indexOf(node.data.name) > -1) {
-            // TODO: for now the maxUploadSize is hardcoded to 5MB, we need to add it to the node properties
-            node.data.inputParams.map((param: INodeParams) => {
-                if (param.name === 'allowImageUploads' && node.data.inputs?.['allowImageUploads']) {
-                    imgUploadSizeAndTypes.push({
-                        fileTypes: 'image/gif;image/jpeg;image/png;image/webp;'.split(';'),
-                        maxUploadSize: 5
-                    })
-                    isImageUploadAllowed = true
-                }
-            })
-        }
-    })
     return {
         isSpeechToTextEnabled,
         isImageUploadAllowed,
-        imgUploadSizeAndTypes
+        isFileUploadAllowed,
+        imgUploadSizeAndTypes,
+        fileUploadSizeAndTypes
     }
 }

packages/server/src/utils/index.ts

@@ -1074,35 +1074,16 @@ export const isSameOverrideConfig = (
 }
 
 /**
- * Map MimeType to InputField
- * @param {string} mimeType
- * @returns {Promise<string>}
+ * @param {string} existingChatId
+ * @param {string} newChatId
+ * @returns {boolean}
  */
-export const mapMimeTypeToInputField = (mimeType: string) => {
-    switch (mimeType) {
-        case 'text/plain':
-            return 'txtFile'
-        case 'application/pdf':
-            return 'pdfFile'
-        case 'application/json':
-            return 'jsonFile'
-        case 'text/csv':
-            return 'csvFile'
-        case 'application/json-lines':
-        case 'application/jsonl':
-        case 'text/jsonl':
-            return 'jsonlinesFile'
-        case 'application/vnd.openxmlformats-officedocument.wordprocessingml.document':
-            return 'docxFile'
-        case 'application/vnd.yaml':
-        case 'application/x-yaml':
-        case 'text/vnd.yaml':
-        case 'text/x-yaml':
-        case 'text/yaml':
-            return 'yamlFile'
-        default:
-            return 'txtFile'
+export const isSameChatId = (existingChatId?: string, newChatId?: string): boolean => {
+    if (isEqual(existingChatId, newChatId)) {
+        return true
     }
+    if (!existingChatId && !newChatId) return true
+    return false
 }
 
 /**

packages/server/src/utils/upsertVector.ts

@@ -1,14 +1,13 @@
 import { Request } from 'express'
 import * as fs from 'fs'
 import { cloneDeep, omit } from 'lodash'
-import { ICommonObject, IMessage, addArrayFilesToStorage } from 'flowise-components'
+import { ICommonObject, IMessage, addArrayFilesToStorage, mapMimeTypeToInputField } from 'flowise-components'
 import telemetryService from '../services/telemetry'
 import logger from '../utils/logger'
 import {
     buildFlow,
     constructGraphs,
     getAllConnectedNodes,
-    mapMimeTypeToInputField,
     findMemoryNode,
     getMemorySessionId,
     getAppVersion,
@@ -70,6 +69,9 @@ export const upsertVector = async (req: Request, isInternal: boolean = false) =>
                 overrideConfig,
                 stopNodeId: req.body.stopNodeId
             }
+            if (req.body.chatId) {
+                incomingInput.chatId = req.body.chatId
+            }
         }
 
         /*** Get chatflows and prepare data  ***/
@@ -87,10 +89,15 @@ export const upsertVector = async (req: Request, isInternal: boolean = false) =>
         const memoryNode = findMemoryNode(nodes, edges)
         let sessionId = getMemorySessionId(memoryNode, incomingInput, chatId, isInternal)
 
-        const vsNodes = nodes.filter(
-            (node) =>
-                node.data.category === 'Vector Stores' && !node.data.label.includes('Upsert') && !node.data.label.includes('Load Existing')
-        )
+        const vsNodes = nodes.filter((node) => node.data.category === 'Vector Stores')
+
+        // Get StopNodeId for vector store which has fielUpload
+        const vsNodesWithFileUpload = vsNodes.filter((node) => node.data.inputs?.fileUpload)
+        if (vsNodesWithFileUpload.length > 1) {
+            throw new InternalFlowiseError(StatusCodes.INTERNAL_SERVER_ERROR, 'Multiple vector store nodes with fileUpload enabled')
+        } else if (vsNodesWithFileUpload.length === 1 && !stopNodeId) {
+            stopNodeId = vsNodesWithFileUpload[0].data.id
+        }
 
         // Check if multiple vector store nodes exist, and if stopNodeId is specified
         if (vsNodes.length > 1 && !stopNodeId) {
@@ -138,7 +145,7 @@ export const upsertVector = async (req: Request, isInternal: boolean = false) =>
 
         const startingNodes = nodes.filter((nd) => startingNodeIds.includes(nd.data.id))
 
-        await appServer.chatflowPool.add(chatflowid, undefined, startingNodes, incomingInput?.overrideConfig)
+        await appServer.chatflowPool.add(chatflowid, undefined, startingNodes, incomingInput?.overrideConfig, chatId)
 
         // Save to DB
         if (upsertedResult['flowData'] && upsertedResult['result']) {