feat(web): 引入dify的流程检查算法

service-web/client/src/pages/ai/flow/FlowChecker.tsx

@@ -1,5 +1,5 @@
 import {find, findIdx, isEqual, lpad, toStr} from 'licia'
-import {type Connection, type Edge, getOutgoers, type Node} from '@xyflow/react'
+import {type Connection, type Edge, getConnectedEdges, getIncomers, getOutgoers, type Node} from '@xyflow/react'
 
 export class CheckError extends Error {
   readonly id: string
@@ -40,6 +40,183 @@ export const hasCycleError = () => new CheckError(204, '禁止流程循环')
 export const nodeNotOnlyToEndNode = () => new CheckError(206, '直连结束节点的节点不允许连接其他节点')
 export const hasRedundantEdgeError = () => new CheckError(207, '禁止出现冗余边')
 
+const hasCycle = (sourceNode: Node, targetNode: Node, nodes: Node[], edges: Edge[], visited = new Set<string>()) => {
+  if (visited.has(targetNode.id)) return false
+  visited.add(targetNode.id)
+  for (const outgoer of getOutgoers(targetNode, nodes, edges)) {
+    if (isEqual(outgoer.id, sourceNode.id)) return true
+    if (hasCycle(sourceNode, outgoer, nodes, edges, visited)) return true
+  }
+}
+
+type ParallelInfoItem = {
+  parallelNodeId: string
+  depth: number
+  isBranch?: boolean
+}
+type NodeParallelInfo = {
+  parallelNodeId: string
+  edgeHandleId: string
+  depth: number
+}
+type NodeHandle = {
+  node: Node
+  handle: string
+}
+type NodeStreamInfo = {
+  upstreamNodes: Set<string>
+  downstreamEdges: Set<string>
+}
+
+const getParallelInfo = (nodes: Node[], edges: Edge[]) => {
+  let startNode
+
+  startNode = nodes.find(node => isEqual(node.type, 'start-node'))
+  if (!startNode)
+    throw new Error('Start node not found')
+
+  const parallelList = [] as ParallelInfoItem[]
+  const nextNodeHandles = [{node: startNode, handle: 'source'}]
+  let hasAbnormalEdges = false
+
+  const groupBy = (array: Record<string, any>[], iteratee: string) => {
+    const result: Record<string, any[]> = {}
+    for (const item of array) {
+      // 获取属性值并转换为字符串键
+      const key = item[iteratee]
+      if (!result[key]) {
+        result[key] = []
+      }
+      result[key].push(item)
+    }
+    return result
+  }
+
+  const traverse = (firstNodeHandle: NodeHandle) => {
+    const nodeEdgesSet = {} as Record<string, Set<string>>
+    const totalEdgesSet = new Set<string>()
+    const nextHandles = [firstNodeHandle]
+    const streamInfo = {} as Record<string, NodeStreamInfo>
+    const parallelListItem = {
+      parallelNodeId: '',
+      depth: 0,
+    } as ParallelInfoItem
+    const nodeParallelInfoMap = {} as Record<string, NodeParallelInfo>
+    nodeParallelInfoMap[firstNodeHandle.node.id] = {
+      parallelNodeId: '',
+      edgeHandleId: '',
+      depth: 0,
+    }
+
+    while (nextHandles.length) {
+      const currentNodeHandle = nextHandles.shift()!
+      const {node: currentNode, handle: currentHandle = 'source'} = currentNodeHandle
+      const currentNodeHandleKey = currentNode.id
+      const connectedEdges = edges.filter(edge => edge.source === currentNode.id && edge.sourceHandle === currentHandle)
+      const connectedEdgesLength = connectedEdges.length
+      const outgoers = nodes.filter(node => connectedEdges.some(edge => edge.target === node.id))
+      const incomers = getIncomers(currentNode, nodes, edges)
+
+      if (!streamInfo[currentNodeHandleKey]) {
+        streamInfo[currentNodeHandleKey] = {
+          upstreamNodes: new Set<string>(),
+          downstreamEdges: new Set<string>(),
+        }
+      }
+
+      if (nodeEdgesSet[currentNodeHandleKey]?.size > 0 && incomers.length > 1) {
+        const newSet = new Set<string>()
+        for (const item of totalEdgesSet) {
+          if (!streamInfo[currentNodeHandleKey].downstreamEdges.has(item))
+            newSet.add(item)
+        }
+        if (isEqual(nodeEdgesSet[currentNodeHandleKey], newSet)) {
+          parallelListItem.depth = nodeParallelInfoMap[currentNode.id].depth
+          nextNodeHandles.push({node: currentNode, handle: currentHandle})
+          break
+        }
+      }
+
+      if (nodeParallelInfoMap[currentNode.id].depth > parallelListItem.depth)
+        parallelListItem.depth = nodeParallelInfoMap[currentNode.id].depth
+
+      outgoers.forEach((outgoer) => {
+        const outgoerConnectedEdges = getConnectedEdges([outgoer], edges).filter(edge => edge.source === outgoer.id)
+        const sourceEdgesGroup = groupBy(outgoerConnectedEdges, 'sourceHandle')
+        const incomers = getIncomers(outgoer, nodes, edges)
+
+        if (outgoers.length > 1 && incomers.length > 1)
+          hasAbnormalEdges = true
+
+        Object.keys(sourceEdgesGroup).forEach((sourceHandle) => {
+          nextHandles.push({node: outgoer, handle: sourceHandle})
+        })
+        if (!outgoerConnectedEdges.length)
+          nextHandles.push({node: outgoer, handle: 'source'})
+
+        const outgoerKey = outgoer.id
+        if (!nodeEdgesSet[outgoerKey])
+          nodeEdgesSet[outgoerKey] = new Set<string>()
+
+        if (nodeEdgesSet[currentNodeHandleKey]) {
+          for (const item of nodeEdgesSet[currentNodeHandleKey])
+            nodeEdgesSet[outgoerKey].add(item)
+        }
+
+        if (!streamInfo[outgoerKey]) {
+          streamInfo[outgoerKey] = {
+            upstreamNodes: new Set<string>(),
+            downstreamEdges: new Set<string>(),
+          }
+        }
+
+        if (!nodeParallelInfoMap[outgoer.id]) {
+          nodeParallelInfoMap[outgoer.id] = {
+            ...nodeParallelInfoMap[currentNode.id],
+          }
+        }
+
+        if (connectedEdgesLength > 1) {
+          const edge = connectedEdges.find(edge => edge.target === outgoer.id)!
+          nodeEdgesSet[outgoerKey].add(edge.id)
+          totalEdgesSet.add(edge.id)
+
+          streamInfo[currentNodeHandleKey].downstreamEdges.add(edge.id)
+          streamInfo[outgoerKey].upstreamNodes.add(currentNodeHandleKey)
+
+          for (const item of streamInfo[currentNodeHandleKey].upstreamNodes)
+            streamInfo[item].downstreamEdges.add(edge.id)
+
+          if (!parallelListItem.parallelNodeId)
+            parallelListItem.parallelNodeId = currentNode.id
+
+          const prevDepth = nodeParallelInfoMap[currentNode.id].depth + 1
+          const currentDepth = nodeParallelInfoMap[outgoer.id].depth
+
+          nodeParallelInfoMap[outgoer.id].depth = Math.max(prevDepth, currentDepth)
+        } else {
+          for (const item of streamInfo[currentNodeHandleKey].upstreamNodes)
+            streamInfo[outgoerKey].upstreamNodes.add(item)
+
+          nodeParallelInfoMap[outgoer.id].depth = nodeParallelInfoMap[currentNode.id].depth
+        }
+      })
+    }
+
+    parallelList.push(parallelListItem)
+  }
+
+  while (nextNodeHandles.length) {
+    const nodeHandle = nextNodeHandles.shift()!
+    traverse(nodeHandle)
+  }
+
+  return {
+    parallelList,
+    hasAbnormalEdges,
+  }
+}
+
 export const checkAddConnection: (connection: Connection, nodes: Node[], edges: Edge[]) => void = (connection, nodes, edges) => {
   let sourceNode = getNodeById(connection.source, nodes)
   if (!sourceNode) {
@@ -55,17 +232,9 @@ export const checkAddConnection: (connection: Connection, nodes: Node[], edges:
   }
 
   // 禁止流程出现环，必须是有向无环图
-  const hasCycle = (node: Node, visited = new Set<string>()) => {
-    if (visited.has(node.id)) return false
-    visited.add(node.id)
-    for (const outgoer of getOutgoers(node, nodes, edges)) {
-      if (isEqual(outgoer.id, sourceNode?.id)) return true
-      if (hasCycle(outgoer, visited)) return true
-    }
-  }
   if (isEqual(sourceNode.id, targetNode.id)) {
     throw nodeToSelfError()
-  } else if (hasCycle(targetNode)) {
+  } else if (hasCycle(sourceNode, targetNode, nodes, edges)) {
     throw hasCycleError()
   }