Initial push

packages/server/src/utils/index.ts

@@ -0,0 +1,264 @@
+import path from 'path'
+import fs from 'fs'
+import {
+    IComponentNodesPool,
+    IExploredNode,
+    INodeDependencies,
+    INodeDirectedGraph,
+    INodeQueue,
+    IReactFlowEdge,
+    IReactFlowNode
+} from '../Interface'
+import { cloneDeep, get } from 'lodash'
+import { ICommonObject, INodeData } from 'flowise-components'
+
+/**
+ * Returns the home folder path of the user if
+ * none can be found it falls back to the current
+ * working directory
+ *
+ */
+export const getUserHome = (): string => {
+    let variableName = 'HOME'
+    if (process.platform === 'win32') {
+        variableName = 'USERPROFILE'
+    }
+
+    if (process.env[variableName] === undefined) {
+        // If for some reason the variable does not exist
+        // fall back to current folder
+        return process.cwd()
+    }
+    return process.env[variableName] as string
+}
+
+/**
+ * Returns the path of node modules package
+ * @param {string} packageName
+ * @returns {string}
+ */
+export const getNodeModulesPackagePath = (packageName: string): string => {
+    const checkPaths = [
+        path.join(__dirname, '..', 'node_modules', packageName),
+        path.join(__dirname, '..', '..', 'node_modules', packageName),
+        path.join(__dirname, '..', '..', '..', 'node_modules', packageName),
+        path.join(__dirname, '..', '..', '..', '..', 'node_modules', packageName),
+        path.join(__dirname, '..', '..', '..', '..', '..', 'node_modules', packageName)
+    ]
+    for (const checkPath of checkPaths) {
+        if (fs.existsSync(checkPath)) {
+            return checkPath
+        }
+    }
+    return ''
+}
+
+/**
+ * Construct directed graph and node dependencies score
+ * @param {IReactFlowNode[]} reactFlowNodes
+ * @param {IReactFlowEdge[]} reactFlowEdges
+ */
+export const constructGraphs = (reactFlowNodes: IReactFlowNode[], reactFlowEdges: IReactFlowEdge[]) => {
+    const nodeDependencies = {} as INodeDependencies
+    const graph = {} as INodeDirectedGraph
+
+    for (let i = 0; i < reactFlowNodes.length; i += 1) {
+        const nodeId = reactFlowNodes[i].id
+        nodeDependencies[nodeId] = 0
+        graph[nodeId] = []
+    }
+
+    for (let i = 0; i < reactFlowEdges.length; i += 1) {
+        const source = reactFlowEdges[i].source
+        const target = reactFlowEdges[i].target
+
+        if (Object.prototype.hasOwnProperty.call(graph, source)) {
+            graph[source].push(target)
+        } else {
+            graph[source] = [target]
+        }
+        nodeDependencies[target] += 1
+    }
+
+    return { graph, nodeDependencies }
+}
+
+/**
+ * Get starting node and check if flow is valid
+ * @param {INodeDependencies} nodeDependencies
+ */
+export const getStartingNode = (nodeDependencies: INodeDependencies) => {
+    // Find starting node
+    const startingNodeIds = [] as string[]
+    Object.keys(nodeDependencies).forEach((nodeId) => {
+        if (nodeDependencies[nodeId] === 0) {
+            startingNodeIds.push(nodeId)
+        }
+    })
+    return startingNodeIds
+}
+
+export const getEndingNode = (nodeDependencies: INodeDependencies, graph: INodeDirectedGraph) => {
+    // Find starting node
+    let endingNodeId = ''
+    Object.keys(graph).forEach((nodeId) => {
+        if (!graph[nodeId].length && nodeDependencies[nodeId] > 0) {
+            endingNodeId = nodeId
+        }
+    })
+    return endingNodeId
+}
+
+/**
+ * Build langchain from start to end
+ * @param {string} startingNodeId
+ * @param {IReactFlowNode[]} reactFlowNodes
+ * @param {IReactFlowEdge[]} reactFlowEdges
+ * @param {INodeDirectedGraph} graph
+ * @param {IComponentNodesPool} componentNodes
+ * @param {string} clientId
+ * @param {any} io
+ */
+export const buildLangchain = async (
+    startingNodeIds: string[],
+    reactFlowNodes: IReactFlowNode[],
+    graph: INodeDirectedGraph,
+    componentNodes: IComponentNodesPool
+) => {
+    const flowNodes = cloneDeep(reactFlowNodes)
+
+    // Create a Queue and add our initial node in it
+    const nodeQueue = [] as INodeQueue[]
+    const exploredNode = {} as IExploredNode
+
+    // In the case of infinite loop, only max 3 loops will be executed
+    const maxLoop = 3
+
+    for (let i = 0; i < startingNodeIds.length; i += 1) {
+        nodeQueue.push({ nodeId: startingNodeIds[i], depth: 0 })
+        exploredNode[startingNodeIds[i]] = { remainingLoop: maxLoop, lastSeenDepth: 0 }
+    }
+
+    while (nodeQueue.length) {
+        const { nodeId, depth } = nodeQueue.shift() as INodeQueue
+
+        const reactFlowNode = flowNodes.find((nd) => nd.id === nodeId)
+        const nodeIndex = flowNodes.findIndex((nd) => nd.id === nodeId)
+        if (!reactFlowNode || reactFlowNode === undefined || nodeIndex < 0) continue
+
+        try {
+            const nodeInstanceFilePath = componentNodes[reactFlowNode.data.name].filePath as string
+            const nodeModule = await import(nodeInstanceFilePath)
+            const newNodeInstance = new nodeModule.nodeClass()
+
+            const reactFlowNodeData: INodeData = resolveVariables(reactFlowNode.data, flowNodes)
+
+            flowNodes[nodeIndex].data.instance = await newNodeInstance.init(reactFlowNodeData)
+        } catch (e: any) {
+            console.error(e)
+            throw new Error(e)
+        }
+
+        const neighbourNodeIds = graph[nodeId]
+        const nextDepth = depth + 1
+
+        for (let i = 0; i < neighbourNodeIds.length; i += 1) {
+            const neighNodeId = neighbourNodeIds[i]
+
+            // If nodeId has been seen, cycle detected
+            if (Object.prototype.hasOwnProperty.call(exploredNode, neighNodeId)) {
+                const { remainingLoop, lastSeenDepth } = exploredNode[neighNodeId]
+
+                if (lastSeenDepth === nextDepth) continue
+
+                if (remainingLoop === 0) {
+                    break
+                }
+                const remainingLoopMinusOne = remainingLoop - 1
+                exploredNode[neighNodeId] = { remainingLoop: remainingLoopMinusOne, lastSeenDepth: nextDepth }
+                nodeQueue.push({ nodeId: neighNodeId, depth: nextDepth })
+            } else {
+                exploredNode[neighNodeId] = { remainingLoop: maxLoop, lastSeenDepth: nextDepth }
+                nodeQueue.push({ nodeId: neighNodeId, depth: nextDepth })
+            }
+        }
+    }
+    return flowNodes
+}
+
+/**
+ * Get variable value from outputResponses.output
+ * @param {string} paramValue
+ * @param {IReactFlowNode[]} reactFlowNodes
+ * @param {string} key
+ * @param {number} loopIndex
+ * @returns {string}
+ */
+export const getVariableValue = (paramValue: string, reactFlowNodes: IReactFlowNode[]) => {
+    let returnVal = paramValue
+    const variableStack = []
+    let startIdx = 0
+    const endIdx = returnVal.length - 1
+
+    while (startIdx < endIdx) {
+        const substr = returnVal.substring(startIdx, startIdx + 2)
+
+        // Store the opening double curly bracket
+        if (substr === '{{') {
+            variableStack.push({ substr, startIdx: startIdx + 2 })
+        }
+
+        // Found the complete variable
+        if (substr === '}}' && variableStack.length > 0 && variableStack[variableStack.length - 1].substr === '{{') {
+            const variableStartIdx = variableStack[variableStack.length - 1].startIdx
+            const variableEndIdx = startIdx
+            const variableFullPath = returnVal.substring(variableStartIdx, variableEndIdx)
+
+            // 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
+            }
+            variableStack.pop()
+        }
+        startIdx += 1
+    }
+    return returnVal
+}
+
+/**
+ * Loop through each inputs and resolve variable if neccessary
+ * @param {INodeData} reactFlowNodeData
+ * @param {IReactFlowNode[]} reactFlowNodes
+ * @returns {INodeData}
+ */
+export const resolveVariables = (reactFlowNodeData: INodeData, reactFlowNodes: IReactFlowNode[]): 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)
+                    resolvedInstances.push(resolvedInstance)
+                }
+                paramsObj[key] = resolvedInstances
+            } else {
+                const resolvedInstance = getVariableValue(paramValue, reactFlowNodes)
+                paramsObj[key] = resolvedInstance
+            }
+        }
+    }
+
+    const paramsObj = (flowNodeData as any)[types]
+
+    getParamValues(paramsObj)
+
+    return flowNodeData
+}