diff --git a/src/backend/base/langflow/initial_setup/starter_projects/Research Translation Loop.json b/src/backend/base/langflow/initial_setup/starter_projects/Research Translation Loop.json index 70d28b0fa8..f83b4c8fed 100644 --- a/src/backend/base/langflow/initial_setup/starter_projects/Research Translation Loop.json +++ b/src/backend/base/langflow/initial_setup/starter_projects/Research Translation Loop.json @@ -127,7 +127,9 @@ "id": "LoopComponent-GtPZT", "inputTypes": [ "DataFrame", - "Table" + "Table", + "Data", + "Message" ], "type": "other" } @@ -136,7 +138,7 @@ "source": "ArXivComponent-tAHR5", "sourceHandle": "{œdataTypeœ: œArXivComponentœ, œidœ: œArXivComponent-tAHR5œ, œnameœ: œdataframeœ, œoutput_typesœ: [œTableœ]}", "target": "LoopComponent-GtPZT", - "targetHandle": "{œfieldNameœ: œdataœ, œidœ: œLoopComponent-GtPZTœ, œinputTypesœ: [œDataFrameœ, œTableœ], œtypeœ: œotherœ}" + "targetHandle": "{œfieldNameœ: œdataœ, œidœ: œLoopComponent-GtPZTœ, œinputTypesœ: [œDataFrameœ, œTableœ, œDataœ, œMessageœ], œtypeœ: œotherœ}" }, { "className": "", @@ -1327,7 +1329,7 @@ "icon": "infinity", "legacy": false, "metadata": { - "code_hash": "0b08e58a5c70", + "code_hash": "2712a1de7210", "dependencies": { "dependencies": [ { @@ -1355,6 +1357,7 @@ "selected": "JSON", "tool_mode": true, "types": [ + "Data", "JSON" ], "value": "__UNDEFINED__" @@ -1369,6 +1372,7 @@ "selected": "Table", "tool_mode": true, "types": [ + "DataFrame", "Table" ], "value": "__UNDEFINED__" @@ -1393,17 +1397,19 @@ "show": true, "title_case": false, "type": "code", - "value": "from lfx.base.flow_controls.loop_utils import (\n execute_loop_body,\n extract_loop_output,\n get_loop_body_start_edge,\n get_loop_body_start_vertex,\n get_loop_body_vertices,\n validate_data_input,\n)\nfrom lfx.components.processing.converter import convert_to_data\nfrom lfx.custom.custom_component.component import Component\nfrom lfx.inputs.inputs import HandleInput\nfrom lfx.schema.data import Data\nfrom lfx.schema.dataframe import DataFrame\nfrom lfx.schema.message import Message\nfrom lfx.template.field.base import Output\n\n\nclass LoopComponent(Component):\n display_name = \"Loop\"\n description = (\n \"Iterates through Data or Message objects, processing items individually \"\n \"and aggregating results from loop inputs.\"\n )\n documentation: str = \"https://docs.langflow.org/loop\"\n icon = \"infinity\"\n\n inputs = [\n HandleInput(\n name=\"data\",\n display_name=\"Inputs\",\n info=\"The initial DataFrame to iterate over.\",\n input_types=[\"DataFrame\", \"Table\"],\n ),\n ]\n\n outputs = [\n Output(\n display_name=\"Item\",\n name=\"item\",\n method=\"item_output\",\n allows_loop=True,\n loop_types=[\"Message\"],\n group_outputs=True,\n ),\n Output(display_name=\"Done\", name=\"done\", method=\"done_output\", group_outputs=True),\n ]\n\n def initialize_data(self) -> None:\n \"\"\"Initialize the data list and context index.\n\n Seeds the input list and index counter in ctx. The aggregated results\n are owned by `_iterate`, which writes them once the subgraph finishes.\n \"\"\"\n if self.ctx.get(f\"{self._id}_initialized\", False):\n return\n\n # Ensure data is a list of Data objects\n data_list = self._validate_data(self.data)\n\n # Store the initial data and context variables\n self.update_ctx(\n {\n f\"{self._id}_data\": data_list,\n f\"{self._id}_index\": 0,\n f\"{self._id}_initialized\": True,\n }\n )\n\n def _convert_message_to_data(self, message: Message) -> Data:\n \"\"\"Convert a Message object to a Data object using Type Convert logic.\"\"\"\n return convert_to_data(message, auto_parse=False)\n\n def _validate_data(self, data):\n \"\"\"Validate and return a list of Data objects.\"\"\"\n return validate_data_input(data)\n\n def get_loop_body_vertices(self) -> set[str]:\n \"\"\"Identify vertices in this loop's body via graph traversal.\n\n Traverses from the loop's \"item\" output to the vertex that feeds back\n to the loop's \"item\" input, collecting all vertices in between.\n This naturally handles nested loops by stopping at this loop's feedback edge.\n\n Returns:\n Set of vertex IDs that form this loop's body\n \"\"\"\n # Check if we have a proper graph context\n if not hasattr(self, \"_vertex\") or self._vertex is None:\n return set()\n\n return get_loop_body_vertices(\n vertex=self._vertex,\n graph=self.graph,\n get_incoming_edge_by_target_param_fn=self.get_incoming_edge_by_target_param,\n )\n\n def _get_loop_body_start_vertex(self) -> str | None:\n \"\"\"Get the first vertex in the loop body (connected to loop's item output).\n\n Returns:\n The vertex ID of the first vertex in the loop body, or None if not found\n \"\"\"\n # Check if we have a proper graph context\n if not hasattr(self, \"_vertex\") or self._vertex is None:\n return None\n\n return get_loop_body_start_vertex(vertex=self._vertex)\n\n def _extract_loop_output(self, results: list) -> Data:\n \"\"\"Extract the output from subgraph execution results.\n\n Args:\n results: List of VertexBuildResult objects from subgraph execution\n\n Returns:\n Data object containing the loop iteration output\n \"\"\"\n # Get the vertex ID that feeds back to the item input (end of loop body)\n end_vertex_id = self.get_incoming_edge_by_target_param(\"item\")\n return extract_loop_output(results=results, end_vertex_id=end_vertex_id)\n\n async def execute_loop_body(self, data_list: list[Data], event_manager=None) -> list[Data]:\n \"\"\"Execute loop body for each data item.\n\n Creates an isolated subgraph for the loop body and executes it\n for each item in the data list, collecting results.\n\n Args:\n data_list: List of Data objects to iterate over\n event_manager: Optional event manager to pass to subgraph execution for UI events\n\n Returns:\n List of Data objects containing results from each iteration\n \"\"\"\n # Get the loop body configuration once\n loop_body_vertex_ids = self.get_loop_body_vertices()\n start_vertex_id = self._get_loop_body_start_vertex()\n start_edge = get_loop_body_start_edge(self._vertex)\n end_vertex_id = self.get_incoming_edge_by_target_param(\"item\")\n\n return await execute_loop_body(\n graph=self.graph,\n data_list=data_list,\n loop_body_vertex_ids=loop_body_vertex_ids,\n start_vertex_id=start_vertex_id,\n start_edge=start_edge,\n end_vertex_id=end_vertex_id,\n event_manager=event_manager,\n )\n\n async def _iterate(self) -> list[Data]:\n \"\"\"Run the loop body subgraph once and cache the aggregated results.\n\n Both `item_output` and `done_output` may be called during the same\n vertex build (if both outputs have downstream consumers), so this\n helper is idempotent: it caches the aggregated results in ctx under\n `{self._id}_aggregated` and guards re-entry with `{self._id}_iterated`.\n If a prior call raised, the exception is cached in\n `{self._id}_iteration_error` and re-raised on subsequent calls so\n repeat invocations surface the same failure instead of silently\n returning empty data or re-running the subgraph.\n\n Production constructs a fresh Graph per request (via\n `Graph.from_payload`), so `ctx` is effectively per-run and the\n cached `_iterated` flag does not leak across executions.\n \"\"\"\n if self.ctx.get(f\"{self._id}_iterated\", False):\n cached_error = self.ctx.get(f\"{self._id}_iteration_error\")\n if cached_error is not None:\n raise cached_error\n return self.ctx.get(f\"{self._id}_aggregated\", [])\n\n import time\n\n started_at = time.perf_counter()\n try:\n self.initialize_data()\n data_list = self.ctx.get(f\"{self._id}_data\", [])\n self.log(f\"Starting loop over {len(data_list)} item(s)\", name=\"Start\")\n\n if not data_list:\n self.update_ctx({f\"{self._id}_aggregated\": [], f\"{self._id}_iterated\": True})\n self.log(\"No items to iterate, skipping loop body\", name=\"Skipped\")\n return []\n\n aggregated_results = await self.execute_loop_body(data_list, event_manager=self._event_manager)\n except Exception as exc:\n from lfx.log.logger import logger\n\n elapsed = time.perf_counter() - started_at\n self.log(f\"Loop failed after {elapsed:.3f}s: {exc}\", name=\"Error\")\n await logger.aexception(f\"Loop {self._id} failed while executing loop body\")\n self.update_ctx({f\"{self._id}_iteration_error\": exc, f\"{self._id}_iterated\": True})\n raise\n\n elapsed = time.perf_counter() - started_at\n self.log(\n f\"Completed {len(aggregated_results)} iteration(s) in {elapsed:.3f}s\",\n name=\"Complete\",\n )\n self.update_ctx({f\"{self._id}_aggregated\": aggregated_results, f\"{self._id}_iterated\": True})\n return aggregated_results\n\n async def item_output(self) -> Data:\n \"\"\"Display the inputs dispatched to the loop body.\n\n Also triggers the iteration (idempotent) when `done` has no\n downstream consumer, so the loop still runs if only the Item\n output is connected. When `done` IS connected we leave iteration\n to `done_output`. The `item` branch is stopped in both cases so\n downstream vertices aren't executed by the outer graph (the loop\n body runs internally via the subgraph in `_iterate`).\n\n Returns a `Data` envelope so the outer item edge remains\n compatible with Data-typed consumers in the loop body. The\n wrapped payload exposes the iterated rows for inspection.\n \"\"\"\n self.stop(\"item\")\n if self._vertex is not None and \"done\" not in self._vertex.edges_source_names:\n await self._iterate()\n data_list = self.ctx.get(f\"{self._id}_data\", [])\n return Data(data={\"count\": len(data_list), \"items\": [d.data for d in data_list]})\n\n async def done_output(self) -> DataFrame:\n \"\"\"Return the aggregated results from the loop iteration.\n\n Triggers the iteration if it hasn't run yet (for example when only\n `done` is connected and `item_output` never executed). In the common\n case where `item` is also connected, `_iterate` has already cached\n the aggregated results in ctx and this call is a cheap read, so the\n order in which the two outputs are evaluated does not matter.\n \"\"\"\n aggregated_results = await self._iterate()\n return DataFrame(aggregated_results)\n" + "value": "from lfx.base.flow_controls.loop_utils import (\n execute_loop_body,\n extract_loop_output,\n get_loop_body_start_edge,\n get_loop_body_start_vertex,\n get_loop_body_vertices,\n validate_data_input,\n)\nfrom lfx.components.processing.converter import convert_to_data\nfrom lfx.custom.custom_component.component import Component\nfrom lfx.inputs.inputs import HandleInput\nfrom lfx.schema.data import Data\nfrom lfx.schema.dataframe import DataFrame\nfrom lfx.schema.message import Message\nfrom lfx.template.field.base import Output\n\n\nclass LoopComponent(Component):\n display_name = \"Loop\"\n description = (\n \"Iterates through Data or Message objects, processing items individually \"\n \"and aggregating results from loop inputs.\"\n )\n documentation: str = \"https://docs.langflow.org/loop\"\n icon = \"infinity\"\n\n inputs = [\n HandleInput(\n name=\"data\",\n display_name=\"Inputs\",\n info=\"Data to iterate over. Accepts DataFrame, Table, Data, or Message.\",\n input_types=[\"DataFrame\", \"Table\", \"Data\", \"Message\"],\n ),\n ]\n\n outputs = [\n Output(\n display_name=\"Item\",\n name=\"item\",\n method=\"item_output\",\n types=[\"Data\"],\n allows_loop=True,\n loop_types=[\"Message\"],\n group_outputs=True,\n ),\n Output(\n display_name=\"Done\",\n name=\"done\",\n method=\"done_output\",\n types=[\"DataFrame\", \"Table\"],\n group_outputs=True,\n ),\n ]\n\n def initialize_data(self) -> None:\n \"\"\"Initialize the data list and context index.\n\n Seeds the input list and index counter in ctx. The aggregated results\n are owned by `_iterate`, which writes them once the subgraph finishes.\n \"\"\"\n if self.ctx.get(f\"{self._id}_initialized\", False):\n return\n\n # Ensure data is a list of Data objects\n data_list = self._validate_data(self.data)\n\n # Store the initial data and context variables\n self.update_ctx(\n {\n f\"{self._id}_data\": data_list,\n f\"{self._id}_index\": 0,\n f\"{self._id}_initialized\": True,\n }\n )\n\n def _convert_message_to_data(self, message: Message) -> Data:\n \"\"\"Convert a Message object to a Data object using Type Convert logic.\"\"\"\n return convert_to_data(message, auto_parse=False)\n\n def _validate_data(self, data):\n \"\"\"Validate and normalize the input to a flat list of Data objects.\n\n Coerces each accepted input type into Data so `validate_data_input`\n always receives a homogeneous payload:\n - Message -> a clean Data via `_convert_message_to_data` (just the\n message payload, not the full envelope). `Message` subclasses\n `Data`, so without this it would slip through as a single Data\n carrying all of the message metadata.\n - DataFrame/Table -> expanded to its rows.\n - Data / list[Data] -> passed through unchanged.\n\n Lists are normalized item-by-item so a mixed `[Message, DataFrame,\n Data]` still yields `list[Data]` instead of being rejected as a\n mixed-type list (`validate_data_input` requires every list item to be\n a Data, and DataFrame is not a Data subclass).\n \"\"\"\n if isinstance(data, Message):\n data = self._convert_message_to_data(data)\n elif isinstance(data, list):\n normalized: list = []\n for item in data:\n if isinstance(item, Message):\n normalized.append(self._convert_message_to_data(item))\n elif isinstance(item, DataFrame):\n normalized.extend(item.to_data_list())\n else:\n normalized.append(item)\n data = normalized\n return validate_data_input(data)\n\n def get_loop_body_vertices(self) -> set[str]:\n \"\"\"Identify vertices in this loop's body via graph traversal.\n\n Traverses from the loop's \"item\" output to the vertex that feeds back\n to the loop's \"item\" input, collecting all vertices in between.\n This naturally handles nested loops by stopping at this loop's feedback edge.\n\n Returns:\n Set of vertex IDs that form this loop's body\n \"\"\"\n # Check if we have a proper graph context\n if not hasattr(self, \"_vertex\") or self._vertex is None:\n return set()\n\n return get_loop_body_vertices(\n vertex=self._vertex,\n graph=self.graph,\n get_incoming_edge_by_target_param_fn=self.get_incoming_edge_by_target_param,\n )\n\n def _get_loop_body_start_vertex(self) -> str | None:\n \"\"\"Get the first vertex in the loop body (connected to loop's item output).\n\n Returns:\n The vertex ID of the first vertex in the loop body, or None if not found\n \"\"\"\n # Check if we have a proper graph context\n if not hasattr(self, \"_vertex\") or self._vertex is None:\n return None\n\n return get_loop_body_start_vertex(vertex=self._vertex)\n\n def _extract_loop_output(self, results: list) -> Data:\n \"\"\"Extract the output from subgraph execution results.\n\n Args:\n results: List of VertexBuildResult objects from subgraph execution\n\n Returns:\n Data object containing the loop iteration output\n \"\"\"\n # Get the vertex ID that feeds back to the item input (end of loop body)\n end_vertex_id = self.get_incoming_edge_by_target_param(\"item\")\n return extract_loop_output(results=results, end_vertex_id=end_vertex_id)\n\n async def execute_loop_body(self, data_list: list[Data], event_manager=None) -> list[Data]:\n \"\"\"Execute loop body for each data item.\n\n Creates an isolated subgraph for the loop body and executes it\n for each item in the data list, collecting results.\n\n Args:\n data_list: List of Data objects to iterate over\n event_manager: Optional event manager to pass to subgraph execution for UI events\n\n Returns:\n List of Data objects containing results from each iteration\n \"\"\"\n # Get the loop body configuration once\n loop_body_vertex_ids = self.get_loop_body_vertices()\n start_vertex_id = self._get_loop_body_start_vertex()\n start_edge = get_loop_body_start_edge(self._vertex)\n end_vertex_id = self.get_incoming_edge_by_target_param(\"item\")\n\n return await execute_loop_body(\n graph=self.graph,\n data_list=data_list,\n loop_body_vertex_ids=loop_body_vertex_ids,\n start_vertex_id=start_vertex_id,\n start_edge=start_edge,\n end_vertex_id=end_vertex_id,\n event_manager=event_manager,\n )\n\n async def _iterate(self) -> list[Data]:\n \"\"\"Run the loop body subgraph once and cache the aggregated results.\n\n Both `item_output` and `done_output` may be called during the same\n vertex build (if both outputs have downstream consumers), so this\n helper is idempotent: it caches the aggregated results in ctx under\n `{self._id}_aggregated` and guards re-entry with `{self._id}_iterated`.\n If a prior call raised, the exception is cached in\n `{self._id}_iteration_error` and re-raised on subsequent calls so\n repeat invocations surface the same failure instead of silently\n returning empty data or re-running the subgraph.\n\n Production constructs a fresh Graph per request (via\n `Graph.from_payload`), so `ctx` is effectively per-run and the\n cached `_iterated` flag does not leak across executions.\n \"\"\"\n if self.ctx.get(f\"{self._id}_iterated\", False):\n cached_error = self.ctx.get(f\"{self._id}_iteration_error\")\n if cached_error is not None:\n raise cached_error\n return self.ctx.get(f\"{self._id}_aggregated\", [])\n\n import time\n\n started_at = time.perf_counter()\n try:\n self.initialize_data()\n data_list = self.ctx.get(f\"{self._id}_data\", [])\n self.log(f\"Starting loop over {len(data_list)} item(s)\", name=\"Start\")\n\n if not data_list:\n self.update_ctx({f\"{self._id}_aggregated\": [], f\"{self._id}_iterated\": True})\n self.log(\"No items to iterate, skipping loop body\", name=\"Skipped\")\n return []\n\n aggregated_results = await self.execute_loop_body(data_list, event_manager=self._event_manager)\n except Exception as exc:\n from lfx.log.logger import logger\n\n elapsed = time.perf_counter() - started_at\n self.log(f\"Loop failed after {elapsed:.3f}s: {exc}\", name=\"Error\")\n await logger.aexception(f\"Loop {self._id} failed while executing loop body\")\n self.update_ctx({f\"{self._id}_iteration_error\": exc, f\"{self._id}_iterated\": True})\n raise\n\n elapsed = time.perf_counter() - started_at\n self.log(\n f\"Completed {len(aggregated_results)} iteration(s) in {elapsed:.3f}s\",\n name=\"Complete\",\n )\n self.update_ctx({f\"{self._id}_aggregated\": aggregated_results, f\"{self._id}_iterated\": True})\n return aggregated_results\n\n async def item_output(self) -> Data:\n \"\"\"Display the inputs dispatched to the loop body.\n\n Also triggers the iteration (idempotent) when `done` has no\n downstream consumer, so the loop still runs if only the Item\n output is connected. When `done` IS connected we leave iteration\n to `done_output`. The `item` branch is stopped in both cases so\n downstream vertices aren't executed by the outer graph (the loop\n body runs internally via the subgraph in `_iterate`).\n\n Returns a `Data` envelope so the outer item edge remains\n compatible with Data-typed consumers in the loop body. The\n wrapped payload exposes the iterated rows for inspection.\n \"\"\"\n self.stop(\"item\")\n if self._vertex is not None and \"done\" not in self._vertex.edges_source_names:\n await self._iterate()\n data_list = self.ctx.get(f\"{self._id}_data\", [])\n return Data(data={\"count\": len(data_list), \"items\": [d.data for d in data_list]})\n\n async def done_output(self) -> DataFrame:\n \"\"\"Return the aggregated results from the loop iteration.\n\n Triggers the iteration if it hasn't run yet (for example when only\n `done` is connected and `item_output` never executed). In the common\n case where `item` is also connected, `_iterate` has already cached\n the aggregated results in ctx and this call is a cheap read, so the\n order in which the two outputs are evaluated does not matter.\n \"\"\"\n aggregated_results = await self._iterate()\n return DataFrame(aggregated_results)\n" }, "data": { "_input_type": "HandleInput", "advanced": false, "display_name": "Inputs", "dynamic": false, - "info": "The initial DataFrame to iterate over.", + "info": "Data to iterate over. Accepts DataFrame, Table, Data, or Message.", "input_types": [ "DataFrame", - "Table" + "Table", + "Data", + "Message" ], "list": false, "list_add_label": "Add More", diff --git a/src/backend/base/langflow/locales/en.json b/src/backend/base/langflow/locales/en.json index 5955936bc2..7a37a3132e 100644 --- a/src/backend/base/langflow/locales/en.json +++ b/src/backend/base/langflow/locales/en.json @@ -4411,7 +4411,7 @@ "components.loopcomponent.description.dc14755d": "Iterates through Data or Message objects, processing items individually and aggregating results from loop inputs.", "components.loopcomponent.display_name.f2f6a018": "Loop", "components.loopcomponent.inputs.data.display_name.7abc49df": "Inputs", - "components.loopcomponent.inputs.data.info.bb8c0783": "The initial DataFrame to iterate over.", + "components.loopcomponent.inputs.data.info.cb3819f9": "Data to iterate over. Accepts DataFrame, Table, Data, or Message.", "components.loopcomponent.outputs.done.display_name.11a6767d": "Done", "components.loopcomponent.outputs.item.display_name.652bcc3a": "Item", "components.maritalk.description.f2ad2e6b": "Generates text using MariTalk LLMs.", diff --git a/src/lfx/src/lfx/_assets/component_index.json b/src/lfx/src/lfx/_assets/component_index.json index f501507a67..53216d7049 100644 --- a/src/lfx/src/lfx/_assets/component_index.json +++ b/src/lfx/src/lfx/_assets/component_index.json @@ -73116,6 +73116,8 @@ }, "LoopComponent": { "base_classes": [ + "Data", + "DataFrame", "JSON", "Table" ], @@ -73133,7 +73135,7 @@ "icon": "infinity", "legacy": false, "metadata": { - "code_hash": "0b08e58a5c70", + "code_hash": "2712a1de7210", "dependencies": { "dependencies": [ { @@ -73158,9 +73160,10 @@ ], "method": "item_output", "name": "item", - "selected": "JSON", + "selected": "Data", "tool_mode": true, "types": [ + "Data", "JSON" ], "value": "__UNDEFINED__" @@ -73172,9 +73175,10 @@ "group_outputs": true, "method": "done_output", "name": "done", - "selected": "Table", + "selected": "DataFrame", "tool_mode": true, "types": [ + "DataFrame", "Table" ], "value": "__UNDEFINED__" @@ -73199,17 +73203,19 @@ "show": true, "title_case": false, "type": "code", - "value": "from lfx.base.flow_controls.loop_utils import (\n execute_loop_body,\n extract_loop_output,\n get_loop_body_start_edge,\n get_loop_body_start_vertex,\n get_loop_body_vertices,\n validate_data_input,\n)\nfrom lfx.components.processing.converter import convert_to_data\nfrom lfx.custom.custom_component.component import Component\nfrom lfx.inputs.inputs import HandleInput\nfrom lfx.schema.data import Data\nfrom lfx.schema.dataframe import DataFrame\nfrom lfx.schema.message import Message\nfrom lfx.template.field.base import Output\n\n\nclass LoopComponent(Component):\n display_name = \"Loop\"\n description = (\n \"Iterates through Data or Message objects, processing items individually \"\n \"and aggregating results from loop inputs.\"\n )\n documentation: str = \"https://docs.langflow.org/loop\"\n icon = \"infinity\"\n\n inputs = [\n HandleInput(\n name=\"data\",\n display_name=\"Inputs\",\n info=\"The initial DataFrame to iterate over.\",\n input_types=[\"DataFrame\", \"Table\"],\n ),\n ]\n\n outputs = [\n Output(\n display_name=\"Item\",\n name=\"item\",\n method=\"item_output\",\n allows_loop=True,\n loop_types=[\"Message\"],\n group_outputs=True,\n ),\n Output(display_name=\"Done\", name=\"done\", method=\"done_output\", group_outputs=True),\n ]\n\n def initialize_data(self) -> None:\n \"\"\"Initialize the data list and context index.\n\n Seeds the input list and index counter in ctx. The aggregated results\n are owned by `_iterate`, which writes them once the subgraph finishes.\n \"\"\"\n if self.ctx.get(f\"{self._id}_initialized\", False):\n return\n\n # Ensure data is a list of Data objects\n data_list = self._validate_data(self.data)\n\n # Store the initial data and context variables\n self.update_ctx(\n {\n f\"{self._id}_data\": data_list,\n f\"{self._id}_index\": 0,\n f\"{self._id}_initialized\": True,\n }\n )\n\n def _convert_message_to_data(self, message: Message) -> Data:\n \"\"\"Convert a Message object to a Data object using Type Convert logic.\"\"\"\n return convert_to_data(message, auto_parse=False)\n\n def _validate_data(self, data):\n \"\"\"Validate and return a list of Data objects.\"\"\"\n return validate_data_input(data)\n\n def get_loop_body_vertices(self) -> set[str]:\n \"\"\"Identify vertices in this loop's body via graph traversal.\n\n Traverses from the loop's \"item\" output to the vertex that feeds back\n to the loop's \"item\" input, collecting all vertices in between.\n This naturally handles nested loops by stopping at this loop's feedback edge.\n\n Returns:\n Set of vertex IDs that form this loop's body\n \"\"\"\n # Check if we have a proper graph context\n if not hasattr(self, \"_vertex\") or self._vertex is None:\n return set()\n\n return get_loop_body_vertices(\n vertex=self._vertex,\n graph=self.graph,\n get_incoming_edge_by_target_param_fn=self.get_incoming_edge_by_target_param,\n )\n\n def _get_loop_body_start_vertex(self) -> str | None:\n \"\"\"Get the first vertex in the loop body (connected to loop's item output).\n\n Returns:\n The vertex ID of the first vertex in the loop body, or None if not found\n \"\"\"\n # Check if we have a proper graph context\n if not hasattr(self, \"_vertex\") or self._vertex is None:\n return None\n\n return get_loop_body_start_vertex(vertex=self._vertex)\n\n def _extract_loop_output(self, results: list) -> Data:\n \"\"\"Extract the output from subgraph execution results.\n\n Args:\n results: List of VertexBuildResult objects from subgraph execution\n\n Returns:\n Data object containing the loop iteration output\n \"\"\"\n # Get the vertex ID that feeds back to the item input (end of loop body)\n end_vertex_id = self.get_incoming_edge_by_target_param(\"item\")\n return extract_loop_output(results=results, end_vertex_id=end_vertex_id)\n\n async def execute_loop_body(self, data_list: list[Data], event_manager=None) -> list[Data]:\n \"\"\"Execute loop body for each data item.\n\n Creates an isolated subgraph for the loop body and executes it\n for each item in the data list, collecting results.\n\n Args:\n data_list: List of Data objects to iterate over\n event_manager: Optional event manager to pass to subgraph execution for UI events\n\n Returns:\n List of Data objects containing results from each iteration\n \"\"\"\n # Get the loop body configuration once\n loop_body_vertex_ids = self.get_loop_body_vertices()\n start_vertex_id = self._get_loop_body_start_vertex()\n start_edge = get_loop_body_start_edge(self._vertex)\n end_vertex_id = self.get_incoming_edge_by_target_param(\"item\")\n\n return await execute_loop_body(\n graph=self.graph,\n data_list=data_list,\n loop_body_vertex_ids=loop_body_vertex_ids,\n start_vertex_id=start_vertex_id,\n start_edge=start_edge,\n end_vertex_id=end_vertex_id,\n event_manager=event_manager,\n )\n\n async def _iterate(self) -> list[Data]:\n \"\"\"Run the loop body subgraph once and cache the aggregated results.\n\n Both `item_output` and `done_output` may be called during the same\n vertex build (if both outputs have downstream consumers), so this\n helper is idempotent: it caches the aggregated results in ctx under\n `{self._id}_aggregated` and guards re-entry with `{self._id}_iterated`.\n If a prior call raised, the exception is cached in\n `{self._id}_iteration_error` and re-raised on subsequent calls so\n repeat invocations surface the same failure instead of silently\n returning empty data or re-running the subgraph.\n\n Production constructs a fresh Graph per request (via\n `Graph.from_payload`), so `ctx` is effectively per-run and the\n cached `_iterated` flag does not leak across executions.\n \"\"\"\n if self.ctx.get(f\"{self._id}_iterated\", False):\n cached_error = self.ctx.get(f\"{self._id}_iteration_error\")\n if cached_error is not None:\n raise cached_error\n return self.ctx.get(f\"{self._id}_aggregated\", [])\n\n import time\n\n started_at = time.perf_counter()\n try:\n self.initialize_data()\n data_list = self.ctx.get(f\"{self._id}_data\", [])\n self.log(f\"Starting loop over {len(data_list)} item(s)\", name=\"Start\")\n\n if not data_list:\n self.update_ctx({f\"{self._id}_aggregated\": [], f\"{self._id}_iterated\": True})\n self.log(\"No items to iterate, skipping loop body\", name=\"Skipped\")\n return []\n\n aggregated_results = await self.execute_loop_body(data_list, event_manager=self._event_manager)\n except Exception as exc:\n from lfx.log.logger import logger\n\n elapsed = time.perf_counter() - started_at\n self.log(f\"Loop failed after {elapsed:.3f}s: {exc}\", name=\"Error\")\n await logger.aexception(f\"Loop {self._id} failed while executing loop body\")\n self.update_ctx({f\"{self._id}_iteration_error\": exc, f\"{self._id}_iterated\": True})\n raise\n\n elapsed = time.perf_counter() - started_at\n self.log(\n f\"Completed {len(aggregated_results)} iteration(s) in {elapsed:.3f}s\",\n name=\"Complete\",\n )\n self.update_ctx({f\"{self._id}_aggregated\": aggregated_results, f\"{self._id}_iterated\": True})\n return aggregated_results\n\n async def item_output(self) -> Data:\n \"\"\"Display the inputs dispatched to the loop body.\n\n Also triggers the iteration (idempotent) when `done` has no\n downstream consumer, so the loop still runs if only the Item\n output is connected. When `done` IS connected we leave iteration\n to `done_output`. The `item` branch is stopped in both cases so\n downstream vertices aren't executed by the outer graph (the loop\n body runs internally via the subgraph in `_iterate`).\n\n Returns a `Data` envelope so the outer item edge remains\n compatible with Data-typed consumers in the loop body. The\n wrapped payload exposes the iterated rows for inspection.\n \"\"\"\n self.stop(\"item\")\n if self._vertex is not None and \"done\" not in self._vertex.edges_source_names:\n await self._iterate()\n data_list = self.ctx.get(f\"{self._id}_data\", [])\n return Data(data={\"count\": len(data_list), \"items\": [d.data for d in data_list]})\n\n async def done_output(self) -> DataFrame:\n \"\"\"Return the aggregated results from the loop iteration.\n\n Triggers the iteration if it hasn't run yet (for example when only\n `done` is connected and `item_output` never executed). In the common\n case where `item` is also connected, `_iterate` has already cached\n the aggregated results in ctx and this call is a cheap read, so the\n order in which the two outputs are evaluated does not matter.\n \"\"\"\n aggregated_results = await self._iterate()\n return DataFrame(aggregated_results)\n" + "value": "from lfx.base.flow_controls.loop_utils import (\n execute_loop_body,\n extract_loop_output,\n get_loop_body_start_edge,\n get_loop_body_start_vertex,\n get_loop_body_vertices,\n validate_data_input,\n)\nfrom lfx.components.processing.converter import convert_to_data\nfrom lfx.custom.custom_component.component import Component\nfrom lfx.inputs.inputs import HandleInput\nfrom lfx.schema.data import Data\nfrom lfx.schema.dataframe import DataFrame\nfrom lfx.schema.message import Message\nfrom lfx.template.field.base import Output\n\n\nclass LoopComponent(Component):\n display_name = \"Loop\"\n description = (\n \"Iterates through Data or Message objects, processing items individually \"\n \"and aggregating results from loop inputs.\"\n )\n documentation: str = \"https://docs.langflow.org/loop\"\n icon = \"infinity\"\n\n inputs = [\n HandleInput(\n name=\"data\",\n display_name=\"Inputs\",\n info=\"Data to iterate over. Accepts DataFrame, Table, Data, or Message.\",\n input_types=[\"DataFrame\", \"Table\", \"Data\", \"Message\"],\n ),\n ]\n\n outputs = [\n Output(\n display_name=\"Item\",\n name=\"item\",\n method=\"item_output\",\n types=[\"Data\"],\n allows_loop=True,\n loop_types=[\"Message\"],\n group_outputs=True,\n ),\n Output(\n display_name=\"Done\",\n name=\"done\",\n method=\"done_output\",\n types=[\"DataFrame\", \"Table\"],\n group_outputs=True,\n ),\n ]\n\n def initialize_data(self) -> None:\n \"\"\"Initialize the data list and context index.\n\n Seeds the input list and index counter in ctx. The aggregated results\n are owned by `_iterate`, which writes them once the subgraph finishes.\n \"\"\"\n if self.ctx.get(f\"{self._id}_initialized\", False):\n return\n\n # Ensure data is a list of Data objects\n data_list = self._validate_data(self.data)\n\n # Store the initial data and context variables\n self.update_ctx(\n {\n f\"{self._id}_data\": data_list,\n f\"{self._id}_index\": 0,\n f\"{self._id}_initialized\": True,\n }\n )\n\n def _convert_message_to_data(self, message: Message) -> Data:\n \"\"\"Convert a Message object to a Data object using Type Convert logic.\"\"\"\n return convert_to_data(message, auto_parse=False)\n\n def _validate_data(self, data):\n \"\"\"Validate and normalize the input to a flat list of Data objects.\n\n Coerces each accepted input type into Data so `validate_data_input`\n always receives a homogeneous payload:\n - Message -> a clean Data via `_convert_message_to_data` (just the\n message payload, not the full envelope). `Message` subclasses\n `Data`, so without this it would slip through as a single Data\n carrying all of the message metadata.\n - DataFrame/Table -> expanded to its rows.\n - Data / list[Data] -> passed through unchanged.\n\n Lists are normalized item-by-item so a mixed `[Message, DataFrame,\n Data]` still yields `list[Data]` instead of being rejected as a\n mixed-type list (`validate_data_input` requires every list item to be\n a Data, and DataFrame is not a Data subclass).\n \"\"\"\n if isinstance(data, Message):\n data = self._convert_message_to_data(data)\n elif isinstance(data, list):\n normalized: list = []\n for item in data:\n if isinstance(item, Message):\n normalized.append(self._convert_message_to_data(item))\n elif isinstance(item, DataFrame):\n normalized.extend(item.to_data_list())\n else:\n normalized.append(item)\n data = normalized\n return validate_data_input(data)\n\n def get_loop_body_vertices(self) -> set[str]:\n \"\"\"Identify vertices in this loop's body via graph traversal.\n\n Traverses from the loop's \"item\" output to the vertex that feeds back\n to the loop's \"item\" input, collecting all vertices in between.\n This naturally handles nested loops by stopping at this loop's feedback edge.\n\n Returns:\n Set of vertex IDs that form this loop's body\n \"\"\"\n # Check if we have a proper graph context\n if not hasattr(self, \"_vertex\") or self._vertex is None:\n return set()\n\n return get_loop_body_vertices(\n vertex=self._vertex,\n graph=self.graph,\n get_incoming_edge_by_target_param_fn=self.get_incoming_edge_by_target_param,\n )\n\n def _get_loop_body_start_vertex(self) -> str | None:\n \"\"\"Get the first vertex in the loop body (connected to loop's item output).\n\n Returns:\n The vertex ID of the first vertex in the loop body, or None if not found\n \"\"\"\n # Check if we have a proper graph context\n if not hasattr(self, \"_vertex\") or self._vertex is None:\n return None\n\n return get_loop_body_start_vertex(vertex=self._vertex)\n\n def _extract_loop_output(self, results: list) -> Data:\n \"\"\"Extract the output from subgraph execution results.\n\n Args:\n results: List of VertexBuildResult objects from subgraph execution\n\n Returns:\n Data object containing the loop iteration output\n \"\"\"\n # Get the vertex ID that feeds back to the item input (end of loop body)\n end_vertex_id = self.get_incoming_edge_by_target_param(\"item\")\n return extract_loop_output(results=results, end_vertex_id=end_vertex_id)\n\n async def execute_loop_body(self, data_list: list[Data], event_manager=None) -> list[Data]:\n \"\"\"Execute loop body for each data item.\n\n Creates an isolated subgraph for the loop body and executes it\n for each item in the data list, collecting results.\n\n Args:\n data_list: List of Data objects to iterate over\n event_manager: Optional event manager to pass to subgraph execution for UI events\n\n Returns:\n List of Data objects containing results from each iteration\n \"\"\"\n # Get the loop body configuration once\n loop_body_vertex_ids = self.get_loop_body_vertices()\n start_vertex_id = self._get_loop_body_start_vertex()\n start_edge = get_loop_body_start_edge(self._vertex)\n end_vertex_id = self.get_incoming_edge_by_target_param(\"item\")\n\n return await execute_loop_body(\n graph=self.graph,\n data_list=data_list,\n loop_body_vertex_ids=loop_body_vertex_ids,\n start_vertex_id=start_vertex_id,\n start_edge=start_edge,\n end_vertex_id=end_vertex_id,\n event_manager=event_manager,\n )\n\n async def _iterate(self) -> list[Data]:\n \"\"\"Run the loop body subgraph once and cache the aggregated results.\n\n Both `item_output` and `done_output` may be called during the same\n vertex build (if both outputs have downstream consumers), so this\n helper is idempotent: it caches the aggregated results in ctx under\n `{self._id}_aggregated` and guards re-entry with `{self._id}_iterated`.\n If a prior call raised, the exception is cached in\n `{self._id}_iteration_error` and re-raised on subsequent calls so\n repeat invocations surface the same failure instead of silently\n returning empty data or re-running the subgraph.\n\n Production constructs a fresh Graph per request (via\n `Graph.from_payload`), so `ctx` is effectively per-run and the\n cached `_iterated` flag does not leak across executions.\n \"\"\"\n if self.ctx.get(f\"{self._id}_iterated\", False):\n cached_error = self.ctx.get(f\"{self._id}_iteration_error\")\n if cached_error is not None:\n raise cached_error\n return self.ctx.get(f\"{self._id}_aggregated\", [])\n\n import time\n\n started_at = time.perf_counter()\n try:\n self.initialize_data()\n data_list = self.ctx.get(f\"{self._id}_data\", [])\n self.log(f\"Starting loop over {len(data_list)} item(s)\", name=\"Start\")\n\n if not data_list:\n self.update_ctx({f\"{self._id}_aggregated\": [], f\"{self._id}_iterated\": True})\n self.log(\"No items to iterate, skipping loop body\", name=\"Skipped\")\n return []\n\n aggregated_results = await self.execute_loop_body(data_list, event_manager=self._event_manager)\n except Exception as exc:\n from lfx.log.logger import logger\n\n elapsed = time.perf_counter() - started_at\n self.log(f\"Loop failed after {elapsed:.3f}s: {exc}\", name=\"Error\")\n await logger.aexception(f\"Loop {self._id} failed while executing loop body\")\n self.update_ctx({f\"{self._id}_iteration_error\": exc, f\"{self._id}_iterated\": True})\n raise\n\n elapsed = time.perf_counter() - started_at\n self.log(\n f\"Completed {len(aggregated_results)} iteration(s) in {elapsed:.3f}s\",\n name=\"Complete\",\n )\n self.update_ctx({f\"{self._id}_aggregated\": aggregated_results, f\"{self._id}_iterated\": True})\n return aggregated_results\n\n async def item_output(self) -> Data:\n \"\"\"Display the inputs dispatched to the loop body.\n\n Also triggers the iteration (idempotent) when `done` has no\n downstream consumer, so the loop still runs if only the Item\n output is connected. When `done` IS connected we leave iteration\n to `done_output`. The `item` branch is stopped in both cases so\n downstream vertices aren't executed by the outer graph (the loop\n body runs internally via the subgraph in `_iterate`).\n\n Returns a `Data` envelope so the outer item edge remains\n compatible with Data-typed consumers in the loop body. The\n wrapped payload exposes the iterated rows for inspection.\n \"\"\"\n self.stop(\"item\")\n if self._vertex is not None and \"done\" not in self._vertex.edges_source_names:\n await self._iterate()\n data_list = self.ctx.get(f\"{self._id}_data\", [])\n return Data(data={\"count\": len(data_list), \"items\": [d.data for d in data_list]})\n\n async def done_output(self) -> DataFrame:\n \"\"\"Return the aggregated results from the loop iteration.\n\n Triggers the iteration if it hasn't run yet (for example when only\n `done` is connected and `item_output` never executed). In the common\n case where `item` is also connected, `_iterate` has already cached\n the aggregated results in ctx and this call is a cheap read, so the\n order in which the two outputs are evaluated does not matter.\n \"\"\"\n aggregated_results = await self._iterate()\n return DataFrame(aggregated_results)\n" }, "data": { "_input_type": "HandleInput", "advanced": false, "display_name": "Inputs", "dynamic": false, - "info": "The initial DataFrame to iterate over.", + "info": "Data to iterate over. Accepts DataFrame, Table, Data, or Message.", "input_types": [ "DataFrame", - "Table" + "Table", + "Data", + "Message" ], "list": false, "list_add_label": "Add More", @@ -118467,6 +118473,6 @@ "num_components": 354, "num_modules": 95 }, - "sha256": "843ba7bdc516522eccbfc466970f8b40cc260abeccfda302436404d505fe8565", + "sha256": "2d7317fbdab5068ab3a4487b7863203127e6aae0af85c5c40ffd5404a7cb1ce4", "version": "1.10.1" } diff --git a/src/lfx/src/lfx/components/flow_controls/loop.py b/src/lfx/src/lfx/components/flow_controls/loop.py index 0553a5418b..6a92490949 100644 --- a/src/lfx/src/lfx/components/flow_controls/loop.py +++ b/src/lfx/src/lfx/components/flow_controls/loop.py @@ -28,8 +28,8 @@ class LoopComponent(Component): HandleInput( name="data", display_name="Inputs", - info="The initial DataFrame to iterate over.", - input_types=["DataFrame", "Table"], + info="Data to iterate over. Accepts DataFrame, Table, Data, or Message.", + input_types=["DataFrame", "Table", "Data", "Message"], ), ] @@ -38,11 +38,18 @@ class LoopComponent(Component): display_name="Item", name="item", method="item_output", + types=["Data"], allows_loop=True, loop_types=["Message"], group_outputs=True, ), - Output(display_name="Done", name="done", method="done_output", group_outputs=True), + Output( + display_name="Done", + name="done", + method="done_output", + types=["DataFrame", "Table"], + group_outputs=True, + ), ] def initialize_data(self) -> None: @@ -71,7 +78,34 @@ class LoopComponent(Component): return convert_to_data(message, auto_parse=False) def _validate_data(self, data): - """Validate and return a list of Data objects.""" + """Validate and normalize the input to a flat list of Data objects. + + Coerces each accepted input type into Data so `validate_data_input` + always receives a homogeneous payload: + - Message -> a clean Data via `_convert_message_to_data` (just the + message payload, not the full envelope). `Message` subclasses + `Data`, so without this it would slip through as a single Data + carrying all of the message metadata. + - DataFrame/Table -> expanded to its rows. + - Data / list[Data] -> passed through unchanged. + + Lists are normalized item-by-item so a mixed `[Message, DataFrame, + Data]` still yields `list[Data]` instead of being rejected as a + mixed-type list (`validate_data_input` requires every list item to be + a Data, and DataFrame is not a Data subclass). + """ + if isinstance(data, Message): + data = self._convert_message_to_data(data) + elif isinstance(data, list): + normalized: list = [] + for item in data: + if isinstance(item, Message): + normalized.append(self._convert_message_to_data(item)) + elif isinstance(item, DataFrame): + normalized.extend(item.to_data_list()) + else: + normalized.append(item) + data = normalized return validate_data_input(data) def get_loop_body_vertices(self) -> set[str]: diff --git a/src/lfx/tests/unit/components/flow_controls/test_loop_input_types.py b/src/lfx/tests/unit/components/flow_controls/test_loop_input_types.py new file mode 100644 index 0000000000..95c3770028 --- /dev/null +++ b/src/lfx/tests/unit/components/flow_controls/test_loop_input_types.py @@ -0,0 +1,154 @@ +"""Regression tests for LoopComponent input/output type metadata. + +Covers the bug where the Loop's ``data`` handle only declared +``input_types=["DataFrame", "Table"]`` even though the component is +documented to iterate over ``Data`` or ``Message`` objects (and ships a +``_convert_message_to_data`` helper). The missing types caused any +Message/Data-producing component (ChatInput, Agent, ...) to be rejected at +connect time, which made agent-assisted flow builders retry until they hit +the LangGraph recursion limit. See issue #13636. + +Also covers normalization of mixed-type lists in ``_validate_data`` so a +``[Message, DataFrame, Data]`` input still yields a homogeneous ``list[Data]`` +instead of being rejected by ``validate_data_input``. +""" + +import pytest +from lfx.components.flow_controls.loop import LoopComponent +from lfx.graph.edge.base import types_compatible +from lfx.schema.data import Data +from lfx.schema.dataframe import DataFrame +from lfx.schema.message import Message + + +def _data_input(loop: LoopComponent): + return next(i for i in loop.inputs if i.name == "data") + + +def _output(loop: LoopComponent, name: str): + return next(o for o in loop.outputs if o.name == name) + + +def test_data_input_accepts_data_and_message(): + """The data handle must accept Data and Message alongside DataFrame/Table.""" + loop = LoopComponent() + input_types = _data_input(loop).input_types + assert "DataFrame" in input_types + assert "Table" in input_types + assert "Data" in input_types + assert "Message" in input_types + + +def test_outputs_declare_types(): + """Both outputs must advertise their types so agents can plan downstream.""" + loop = LoopComponent() + assert _output(loop, "item").types == ["Data"] + assert _output(loop, "done").types == ["DataFrame", "Table"] + + +def test_message_outputs_are_connectable_to_loop(): + """A Message- or Data-producing output must be compatible with Loop.data. + + This is the connect-time check that previously failed and triggered the + agent-builder retry loop. + """ + loop_input = _data_input(LoopComponent()).input_types + # ChatInput / Agent emit Message; Agent.structured_response emits Data/JSON. + assert types_compatible(["Message"], loop_input) + assert types_compatible(["Data", "JSON"], loop_input) + assert types_compatible(["JSON"], loop_input) + assert types_compatible(["DataFrame"], loop_input) + + +def test_validate_data_converts_message_to_clean_data(): + """A Message input is converted to a Data carrying just its payload. + + Message subclasses Data, so without the explicit conversion it would be + accepted verbatim as a single Data object holding the whole message + envelope (sender, session_id, timestamp, ...) instead of the payload. + """ + loop = LoopComponent() + validated = loop._validate_data(Message(text="hello world")) + + assert isinstance(validated, list) + assert len(validated) == 1 + assert isinstance(validated[0], Data) + # Must be a plain Data, not the Message passed through verbatim (Message + # subclasses Data, so the isinstance check above is not enough on its own). + assert not isinstance(validated[0], Message) + assert validated[0].data == {"text": "hello world"} + + +def test_validate_data_still_handles_dataframe(): + """The existing DataFrame path is preserved (expanded to its rows).""" + loop = LoopComponent() + validated = loop._validate_data(DataFrame([Data(text="a"), Data(text="b")])) + + assert isinstance(validated, list) + assert len(validated) == 2 + assert all(isinstance(item, Data) for item in validated) + + +def test_validate_data_still_handles_single_data(): + """A single Data input yields a one-item list.""" + loop = LoopComponent() + single = Data(text="single") + validated = loop._validate_data(single) + + assert validated == [single] + + +def test_validate_data_handles_list_of_messages(): + """A list of Message objects is converted item-by-item to clean Data.""" + loop = LoopComponent() + validated = loop._validate_data([Message(text="m1"), Message(text="m2")]) + + assert len(validated) == 2 + assert all(isinstance(item, Data) for item in validated) + assert [d.data for d in validated] == [{"text": "m1"}, {"text": "m2"}] + + +def test_validate_data_normalizes_mixed_list(): + """A mixed [Message, DataFrame, Data] list yields a flat list[Data]. + + Regression for the CodeRabbit review on #13646: converting only the + Message items produced ``[Data, DataFrame, Data]``, which + ``validate_data_input`` rejects because DataFrame is not a Data subclass. + DataFrame items must be expanded to their rows so the list stays + homogeneous. + """ + loop = LoopComponent() + mixed = [ + Message(text="m"), + DataFrame([Data(text="r1"), Data(text="r2")]), + Data(text="d"), + ] + validated = loop._validate_data(mixed) + + assert len(validated) == 4 + assert all(isinstance(item, Data) for item in validated) + assert [d.data for d in validated] == [ + {"text": "m"}, + {"text": "r1"}, + {"text": "r2"}, + {"text": "d"}, + ] + + +def test_validate_data_rejects_invalid_input(): + """Invalid input types still raise a clear TypeError.""" + loop = LoopComponent() + with pytest.raises(TypeError, match="must be a DataFrame"): + loop._validate_data("not data") + + +def test_validate_data_rejects_list_with_non_coercible_item(): + """A list containing a non-Data/Message/DataFrame item is still rejected. + + The list branch only coerces Message and DataFrame items; anything else + falls through unchanged and must be caught by validate_data_input so the + documented contract (every item resolves to Data) is preserved. + """ + loop = LoopComponent() + with pytest.raises(TypeError, match="must be a DataFrame"): + loop._validate_data([Data(text="ok"), "bad"])