fix: update model output type to message instead of string (#8489)

* Changed conversion methods to support the fact that LLM hint on a Message return type but return a string, causing a runtime error while converting types

* Convert to Message for model component text response

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes

---------

Co-authored-by: Pedro Pacheco <pedro.pacheco@yahoo.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
Jordan Frazier 2025-06-12 04:23:25 -07:00 committed by GitHub
commit e739db5143
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 47 additions and 13 deletions

View file

@ -174,7 +174,7 @@ class LCModelComponent(Component):
stream: bool,
input_value: str | Message,
system_message: str | None = None,
):
) -> Message:
if getattr(self, "detailed_thinking", False):
system_message = DETAILED_THINKING_PREFIX + (system_message or "")
@ -192,7 +192,7 @@ class LCModelComponent(Component):
stream: bool,
input_value: str | Message,
system_message: str | None = None,
):
) -> Message:
messages: list[BaseMessage] = []
if not input_value and not system_message:
msg = "The message you want to send to the model is empty."
@ -248,7 +248,7 @@ class LCModelComponent(Component):
raise ValueError(message) from e
raise
return result
return Message(text=result)
@abstractmethod
def build_model(self) -> LanguageModel: # type: ignore[type-var]

View file

@ -68,7 +68,13 @@ class TypeConverterComponent(Component):
),
]
outputs = [Output(display_name="Message Output", name="message_output", method="convert_to_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."""
@ -79,16 +85,26 @@ class TypeConverterComponent(Component):
# 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()
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()
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"
display_name="DataFrame Output",
name="dataframe_output",
method="convert_to_dataframe",
).to_dict()
)
@ -96,12 +112,30 @@ class TypeConverterComponent(Component):
def convert_to_message(self) -> Message:
"""Convert input to Message type."""
return convert_to_message(self.input_data[0] if isinstance(self.input_data, list) else self.input_data)
input_value = self.input_data[0] if isinstance(self.input_data, list) else self.input_data
# Handle string input by converting to Message first
if isinstance(input_value, str):
input_value = Message(text=input_value)
return convert_to_message(input_value)
def convert_to_data(self) -> Data:
"""Convert input to Data type."""
return convert_to_data(self.input_data[0] if isinstance(self.input_data, list) else self.input_data)
input_value = self.input_data[0] if isinstance(self.input_data, list) else self.input_data
# Handle string input by converting to Message first
if isinstance(input_value, str):
input_value = Message(text=input_value)
return convert_to_data(input_value)
def convert_to_dataframe(self) -> DataFrame:
"""Convert input to DataFrame type."""
return convert_to_dataframe(self.input_data[0] if isinstance(self.input_data, list) else self.input_data)
input_value = self.input_data[0] if isinstance(self.input_data, list) else self.input_data
# Handle string input by converting to Message first
if isinstance(input_value, str):
input_value = Message(text=input_value)
return convert_to_dataframe(input_value)

View file

@ -2954,7 +2954,7 @@
"show": true,
"title_case": false,
"type": "code",
"value": "from typing import Any\n\nfrom langflow.custom import Component\nfrom langflow.io import HandleInput, Output, TabInput\nfrom langflow.schema import Data, DataFrame, Message\n\n\ndef convert_to_message(v) -> Message:\n \"\"\"Convert input to Message type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n Message: Converted Message object\n \"\"\"\n return v if isinstance(v, Message) else v.to_message()\n\n\ndef convert_to_data(v: DataFrame | Data | Message | dict) -> Data:\n \"\"\"Convert input to Data type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n Data: Converted Data object\n \"\"\"\n if isinstance(v, dict):\n return Data(v)\n return v if isinstance(v, Data) else v.to_data()\n\n\ndef convert_to_dataframe(v: DataFrame | Data | Message | dict) -> DataFrame:\n \"\"\"Convert input to DataFrame type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n DataFrame: Converted DataFrame object\n \"\"\"\n if isinstance(v, dict):\n return DataFrame([v])\n return v if isinstance(v, DataFrame) else v.to_dataframe()\n\n\nclass TypeConverterComponent(Component):\n display_name = \"Type Convert\"\n description = \"Convert between different types (Message, Data, DataFrame)\"\n icon = \"repeat\"\n\n inputs = [\n HandleInput(\n name=\"input_data\",\n display_name=\"Input\",\n input_types=[\"Message\", \"Data\", \"DataFrame\"],\n info=\"Accept Message, Data or DataFrame as input\",\n required=True,\n ),\n TabInput(\n name=\"output_type\",\n display_name=\"Output Type\",\n options=[\"Message\", \"Data\", \"DataFrame\"],\n info=\"Select the desired output data type\",\n real_time_refresh=True,\n value=\"Message\",\n ),\n ]\n\n outputs = [Output(display_name=\"Message Output\", name=\"message_output\", method=\"convert_to_message\")]\n\n def update_outputs(self, frontend_node: dict, field_name: str, field_value: Any) -> dict:\n \"\"\"Dynamically show only the relevant output based on the selected output type.\"\"\"\n if field_name == \"output_type\":\n # Start with empty outputs\n frontend_node[\"outputs\"] = []\n\n # Add only the selected output type\n if field_value == \"Message\":\n frontend_node[\"outputs\"].append(\n Output(display_name=\"Message Output\", name=\"message_output\", method=\"convert_to_message\").to_dict()\n )\n elif field_value == \"Data\":\n frontend_node[\"outputs\"].append(\n Output(display_name=\"Data Output\", name=\"data_output\", method=\"convert_to_data\").to_dict()\n )\n elif field_value == \"DataFrame\":\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"DataFrame Output\", name=\"dataframe_output\", method=\"convert_to_dataframe\"\n ).to_dict()\n )\n\n return frontend_node\n\n def convert_to_message(self) -> Message:\n \"\"\"Convert input to Message type.\"\"\"\n return convert_to_message(self.input_data[0] if isinstance(self.input_data, list) else self.input_data)\n\n def convert_to_data(self) -> Data:\n \"\"\"Convert input to Data type.\"\"\"\n return convert_to_data(self.input_data[0] if isinstance(self.input_data, list) else self.input_data)\n\n def convert_to_dataframe(self) -> DataFrame:\n \"\"\"Convert input to DataFrame type.\"\"\"\n return convert_to_dataframe(self.input_data[0] if isinstance(self.input_data, list) else self.input_data)\n"
"value": "from typing import Any\n\nfrom langflow.custom import Component\nfrom langflow.io import HandleInput, Output, TabInput\nfrom langflow.schema import Data, DataFrame, Message\n\n\ndef convert_to_message(v) -> Message:\n \"\"\"Convert input to Message type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n Message: Converted Message object\n \"\"\"\n return v if isinstance(v, Message) else v.to_message()\n\n\ndef convert_to_data(v: DataFrame | Data | Message | dict) -> Data:\n \"\"\"Convert input to Data type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n Data: Converted Data object\n \"\"\"\n if isinstance(v, dict):\n return Data(v)\n return v if isinstance(v, Data) else v.to_data()\n\n\ndef convert_to_dataframe(v: DataFrame | Data | Message | dict) -> DataFrame:\n \"\"\"Convert input to DataFrame type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n DataFrame: Converted DataFrame object\n \"\"\"\n if isinstance(v, dict):\n return DataFrame([v])\n return v if isinstance(v, DataFrame) else v.to_dataframe()\n\n\nclass TypeConverterComponent(Component):\n display_name = \"Type Convert\"\n description = \"Convert between different types (Message, Data, DataFrame)\"\n icon = \"repeat\"\n\n inputs = [\n HandleInput(\n name=\"input_data\",\n display_name=\"Input\",\n input_types=[\"Message\", \"Data\", \"DataFrame\"],\n info=\"Accept Message, Data or DataFrame as input\",\n required=True,\n ),\n TabInput(\n name=\"output_type\",\n display_name=\"Output Type\",\n options=[\"Message\", \"Data\", \"DataFrame\"],\n info=\"Select the desired output data type\",\n real_time_refresh=True,\n value=\"Message\",\n ),\n ]\n\n outputs = [\n Output(\n display_name=\"Message Output\",\n name=\"message_output\",\n method=\"convert_to_message\",\n )\n ]\n\n def update_outputs(self, frontend_node: dict, field_name: str, field_value: Any) -> dict:\n \"\"\"Dynamically show only the relevant output based on the selected output type.\"\"\"\n if field_name == \"output_type\":\n # Start with empty outputs\n frontend_node[\"outputs\"] = []\n\n # Add only the selected output type\n if field_value == \"Message\":\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"Message Output\",\n name=\"message_output\",\n method=\"convert_to_message\",\n ).to_dict()\n )\n elif field_value == \"Data\":\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"Data Output\",\n name=\"data_output\",\n method=\"convert_to_data\",\n ).to_dict()\n )\n elif field_value == \"DataFrame\":\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"DataFrame Output\",\n name=\"dataframe_output\",\n method=\"convert_to_dataframe\",\n ).to_dict()\n )\n\n return frontend_node\n\n def convert_to_message(self) -> Message:\n \"\"\"Convert input to Message type.\"\"\"\n input_value = self.input_data[0] if isinstance(self.input_data, list) else self.input_data\n\n # Handle string input by converting to Message first\n if isinstance(input_value, str):\n input_value = Message(text=input_value)\n\n return convert_to_message(input_value)\n\n def convert_to_data(self) -> Data:\n \"\"\"Convert input to Data type.\"\"\"\n input_value = self.input_data[0] if isinstance(self.input_data, list) else self.input_data\n\n # Handle string input by converting to Message first\n if isinstance(input_value, str):\n input_value = Message(text=input_value)\n\n return convert_to_data(input_value)\n\n def convert_to_dataframe(self) -> DataFrame:\n \"\"\"Convert input to DataFrame type.\"\"\"\n input_value = self.input_data[0] if isinstance(self.input_data, list) else self.input_data\n\n # Handle string input by converting to Message first\n if isinstance(input_value, str):\n input_value = Message(text=input_value)\n\n return convert_to_dataframe(input_value)\n"
},
"input_data": {
"_input_type": "HandleInput",

View file

@ -3964,7 +3964,7 @@
"show": true,
"title_case": false,
"type": "code",
"value": "from typing import Any\n\nfrom langflow.custom import Component\nfrom langflow.io import HandleInput, Output, TabInput\nfrom langflow.schema import Data, DataFrame, Message\n\n\ndef convert_to_message(v) -> Message:\n \"\"\"Convert input to Message type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n Message: Converted Message object\n \"\"\"\n return v if isinstance(v, Message) else v.to_message()\n\n\ndef convert_to_data(v: DataFrame | Data | Message | dict) -> Data:\n \"\"\"Convert input to Data type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n Data: Converted Data object\n \"\"\"\n if isinstance(v, dict):\n return Data(v)\n return v if isinstance(v, Data) else v.to_data()\n\n\ndef convert_to_dataframe(v: DataFrame | Data | Message | dict) -> DataFrame:\n \"\"\"Convert input to DataFrame type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n DataFrame: Converted DataFrame object\n \"\"\"\n if isinstance(v, dict):\n return DataFrame([v])\n return v if isinstance(v, DataFrame) else v.to_dataframe()\n\n\nclass TypeConverterComponent(Component):\n display_name = \"Type Convert\"\n description = \"Convert between different types (Message, Data, DataFrame)\"\n icon = \"repeat\"\n\n inputs = [\n HandleInput(\n name=\"input_data\",\n display_name=\"Input\",\n input_types=[\"Message\", \"Data\", \"DataFrame\"],\n info=\"Accept Message, Data or DataFrame as input\",\n required=True,\n ),\n TabInput(\n name=\"output_type\",\n display_name=\"Output Type\",\n options=[\"Message\", \"Data\", \"DataFrame\"],\n info=\"Select the desired output data type\",\n real_time_refresh=True,\n value=\"Message\",\n ),\n ]\n\n outputs = [Output(display_name=\"Message Output\", name=\"message_output\", method=\"convert_to_message\")]\n\n def update_outputs(self, frontend_node: dict, field_name: str, field_value: Any) -> dict:\n \"\"\"Dynamically show only the relevant output based on the selected output type.\"\"\"\n if field_name == \"output_type\":\n # Start with empty outputs\n frontend_node[\"outputs\"] = []\n\n # Add only the selected output type\n if field_value == \"Message\":\n frontend_node[\"outputs\"].append(\n Output(display_name=\"Message Output\", name=\"message_output\", method=\"convert_to_message\").to_dict()\n )\n elif field_value == \"Data\":\n frontend_node[\"outputs\"].append(\n Output(display_name=\"Data Output\", name=\"data_output\", method=\"convert_to_data\").to_dict()\n )\n elif field_value == \"DataFrame\":\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"DataFrame Output\", name=\"dataframe_output\", method=\"convert_to_dataframe\"\n ).to_dict()\n )\n\n return frontend_node\n\n def convert_to_message(self) -> Message:\n \"\"\"Convert input to Message type.\"\"\"\n return convert_to_message(self.input_data[0] if isinstance(self.input_data, list) else self.input_data)\n\n def convert_to_data(self) -> Data:\n \"\"\"Convert input to Data type.\"\"\"\n return convert_to_data(self.input_data[0] if isinstance(self.input_data, list) else self.input_data)\n\n def convert_to_dataframe(self) -> DataFrame:\n \"\"\"Convert input to DataFrame type.\"\"\"\n return convert_to_dataframe(self.input_data[0] if isinstance(self.input_data, list) else self.input_data)\n"
"value": "from typing import Any\n\nfrom langflow.custom import Component\nfrom langflow.io import HandleInput, Output, TabInput\nfrom langflow.schema import Data, DataFrame, Message\n\n\ndef convert_to_message(v) -> Message:\n \"\"\"Convert input to Message type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n Message: Converted Message object\n \"\"\"\n return v if isinstance(v, Message) else v.to_message()\n\n\ndef convert_to_data(v: DataFrame | Data | Message | dict) -> Data:\n \"\"\"Convert input to Data type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n Data: Converted Data object\n \"\"\"\n if isinstance(v, dict):\n return Data(v)\n return v if isinstance(v, Data) else v.to_data()\n\n\ndef convert_to_dataframe(v: DataFrame | Data | Message | dict) -> DataFrame:\n \"\"\"Convert input to DataFrame type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n DataFrame: Converted DataFrame object\n \"\"\"\n if isinstance(v, dict):\n return DataFrame([v])\n return v if isinstance(v, DataFrame) else v.to_dataframe()\n\n\nclass TypeConverterComponent(Component):\n display_name = \"Type Convert\"\n description = \"Convert between different types (Message, Data, DataFrame)\"\n icon = \"repeat\"\n\n inputs = [\n HandleInput(\n name=\"input_data\",\n display_name=\"Input\",\n input_types=[\"Message\", \"Data\", \"DataFrame\"],\n info=\"Accept Message, Data or DataFrame as input\",\n required=True,\n ),\n TabInput(\n name=\"output_type\",\n display_name=\"Output Type\",\n options=[\"Message\", \"Data\", \"DataFrame\"],\n info=\"Select the desired output data type\",\n real_time_refresh=True,\n value=\"Message\",\n ),\n ]\n\n outputs = [\n Output(\n display_name=\"Message Output\",\n name=\"message_output\",\n method=\"convert_to_message\",\n )\n ]\n\n def update_outputs(self, frontend_node: dict, field_name: str, field_value: Any) -> dict:\n \"\"\"Dynamically show only the relevant output based on the selected output type.\"\"\"\n if field_name == \"output_type\":\n # Start with empty outputs\n frontend_node[\"outputs\"] = []\n\n # Add only the selected output type\n if field_value == \"Message\":\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"Message Output\",\n name=\"message_output\",\n method=\"convert_to_message\",\n ).to_dict()\n )\n elif field_value == \"Data\":\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"Data Output\",\n name=\"data_output\",\n method=\"convert_to_data\",\n ).to_dict()\n )\n elif field_value == \"DataFrame\":\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"DataFrame Output\",\n name=\"dataframe_output\",\n method=\"convert_to_dataframe\",\n ).to_dict()\n )\n\n return frontend_node\n\n def convert_to_message(self) -> Message:\n \"\"\"Convert input to Message type.\"\"\"\n input_value = self.input_data[0] if isinstance(self.input_data, list) else self.input_data\n\n # Handle string input by converting to Message first\n if isinstance(input_value, str):\n input_value = Message(text=input_value)\n\n return convert_to_message(input_value)\n\n def convert_to_data(self) -> Data:\n \"\"\"Convert input to Data type.\"\"\"\n input_value = self.input_data[0] if isinstance(self.input_data, list) else self.input_data\n\n # Handle string input by converting to Message first\n if isinstance(input_value, str):\n input_value = Message(text=input_value)\n\n return convert_to_data(input_value)\n\n def convert_to_dataframe(self) -> DataFrame:\n \"\"\"Convert input to DataFrame type.\"\"\"\n input_value = self.input_data[0] if isinstance(self.input_data, list) else self.input_data\n\n # Handle string input by converting to Message first\n if isinstance(input_value, str):\n input_value = Message(text=input_value)\n\n return convert_to_dataframe(input_value)\n"
},
"input_data": {
"_input_type": "HandleInput",

View file

@ -1703,7 +1703,7 @@
"show": true,
"title_case": false,
"type": "code",
"value": "from typing import Any\n\nfrom langflow.custom import Component\nfrom langflow.io import HandleInput, Output, TabInput\nfrom langflow.schema import Data, DataFrame, Message\n\n\ndef convert_to_message(v) -> Message:\n \"\"\"Convert input to Message type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n Message: Converted Message object\n \"\"\"\n return v if isinstance(v, Message) else v.to_message()\n\n\ndef convert_to_data(v: DataFrame | Data | Message | dict) -> Data:\n \"\"\"Convert input to Data type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n Data: Converted Data object\n \"\"\"\n if isinstance(v, dict):\n return Data(v)\n return v if isinstance(v, Data) else v.to_data()\n\n\ndef convert_to_dataframe(v: DataFrame | Data | Message | dict) -> DataFrame:\n \"\"\"Convert input to DataFrame type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n DataFrame: Converted DataFrame object\n \"\"\"\n if isinstance(v, dict):\n return DataFrame([v])\n return v if isinstance(v, DataFrame) else v.to_dataframe()\n\n\nclass TypeConverterComponent(Component):\n display_name = \"Type Convert\"\n description = \"Convert between different types (Message, Data, DataFrame)\"\n icon = \"repeat\"\n\n inputs = [\n HandleInput(\n name=\"input_data\",\n display_name=\"Input\",\n input_types=[\"Message\", \"Data\", \"DataFrame\"],\n info=\"Accept Message, Data or DataFrame as input\",\n required=True,\n ),\n TabInput(\n name=\"output_type\",\n display_name=\"Output Type\",\n options=[\"Message\", \"Data\", \"DataFrame\"],\n info=\"Select the desired output data type\",\n real_time_refresh=True,\n value=\"Message\",\n ),\n ]\n\n outputs = [Output(display_name=\"Message Output\", name=\"message_output\", method=\"convert_to_message\")]\n\n def update_outputs(self, frontend_node: dict, field_name: str, field_value: Any) -> dict:\n \"\"\"Dynamically show only the relevant output based on the selected output type.\"\"\"\n if field_name == \"output_type\":\n # Start with empty outputs\n frontend_node[\"outputs\"] = []\n\n # Add only the selected output type\n if field_value == \"Message\":\n frontend_node[\"outputs\"].append(\n Output(display_name=\"Message Output\", name=\"message_output\", method=\"convert_to_message\").to_dict()\n )\n elif field_value == \"Data\":\n frontend_node[\"outputs\"].append(\n Output(display_name=\"Data Output\", name=\"data_output\", method=\"convert_to_data\").to_dict()\n )\n elif field_value == \"DataFrame\":\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"DataFrame Output\", name=\"dataframe_output\", method=\"convert_to_dataframe\"\n ).to_dict()\n )\n\n return frontend_node\n\n def convert_to_message(self) -> Message:\n \"\"\"Convert input to Message type.\"\"\"\n return convert_to_message(self.input_data[0] if isinstance(self.input_data, list) else self.input_data)\n\n def convert_to_data(self) -> Data:\n \"\"\"Convert input to Data type.\"\"\"\n return convert_to_data(self.input_data[0] if isinstance(self.input_data, list) else self.input_data)\n\n def convert_to_dataframe(self) -> DataFrame:\n \"\"\"Convert input to DataFrame type.\"\"\"\n return convert_to_dataframe(self.input_data[0] if isinstance(self.input_data, list) else self.input_data)\n"
"value": "from typing import Any\n\nfrom langflow.custom import Component\nfrom langflow.io import HandleInput, Output, TabInput\nfrom langflow.schema import Data, DataFrame, Message\n\n\ndef convert_to_message(v) -> Message:\n \"\"\"Convert input to Message type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n Message: Converted Message object\n \"\"\"\n return v if isinstance(v, Message) else v.to_message()\n\n\ndef convert_to_data(v: DataFrame | Data | Message | dict) -> Data:\n \"\"\"Convert input to Data type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n Data: Converted Data object\n \"\"\"\n if isinstance(v, dict):\n return Data(v)\n return v if isinstance(v, Data) else v.to_data()\n\n\ndef convert_to_dataframe(v: DataFrame | Data | Message | dict) -> DataFrame:\n \"\"\"Convert input to DataFrame type.\n\n Args:\n v: Input to convert (Message, Data, DataFrame, or dict)\n\n Returns:\n DataFrame: Converted DataFrame object\n \"\"\"\n if isinstance(v, dict):\n return DataFrame([v])\n return v if isinstance(v, DataFrame) else v.to_dataframe()\n\n\nclass TypeConverterComponent(Component):\n display_name = \"Type Convert\"\n description = \"Convert between different types (Message, Data, DataFrame)\"\n icon = \"repeat\"\n\n inputs = [\n HandleInput(\n name=\"input_data\",\n display_name=\"Input\",\n input_types=[\"Message\", \"Data\", \"DataFrame\"],\n info=\"Accept Message, Data or DataFrame as input\",\n required=True,\n ),\n TabInput(\n name=\"output_type\",\n display_name=\"Output Type\",\n options=[\"Message\", \"Data\", \"DataFrame\"],\n info=\"Select the desired output data type\",\n real_time_refresh=True,\n value=\"Message\",\n ),\n ]\n\n outputs = [\n Output(\n display_name=\"Message Output\",\n name=\"message_output\",\n method=\"convert_to_message\",\n )\n ]\n\n def update_outputs(self, frontend_node: dict, field_name: str, field_value: Any) -> dict:\n \"\"\"Dynamically show only the relevant output based on the selected output type.\"\"\"\n if field_name == \"output_type\":\n # Start with empty outputs\n frontend_node[\"outputs\"] = []\n\n # Add only the selected output type\n if field_value == \"Message\":\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"Message Output\",\n name=\"message_output\",\n method=\"convert_to_message\",\n ).to_dict()\n )\n elif field_value == \"Data\":\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"Data Output\",\n name=\"data_output\",\n method=\"convert_to_data\",\n ).to_dict()\n )\n elif field_value == \"DataFrame\":\n frontend_node[\"outputs\"].append(\n Output(\n display_name=\"DataFrame Output\",\n name=\"dataframe_output\",\n method=\"convert_to_dataframe\",\n ).to_dict()\n )\n\n return frontend_node\n\n def convert_to_message(self) -> Message:\n \"\"\"Convert input to Message type.\"\"\"\n input_value = self.input_data[0] if isinstance(self.input_data, list) else self.input_data\n\n # Handle string input by converting to Message first\n if isinstance(input_value, str):\n input_value = Message(text=input_value)\n\n return convert_to_message(input_value)\n\n def convert_to_data(self) -> Data:\n \"\"\"Convert input to Data type.\"\"\"\n input_value = self.input_data[0] if isinstance(self.input_data, list) else self.input_data\n\n # Handle string input by converting to Message first\n if isinstance(input_value, str):\n input_value = Message(text=input_value)\n\n return convert_to_data(input_value)\n\n def convert_to_dataframe(self) -> DataFrame:\n \"\"\"Convert input to DataFrame type.\"\"\"\n input_value = self.input_data[0] if isinstance(self.input_data, list) else self.input_data\n\n # Handle string input by converting to Message first\n if isinstance(input_value, str):\n input_value = Message(text=input_value)\n\n return convert_to_dataframe(input_value)\n"
},
"input_data": {
"_input_type": "HandleInput",