feat: add convert component with dynamic output support

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

@@ -1,5 +1,6 @@
 from .alter_metadata import AlterMetadataComponent
 from .combine_text import CombineTextComponent
+from .convertor import TypeConverterComponent
 from .create_data import CreateDataComponent
 from .data_operations import DataOperationsComponent
 from .extract_key import ExtractDataKeyComponent
@@ -38,5 +39,6 @@
     "RegexExtractorComponent",
     "SelectDataComponent",
     "SplitTextComponent",
+    "TypeConverterComponent",
     "UpdateDataComponent",
 ]

src/backend/base/langflow/components/processing/convertor.py

@@ -0,0 +1,138 @@
+from collections.abc import AsyncIterator, Iterator
+from typing import Any
+
+from langflow.custom import Component
+from langflow.io import HandleInput, Output, TabInput
+from langflow.schema import Data, DataFrame, Message
+from langflow.services.database.models.message.model import MessageBase
+
+
+def get_message_converter(v) -> Message:
+    # If v is a instance of Message, then its fine
+    if isinstance(v, dict):
+        return Message(**v)
+    if isinstance(v, Message):
+        return v
+    if isinstance(v, str | AsyncIterator | Iterator):
+        return Message(text=v)
+    if isinstance(v, MessageBase):
+        return Message(**v.model_dump())
+    if isinstance(v, DataFrame):
+        # Process DataFrame similar to the _safe_convert method
+        # Remove empty rows
+        processed_df = v.dropna(how="all")
+        # Remove empty lines in each cell
+        processed_df = processed_df.replace(r"^\s*$", "", regex=True)
+        # Replace multiple newlines with a single newline
+        processed_df = processed_df.replace(r"\n+", "\n", regex=True)
+        # Replace pipe characters to avoid markdown table issues
+        processed_df = processed_df.replace(r"\|", r"\\|", regex=True)
+        processed_df = processed_df.map(lambda x: str(x).replace("\n", "<br/>") if isinstance(x, str) else x)
+        # Convert to markdown and wrap in a Message
+        return Message(text=processed_df.to_markdown(index=False))
+    if isinstance(v, Data):
+        if v.text_key in v.data:
+            return Message(text=v.get_text())
+        return Message(text=str(v.data))
+    msg = f"Invalid value type {type(v)}"
+    raise ValueError(msg)
+
+
+def get_data_converter(v: DataFrame | Data | Message | dict) -> Data:
+    """Get the data conversion dispatcher."""
+    if isinstance(v, DataFrame):
+        # Convert DataFrame to a list of dictionaries and wrap in a Data object
+        dict_list = v.to_dict(orient="records")
+        return Data(data={"results": dict_list})
+    if isinstance(v, Message):
+        return Data(data=v.data)
+    if isinstance(v, dict):
+        return Data(data=v)
+    if not isinstance(v, Data):
+        msg = f"Invalid value type {type(v)} for input Expected Data."
+        raise ValueError(msg)  # noqa: TRY004
+    return v
+
+
+def get_dataframe_converter(v: DataFrame | Data | Message | dict) -> DataFrame:
+    """Get the dataframe conversion dispatcher."""
+    if isinstance(v, Data):
+        data_dict = v.data
+        # If data contains only one key and the value is a list of dictionaries, convert to DataFrame
+        if (
+            len(data_dict) == 1
+            and isinstance(next(iter(data_dict.values())), list)
+            and all(isinstance(item, dict) for item in next(iter(data_dict.values())))
+        ):
+            return DataFrame(data=next(iter(data_dict.values())))
+        return DataFrame(data=[v])
+    if isinstance(v, Message):
+        return DataFrame(data=[v])
+    if isinstance(v, dict):
+        return DataFrame(data=[v])
+    if not isinstance(v, DataFrame):
+        msg = f"Invalid value type {type(v)}. Expected DataFrame."
+        raise ValueError(msg)  # noqa: TRY004
+    return v
+
+
+class TypeConverterComponent(Component):
+    display_name = "Type Convert"
+    description = "Convert between different types (Message, Data, DataFrame)"
+    icon = "repeat"
+
+    inputs = [
+        HandleInput(
+            name="input_data",
+            display_name="Input",
+            input_types=["Message", "Data", "DataFrame"],
+            info="Accept Message, Data or DataFrame as input",
+            required=True,
+        ),
+        TabInput(
+            name="output_type",
+            display_name="Output Type",
+            options=["Message", "Data", "DataFrame"],
+            info="Select the desired output data type",
+            real_time_refresh=True,
+            value="Message",
+        ),
+    ]
+
+    outputs = [Output(display_name="Message Output", name="message_output", method="convert_to_message")]
+
+    def update_outputs(self, frontend_node: dict, field_name: str, field_value: Any) -> dict:
+        """Dynamically show only the relevant output based on the selected output type."""
+        if field_name == "output_type":
+            # Start with empty outputs
+            frontend_node["outputs"] = []
+
+            # Add only the selected output type
+            if field_value == "Message":
+                frontend_node["outputs"].append(
+                    Output(display_name="Message Output", name="message_output", method="convert_to_message").to_dict()
+                )
+            elif field_value == "Data":
+                frontend_node["outputs"].append(
+                    Output(display_name="Data Output", name="data_output", method="convert_to_data").to_dict()
+                )
+            elif field_value == "DataFrame":
+                frontend_node["outputs"].append(
+                    Output(
+                        display_name="DataFrame Output", name="dataframe_output", method="convert_to_dataframe"
+                    ).to_dict()
+                )
+
+        return frontend_node
+
+    def convert_to_message(self) -> Message:
+        """Convert input to Message type."""
+        return get_message_converter(self.input_data)
+
+    def convert_to_data(self) -> Data:
+        """Convert input to Data type."""
+        return get_data_converter(self.input_data)
+
+    def convert_to_dataframe(self) -> DataFrame:
+        """Convert input to DataFrame type."""
+        return get_dataframe_converter(self.input_data)

src/backend/tests/unit/components/processing/test_type_convertor_component.py

@@ -0,0 +1,113 @@
+import pandas as pd
+import pytest
+from langflow.components.processing.convertor import TypeConverterComponent
+from langflow.schema import Data, DataFrame, Message
+
+from tests.base import ComponentTestBaseWithoutClient
+
+
+class TestTypeConverterComponent(ComponentTestBaseWithoutClient):
+    @pytest.fixture
+    def component_class(self):
+        """Return the component class to test."""
+        return TypeConverterComponent
+
+    @pytest.fixture
+    def file_names_mapping(self):
+        """Return an empty list since this component doesn't have version-specific files."""
+        return []
+
+    # Message to other types
+    def test_message_to_message(self, component_class):
+        """Test converting Message to Message."""
+        component = component_class(input_data=Message(text="Hello World"), output_type="Message")
+        result = component.convert_to_message()
+        assert isinstance(result, Message)
+        assert result.text == "Hello World"
+
+    def test_message_to_data(self, component_class):
+        """Test converting Message to Data."""
+        component = component_class(input_data=Message(text="Hello"), output_type="Data")
+        result = component.convert_to_data()
+        assert isinstance(result, Data)
+        assert "text" in result.data
+        assert result.data["text"] == "Hello"
+
+    def test_message_to_dataframe(self, component_class):
+        """Test converting Message to DataFrame."""
+        component = component_class(input_data=Message(text="Hello"), output_type="DataFrame")
+        result = component.convert_to_dataframe()
+        assert isinstance(result, DataFrame)
+        assert "text" in result.columns
+        assert result.iloc[0]["text"] == "Hello"
+
+    # Data to other types
+    def test_data_to_message(self, component_class):
+        """Test converting Data to Message."""
+        component = component_class(input_data=Data(data={"text": "Hello World"}), output_type="Message")
+        result = component.convert_to_message()
+        assert isinstance(result, Message)
+        assert result.text == "Hello World"
+
+    def test_data_to_data(self, component_class):
+        """Test converting Data to Data."""
+        component = component_class(input_data=Data(data={"key": "value"}), output_type="Data")
+        result = component.convert_to_data()
+        assert isinstance(result, Data)
+        assert result.data == {"key": "value"}
+
+    def test_data_to_dataframe(self, component_class):
+        """Test converting Data to DataFrame."""
+        component = component_class(input_data=Data(data={"text": "Hello World"}), output_type="DataFrame")
+        result = component.convert_to_dataframe()
+        assert isinstance(result, DataFrame)
+        assert "text" in result.columns
+        assert result.iloc[0]["text"] == "Hello World"
+
+    # DataFrame to other types
+    def test_dataframe_to_message(self, component_class):
+        """Test converting DataFrame to Message."""
+        df_data = pd.DataFrame({"col1": ["Hello"], "col2": ["World"]})
+        component = component_class(input_data=DataFrame(data=df_data), output_type="Message")
+        result = component.convert_to_message()
+        assert isinstance(result, Message)
+        assert result.text == "| col1   | col2   |\n|:-------|:-------|\n| Hello  | World  |"
+
+    def test_dataframe_to_data(self, component_class):
+        """Test converting DataFrame to Data."""
+        df_data = pd.DataFrame({"col1": ["Hello"]})
+        component = component_class(input_data=DataFrame(data=df_data), output_type="Data")
+        result = component.convert_to_data()
+        assert isinstance(result, Data)
+        assert isinstance(result.data, dict)
+
+    def test_dataframe_to_dataframe(self, component_class):
+        """Test converting DataFrame to DataFrame."""
+        df_data = pd.DataFrame({"col1": ["Hello"], "col2": ["World"]})
+        component = component_class(input_data=DataFrame(data=df_data), output_type="DataFrame")
+        result = component.convert_to_dataframe()
+        assert isinstance(result, DataFrame)
+        assert "col1" in result.columns
+        assert "col2" in result.columns
+        assert result.iloc[0]["col1"] == "Hello"
+        assert result.iloc[0]["col2"] == "World"
+
+    def test_update_outputs(self, component_class):
+        """Test the update_outputs method."""
+        component = component_class(input_data=Message(text="Hello"), output_type="Message")
+        frontend_node = {"outputs": []}
+
+        # Test with Message output
+        updated = component.update_outputs(frontend_node, "output_type", "Message")
+        assert len(updated["outputs"]) == 1
+        assert updated["outputs"][0]["name"] == "message_output"
+
+        # Test with Data output
+        updated = component.update_outputs(frontend_node, "output_type", "Data")
+        assert len(updated["outputs"]) == 1
+        assert updated["outputs"][0]["name"] == "data_output"
+
+        # Test with DataFrame output
+        updated = component.update_outputs(frontend_node, "output_type", "DataFrame")
+        assert len(updated["outputs"]) == 1
+        assert updated["outputs"][0]["name"] == "dataframe_output"