feat(api): Implement variable mapping extraction for VariableAssignerNode

api/core/workflow/nodes/variable_assigner/v1/node.py

@@ -18,6 +18,23 @@ class VariableAssignerNode(BaseNode[VariableAssignerData]):
     def version(cls) -> str:
         return "1"
 
+    @classmethod
+    def _extract_variable_selector_to_variable_mapping(
+        cls,
+        *,
+        graph_config: Mapping[str, Any],
+        node_id: str,
+        node_data: VariableAssignerData,
+    ) -> Mapping[str, Sequence[str]]:
+        assigned_variable_node_id = node_data.assigned_variable_selector[0]
+        if assigned_variable_node_id != CONVERSATION_VARIABLE_NODE_ID:
+            return {}
+        selector_key = ".".join(node_data.assigned_variable_selector)
+        key = f"{node_id}.#{selector_key}#"
+        return {
+            key: node_data.assigned_variable_selector,
+        }
+
     def _run(self) -> NodeRunResult:
         assigned_variable_selector = self.node_data.assigned_variable_selector
         # Should be String, Number, Object, ArrayString, ArrayNumber, ArrayObject

api/core/workflow/nodes/variable_assigner/v2/node.py

@@ -33,6 +33,24 @@ class VariableAssignerNode(BaseNode[VariableAssignerNodeData]):
     def version(cls) -> str:
         return "2"
 
+    @classmethod
+    def _extract_variable_selector_to_variable_mapping(
+        cls,
+        *,
+        graph_config: Mapping[str, Any],
+        node_id: str,
+        node_data: VariableAssignerNodeData,
+    ) -> Mapping[str, Sequence[str]]:
+        var_mapping: dict[str, Sequence[str]] = {}
+        for item in node_data.items:
+            selector_node_id = item.variable_selector[0]
+            if selector_node_id != CONVERSATION_VARIABLE_NODE_ID:
+                continue
+            selector_str = ".".join(item.variable_selector)
+            key = f"{node_id}.#{selector_str}#"
+            var_mapping[key] = item.variable_selector
+        return var_mapping
+
     def _run(self) -> NodeRunResult:
         inputs = self.node_data.model_dump()
         process_data: dict[str, Any] = {}