diff --git a/src/backend/base/langflow/custom/custom_component/component.py b/src/backend/base/langflow/custom/custom_component/component.py index a5373b449..4af5f65b8 100644 --- a/src/backend/base/langflow/custom/custom_component/component.py +++ b/src/backend/base/langflow/custom/custom_component/component.py @@ -89,6 +89,40 @@ class Component(CustomComponent): raise ValueError("Output name cannot be None.") self._outputs[output.name] = output + def get_input(self, name: str) -> Any: + """ + Retrieves the value of the input with the specified name. + Args: + name (str): The name of the input. + Returns: + Any: The value of the input. + Raises: + ValueError: If the input with the specified name is not found. + """ + if name in self._inputs: + return self._inputs[name] + raise ValueError(f"Input {name} not found in {self.__class__.__name__}") + + def get_output(self, name: str) -> Any: + """ + Retrieves the output with the specified name. + Args: + name (str): The name of the output to retrieve. + Returns: + Any: The output value. + Raises: + ValueError: If the output with the specified name is not found. + """ + if name in self._outputs: + return self._outputs[name] + raise ValueError(f"Output {name} not found in {self.__class__.__name__}") + + def set_output_value(self, name: str, value: Any): + if name in self._outputs: + self._outputs[name].value = value + else: + raise ValueError(f"Output {name} not found in {self.__class__.__name__}") + def validate(self, params: dict): self._validate_inputs(params) self._validate_outputs() diff --git a/src/backend/base/langflow/graph/vertex/types.py b/src/backend/base/langflow/graph/vertex/types.py index c36487dd9..2078e341c 100644 --- a/src/backend/base/langflow/graph/vertex/types.py +++ b/src/backend/base/langflow/graph/vertex/types.py @@ -9,11 +9,12 @@ from loguru import logger from langflow.graph.schema import CHAT_COMPONENTS, RECORDS_COMPONENTS, InterfaceComponentTypes, ResultData from langflow.graph.utils import UnbuiltObject, log_transaction, log_vertex_build, serialize_field from langflow.graph.vertex.base import Vertex +from langflow.inputs.inputs import InputTypes from langflow.schema import Data from langflow.schema.artifact import ArtifactType from langflow.schema.message import Message from langflow.schema.schema import INPUT_FIELD_NAME -from langflow.template.field.base import UNDEFINED +from langflow.template.field.base import UNDEFINED, Output from langflow.utils.schemas import ChatOutputResponse, DataOutputResponse from langflow.utils.util import unescape_string @@ -57,6 +58,16 @@ class ComponentVertex(Vertex): for key, value in self._built_object.items(): self.add_result(key, value) + def get_input(self, name: str) -> InputTypes: + if self._custom_component is None: + raise ValueError(f"Vertex {self.id} does not have a component instance.") + return self._custom_component.get_input(name) + + def get_output(self, name: str) -> Output: + if self._custom_component is None: + raise ValueError(f"Vertex {self.id} does not have a component instance.") + return self._custom_component.get_output(name) + def get_edge_with_target(self, target_id: str) -> Generator["ContractEdge", None, None]: """ Get the edge with the target id.