Feature/Update Loop Agentflow (#4957)

* Feature: Update Loop Agentflow to include fallback message and version increment to 1.1

- Added a new input parameter 'fallbackMessage' to the Loop Agentflow for displaying a message when the loop count is exceeded.
- Incremented the version of Loop Agentflow from 1.0 to 1.1.
- Updated the processing logic to handle the fallback message appropriately when the maximum loop count is reached.

* - Introduced a new input parameter 'loopUpdateState' to allow updating the runtime state during workflow execution.
- Added a method to list runtime state keys for dynamic state management.
- Implemented logic to retrieve and utilize the current loop count in variable resolution.
- Updated the Loop Agentflow output to reflect the new state and final output content.

packages/server/src/utils/buildAgentflow.ts

@@ -43,7 +43,8 @@ import {
     QUESTION_VAR_PREFIX,
     CURRENT_DATE_TIME_VAR_PREFIX,
     _removeCredentialId,
-    validateHistorySchema
+    validateHistorySchema,
+    LOOP_COUNT_VAR_PREFIX
 } from '.'
 import { ChatFlow } from '../database/entities/ChatFlow'
 import { Variable } from '../database/entities/Variable'
@@ -84,6 +85,8 @@ interface IProcessNodeOutputsParams {
     waitingNodes: Map<string, IWaitingNode>
     loopCounts: Map<string, number>
     abortController?: AbortController
+    sseStreamer?: IServerSideEventStreamer
+    chatId: string
 }
 
 interface IAgentFlowRuntime {
@@ -130,6 +133,7 @@ interface IExecuteNodeParams {
     parentExecutionId?: string
     isRecursive?: boolean
     iterationContext?: ICommonObject
+    loopCounts?: Map<string, number>
     orgId: string
     workspaceId: string
     subscriptionId: string
@@ -218,7 +222,8 @@ export const resolveVariables = async (
     chatHistory: IMessage[],
     componentNodes: IComponentNodes,
     agentFlowExecutedData?: IAgentflowExecutedData[],
-    iterationContext?: ICommonObject
+    iterationContext?: ICommonObject,
+    loopCounts?: Map<string, number>
 ): Promise<INodeData> => {
     let flowNodeData = cloneDeep(reactFlowNodeData)
     const types = 'inputs'
@@ -285,6 +290,20 @@ export const resolveVariables = async (
                 resolvedValue = resolvedValue.replace(match, flowConfig?.runtimeChatHistoryLength ?? 0)
             }
 
+            if (variableFullPath === LOOP_COUNT_VAR_PREFIX) {
+                // Get the current loop count from the most recent loopAgentflow node execution
+                let currentLoopCount = 0
+                if (loopCounts && agentFlowExecutedData) {
+                    // Find the most recent loopAgentflow node execution to get its loop count
+                    const loopNodes = [...agentFlowExecutedData].reverse().filter((data) => data.data?.name === 'loopAgentflow')
+                    if (loopNodes.length > 0) {
+                        const latestLoopNode = loopNodes[0]
+                        currentLoopCount = loopCounts.get(latestLoopNode.nodeId) || 0
+                    }
+                }
+                resolvedValue = resolvedValue.replace(match, currentLoopCount.toString())
+            }
+
             if (variableFullPath === CURRENT_DATE_TIME_VAR_PREFIX) {
                 resolvedValue = resolvedValue.replace(match, new Date().toISOString())
             }
@@ -742,7 +761,9 @@ async function processNodeOutputs({
     edges,
     nodeExecutionQueue,
     waitingNodes,
-    loopCounts
+    loopCounts,
+    sseStreamer,
+    chatId
 }: IProcessNodeOutputsParams): Promise<{ humanInput?: IHumanInput }> {
     logger.debug(`\n🔄 Processing outputs from node: ${nodeId}`)
 
@@ -823,6 +844,11 @@ async function processNodeOutputs({
             }
         } else {
             logger.debug(`    ⚠️ Maximum loop count (${maxLoop}) reached, stopping loop`)
+            const fallbackMessage = result.output.fallbackMessage || `Loop completed after reaching maximum iteration count of ${maxLoop}.`
+            if (sseStreamer) {
+                sseStreamer.streamTokenEvent(chatId, fallbackMessage)
+            }
+            result.output = { ...result.output, content: fallbackMessage }
         }
     }
 
@@ -967,6 +993,7 @@ const executeNode = async ({
     isInternal,
     isRecursive,
     iterationContext,
+    loopCounts,
     orgId,
     workspaceId,
     subscriptionId,
@@ -1045,7 +1072,8 @@ const executeNode = async ({
             chatHistory,
             componentNodes,
             agentFlowExecutedData,
-            iterationContext
+            iterationContext,
+            loopCounts
         )
 
         // Handle human input if present
@@ -1889,6 +1917,7 @@ export const executeAgentFlow = async ({
                 analyticHandlers,
                 isRecursive,
                 iterationContext,
+                loopCounts,
                 orgId,
                 workspaceId,
                 subscriptionId,
@@ -1957,7 +1986,8 @@ export const executeAgentFlow = async ({
                 nodeExecutionQueue,
                 waitingNodes,
                 loopCounts,
-                abortController
+                sseStreamer,
+                chatId
             })
 
             // Update humanInput if it was changed

packages/server/src/utils/index.ts

@@ -69,6 +69,7 @@ export const QUESTION_VAR_PREFIX = 'question'
 export const FILE_ATTACHMENT_PREFIX = 'file_attachment'
 export const CHAT_HISTORY_VAR_PREFIX = 'chat_history'
 export const RUNTIME_MESSAGES_LENGTH_VAR_PREFIX = 'runtime_messages_length'
+export const LOOP_COUNT_VAR_PREFIX = 'loop_count'
 export const CURRENT_DATE_TIME_VAR_PREFIX = 'current_date_time'
 export const REDACTED_CREDENTIAL_VALUE = '_FLOWISE_BLANK_07167752-1a71-43b1-bf8f-4f32252165db'