From 235b33c335ace2539477be2fe1f26552ff90ecf8 Mon Sep 17 00:00:00 2001 From: RockChinQ Date: Wed, 1 Jul 2026 22:01:06 +0800 Subject: [PATCH] separate workflow control and data edges --- src/langbot/pkg/api/http/service/workflow.py | 1 + src/langbot/pkg/workflow/debug.py | 7 +- src/langbot/pkg/workflow/entities.py | 1 + src/langbot/pkg/workflow/executor.py | 46 +++- .../workflow-editor/PropertyPanel.tsx | 124 +++++++--- .../WorkflowEditorComponent.tsx | 1 - .../workflow-editor/WorkflowNodeComponent.tsx | 219 ++++++++++++------ .../workflow-editor/workflow-constants.ts | 38 +++ .../home/workflows/store/useWorkflowStore.ts | 142 +++++++++++- web/src/app/infra/entities/api/index.ts | 1 + web/src/app/infra/entities/workflow/index.ts | 1 + 11 files changed, 457 insertions(+), 124 deletions(-) diff --git a/src/langbot/pkg/api/http/service/workflow.py b/src/langbot/pkg/api/http/service/workflow.py index a7ddd028e..92e2b00b7 100644 --- a/src/langbot/pkg/api/http/service/workflow.py +++ b/src/langbot/pkg/api/http/service/workflow.py @@ -323,6 +323,7 @@ class WorkflowService: source_port=e.get('source_port', '') or e.get('sourceHandle', 'output'), target_node=e.get('target_node', '') or e.get('target', ''), target_port=e.get('target_port', '') or e.get('targetHandle', 'input'), + edge_type=e.get('edge_type') or (e.get('data', {}) or {}).get('edgeType') or 'legacy', condition=e.get('condition') or (e.get('data', {}) or {}).get('condition'), ) for e in definition.get('edges', []) diff --git a/src/langbot/pkg/workflow/debug.py b/src/langbot/pkg/workflow/debug.py index 383712037..31bc127df 100644 --- a/src/langbot/pkg/workflow/debug.py +++ b/src/langbot/pkg/workflow/debug.py @@ -441,12 +441,7 @@ class DebugWorkflowExecutor(WorkflowExecutor): ): continue - # Check if this node's inputs are ready - incoming_nodes = set() - for source_id, edges in edge_map.items(): - for edge in edges: - if edge.target_node == node.id: - incoming_nodes.add(source_id) + incoming_nodes = self._incoming_dependency_nodes(node.id, edge_map) # If no incoming nodes, it's a start node if not incoming_nodes: diff --git a/src/langbot/pkg/workflow/entities.py b/src/langbot/pkg/workflow/entities.py index b9a12b359..7b50faa5e 100644 --- a/src/langbot/pkg/workflow/entities.py +++ b/src/langbot/pkg/workflow/entities.py @@ -52,6 +52,7 @@ class EdgeDefinition(pydantic.BaseModel): source_port: str = 'output' target_node: str target_port: str = 'input' + edge_type: str = 'legacy' # control, data, or legacy (old mixed semantics) condition: Optional[str] = None # Optional condition expression diff --git a/src/langbot/pkg/workflow/executor.py b/src/langbot/pkg/workflow/executor.py index 6dcd8a89b..7f0540ed6 100644 --- a/src/langbot/pkg/workflow/executor.py +++ b/src/langbot/pkg/workflow/executor.py @@ -390,10 +390,12 @@ class WorkflowExecutor: else: inputs[key] = value - # Get inputs from connected upstream nodes via edges + # Get inputs from connected upstream nodes via data edges. # Build a reverse map: for each incoming edge to this node, find the # source node and the specific source/target port. for edge in self._edges: + if not self._is_data_edge(edge): + continue if edge.target_node != node.id: continue source_state = context.node_states.get(edge.source_node) @@ -519,13 +521,8 @@ class WorkflowExecutor: async def _inputs_ready( self, node: NodeDefinition, edge_map: dict[str, list[EdgeDefinition]], context: ExecutionContext ) -> bool: - """Check if all inputs for a node are ready""" - # Find all edges that connect to this node - incoming_nodes = set() - for source_id, edges in edge_map.items(): - for edge in edges: - if edge.target_node == node.id: - incoming_nodes.add(source_id) + """Check if all control predecessors and data providers are ready.""" + incoming_nodes = self._incoming_dependency_nodes(node.id, edge_map) # Check if all incoming nodes have completed for source_id in incoming_nodes: @@ -537,7 +534,7 @@ class WorkflowExecutor: def _find_start_nodes(self, nodes: list[NodeDefinition], edges: list[EdgeDefinition]) -> list[NodeDefinition]: """Find nodes that have no incoming edges (start nodes)""" - target_nodes = {edge.target_node for edge in edges} + target_nodes = {edge.target_node for edge in edges if self._is_control_edge(edge) or self._is_data_edge(edge)} start_nodes = [node for node in nodes if node.id not in target_nodes] # Also check for trigger nodes @@ -549,14 +546,43 @@ class WorkflowExecutor: return start_nodes def _build_edge_map(self, edges: list[EdgeDefinition]) -> dict[str, list[EdgeDefinition]]: - """Build a map of source node ID to outgoing edges""" + """Build a map of source node ID to outgoing control edges.""" edge_map: dict[str, list[EdgeDefinition]] = {} for edge in edges: + if not self._is_control_edge(edge): + continue if edge.source_node not in edge_map: edge_map[edge.source_node] = [] edge_map[edge.source_node].append(edge) return edge_map + def _edge_type(self, edge: EdgeDefinition) -> str: + edge_type = (getattr(edge, 'edge_type', None) or 'legacy').strip().lower() + if edge_type not in {'control', 'data', 'legacy'}: + return 'legacy' + return edge_type + + def _is_control_edge(self, edge: EdgeDefinition) -> bool: + return self._edge_type(edge) in {'control', 'legacy'} + + def _is_data_edge(self, edge: EdgeDefinition) -> bool: + return self._edge_type(edge) in {'data', 'legacy'} + + def _incoming_dependency_nodes( + self, node_id: str, edge_map: dict[str, list[EdgeDefinition]] + ) -> set[str]: + incoming_nodes: set[str] = set() + for source_id, edges in edge_map.items(): + for edge in edges: + if edge.target_node == node_id: + incoming_nodes.add(source_id) + + for edge in self._edges: + if self._is_data_edge(edge) and edge.target_node == node_id: + incoming_nodes.add(edge.source_node) + + return incoming_nodes + def _record_execution_step(self, node: NodeDefinition, node_state: NodeState, context: ExecutionContext): """Record an execution step in the history""" duration_ms = 0 diff --git a/web/src/app/home/workflows/components/workflow-editor/PropertyPanel.tsx b/web/src/app/home/workflows/components/workflow-editor/PropertyPanel.tsx index a3aefd64a..bfdfa2730 100644 --- a/web/src/app/home/workflows/components/workflow-editor/PropertyPanel.tsx +++ b/web/src/app/home/workflows/components/workflow-editor/PropertyPanel.tsx @@ -51,6 +51,11 @@ import { ScrollArea } from '@/components/ui/scroll-area'; import { toast } from 'sonner'; import { resolveI18nLabel, maybeTranslateKey } from './workflow-i18n'; +import { + WorkflowEdgeType, + isControlHandle, + normalizeWorkflowEdgeType, +} from './workflow-constants'; // Delegate to shared utility const translateIfKey = (value: string | undefined): string | undefined => { @@ -76,6 +81,29 @@ const getTypeLabel = (type: string | undefined): string => { return translated === i18nKey ? type : translated; }; +const getEdgeTypeLabel = ( + edgeType: WorkflowEdgeType, + t: (key: string, options?: { defaultValue: string }) => string, +): string => { + if (edgeType === 'control') { + return t('workflows.edgeTypes.control', { defaultValue: 'Control' }); + } + if (edgeType === 'data') { + return t('workflows.edgeTypes.data', { defaultValue: 'Data' }); + } + return t('workflows.edgeTypes.legacy', { defaultValue: 'Legacy' }); +}; + +const getEdgeTypeBadgeClass = (edgeType: WorkflowEdgeType): string => { + if (edgeType === 'control') { + return 'border-orange-300 bg-orange-50 text-orange-700 dark:border-orange-800 dark:bg-orange-950/40 dark:text-orange-300'; + } + if (edgeType === 'data') { + return 'border-sky-300 bg-sky-50 text-sky-700 dark:border-sky-800 dark:bg-sky-950/40 dark:text-sky-300'; + } + return 'border-slate-300 bg-slate-50 text-slate-700 dark:border-slate-700 dark:bg-slate-900/60 dark:text-slate-300'; +}; + interface PropertyPanelProps { selectedNodeId: string | null; selectedEdgeId: string | null; @@ -275,7 +303,11 @@ export default function PropertyPanel({ }[] = []; // Find all upstream nodes - const incomingEdges = edges.filter((e) => e.target === selectedNode.id); + const incomingEdges = edges.filter( + (e) => + e.target === selectedNode.id && + normalizeWorkflowEdgeType(e.data?.edgeType) !== 'control', + ); const upstreamNodeIds = incomingEdges.map((e) => e.source); for (const nodeId of upstreamNodeIds) { @@ -414,6 +446,17 @@ export default function PropertyPanel({ if (selectedEdge) { const sourceNode = nodes.find((n) => n.id === selectedEdge.source); const targetNode = nodes.find((n) => n.id === selectedEdge.target); + const edgeType = normalizeWorkflowEdgeType(selectedEdge.data?.edgeType); + const isDataLikeEdge = edgeType !== 'control'; + const isControlLikeEdge = edgeType !== 'data'; + const sourcePort = + selectedEdge.sourceHandle && !isControlHandle(selectedEdge.sourceHandle) + ? selectedEdge.sourceHandle + : undefined; + const targetPort = + selectedEdge.targetHandle && !isControlHandle(selectedEdge.targetHandle) + ? selectedEdge.targetHandle + : undefined; return ( @@ -422,6 +465,12 @@ export default function PropertyPanel({

{t('workflows.edgeProperties')} + + {getEdgeTypeLabel(edgeType, t)} +

@@ -444,37 +493,56 @@ export default function PropertyPanel({ {targetNode?.data.label || selectedEdge.target} + {isDataLikeEdge && ( +
+ + {sourcePort || 'output'} + + + + {targetPort || 'input'} + +
+ )} {/* Condition */} - - {t('workflows.hasCondition')} - - ) : null - } - > -
-