feat: add dataframe to Loop component, fixes loop input error

src/backend/base/langflow/components/logic/__init__.py

@@ -3,12 +3,14 @@
 from .flow_tool import FlowToolComponent
 from .listen import ListenComponent
 from .loop import LoopComponent
+from .loop_basic import BasicLoopComponent
 from .notify import NotifyComponent
 from .pass_message import PassMessageComponent
 from .run_flow import RunFlowComponent
 from .sub_flow import SubFlowComponent
 
 __all__ = [
+    "BasicLoopComponent",
     "ConditionalRouterComponent",
     "DataConditionalRouterComponent",
     "FlowToolComponent",

src/backend/base/langflow/components/logic/loop.py

@@ -9,6 +9,7 @@ class LoopComponent(Component):
         "Iterates over a list of Data objects, outputting one item at a time and aggregating results from loop inputs."
     )
     icon = "infinity"
+    legacy = True
 
     inputs = [
         DataInput(

src/backend/base/langflow/components/logic/loop_basic.py

@@ -0,0 +1,116 @@
+from langflow.custom import Component
+from langflow.io import HandleInput, Output
+from langflow.schema import Data
+from langflow.schema.dataframe import DataFrame
+
+
+class BasicLoopComponent(Component):
+    display_name = "Loop"
+    description = (
+        "Iterates over a list of Data objects, outputting one item at a time and aggregating results from loop inputs."
+    )
+    icon = "infinity"
+
+    inputs = [
+        HandleInput(
+            name="data",
+            display_name="Data or DataFrame",
+            info="The initial list of Data objects or DataFrame to iterate over.",
+            input_types=["Data", "DataFrame"],
+        ),
+    ]
+
+    outputs = [
+        Output(display_name="Item", name="item", method="item_output", allows_loop=True),
+        Output(display_name="Done", name="done", method="done_output"),
+    ]
+
+    def initialize_data(self) -> None:
+        """Initialize the data list, context index, and aggregated list."""
+        if self.ctx.get(f"{self._id}_initialized", False):
+            return
+
+        # Ensure data is a list of Data objects
+        data_list = self._validate_data(self.data)
+
+        # Store the initial data and context variables
+        self.update_ctx(
+            {
+                f"{self._id}_data": data_list,
+                f"{self._id}_index": 0,
+                f"{self._id}_aggregated": [],
+                f"{self._id}_initialized": True,
+            }
+        )
+
+    def _validate_data(self, data):
+        """Validate and return a list of Data objects."""
+        if isinstance(data, DataFrame):
+            return data.to_data_list()
+        if isinstance(data, Data):
+            return [data]
+        if isinstance(data, list) and all(isinstance(item, Data) for item in data):
+            return data
+        msg = "The 'data' input must be a DataFrame, a list of Data objects, or a single Data object."
+        raise TypeError(msg)
+
+    def evaluate_stop_loop(self) -> bool:
+        """Evaluate whether to stop item or done output."""
+        current_index = self.ctx.get(f"{self._id}_index", 0)
+        data_length = len(self.ctx.get(f"{self._id}_data", []))
+        return current_index > data_length
+
+    def item_output(self) -> Data:
+        """Output the next item in the list or stop if done."""
+        self.initialize_data()
+        current_item = Data(text="")
+
+        if self.evaluate_stop_loop():
+            self.stop("item")
+            return Data(text="")
+
+        # Get data list and current index
+        data_list, current_index = self.loop_variables()
+        if current_index < len(data_list):
+            # Output current item and increment index
+            try:
+                current_item = data_list[current_index]
+            except IndexError:
+                current_item = Data(text="")
+        self.aggregated_output()
+        self.update_ctx({f"{self._id}_index": current_index + 1})
+        return current_item
+
+    def done_output(self) -> DataFrame:
+        """Trigger the done output when iteration is complete."""
+        self.initialize_data()
+
+        if self.evaluate_stop_loop():
+            self.stop("item")
+            self.start("done")
+
+            aggregated = self.ctx.get(f"{self._id}_aggregated", [])
+
+            return DataFrame(aggregated)
+        self.stop("done")
+        return DataFrame([])
+
+    def loop_variables(self):
+        """Retrieve loop variables from context."""
+        return (
+            self.ctx.get(f"{self._id}_data", []),
+            self.ctx.get(f"{self._id}_index", 0),
+        )
+
+    def aggregated_output(self) -> list[Data]:
+        """Return the aggregated list once all items are processed."""
+        self.initialize_data()
+
+        # Get data list and aggregated list
+        data_list = self.ctx.get(f"{self._id}_data", [])
+        aggregated = self.ctx.get(f"{self._id}_aggregated", [])
+        loop_input = self.get_loop_output_value("item")
+        if loop_input is not None and not isinstance(loop_input, str) and len(aggregated) <= len(data_list):
+            aggregated.append(loop_input)
+            self.update_ctx({f"{self._id}_aggregated": aggregated})
+        return aggregated

src/backend/base/langflow/custom/custom_component/component.py

@@ -717,7 +717,27 @@ async def _run(self):
 
         return await self.build_results()
 
+    def get_loop_output_value(self, name: str) -> Any:
+        """Get the value of an output that allows looping.
+
+        Args:
+            name (str): The name of the output to check.
+
+        Returns:
+            Any: The value from the vertex if the output allows looping,
+                 otherwise returns None.
+        """
+        if not hasattr(self, "_vertex") or self._vertex is None:
+            return None
+
+        if any(
+            getattr(output, "allows_loop", False) and output.name == name for output in getattr(self, "outputs", [])
+        ):
+            return self._vertex.get_value_from_output_names(name)
+        return None
+
     def __getattr__(self, name: str) -> Any:
+        # First check if it's a loop output
         if "_attributes" in self.__dict__ and name in self.__dict__["_attributes"]:
             return self.__dict__["_attributes"][name]
         if "_inputs" in self.__dict__ and name in self.__dict__["_inputs"]:
@@ -737,6 +757,7 @@ def __getattr__(self, name: str) -> Any:
             return PlaceholderGraph(
                 flow_id=flow_id, user_id=str(user_id), session_id=session_id, context={}, flow_name=flow_name
             )
+
         msg = f"Attribute {name} not found in {self.__class__.__name__}"
         raise AttributeError(msg)
 

src/backend/base/langflow/graph/vertex/base.py

@@ -270,10 +270,33 @@ def parse_data(self) -> None:
                     self.base_type = base_type
                     break
 
-    def get_value_from_output_names(self, key: str):
-        if key in self.output_names:
-            return self.graph.get_vertex(key)
-        return None
+    def get_value_from_output_names(self, key: str) -> Any:
+        """Get value from output names.
+
+        Args:
+            key (str): Name of output.
+
+        Returns:
+            Any: Value of output.
+        """
+        # Find edges where this vertex is the target and the target handle's field name matches the key
+        edges = [edge for edge in self.edges if edge.target_id == self.id and edge.target_handle.field_name == key]
+
+        if not edges:
+            return None
+
+        # Get the source vertex and its output value
+        edge = edges[0]  # Take the first matching edge
+        source_vertex = self.graph.get_vertex(edge.source_id)
+        if not source_vertex:
+            return None
+
+        # Get the output value from the source vertex's results
+        source_handle = edge.source_handle
+        if not source_handle or not source_handle.name:
+            return None
+
+        return source_vertex.results.get(source_handle.name)
 
     def get_value_from_template_dict(self, key: str):
         template_dict = self.data.get("node", {}).get("template", {})
@@ -307,6 +330,9 @@ def _set_params_from_normal_edge(self, params: dict, edge: Edge, template_dict:
                 else:
                     params[param_key] = self.graph.get_vertex(edge.source_id)
         elif param_key in self.output_names:
+            # If this is a loop output being used as an input
+            params[param_key] = self.get_value_from_output_names(param_key)
+        else:
             params[param_key] = self.graph.get_vertex(edge.source_id)
         return params