Add feature to be able to chain prompt values

packages/server/src/ChatflowPool.ts

@@ -1,9 +1,8 @@
-import { INodeData } from 'flowise-components'
-import { IActiveChatflows } from './Interface'
+import { IActiveChatflows, INodeData } from './Interface'
 
 /**
- * This pool is to keep track of active test triggers (event listeners),
- * so we can clear the event listeners whenever user refresh or exit page
+ * This pool is to keep track of active chatflow pools
+ * so we can prevent building langchain flow all over again
  */
 export class ChatflowPool {
     activeChatflows: IActiveChatflows = {}

packages/server/src/Interface.ts

@@ -1,4 +1,4 @@
-import { INode, INodeData } from 'flowise-components'
+import { INode, INodeData as INodeDataFromComponent, INodeParams } from 'flowise-components'
 
 export type MessageType = 'apiMessage' | 'userMessage'
 
@@ -38,6 +38,12 @@ export interface INodeDirectedGraph {
     [key: string]: string[]
 }
 
+export interface INodeData extends INodeDataFromComponent {
+    inputAnchors: INodeParams[]
+    inputParams: INodeParams[]
+    outputAnchors: INodeParams[]
+}
+
 export interface IReactFlowNode {
     id: string
     position: {

packages/server/src/index.ts

@@ -4,15 +4,22 @@ import cors from 'cors'
 import http from 'http'
 import * as fs from 'fs'
 
-import { IChatFlow, IncomingInput, IReactFlowNode, IReactFlowObject } from './Interface'
-import { getNodeModulesPackagePath, getStartingNodes, buildLangchain, getEndingNode, constructGraphs } from './utils'
+import { IChatFlow, IncomingInput, IReactFlowNode, IReactFlowObject, INodeData } from './Interface'
+import {
+    getNodeModulesPackagePath,
+    getStartingNodes,
+    buildLangchain,
+    getEndingNode,
+    constructGraphs,
+    resolveVariables,
+    checkIfFlowNeedToRebuild
+} from './utils'
 import { cloneDeep } from 'lodash'
 import { getDataSource } from './DataSource'
 import { NodesPool } from './NodesPool'
 import { ChatFlow } from './entity/ChatFlow'
 import { ChatMessage } from './entity/ChatMessage'
 import { ChatflowPool } from './ChatflowPool'
-import { INodeData } from 'flowise-components'
 
 export class App {
     app: express.Application
@@ -196,44 +203,61 @@ export class App {
 
                 let nodeToExecuteData: INodeData
 
+                const chatflow = await this.AppDataSource.getRepository(ChatFlow).findOneBy({
+                    id: chatflowid
+                })
+                if (!chatflow) return res.status(404).send(`Chatflow ${chatflowid} not found`)
+
+                const flowData = chatflow.flowData
+                const parsedFlowData: IReactFlowObject = JSON.parse(flowData)
+                const nodes = parsedFlowData.nodes
+                const edges = parsedFlowData.edges
+
+                // Check if node data exists in pool && not out of sync, prevent building whole flow again
                 if (
                     Object.prototype.hasOwnProperty.call(this.chatflowPool.activeChatflows, chatflowid) &&
-                    this.chatflowPool.activeChatflows[chatflowid].inSync
+                    this.chatflowPool.activeChatflows[chatflowid].inSync &&
+                    !checkIfFlowNeedToRebuild(nodes, this.chatflowPool.activeChatflows[chatflowid].endingNodeData)
                 ) {
                     nodeToExecuteData = this.chatflowPool.activeChatflows[chatflowid].endingNodeData
                 } else {
-                    const chatflow = await this.AppDataSource.getRepository(ChatFlow).findOneBy({
-                        id: chatflowid
-                    })
-                    if (!chatflow) return res.status(404).send(`Chatflow ${chatflowid} not found`)
-
-                    const flowData = chatflow.flowData
-                    const parsedFlowData: IReactFlowObject = JSON.parse(flowData)
-
                     /*** Get Ending Node with Directed Graph  ***/
-                    const { graph, nodeDependencies } = constructGraphs(parsedFlowData.nodes, parsedFlowData.edges)
+                    const { graph, nodeDependencies } = constructGraphs(nodes, edges)
                     const directedGraph = graph
                     const endingNodeId = getEndingNode(nodeDependencies, directedGraph)
                     if (!endingNodeId) return res.status(500).send(`Ending node must be either a Chain or Agent`)
 
+                    const endingNodeData = nodes.find((nd) => nd.id === endingNodeId)?.data
+                    if (!endingNodeData) return res.status(500).send(`Ending node must be either a Chain or Agent`)
+
+                    if (!Object.values(endingNodeData.outputs ?? {}).includes(endingNodeData.name)) {
+                        return res
+                            .status(500)
+                            .send(
+                                `Output of ${endingNodeData.label} (${endingNodeData.id}) must be ${endingNodeData.label}, can't be an Output Prediction`
+                            )
+                    }
+
                     /*** Get Starting Nodes with Non-Directed Graph ***/
-                    const constructedObj = constructGraphs(parsedFlowData.nodes, parsedFlowData.edges, true)
+                    const constructedObj = constructGraphs(nodes, edges, true)
                     const nonDirectedGraph = constructedObj.graph
                     const { startingNodeIds, depthQueue } = getStartingNodes(nonDirectedGraph, endingNodeId)
 
                     /*** BFS to traverse from Starting Nodes to Ending Node ***/
                     const reactFlowNodes = await buildLangchain(
                         startingNodeIds,
-                        parsedFlowData.nodes,
+                        nodes,
                         graph,
                         depthQueue,
-                        this.nodesPool.componentNodes
+                        this.nodesPool.componentNodes,
+                        incomingInput.question
                     )
 
                     const nodeToExecute = reactFlowNodes.find((node: IReactFlowNode) => node.id === endingNodeId)
                     if (!nodeToExecute) return res.status(404).send(`Node ${endingNodeId} not found`)
 
-                    nodeToExecuteData = nodeToExecute.data
+                    const reactFlowNodeData: INodeData = resolveVariables(nodeToExecute.data, reactFlowNodes, incomingInput.question)
+                    nodeToExecuteData = reactFlowNodeData
 
                     this.chatflowPool.add(chatflowid, nodeToExecuteData)
                 }

packages/server/src/utils/index.ts

@@ -8,10 +8,14 @@ import {
     INodeDirectedGraph,
     INodeQueue,
     IReactFlowEdge,
-    IReactFlowNode
+    IReactFlowNode,
+    IVariableDict,
+    INodeData
 } from '../Interface'
 import { cloneDeep, get } from 'lodash'
-import { ICommonObject, INodeData } from 'flowise-components'
+import { ICommonObject } from 'flowise-components'
+
+const QUESTION_VAR_PREFIX = 'question'
 
 /**
  * Returns the home folder path of the user if
@@ -166,13 +170,15 @@ export const getEndingNode = (nodeDependencies: INodeDependencies, graph: INodeD
  * @param {INodeDirectedGraph} graph
  * @param {IDepthQueue} depthQueue
  * @param {IComponentNodes} componentNodes
+ * @param {string} question
  */
 export const buildLangchain = async (
     startingNodeIds: string[],
     reactFlowNodes: IReactFlowNode[],
     graph: INodeDirectedGraph,
     depthQueue: IDepthQueue,
-    componentNodes: IComponentNodes
+    componentNodes: IComponentNodes,
+    question: string
 ) => {
     const flowNodes = cloneDeep(reactFlowNodes)
 
@@ -200,9 +206,9 @@ export const buildLangchain = async (
             const nodeModule = await import(nodeInstanceFilePath)
             const newNodeInstance = new nodeModule.nodeClass()
 
-            const reactFlowNodeData: INodeData = resolveVariables(reactFlowNode.data, flowNodes)
+            const reactFlowNodeData: INodeData = resolveVariables(reactFlowNode.data, flowNodes, question)
 
-            flowNodes[nodeIndex].data.instance = await newNodeInstance.init(reactFlowNodeData)
+            flowNodes[nodeIndex].data.instance = await newNodeInstance.init(reactFlowNodeData, question)
         } catch (e: any) {
             console.error(e)
             throw new Error(e)
@@ -247,11 +253,14 @@ export const buildLangchain = async (
  * Get variable value from outputResponses.output
  * @param {string} paramValue
  * @param {IReactFlowNode[]} reactFlowNodes
+ * @param {string} question
+ * @param {boolean} isAcceptVariable
  * @returns {string}
  */
-export const getVariableValue = (paramValue: string, reactFlowNodes: IReactFlowNode[]) => {
+export const getVariableValue = (paramValue: string, reactFlowNodes: IReactFlowNode[], question: string, isAcceptVariable = false) => {
     let returnVal = paramValue
     const variableStack = []
+    const variableDict = {} as IVariableDict
     let startIdx = 0
     const endIdx = returnVal.length - 1
 
@@ -269,17 +278,36 @@ export const getVariableValue = (paramValue: string, reactFlowNodes: IReactFlowN
             const variableEndIdx = startIdx
             const variableFullPath = returnVal.substring(variableStartIdx, variableEndIdx)
 
+            if (isAcceptVariable && variableFullPath === QUESTION_VAR_PREFIX) {
+                variableDict[`{{${variableFullPath}}}`] = question
+            }
+
             // Split by first occurence of '.' to get just nodeId
             const [variableNodeId, _] = variableFullPath.split('.')
             const executedNode = reactFlowNodes.find((nd) => nd.id === variableNodeId)
             if (executedNode) {
-                const variableInstance = get(executedNode.data, 'instance')
-                returnVal = variableInstance
+                const variableValue = get(executedNode.data, 'instance')
+                if (isAcceptVariable) {
+                    variableDict[`{{${variableFullPath}}}`] = variableValue
+                } else {
+                    returnVal = variableValue
+                }
             }
             variableStack.pop()
         }
         startIdx += 1
     }
+
+    if (isAcceptVariable) {
+        const variablePaths = Object.keys(variableDict)
+        variablePaths.sort() // Sort by length of variable path because longer path could possibly contains nested variable
+        variablePaths.forEach((path) => {
+            const variableValue = variableDict[path]
+            // Replace all occurence
+            returnVal = returnVal.split(path).join(variableValue)
+        })
+        return returnVal
+    }
     return returnVal
 }
 
@@ -287,25 +315,26 @@ export const getVariableValue = (paramValue: string, reactFlowNodes: IReactFlowN
  * Loop through each inputs and resolve variable if neccessary
  * @param {INodeData} reactFlowNodeData
  * @param {IReactFlowNode[]} reactFlowNodes
+ * @param {string} question
  * @returns {INodeData}
  */
-export const resolveVariables = (reactFlowNodeData: INodeData, reactFlowNodes: IReactFlowNode[]): INodeData => {
+export const resolveVariables = (reactFlowNodeData: INodeData, reactFlowNodes: IReactFlowNode[], question: string): INodeData => {
     const flowNodeData = cloneDeep(reactFlowNodeData)
     const types = 'inputs'
 
     const getParamValues = (paramsObj: ICommonObject) => {
         for (const key in paramsObj) {
             const paramValue: string = paramsObj[key]
-
             if (Array.isArray(paramValue)) {
                 const resolvedInstances = []
                 for (const param of paramValue) {
-                    const resolvedInstance = getVariableValue(param, reactFlowNodes)
+                    const resolvedInstance = getVariableValue(param, reactFlowNodes, question)
                     resolvedInstances.push(resolvedInstance)
                 }
                 paramsObj[key] = resolvedInstances
             } else {
-                const resolvedInstance = getVariableValue(paramValue, reactFlowNodes)
+                const isAcceptVariable = reactFlowNodeData.inputParams.find((param) => param.name === key)?.acceptVariable ?? false
+                const resolvedInstance = getVariableValue(paramValue, reactFlowNodes, question, isAcceptVariable)
                 paramsObj[key] = resolvedInstance
             }
         }
@@ -317,3 +346,24 @@ export const resolveVariables = (reactFlowNodeData: INodeData, reactFlowNodes: I
 
     return flowNodeData
 }
+
+/**
+ * Rebuild flow if LLMChain has dependency on other chains
+ * User Question => Prompt_0 => LLMChain_0 => Prompt-1 => LLMChain_1
+ * @param {IReactFlowNode[]} nodes
+ * @param {INodeData} nodeData
+ * @returns {boolean}
+ */
+export const checkIfFlowNeedToRebuild = (nodes: IReactFlowNode[], nodeData: INodeData) => {
+    if (nodeData.name !== 'llmChain') return false
+
+    const node = nodes.find((nd) => nd.id === nodeData.id)
+    if (!node) throw new Error(`Node ${nodeData.id} not found`)
+
+    const inputs = node.data.inputs
+    for (const key in inputs) {
+        const isInputAcceptVariable = node.data.inputParams.find((param) => param.name === key)?.acceptVariable || false
+        if (isInputAcceptVariable && inputs[key].includes('{{') && inputs[key].includes('}}')) return true
+    }
+    return false
+}