Chore/refractor (#4454)

* markdown files and env examples cleanup * components update * update jsonlines description * server refractor * update telemetry * add execute custom node * add ui refractor * add username and password authenticate * correctly retrieve past images in agentflowv2 * disable e2e temporarily * add existing username and password authenticate * update migration to default workspace * update todo * blob storage migrating * throw error on agent tool call error * add missing execution import * add referral * chore: add error message when importData is undefined * migrate api keys to db * fix: data too long for column executionData * migrate api keys from json to db at init * add info on account setup * update docstore missing fields --------- Co-authored-by: chungyau97 <chungyau97@gmail.com>
2026-06-28 17:01:00 +03:00 · 2025-05-27 14:29:42 +08:00
parent e35a126b46
commit 5a37227d14
560 changed files with 62127 additions and 4100 deletions
@@ -15,6 +15,7 @@ import {
    mapExtToInputField,
    getFileFromUpload,
    removeSpecificFileFromUpload,
+    EvaluationRunner,
    handleEscapeCharacters
 } from 'flowise-components'
 import { StatusCodes } from 'http-status-codes'
@@ -22,8 +23,8 @@ import {
    IncomingInput,
    IMessage,
    INodeData,
-    IReactFlowObject,
    IReactFlowNode,
+    IReactFlowObject,
    IDepthQueue,
    ChatType,
    IChatMessage,
@@ -59,11 +60,15 @@ import {
 import { validateChatflowAPIKey } from './validateKey'
 import logger from './logger'
 import { utilAddChatMessage } from './addChatMesage'
+import { checkPredictions, checkStorage, updatePredictionsUsage, updateStorageUsage } from './quotaUsage'
 import { buildAgentGraph } from './buildAgentGraph'
 import { getErrorMessage } from '../errors/utils'
 import { FLOWISE_METRIC_COUNTERS, FLOWISE_COUNTER_STATUS, IMetricsProvider } from '../Interface.Metrics'
+import { getWorkspaceSearchOptions } from '../enterprise/utils/ControllerServiceUtils'
 import { OMIT_QUEUE_JOB_DATA } from './constants'
 import { executeAgentFlow } from './buildAgentflow'
+import { Workspace } from '../enterprise/database/entities/workspace.entity'
+import { Organization } from '../enterprise/database/entities/organization.entity'

 /*
 * Initialize the ending node to be executed
@@ -230,15 +235,21 @@ export const executeFlow = async ({
    incomingInput,
    chatflow,
    chatId,
+    isEvaluation,
+    evaluationRunId,
    appDataSource,
    telemetry,
    cachePool,
+    usageCacheManager,
    sseStreamer,
    baseURL,
    isInternal,
    files,
    signal,
-    isTool
+    isTool,
+    orgId,
+    workspaceId,
+    subscriptionId
 }: IExecuteFlowParams) => {
    // Ensure incomingInput has all required properties with default values
    incomingInput = {
@@ -265,6 +276,8 @@ export const executeFlow = async ({
    if (uploads) {
        fileUploads = uploads
        for (let i = 0; i < fileUploads.length; i += 1) {
+            await checkStorage(orgId, subscriptionId, usageCacheManager)
+
            const upload = fileUploads[i]

            // if upload in an image, a rag file, or audio
@@ -273,7 +286,8 @@ export const executeFlow = async ({
                const splitDataURI = upload.data.split(',')
                const bf = Buffer.from(splitDataURI.pop() || '', 'base64')
                const mime = splitDataURI[0].split(':')[1].split(';')[0]
-                await addSingleFileToStorage(mime, bf, filename, chatflowid, chatId)
+                const { totalSize } = await addSingleFileToStorage(mime, bf, filename, orgId, chatflowid, chatId)
+                await updateStorageUsage(orgId, workspaceId, totalSize, usageCacheManager)
                upload.type = 'stored-file'
                // Omit upload.data since we don't store the content in database
                fileUploads[i] = omit(upload, ['data'])
@@ -287,7 +301,7 @@ export const executeFlow = async ({

            // Run Speech to Text conversion
            if (upload.mime === 'audio/webm' || upload.mime === 'audio/mp4' || upload.mime === 'audio/ogg') {
-                logger.debug(`Attempting a speech to text conversion...`)
+                logger.debug(`[server]: [${orgId}]: Attempting a speech to text conversion...`)
                let speechToTextConfig: ICommonObject = {}
                if (chatflow.speechToText) {
                    const speechToTextProviders = JSON.parse(chatflow.speechToText)
@@ -302,13 +316,14 @@ export const executeFlow = async ({
                }
                if (speechToTextConfig) {
                    const options: ICommonObject = {
+                        orgId,
                        chatId,
                        chatflowid,
                        appDataSource,
                        databaseEntities: databaseEntities
                    }
                    const speechToTextResult = await convertSpeechToText(upload, speechToTextConfig, options)
-                    logger.debug(`Speech to text result: ${speechToTextResult}`)
+                    logger.debug(`[server]: [${orgId}]: Speech to text result: ${speechToTextResult}`)
                    if (speechToTextResult) {
                        incomingInput.question = speechToTextResult
                        question = speechToTextResult
@@ -329,11 +344,21 @@ export const executeFlow = async ({
    if (files?.length) {
        overrideConfig = { ...incomingInput }
        for (const file of files) {
+            await checkStorage(orgId, subscriptionId, usageCacheManager)
+
            const fileNames: string[] = []
            const fileBuffer = await getFileFromUpload(file.path ?? file.key)
            // Address file name with special characters: https://github.com/expressjs/multer/issues/1104
            file.originalname = Buffer.from(file.originalname, 'latin1').toString('utf8')
-            const storagePath = await addArrayFilesToStorage(file.mimetype, fileBuffer, file.originalname, fileNames, chatflowid)
+            const { path: storagePath, totalSize } = await addArrayFilesToStorage(
+                file.mimetype,
+                fileBuffer,
+                file.originalname,
+                fileNames,
+                orgId,
+                chatflowid
+            )
+            await updateStorageUsage(orgId, workspaceId, totalSize, usageCacheManager)

            const fileInputFieldFromMimeType = mapMimeTypeToInputField(file.mimetype)

@@ -382,16 +407,21 @@ export const executeFlow = async ({
            incomingInput,
            chatflow,
            chatId,
+            evaluationRunId,
            appDataSource,
            telemetry,
            cachePool,
+            usageCacheManager,
            sseStreamer,
            baseURL,
            isInternal,
            uploadedFilesContent,
            fileUploads,
            signal,
-            isTool
+            isTool,
+            orgId,
+            workspaceId,
+            subscriptionId
        })
    }

@@ -443,7 +473,7 @@ export const executeFlow = async ({
    })

    /*** Get API Config ***/
-    const availableVariables = await appDataSource.getRepository(Variable).find()
+    const availableVariables = await appDataSource.getRepository(Variable).findBy(getWorkspaceSearchOptions(workspaceId))
    const { nodeOverrides, variableOverrides, apiOverrideStatus } = getAPIOverrideConfig(chatflow)

    const flowConfig: IFlowConfig = {
@@ -455,7 +485,7 @@ export const executeFlow = async ({
        ...incomingInput.overrideConfig
    }

-    logger.debug(`[server]: Start building flow ${chatflowid}`)
+    logger.debug(`[server]: [${orgId}]: Start building flow ${chatflowid}`)

    /*** BFS to traverse from Starting Nodes to Ending Node ***/
    const reactFlowNodes = await buildFlow({
@@ -479,9 +509,13 @@ export const executeFlow = async ({
        availableVariables,
        variableOverrides,
        cachePool,
+        usageCacheManager,
        isUpsert: false,
        uploads,
-        baseURL
+        baseURL,
+        orgId,
+        workspaceId,
+        subscriptionId
    })

    const setVariableNodesOutput = getSetVariableNodesOutput(reactFlowNodes)
@@ -506,7 +540,9 @@ export const executeFlow = async ({
            shouldStreamResponse: true, // agentflow is always streamed
            cachePool,
            baseURL,
-            signal
+            signal,
+            orgId,
+            workspaceId
        })

        if (streamResults) {
@@ -556,13 +592,17 @@ export const executeFlow = async ({
            }
            const chatMessage = await utilAddChatMessage(apiMessage, appDataSource)

-            await telemetry.sendTelemetry('agentflow_prediction_sent', {
-                version: await getAppVersion(),
-                agentflowId: agentflow.id,
-                chatId,
-                type: isInternal ? ChatType.INTERNAL : ChatType.EXTERNAL,
-                flowGraph: getTelemetryFlowObj(nodes, edges)
-            })
+            await telemetry.sendTelemetry(
+                'agentflow_prediction_sent',
+                {
+                    version: await getAppVersion(),
+                    agentflowId: agentflow.id,
+                    chatId,
+                    type: isInternal ? ChatType.INTERNAL : ChatType.EXTERNAL,
+                    flowGraph: getTelemetryFlowObj(nodes, edges)
+                },
+                orgId
+            )

            // Find the previous chat message with the same action id and remove the action
            if (incomingInput.action && Object.keys(incomingInput.action).length) {
@@ -596,6 +636,7 @@ export const executeFlow = async ({
            // Prepare response
            let result: ICommonObject = {}
            result.text = finalResult
+
            result.question = incomingInput.question
            result.chatId = chatId
            result.chatMessageId = chatMessage?.id
@@ -605,7 +646,6 @@ export const executeFlow = async ({
            if (finalAction && Object.keys(finalAction).length) result.action = finalAction
            if (Object.keys(setVariableNodesOutput).length) result.flowVariables = setVariableNodesOutput
            result.followUpPrompts = JSON.stringify(apiMessage.followUpPrompts)
-
            return result
        }
        return undefined
@@ -643,16 +683,23 @@ export const executeFlow = async ({

        /*** Prepare run params ***/
        const runParams = {
+            orgId,
+            workspaceId,
+            subscriptionId,
            chatId,
            chatflowid,
            apiMessageId,
            logger,
            appDataSource,
            databaseEntities,
+            usageCacheManager,
            analytic: chatflow.analytic,
            uploads,
            prependMessages,
-            ...(isStreamValid && { sseStreamer, shouldStreamResponse: isStreamValid })
+            ...(isStreamValid && { sseStreamer, shouldStreamResponse: isStreamValid }),
+            evaluationRunId,
+            updateStorageUsage,
+            checkStorage
        }

        /*** Run the ending node ***/
@@ -669,7 +716,7 @@ export const executeFlow = async ({
            role: 'userMessage',
            content: question,
            chatflowid,
-            chatType: isInternal ? ChatType.INTERNAL : ChatType.EXTERNAL,
+            chatType: isEvaluation ? ChatType.EVALUATION : isInternal ? ChatType.INTERNAL : ChatType.EXTERNAL,
            chatId,
            memoryType,
            sessionId,
@@ -725,7 +772,7 @@ export const executeFlow = async ({
            role: 'apiMessage',
            content: resultText,
            chatflowid,
-            chatType: isInternal ? ChatType.INTERNAL : ChatType.EXTERNAL,
+            chatType: isEvaluation ? ChatType.EVALUATION : isInternal ? ChatType.INTERNAL : ChatType.EXTERNAL,
            chatId,
            memoryType,
            sessionId
@@ -749,15 +796,22 @@ export const executeFlow = async ({

        const chatMessage = await utilAddChatMessage(apiMessage, appDataSource)

-        logger.debug(`[server]: Finished running ${endingNodeData.label} (${endingNodeData.id})`)
-
-        await telemetry.sendTelemetry('prediction_sent', {
-            version: await getAppVersion(),
-            chatflowId: chatflowid,
-            chatId,
-            type: isInternal ? ChatType.INTERNAL : ChatType.EXTERNAL,
-            flowGraph: getTelemetryFlowObj(nodes, edges)
-        })
+        logger.debug(`[server]: [${orgId}]: Finished running ${endingNodeData.label} (${endingNodeData.id})`)
+        if (evaluationRunId) {
+            const metrics = await EvaluationRunner.getAndDeleteMetrics(evaluationRunId)
+            result.metrics = metrics
+        }
+        await telemetry.sendTelemetry(
+            'prediction_sent',
+            {
+                version: await getAppVersion(),
+                chatflowId: chatflowid,
+                chatId,
+                type: isInternal ? ChatType.INTERNAL : ChatType.EXTERNAL,
+                flowGraph: getTelemetryFlowObj(nodes, edges)
+            },
+            orgId
+        )

        /*** Prepare response ***/
        result.question = incomingInput.question // return the question in the response, this is used when input text is empty but question is in audio format
@@ -830,6 +884,7 @@ const checkIfStreamValid = async (
 */
 export const utilBuildChatflow = async (req: Request, isInternal: boolean = false): Promise<any> => {
    const appServer = getRunningExpressApp()
+
    const chatflowid = req.params.id

    // Check if chatflow exists
@@ -841,7 +896,6 @@ export const utilBuildChatflow = async (req: Request, isInternal: boolean = fals
    }

    const isAgentFlow = chatflow.type === 'MULTIAGENT'
-
    const httpProtocol = req.get('x-forwarded-proto') || req.protocol
    const baseURL = `${httpProtocol}://${req.get('host')}`
    const incomingInput: IncomingInput = req.body || {} // Ensure incomingInput is never undefined
@@ -849,6 +903,20 @@ export const utilBuildChatflow = async (req: Request, isInternal: boolean = fals
    const files = (req.files as Express.Multer.File[]) || []
    const abortControllerId = `${chatflow.id}_${chatId}`
    const isTool = req.get('flowise-tool') === 'true'
+    const isEvaluation: boolean = req.headers['X-Flowise-Evaluation'] || req.body.evaluation
+    let evaluationRunId = ''
+    if (isEvaluation) {
+        evaluationRunId = req.body.evaluationRunId
+        if (evaluationRunId) {
+            const newEval = {
+                evaluation: {
+                    status: true,
+                    evaluationRunId
+                }
+            }
+            chatflow.analytic = JSON.stringify(newEval)
+        }
+    }

    try {
        // Validate API Key if its external API request
@@ -859,6 +927,28 @@ export const utilBuildChatflow = async (req: Request, isInternal: boolean = fals
            }
        }

+        // This can be public API, so we can only get orgId from the chatflow
+        const chatflowWorkspaceId = chatflow.workspaceId
+        const workspace = await appServer.AppDataSource.getRepository(Workspace).findOneBy({
+            id: chatflowWorkspaceId
+        })
+        if (!workspace) {
+            throw new InternalFlowiseError(StatusCodes.NOT_FOUND, `Workspace ${chatflowWorkspaceId} not found`)
+        }
+        const workspaceId = workspace.id
+
+        const org = await appServer.AppDataSource.getRepository(Organization).findOneBy({
+            id: workspace.organizationId
+        })
+        if (!org) {
+            throw new InternalFlowiseError(StatusCodes.NOT_FOUND, `Organization ${workspace.organizationId} not found`)
+        }
+
+        const orgId = org.id
+        const subscriptionId = org.subscriptionId as string
+
+        await checkPredictions(orgId, subscriptionId, appServer.usageCacheManager)
+
        const executeData: IExecuteFlowParams = {
            incomingInput, // Use the defensively created incomingInput variable
            chatflow,
@@ -866,18 +956,24 @@ export const utilBuildChatflow = async (req: Request, isInternal: boolean = fals
            baseURL,
            isInternal,
            files,
+            isEvaluation,
+            evaluationRunId,
            appDataSource: appServer.AppDataSource,
            sseStreamer: appServer.sseStreamer,
            telemetry: appServer.telemetry,
            cachePool: appServer.cachePool,
            componentNodes: appServer.nodesPool.componentNodes,
-            isTool // used to disable streaming if incoming request its from ChatflowTool
+            isTool, // used to disable streaming if incoming request its from ChatflowTool
+            usageCacheManager: appServer.usageCacheManager,
+            orgId,
+            workspaceId,
+            subscriptionId
        }

        if (process.env.MODE === MODE.QUEUE) {
            const predictionQueue = appServer.queueManager.getQueue('prediction')
            const job = await predictionQueue.addJob(omit(executeData, OMIT_QUEUE_JOB_DATA))
-            logger.debug(`[server]: Job added to queue: ${job.id}`)
+            logger.debug(`[server]: [${orgId}]: Job added to queue: ${job.id}`)

            const queueEvents = predictionQueue.getQueueEvents()
            const result = await job.waitUntilFinished(queueEvents)
@@ -885,7 +981,7 @@ export const utilBuildChatflow = async (req: Request, isInternal: boolean = fals
            if (!result) {
                throw new Error('Job execution failed')
            }
-
+            await updatePredictionsUsage(orgId, subscriptionId, workspaceId, appServer.usageCacheManager)
            incrementSuccessMetricCounter(appServer.metricsProvider, isInternal, isAgentFlow)
            return result
        } else {
@@ -893,9 +989,11 @@ export const utilBuildChatflow = async (req: Request, isInternal: boolean = fals
            const signal = new AbortController()
            appServer.abortControllerPool.add(abortControllerId, signal)
            executeData.signal = signal
+
            const result = await executeFlow(executeData)

            appServer.abortControllerPool.remove(abortControllerId)
+            await updatePredictionsUsage(orgId, subscriptionId, workspaceId, appServer.usageCacheManager)
            incrementSuccessMetricCounter(appServer.metricsProvider, isInternal, isAgentFlow)
            return result
        }