diff --git a/src/backend/langflow/graph/vertex/base.py b/src/backend/langflow/graph/vertex/base.py index 79c3270fb..a7fcf88b5 100644 --- a/src/backend/langflow/graph/vertex/base.py +++ b/src/backend/langflow/graph/vertex/base.py @@ -1,14 +1,11 @@ -from langflow.utils.constants import DIRECT_TYPES from langflow.interface.initialize import loading from langflow.interface.listing import ALL_TYPES_DICT from langflow.utils.logger import logger from langflow.utils.util import sync_to_async -import contextlib import inspect import types -import warnings from typing import Any, Dict, List, Optional from typing import TYPE_CHECKING @@ -69,6 +66,7 @@ class Vertex: break def _build_params(self): + # sourcery skip: merge-list-append, remove-redundant-if # Some params are required, some are optional # but most importantly, some params are python base classes # like str and others are LangChain objects like LLMChain, BasePromptTemplate @@ -89,8 +87,19 @@ class Vertex: if isinstance(value, dict) } params = {} + + for edge in self.edges: + param_key = edge.target_param + if param_key in template_dict: + if template_dict[param_key]["list"]: + if param_key not in params: + params[param_key] = [] + params[param_key].append(edge.source) + else: + params[param_key] = edge.source + for key, value in template_dict.items(): - if key == "_type": + if key == "_type" or not value.get("show"): continue # If the type is not transformable to a python base class # then we need to get the edge that connects to this node @@ -101,45 +110,8 @@ class Vertex: file_path = value.get("file_path") params[key] = file_path - - elif value.get("type") not in DIRECT_TYPES: - # Get the edge that connects to this node - edges = [ - edge - for edge in self.edges - if edge.target == self and edge.matched_type in value["type"] - ] - - # Get the output of the node that the edge connects to - # if the value['list'] is True, then there will be more - # than one time setting to params[key] - # so we need to append to a list if it exists - # or create a new list if it doesn't - - if value["required"] and not edges: - # If a required parameter is not found, raise an error - raise ValueError( - f"Required input {key} for module {self.vertex_type} not found" - ) - elif value["list"]: - # If this is a list parameter, append all sources to a list - params[key] = [edge.source for edge in edges] - elif edges: - # If a single parameter is found, use its source - params[key] = edges[0].source - - elif value["required"] or value.get("value"): - # If value does not have value this still passes - # but then gives a keyError - # so we need to check if value has value - new_value = value.get("value") - if new_value is None: - warnings.warn(f"Value for {key} in {self.vertex_type} is None. ") - if value.get("type") == "int": - with contextlib.suppress(TypeError, ValueError): - new_value = int(new_value) # type: ignore - params[key] = new_value - + elif value.get("type") in ["str", "prompt"] and params.get(key) is None: + params[key] = value.get("value") # Add _type to params self.params = params