feat: add dynamic outputs to Component (#4308)

* Refine condition to check for undefined value in use-handle-new-value hook

* update mutateTemplate to use APIClassType

* Refactor Component class to update inputs and validate outputs

This commit refactors the Component class in the `component.py` file. It introduces two new methods: `update_inputs` and `run_and_validate_update_outputs`. The `update_inputs` method allows for updating the build configuration with new field values, while the `run_and_validate_update_outputs` method updates the frontend node and validates the outputs. Additionally, the `_validate_frontend_node` method is added to check if all outputs are valid. The `_set_output_types` method is modified to accept a list of outputs and set their return types. Overall, these changes improve the functionality and maintainability of the Component class.

* Add dynamic output validation in API endpoint for component updates

* Update build_config to store field_value under "value" key

* Refactor: Convert dict values to list in _set_output_types call

* Add type check for `cc_instance` before calling `run_and_validate_update_outputs`

* Add DynamicOutputComponent with configurable outputs based on input

* Add test for updating component outputs with dynamic code input

* Refactor: Make get_dynamic_output_component_code asynchronous for improved performance

---------

Co-authored-by: Edwin Jose <edwin.jose@datastax.com>

src/backend/base/langflow/api/v1/endpoints.py

@@ -49,12 +49,7 @@ from langflow.services.database.models.flow.utils import (
     get_all_webhook_components_in_flow,
 )
 from langflow.services.database.models.user.model import User, UserRead
-from langflow.services.deps import (
-    get_session_service,
-    get_settings_service,
-    get_task_service,
-    get_telemetry_service,
-)
+from langflow.services.deps import get_session_service, get_settings_service, get_task_service, get_telemetry_service
 from langflow.services.settings.feature_flags import FEATURE_FLAGS
 from langflow.services.telemetry.schema import RunPayload
 from langflow.utils.version import get_version_info
@@ -637,10 +632,16 @@ async def custom_component_update(
             field_value=code_request.field_value,
             field_name=code_request.field,
         )
+        component_node["template"] = updated_build_config
+        if isinstance(cc_instance, Component):
+            cc_instance.run_and_validate_update_outputs(
+                frontend_node=component_node,
+                field_name=code_request.field,
+                field_value=code_request.field_value,
+            )
+
     except Exception as exc:
         raise HTTPException(status_code=400, detail=str(exc)) from exc
-
-    component_node["template"] = updated_build_config
     return component_node
 
 

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

@@ -9,7 +9,7 @@ from typing import TYPE_CHECKING, Any, ClassVar, get_type_hints
 
 import nanoid
 import yaml
-from pydantic import BaseModel
+from pydantic import BaseModel, ValidationError
 
 from langflow.base.tools.constants import TOOL_OUTPUT_NAME
 from langflow.custom.tree_visitor import RequiredInputsVisitor
@@ -35,6 +35,7 @@ if TYPE_CHECKING:
     from langflow.graph.edge.schema import EdgeData
     from langflow.graph.vertex.base import Vertex
     from langflow.inputs.inputs import InputTypes
+    from langflow.schema import dotdict
     from langflow.schema.log import LoggableType
 
 
@@ -104,7 +105,7 @@ class Component(CustomComponent):
         if self.outputs is not None:
             self.map_outputs(self.outputs)
         # Set output types
-        self._set_output_types()
+        self._set_output_types(list(self._outputs_map.values()))
         self.set_class_code()
         self._set_output_required_inputs()
 
@@ -310,14 +311,57 @@ class Component(CustomComponent):
         self._validate_inputs(params)
         self._validate_outputs()
 
-    def _set_output_types(self) -> None:
-        for output in self._outputs_map.values():
-            if output.method is None:
-                msg = f"Output {output.name} does not have a method"
-                raise ValueError(msg)
-            return_types = self._get_method_return_type(output.method)
-            output.add_types(return_types)
-            output.set_selected()
+    def update_inputs(
+        self,
+        build_config: dotdict,
+        field_value: Any,
+        field_name: str | None = None,
+    ):
+        return self.update_build_config(build_config, field_value, field_name)
+
+    def run_and_validate_update_outputs(self, frontend_node: dict, field_name: str, field_value: Any):
+        frontend_node = self.update_outputs(frontend_node, field_name, field_value)
+        return self._validate_frontend_node(frontend_node)
+
+    def _validate_frontend_node(self, frontend_node: dict):
+        # Check if all outputs are either Output or a valid Output model
+        for index, output in enumerate(frontend_node["outputs"]):
+            if isinstance(output, dict):
+                try:
+                    _output = Output(**output)
+                    self._set_output_return_type(_output)
+                    _output_dict = _output.model_dump()
+                except ValidationError as e:
+                    msg = f"Invalid output: {e}"
+                    raise ValueError(msg) from e
+            elif isinstance(output, Output):
+                # we need to serialize it
+                self._set_output_return_type(output)
+                _output_dict = output.model_dump()
+            else:
+                msg = f"Invalid output type: {type(output)}"
+                raise TypeError(msg)
+            frontend_node["outputs"][index] = _output_dict
+        return frontend_node
+
+    def update_outputs(self, frontend_node: dict, field_name: str, field_value: Any) -> dict:  # noqa: ARG002
+        """Default implementation for updating outputs based on field changes.
+
+        Subclasses can override this to modify outputs based on field_name and field_value.
+        """
+        return frontend_node
+
+    def _set_output_types(self, outputs: list[Output]) -> None:
+        for output in outputs:
+            self._set_output_return_type(output)
+
+    def _set_output_return_type(self, output: Output) -> None:
+        if output.method is None:
+            msg = f"Output {output.name} does not have a method"
+            raise ValueError(msg)
+        return_types = self._get_method_return_type(output.method)
+        output.add_types(return_types)
+        output.set_selected()
 
     def _set_output_required_inputs(self) -> None:
         for output in self.outputs:

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

@@ -228,7 +228,7 @@ class CustomComponent(BaseComponent):
         field_value: Any,
         field_name: str | None = None,
     ):
-        build_config[field_name] = field_value
+        build_config[field_name]["value"] = field_value
         return build_config
 
     @property