Definition of output type in run flow

[zhutaoshen]

The current version I am using is 1.4.1.
Define multiple different output types in one sub flow, including data and dataFrame types, and use run flow to call this sub flow in another flow. How to implement this process? I only saw an output that can only output messages, but cannot output dataFrame.

[dosubot[bot]]

Hello @zhutaoshen! I'm Dosu, here to help you with bugs, questions, and contributing while you wait for a human maintainer. Let's get started on solving your issue!

To define multiple different output types, including data and dataFrame, in one sub flow and use run_flow to call this sub flow in another flow in version 1.4.1, you can utilize the RunFlowBaseComponent class. This class provides methods to handle different output types. Here's how you can implement this:

1. Define Outputs in the Sub Flow:
   Use the _base_outputs attribute of the RunFlowBaseComponent class to define outputs for data, dataFrame, and message:

   _base_outputs: list[Output] = [
       Output(
           name="flow_outputs_data",
           display_name="Flow Data Output",
           method="data_output",
           hidden=True,
           tool_mode=False,
       ),
       Output(
           name="flow_outputs_dataframe",
           display_name="Flow Dataframe Output",
           method="dataframe_output",
           hidden=True,
           tool_mode=False,
       ),
       Output(name="flow_outputs_message", display_name="Flow Message Output", method="message_output"),
   ]

2. Implement Output Methods:
   Implement the methods data_output, dataframe_output, and message_output to return the respective output types. You can use the build_structured_output_base method to construct structured outputs:

   async def data_output(self) -> Data:
       output = self.build_structured_output_base()
       return Data(text_key="results", data={"results": output})
   
   async def dataframe_output(self) -> DataFrame:
       output = self.build_structured_output_base()
       if isinstance(output, list):
           return DataFrame(data=output)
       return DataFrame(data=[output])
   
   async def message_output(self) -> Message:
       run_outputs = await self.run_flow_with_tweaks()
       _, message_result = next(iter(run_outputs[0].outputs[0].results.items()))
       if isinstance(message_result, Message):
           return message_result
       if isinstance(message_result, str):
           return Message(text=message_result)
       return Message(text=message_result.data["text"])

3. Use run_flow in Another Flow:
   To call this sub flow in another flow, use the run_flow_with_tweaks method to execute the flow and retrieve the outputs:

   async def run_flow_with_tweaks(self) -> list[Data]:
       """Run the flow with tweaks."""
       # Implement the logic to run the flow and return the outputs

By following these steps, you can define multiple output types in a sub flow and use run_flow to call this sub flow in another flow ^[1][2].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}

[zhutaoshen]

Thank you very much for your reply. Does it mean that I need to modify the referenced component code to implement this feature? I'll try.

[dosubot[bot]]

Yes, to implement the feature of defining multiple output types in a sub flow and using run_flow to call this sub flow in another flow, you would need to modify the component code. Specifically, you would need to define the outputs and implement the methods for handling different output types, as described in the previous response. This involves working with the RunFlowBaseComponent class and its methods to ensure that your flow can handle and return the desired output types like data and dataFrame.

Regarding the image you mentioned, it seems to show the update_build_config and run_flow_with_tweaks methods. The update_build_config method is used to dynamically adjust the build configuration based on field updates, ensuring it reflects the current state of available flow names ^[1]. The run_flow_with_tweaks method processes flow tweaks, parsing data and splitting fields based on a delimiter ^[2]. You might need to ensure these methods are correctly implemented to support your flow management needs.

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Bug Report | Other}