/** * 工作流节点 IO 变量推断引擎 * * 职责: * 1. 从节点数据中提取 inputs / outputs 定义 * 2. 连线时自动匹配源节点输出 → 目标节点输入 * 3. 多源汇聚时计算变量分配 */ // ========== 类型常量 ========== export const IO_TYPES = [ { label: '字符串', value: 'string' }, { label: '数字', value: 'number' }, { label: '布尔值', value: 'boolean' }, { label: '数组', value: 'array' }, { label: '对象', value: 'object' }, { label: '文件路径', value: 'filePath' }, { label: '目录路径', value: 'directoryPath' } ] // 边状态 → 样式 export const EDGE_STATUS_STYLE = { ok: { stroke: '#22c55e', strokeWidth: 2 }, unmapped: { stroke: '#666', strokeWidth: 2 }, partial: { stroke: '#f59e0b', strokeWidth: 2 }, mismatch: { stroke: '#ef4444', strokeWidth: 2 } } // ========== 模板变量提取 ========== /** * 从 {{变量名}} 模板中提取变量名列表(去重) */ export function extractTemplateVariables(text) { if (!text) return [] const matches = text.matchAll(/\{\{\s*([\w.]+)\s*\}\}/g) const seen = new Set() const result = [] for (const m of matches) { if (!seen.has(m[1])) { seen.add(m[1]) result.push(m[1]) } } return result } // ========== 节点 IO 获取 ========== /** * 获取节点的输出字段列表 */ export function getNodeOutputs(node) { const d = node.data || {} switch (node.type) { case 'userInput': return (d.variables || []).map(v => ({ name: v.name, label: v.label || v.name, type: v.type || 'string', description: v.description || '' })) case 'llm': if (d.outputs && d.outputs.length) return d.outputs return [{ name: 'result', label: 'LLM 输出', type: 'string', description: '大模型响应文本' }] case 'agent': return d.outputs || [] case 'skill': return d.outputs || [] case 'knowledgeRetrieval': if (d.outputs && d.outputs.length) return d.outputs return [ { name: 'evidences', label: '检索证据列表', type: 'array', description: '命中的知识片段集合' }, { name: 'evidenceCount', label: '证据数量', type: 'number', description: '命中证据条数' }, { name: 'sourceType', label: '来源类型', type: 'string', description: 'DOCUMENT / STRUCTURED_DATA / GRAPH / HYBRID' }, { name: 'diagnostics', label: '诊断信息', type: 'object', description: '检索过程的诊断信息' } ] case 'output': return [] case 'condition': // 条件节点透传:输出 = 所有输入 return getNodeInputs(node) default: return d.outputs || [] } } /** * 获取节点的输入字段列表 */ export function getNodeInputs(node) { const d = node.data || {} switch (node.type) { case 'userInput': return [] case 'llm': { // 优先使用手动定义的 inputs if (d.inputs && d.inputs.length) return d.inputs // 回退:从模板提取 const vars = new Set([ ...extractTemplateVariables(d.systemPrompt), ...extractTemplateVariables(d.userPrompt) ]) return [...vars].map(name => ({ name, label: name, type: 'string', description: '' })) } case 'agent': return d.inputs || [] case 'skill': return d.inputs || [] case 'smartAction': { // 优先使用手动定义的 inputs if (d.inputs && d.inputs.length) return d.inputs // 回退:从操作要求模板提取 const saVars = new Set(extractTemplateVariables(d.actionPrompt)) return [...saVars].map(name => ({ name, label: name, type: 'string', description: '' })) } case 'knowledgeRetrieval': { // 优先使用手动定义的 inputs if (d.inputs && d.inputs.length) return d.inputs // 回退:从检索语句模板提取 const krVars = new Set(extractTemplateVariables(d.query)) return [...krVars].map(name => ({ name, label: name, type: 'string', description: '' })) } case 'output': return d.fields || [] case 'condition': { // 优先使用手动定义的 inputs if (d.inputs && d.inputs.length) return d.inputs // 回退:从条件表达式提取 const vars = new Set() for (const c of (d.conditions || [])) { for (const v of extractTemplateVariables(c.expression)) { vars.add(v) } } return [...vars].map(name => ({ name, label: name, type: 'string', description: '' })) } default: return d.inputs || [] } } // ========== 类型兼容性 ========== /** * 判断源类型是否可以赋值给目标类型 */ export function isTypeCompatible(sourceType, targetType) { if (sourceType === targetType) return true // 以下类型可隐式转为 string const stringLike = ['filePath', 'directoryPath', 'number', 'boolean'] if (stringLike.includes(sourceType) && targetType === 'string') return true // array → object 兼容 if (sourceType === 'array' && targetType === 'object') return true return false } // ========== 单边匹配 ========== /** * 匹配源节点输出与目标节点输入 * @returns {{ status, mapping, unmatchedSource, unmatchedTarget, typeMismatches }} */ export function matchIO(sourceOutputs, targetInputs, existingMapping) { const mapping = [] const unmatchedSource = [...sourceOutputs] const unmatchedTarget = [...targetInputs] const typeMismatches = [] // 第一轮:按变量名精确匹配 for (let si = unmatchedSource.length - 1; si >= 0; si--) { const srcField = unmatchedSource[si] const ti = unmatchedTarget.findIndex(t => t.name === srcField.name) if (ti !== -1) { const tgtField = unmatchedTarget[ti] if (isTypeCompatible(srcField.type, tgtField.type)) { mapping.push({ sourceField: srcField.name, targetField: tgtField.name }) } else { typeMismatches.push({ source: srcField, target: tgtField }) } unmatchedSource.splice(si, 1) unmatchedTarget.splice(ti, 1) } } // 第二轮:保留已有的手动映射(不与新映射冲突的部分) if (existingMapping) { for (const em of existingMapping) { if (mapping.some(m => m.sourceField === em.sourceField && m.targetField === em.targetField)) continue if (mapping.some(m => m.sourceField === em.sourceField || m.targetField === em.targetField)) continue mapping.push(em) const ti = unmatchedTarget.findIndex(t => t.name === em.targetField) if (ti !== -1) unmatchedTarget.splice(ti, 1) } } // 判定状态 let status if (targetInputs.length === 0 && sourceOutputs.length === 0) { status = 'unmapped' } else if (unmatchedTarget.length === 0 && typeMismatches.length === 0) { status = 'ok' } else if (typeMismatches.length > 0) { status = 'mismatch' } else if (unmatchedTarget.length > 0 && unmatchedSource.length === 0) { status = 'partial' } else if (unmatchedTarget.length === 0 && unmatchedSource.length > 0) { status = 'ok' // 源有多余输出,但目标全部满足 } else { status = 'mismatch' } return { status, mapping, unmatchedSource, unmatchedTarget, typeMismatches } } // ========== 推断调度 ========== /** * 推断一条边的映射状态 * mapping 仍然基于直接源节点的输出(精确映射), * 但状态判断基于所有可达前驱节点的合并输出(因为上下文是累积的) * @param {object} sourceNode - 源节点 * @param {object} targetNode - 目标节点 * @param {object} [existingEdge] - 已有边数据 * @param {Array} [allNodes] - 全部节点(用于可达前驱计算) * @param {Array} [allEdges] - 全部边(用于可达前驱计算) * @returns {{ status, mapping, unmatchedSource, unmatchedTarget, typeMismatches }} */ export function inferEdgeMapping(sourceNode, targetNode, existingEdge, allNodes, allEdges) { const sourceOutputs = getNodeOutputs(sourceNode) const targetInputs = getNodeInputs(targetNode) const existingMapping = existingEdge?.data?.mapping // 直接源→目标的精确映射(用于 mapping 字段) const directResult = matchIO(sourceOutputs, targetInputs, existingMapping) // 如果没有传入全图数据,降级为只看直接源 if (!allNodes || !allEdges) { return directResult } // 收集所有可达前驱节点的合并输出(用于状态判断) const reachableOutputs = collectReachableOutputs(targetNode.id, allNodes, allEdges) if (reachableOutputs.length === 0 && targetInputs.length === 0) { return { ...directResult, status: 'unmapped' } } // 用合并输出判断目标输入是否全部满足 const unmatched = [] const typeMismatches = [] for (const tgtField of targetInputs) { const srcField = reachableOutputs.find(o => o.name === tgtField.name) if (!srcField) { unmatched.push(tgtField) } else if (!isTypeCompatible(srcField.type, tgtField.type)) { typeMismatches.push({ source: srcField, target: tgtField }) } } let status if (targetInputs.length === 0 && sourceOutputs.length === 0) { status = 'unmapped' } else if (unmatched.length === 0 && typeMismatches.length === 0) { status = 'ok' } else if (typeMismatches.length > 0) { status = 'mismatch' } else { status = 'partial' } return { ...directResult, status } } /** * 收集目标节点所有可达前驱节点的合并输出(BFS 回溯边图) * @param {string} targetId - 目标节点 ID * @param {Array} allNodes - 全部节点 * @param {Array} allEdges - 全部边 * @returns {Array} 合并后的输出字段列表(去重,先出现的优先) */ function collectReachableOutputs(targetId, allNodes, allEdges) { const nodeMap = new Map(allNodes.map(n => [n.id, n])) const visited = new Set() const queue = [targetId] const merged = new Map() // name → field while (queue.length > 0) { const currentId = queue.shift() if (visited.has(currentId)) continue visited.add(currentId) // 找到所有指向当前节点的边 for (const edge of allEdges) { if (edge.target === currentId && !visited.has(edge.source)) { const sourceNode = nodeMap.get(edge.source) if (sourceNode) { for (const field of getNodeOutputs(sourceNode)) { if (!merged.has(field.name)) { merged.set(field.name, field) } } queue.push(edge.source) } } } } return [...merged.values()] } /** * 获取目标节点的所有可达前驱节点(通过边反向 BFS) * * 用于"前置数据关联"下拉选项:节点输入不仅可关联直接上游, * 也可关联画布中所有可达的前序节点输出。 * * @param {string} targetId - 目标节点 ID * @param {Array} allNodes - 全部节点 * @param {Array} allEdges - 全部边 * @returns {Array} 可达前驱节点列表(按 BFS 发现顺序,不含 targetId 本身,不重复) */ export function getReachablePredecessors(targetId, allNodes, allEdges) { const nodeMap = new Map(allNodes.map(n => [n.id, n])) const visited = new Set([targetId]) const queue = [targetId] const result = [] while (queue.length > 0) { const currentId = queue.shift() for (const edge of allEdges) { if (edge.target !== currentId) continue if (visited.has(edge.source)) continue visited.add(edge.source) const sourceNode = nodeMap.get(edge.source) if (sourceNode) { result.push(sourceNode) queue.push(edge.source) } } } return result } /** * 获取起始节点的所有可达后继节点(通过边正向 BFS) * * 用于"输出变量引入":节点输出可关联到所有可达后继节点的输入字段。 * * @param {string} sourceId - 起始节点 ID * @param {Array} allNodes - 全部节点 * @param {Array} allEdges - 全部边 * @returns {Array} 可达后继节点列表(按 BFS 发现顺序,不含 sourceId 本身,不重复) */ export function getReachableSuccessors(sourceId, allNodes, allEdges) { const nodeMap = new Map(allNodes.map(n => [n.id, n])) const visited = new Set([sourceId]) const queue = [sourceId] const result = [] while (queue.length > 0) { const currentId = queue.shift() for (const edge of allEdges) { if (edge.source !== currentId) continue if (visited.has(edge.target)) continue visited.add(edge.target) const targetNode = nodeMap.get(edge.target) if (targetNode) { result.push(targetNode) queue.push(edge.target) } } } return result } /** * 推断指定边的样式 */ export function getEdgeStyle(status) { return EDGE_STATUS_STYLE[status] || EDGE_STATUS_STYLE.unmapped } /** * 刷新图中与指定节点关联的所有边的映射 * 返回需要更新的边列表 * @param {string} nodeId - 变更的节点 ID * @param {Array} nodes - 所有节点 * @param {Array} edges - 所有边 * @returns {Array} - 需要更新的边 [{ id, data, style }] */ export function refreshMappingsForNode(nodeId, nodes, edges) { const updates = [] const nodeMap = new Map(nodes.map(n => [n.id, n])) for (const edge of edges) { const isRelevant = edge.source === nodeId || edge.target === nodeId if (!isRelevant) continue const sourceNode = nodeMap.get(edge.source) const targetNode = nodeMap.get(edge.target) if (!sourceNode || !targetNode) continue const result = inferEdgeMapping(sourceNode, targetNode, edge, nodes, edges) updates.push({ id: edge.id, data: { ...edge.data, mapping: result.mapping, status: result.status, unmatchedSource: result.unmatchedSource, unmatchedTarget: result.unmatchedTarget, typeMismatches: result.typeMismatches }, style: getEdgeStyle(result.status) }) } return updates } /** * 推断一条新边的映射(用于 onConnect) */ export function inferNewEdge(sourceNode, targetNode, allNodes, allEdges) { const result = inferEdgeMapping(sourceNode, targetNode, undefined, allNodes, allEdges) return { data: { mapping: result.mapping, status: result.status, unmatchedSource: result.unmatchedSource, unmatchedTarget: result.unmatchedTarget, typeMismatches: result.typeMismatches }, style: getEdgeStyle(result.status) } } // ========== 多源汇聚分配 ========== /** * 计算多源汇聚时的变量分配方案 * @param {Array} sources - 源节点列表 * @param {object} target - 目标节点 * @param {Array} edges - 连接到目标的边列表 * @returns {{ edges: Array<{ edgeId, mapping, provided, missing }> }} */ export function resolveMultiSource(sources, target, edges) { const targetInputs = getNodeInputs(target) if (targetInputs.length === 0) { return { edges: edges.map(e => ({ edgeId: e.id, mapping: [], provided: [], missing: [] })) } } const allSourceOutputs = sources.map(s => getNodeOutputs(s)) const results = [] // 计算所有源的合并输出 const mergedOutputs = new Map() for (let i = 0; i < sources.length; i++) { for (const field of allSourceOutputs[i]) { if (!mergedOutputs.has(field.name)) { mergedOutputs.set(field.name, { field, sourceIndex: i }) } } } // 尝试匹配每个目标输入字段 const assigned = new Map() // targetField -> sourceIndex for (const tgtField of targetInputs) { const entry = mergedOutputs.get(tgtField.name) if (entry && isTypeCompatible(entry.field.type, tgtField.type)) { assigned.set(tgtField.name, entry.sourceIndex) } } // 分配回每条边 for (let i = 0; i < edges.length; i++) { const sourceOutputs = allSourceOutputs[i] || [] const edgeMapping = [] const provided = [] for (const tgtField of targetInputs) { if (assigned.get(tgtField.name) === i) { const srcField = sourceOutputs.find(f => f.name === tgtField.name) if (srcField) { edgeMapping.push({ sourceField: srcField.name, targetField: tgtField.name }) provided.push(tgtField.name) } } } const missing = targetInputs.filter(t => !assigned.has(t.name)).map(t => t.name) results.push({ edgeId: edges[i].id, mapping: edgeMapping, provided, missing }) } return { edges: results } }