diff --git a/src/backend/langflow/graph/__init__.py b/src/backend/langflow/graph/__init__.py index e69de29bb..3afa92b80 100644 --- a/src/backend/langflow/graph/__init__.py +++ b/src/backend/langflow/graph/__init__.py @@ -0,0 +1,4 @@ +from langflow.graph.graph import Graph +from langflow.graph.base import Node, Edge + +__all__ = ["Graph", "Node", "Edge"] diff --git a/src/backend/langflow/graph/base.py b/src/backend/langflow/graph/base.py new file mode 100644 index 000000000..65e81f934 --- /dev/null +++ b/src/backend/langflow/graph/base.py @@ -0,0 +1,213 @@ +# Description: Graph class for building a graph of nodes and edges +# Insights: +# - Defer prompts building to the last moment or when they have all the tools +# - Build each inner agent first, then build the outer agent + +from copy import deepcopy +import types +from typing import Any, Dict, List, Optional, Union +from langflow.utils import payload +from langflow.interface.listing import ALL_TYPES_DICT, ALL_TOOLS_NAMES, TOOLS_DICT +from langflow.interface import loading + + +class Node: + def __init__(self, data: Dict): + self.id: str = data["id"] + self._data = data + self.edges: List[Edge] = [] + self._parse_data() + self._built_object = None + self._built = False + + def _parse_data(self) -> None: + self.data = self._data["data"] + self.output = self.data["node"]["base_classes"] + template_dicts = { + key: value + for key, value in self.data["node"]["template"].items() + if isinstance(value, dict) + } + + self.required_inputs = [ + template_dicts[key]["type"] + for key, value in template_dicts.items() + if value["required"] + ] + self.optional_inputs = [ + template_dicts[key]["type"] + for key, value in template_dicts.items() + if not value["required"] + ] + + template_dict = self.data["node"]["template"] + self.node_type = ( + self.data["type"] if "Tool" not in self.output else template_dict["_type"] + ) + + def _build_params(self): + # 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 + # so we need to be able to distinguish between the two + + # The dicts with "type" == "str" are the ones that are python base classes + # and most likely have a "value" key + + # So for each key besides "_type" in the template dict, we have a dict + # with a "type" key. If the type is not "str", then we need to get the + # edge that connects to that node and get the Node with the required data + # and use that as the value for the param + # If the type is "str", then we need to get the value of the "value" key + # and use that as the value for the param + template_dict = { + key: value + for key, value in self.data["node"]["template"].items() + if isinstance(value, dict) + } + params = {} + for key, value in template_dict.items(): + if key == "_type": + continue + # If the type is not transformable to a python base class + # then we need to get the edge that connects to this node + if value["type"] not in ["str", "bool", "code"]: + # Get the edge that connects to this node + edge = next( + ( + edge + for edge in self.edges + if edge.target == self and edge.matched_type in value["type"] + ), + None, + ) + # 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 edge is None and value["required"]: + # break line + raise ValueError( + f"Required input {key} for module {self.node_type} not found" + ) + if value["list"]: + if key in params: + params[key].append(edge.source) + else: + params[key] = [edge.source] + elif value["required"] or edge is not None: + params[key] = edge.source + elif value["required"] or value.get("value"): + params[key] = value["value"] + + # Add _type to params + self.params = params + + def _build(self): + # The params dict is used to build the module + # it contains values and keys that point to nodes which + # have their own params dict + # When build is called, we iterate through the params dict + # and if the value is a node, we call build on that node + # and use the output of that build as the value for the param + # if the value is not a node, then we use the value as the param + # and continue + # Another aspect is that the node_type is the class that we need to import + # and instantiate with these built params + + # Build each node in the params dict + for key, value in self.params.items(): + # Check if Node or list of Nodes + if isinstance(value, Node): + result = value.build() + # If the key is "func", then we need to use the run method + if key == "func" and not isinstance(result, types.FunctionType): + # func can be PythonFunction(code='\ndef upper_case(text: str) -> str:\n return text.upper()\n') + # so we need to check if there is an attribute called run + if hasattr(result, "run"): + result = result.run # type: ignore + elif hasattr(result, "get_function"): + result = result.get_function() # type: ignore + self.params[key] = result + elif isinstance(value, list) and all( + isinstance(node, Node) for node in value + ): + self.params[key] = [node.build() for node in value] # type: ignore + + # Get the class from LANGCHAIN_TYPES_DICT + # and instantiate it with the params + # and return the instance + for base_type, value in ALL_TYPES_DICT.items(): + if base_type == "tools": + value = TOOLS_DICT + + if self.node_type in value: + self._built_object = loading.instantiate_class( + node_type=self.node_type, + base_type=base_type, + params=self.params, + ) + break + + if self._built_object is None: + raise ValueError(f"Node type {self.node_type} not found") + + self._built = True + + def build(self, force: bool = False) -> Any: + if not self._built or force: + self._build() + return deepcopy(self._built_object) + + def add_edge(self, edge: "Edge") -> None: + self.edges.append(edge) + + def __repr__(self) -> str: + return f"Node(id={self.id}, data={self.data})" + + def __eq__(self, __o: object) -> bool: + return self.id == __o.id if isinstance(__o, Node) else False + + def __hash__(self) -> int: + return id(self) + + +class Edge: + def __init__(self, source: "Node", target: "Node"): + self.source: "Node" = source + self.target: "Node" = target + self.validate_edge() + + def validate_edge(self) -> None: + # Validate that the outputs of the source node are valid inputs + # for the target node + self.source_types = self.source.output + self.target_reqs = self.target.required_inputs + self.target.optional_inputs + # Both lists contain strings and sometimes a string contains the value we are + # looking for e.g. comgin_out=["Chain"] and target_reqs=["LLMChain"] + # so we need to check if any of the strings in source_types is in target_reqs + self.valid = any( + output in target_req + for output in self.source_types + for target_req in self.target_reqs + ) + # Get what type of input the target node is expecting + + self.matched_type = next( + ( + output + for output in self.source_types + for target_req in self.target_reqs + if output in target_req + ), + None, + ) + + def __repr__(self) -> str: + return ( + f"Edge(source={self.source.id}, target={self.target.id}, valid={self.valid}" + f", matched_type={self.matched_type})" + ) + + diff --git a/src/backend/langflow/graph/graph.py b/src/backend/langflow/graph/graph.py index 8e3c06ef5..91b225986 100644 --- a/src/backend/langflow/graph/graph.py +++ b/src/backend/langflow/graph/graph.py @@ -1,296 +1,17 @@ -# Description: Graph class for building a graph of nodes and edges -# Insights: -# - Defer prompts building to the last moment or when they have all the tools -# - Build each inner agent first, then build the outer agent - from copy import deepcopy import types -from typing import Any, Dict, List, Optional, Union +from typing import Dict, List, Union from langflow.utils import payload -from langflow.interface.listing import ALL_TYPES_DICT, ALL_TOOLS_NAMES, TOOLS_DICT -from langflow.interface import loading +from langflow.interface.listing import ALL_TOOLS_NAMES - -class Node: - def __init__(self, data: Dict): - self.id: str = data["id"] - self._data = data - self.edges: List[Edge] = [] - self._parse_data() - self._built_object = None - self._built = False - - def _parse_data(self) -> None: - self.data = self._data["data"] - self.output = self.data["node"]["base_classes"] - template_dicts = { - key: value - for key, value in self.data["node"]["template"].items() - if isinstance(value, dict) - } - - self.required_inputs = [ - template_dicts[key]["type"] - for key, value in template_dicts.items() - if value["required"] - ] - self.optional_inputs = [ - template_dicts[key]["type"] - for key, value in template_dicts.items() - if not value["required"] - ] - - template_dict = self.data["node"]["template"] - self.node_type = ( - self.data["type"] if "Tool" not in self.output else template_dict["_type"] - ) - - def _build_params(self): - # 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 - # so we need to be able to distinguish between the two - - # The dicts with "type" == "str" are the ones that are python base classes - # and most likely have a "value" key - - # So for each key besides "_type" in the template dict, we have a dict - # with a "type" key. If the type is not "str", then we need to get the - # edge that connects to that node and get the Node with the required data - # and use that as the value for the param - # If the type is "str", then we need to get the value of the "value" key - # and use that as the value for the param - template_dict = { - key: value - for key, value in self.data["node"]["template"].items() - if isinstance(value, dict) - } - params = {} - for key, value in template_dict.items(): - if key == "_type": - continue - # If the type is not transformable to a python base class - # then we need to get the edge that connects to this node - if value["type"] not in ["str", "bool", "code"]: - # Get the edge that connects to this node - edge = next( - ( - edge - for edge in self.edges - if edge.target == self and edge.matched_type in value["type"] - ), - None, - ) - # 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 edge is None and value["required"]: - # break line - raise ValueError( - f"Required input {key} for module {self.node_type} not found" - ) - if value["list"]: - if key in params: - params[key].append(edge.source) - else: - params[key] = [edge.source] - elif value["required"] or edge is not None: - params[key] = edge.source - elif value["required"] or value.get("value"): - params[key] = value["value"] - - # Add _type to params - self.params = params - - def _build(self): - # The params dict is used to build the module - # it contains values and keys that point to nodes which - # have their own params dict - # When build is called, we iterate through the params dict - # and if the value is a node, we call build on that node - # and use the output of that build as the value for the param - # if the value is not a node, then we use the value as the param - # and continue - # Another aspect is that the node_type is the class that we need to import - # and instantiate with these built params - - # Build each node in the params dict - for key, value in self.params.items(): - # Check if Node or list of Nodes - if isinstance(value, Node): - result = value.build() - # If the key is "func", then we need to use the run method - if key == "func" and not isinstance(result, types.FunctionType): - # func can be PythonFunction(code='\ndef upper_case(text: str) -> str:\n return text.upper()\n') - # so we need to check if there is an attribute called run - if hasattr(result, "run"): - result = result.run # type: ignore - elif hasattr(result, "get_function"): - result = result.get_function() # type: ignore - self.params[key] = result - elif isinstance(value, list) and all( - isinstance(node, Node) for node in value - ): - self.params[key] = [node.build() for node in value] # type: ignore - - # Get the class from LANGCHAIN_TYPES_DICT - # and instantiate it with the params - # and return the instance - for base_type, value in ALL_TYPES_DICT.items(): - if base_type == "tools": - value = TOOLS_DICT - - if self.node_type in value: - self._built_object = loading.instantiate_class( - node_type=self.node_type, - base_type=base_type, - params=self.params, - ) - break - - if self._built_object is None: - raise ValueError(f"Node type {self.node_type} not found") - - self._built = True - - def build(self, force: bool = False) -> Any: - if not self._built or force: - self._build() - return deepcopy(self._built_object) - - def add_edge(self, edge: "Edge") -> None: - self.edges.append(edge) - - def __repr__(self) -> str: - return f"Node(id={self.id}, data={self.data})" - - def __eq__(self, __o: object) -> bool: - return self.id == __o.id if isinstance(__o, Node) else False - - def __hash__(self) -> int: - return id(self) - - -class AgentNode(Node): - def __init__(self, data: Dict): - super().__init__(data) - self.tools: List[ToolNode] = [] - self.chains: List[ChainNode] = [] - - def _set_tools_and_chains(self) -> None: - for edge in self.edges: - source_node = edge.source - if isinstance(source_node, ToolNode): - self.tools.append(source_node) - elif isinstance(source_node, ChainNode): - self.chains.append(source_node) - - def build(self, force: bool = False) -> Any: - if not self._built or force: - self._set_tools_and_chains() - # First, build the tools - for tool_node in self.tools: - tool_node.build() - - # Next, build the chains and the rest - for chain_node in self.chains: - chain_node.build(tools=self.tools) - - self._build() - return deepcopy(self._built_object) - - -class Edge: - def __init__(self, source: "Node", target: "Node"): - self.source: "Node" = source - self.target: "Node" = target - self.validate_edge() - - def validate_edge(self) -> None: - # Validate that the outputs of the source node are valid inputs - # for the target node - self.source_types = self.source.output - self.target_reqs = self.target.required_inputs + self.target.optional_inputs - # Both lists contain strings and sometimes a string contains the value we are - # looking for e.g. comgin_out=["Chain"] and target_reqs=["LLMChain"] - # so we need to check if any of the strings in source_types is in target_reqs - self.valid = any( - output in target_req - for output in self.source_types - for target_req in self.target_reqs - ) - # Get what type of input the target node is expecting - - self.matched_type = next( - ( - output - for output in self.source_types - for target_req in self.target_reqs - if output in target_req - ), - None, - ) - - def __repr__(self) -> str: - return ( - f"Edge(source={self.source.id}, target={self.target.id}, valid={self.valid}" - f", matched_type={self.matched_type})" - ) - - -class ToolNode(Node): - def __init__(self, data: Dict): - super().__init__(data) - - def build(self, force: bool = False) -> Any: - if not self._built or force: - self._build() - return deepcopy(self._built_object) - - -class PromptNode(Node): - def __init__(self, data: Dict): - super().__init__(data) - - def build( - self, - force: bool = False, - tools: Optional[Union[List[Node], List[ToolNode]]] = None, - ) -> Any: - if not self._built or force: - # Check if it is a ZeroShotPrompt and needs a tool - if self.node_type == "ZeroShotPrompt": - tools = ( - [tool_node.build() for tool_node in tools] - if tools is not None - else [] - ) - self.params["tools"] = tools - - self._build() - return deepcopy(self._built_object) - - -class ChainNode(Node): - def __init__(self, data: Dict): - super().__init__(data) - - def build( - self, - force: bool = False, - tools: Optional[Union[List[Node], List[ToolNode]]] = None, - ) -> Any: - if not self._built or force: - # Check if the chain requires a PromptNode - for key, value in self.params.items(): - if isinstance(value, PromptNode): - # Build the PromptNode, passing the tools if available - self.params[key] = value.build(tools=tools, force=force) - - self._build() - return deepcopy(self._built_object) +from langflow.graph.base import Node, Edge +from langflow.graph.nodes import ( + AgentNode, + ChainNode, + PromptNode, + ToolkitNode, + ToolNode, +) class Graph: @@ -373,6 +94,8 @@ class Graph: nodes.append(ChainNode(node)) elif "tool" in node_type.lower() or node_lc_type in ALL_TOOLS_NAMES: nodes.append(ToolNode(node)) + elif "toolkit" in node_type.lower(): + nodes.append(ToolkitNode(node)) else: nodes.append(Node(node)) return nodes diff --git a/src/backend/langflow/graph/nodes.py b/src/backend/langflow/graph/nodes.py new file mode 100644 index 000000000..963b6cb45 --- /dev/null +++ b/src/backend/langflow/graph/nodes.py @@ -0,0 +1,99 @@ +from copy import deepcopy +import types +from typing import Any, Dict, List, Optional, Union + +from langflow.interface.listing import ALL_TYPES_DICT, TOOLS_DICT +from langflow.interface import loading +from langflow.graph.base import Node + + +class AgentNode(Node): + def __init__(self, data: Dict): + super().__init__(data) + self.tools: List[ToolNode] = [] + self.chains: List[ChainNode] = [] + + def _set_tools_and_chains(self) -> None: + for edge in self.edges: + source_node = edge.source + if isinstance(source_node, ToolNode): + self.tools.append(source_node) + elif isinstance(source_node, ChainNode): + self.chains.append(source_node) + + def build(self, force: bool = False) -> Any: + if not self._built or force: + self._set_tools_and_chains() + # First, build the tools + for tool_node in self.tools: + tool_node.build() + + # Next, build the chains and the rest + for chain_node in self.chains: + chain_node.build(tools=self.tools) + + self._build() + return deepcopy(self._built_object) + + +class ToolNode(Node): + def __init__(self, data: Dict): + super().__init__(data) + + def build(self, force: bool = False) -> Any: + if not self._built or force: + self._build() + return deepcopy(self._built_object) + + +class PromptNode(Node): + def __init__(self, data: Dict): + super().__init__(data) + + def build( + self, + force: bool = False, + tools: Optional[Union[List[Node], List[ToolNode]]] = None, + ) -> Any: + if not self._built or force: + # Check if it is a ZeroShotPrompt and needs a tool + if self.node_type == "ZeroShotPrompt": + tools = ( + [tool_node.build() for tool_node in tools] + if tools is not None + else [] + ) + self.params["tools"] = tools + + self._build() + return deepcopy(self._built_object) + + +class ChainNode(Node): + def __init__(self, data: Dict): + super().__init__(data) + + def build( + self, + force: bool = False, + tools: Optional[Union[List[Node], List[ToolNode]]] = None, + ) -> Any: + if not self._built or force: + # Check if the chain requires a PromptNode + for key, value in self.params.items(): + if isinstance(value, PromptNode): + # Build the PromptNode, passing the tools if available + self.params[key] = value.build(tools=tools, force=force) + + self._build() + return deepcopy(self._built_object) + + +class ToolkitNode(Node): + def __init__(self, data: Dict): + super().__init__(data) + + def build(self, force: bool = False) -> Any: + if not self._built or force: + self._build() + return deepcopy(self._built_object) diff --git a/src/backend/langflow/interface/loading.py b/src/backend/langflow/interface/loading.py index 1c73368e9..d9ed2552a 100644 --- a/src/backend/langflow/interface/loading.py +++ b/src/backend/langflow/interface/loading.py @@ -53,7 +53,7 @@ def instantiate_class(node_type: str, base_type: str, params: Dict) -> Any: def load_flow_from_json(path: str): # This is done to avoid circular imports - from langflow.graph.graph import Graph + from langflow.graph import Graph """Load flow from json file""" with open(path, "r") as f: diff --git a/tests/test_graph.py b/tests/test_graph.py index dfb0e2323..43669d457 100644 --- a/tests/test_graph.py +++ b/tests/test_graph.py @@ -1,5 +1,5 @@ import json -from langflow.graph.graph import Edge, Graph, Node +from langflow.graph import Edge, Node, Graph import pytest from langflow.utils.payload import build_json, get_root_node from langchain.agents import AgentExecutor diff --git a/tests/test_loading.py b/tests/test_loading.py index a824ec4e5..b85d45e86 100644 --- a/tests/test_loading.py +++ b/tests/test_loading.py @@ -1,5 +1,5 @@ import json -from langflow.graph.graph import Graph +from langflow.graph import Graph import pytest from langflow import load_flow_from_json