jyapayne/langflow
Watch
1
0
Fork
You've already forked langflow
 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

🚀 feat(langflow): reorganize graph package to improve modularity and maintainability

Browse source 
The changes include:
- Moved the `Edge` class to a new `edge` package
- Moved the `Graph` class to a new `graph` package
- Moved the `Node` class to a new `node` package
- Moved the `VectorStoreNode` class to the `node/types.py` module
- Moved the `Edge`, `Graph`, and `Node` classes to their respective `base.py` modules
- Added an `__init__.py` file to each package to allow for importing of classes
- Added a `constants.py` module to the `graph` package to store constants used in the `Graph` class
- Refactored the `Graph` class to use the new `Node` and `Edge` classes
- Refactored the `Graph` class to use a dictionary to map node types to their respective classes
- Refactored the `Graph` class to remove invalid nodes from the graph
- Refactored the `Graph` class to handle the LLM node within the graph
- Refactored the `Graph` class to build the nodes before building the edges
- Refactored the `Graph` class to use the `get_node` method to find nodes by id
- Refactored the `Graph` class to use the `get_node_neighbors` method to find the neighbors of a node
- Refactored the `Graph` class to use the `get_children_by_node_type` method to find the children of a node based on the node type

These changes improve the modularity and maintainability of the `langflow` package by separating the classes into their respective packages and modules. The changes also make it easier to add new node types to the `Graph` class by using a dictionary to map node types to their respective classes.

🚀 feat(node): add Node class to represent a node in the graph
🚀 feat(constants.py): add DIRECT_TYPES constant to represent direct types in a node's template
The Node class represents a node in the graph and is responsible for parsing the data and building the module. The DIRECT_TYPES constant is a list of direct types in a node's template.

🚧 chore(types.py): add import statements for typing and Node classes
This commit adds import statements for the typing module and the Node class to the types.py file. This is necessary for the code to run properly as it uses these classes and modules.

🚧 chore(loading.py): remove unnecessary import statement
This commit removes an unnecessary import statement from the loading.py file. The import statement was causing a circular import error and was not needed for the code to run properly.

🚧 chore(run.py): update import statement for Graph class
This commit updates the import statement for the Graph class in the run.py file. The import statement was outdated and was causing an import error.

🚧 chore(conftest.py): update import statement for Graph class
This commit updates the import statement for the Graph class in the conftest.py file. The import statement was outdated and was causing an import error.

🚧 chore(test_graph.py): update import statements for Node and Edge classes
This commit updates the import statements for the Node and Edge classes in the test_graph.py file. The import statements were outdated and were causing import errors.
This commit is contained in:
Gabriel Luiz Freitas Almeida 2023-06-01 15:07:56 -03:00
commit 5b5eea9895
15 changed files with 187 additions and 117 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/backend/langflow/api/validate.py
View file

@ -9,7 +9,7 @@ from langflow.api.base import (
PromptValidationResponse,
validate_prompt,
)
from langflow.graph.nodes import VectorStoreNode
from langflow.graph.node.types import VectorStoreNode
from langflow.interface.run import build_graph
from langflow.utils.logger import logger
from langflow.utils.validate import validate_code

37
src/backend/langflow/graph/__init__.py
View file

@ -1,4 +1,35 @@
from langflow.graph.base import Edge, Node
from langflow.graph.graph import Graph
from langflow.graph.edge.base import Edge
from langflow.graph.graph.base import Graph
from langflow.graph.node.base import Node
from langflow.graph.node.types import (
AgentNode,
ChainNode,
DocumentLoaderNode,
EmbeddingNode,
LLMNode,
MemoryNode,
PromptNode,
TextSplitterNode,
ToolNode,
ToolkitNode,
VectorStoreNode,
WrapperNode,
)
__all__ = ["Graph", "Node", "Edge"]
__all__ = [
"Graph",
"Node",
"Edge",
"AgentNode",
"ChainNode",
"DocumentLoaderNode",
"EmbeddingNode",
"LLMNode",
"MemoryNode",
"PromptNode",
"TextSplitterNode",
"ToolNode",
"ToolkitNode",
"VectorStoreNode",
"WrapperNode",
]

0
src/backend/langflow/graph/edge/__init__.py Normal file
View file

52
src/backend/langflow/graph/edge/base.py Normal file
View file

@ -0,0 +1,52 @@
from langflow.utils.logger import logger
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from langflow.graph.node.base import Node
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,
)
no_matched_type = self.matched_type is None
if no_matched_type:
logger.debug(self.source_types)
logger.debug(self.target_reqs)
if no_matched_type:
raise ValueError(
f"Edge between {self.source.node_type} and {self.target.node_type} "
f"has no matched type"
)
def __repr__(self) -> str:
return (
f"Edge(source={self.source.id}, target={self.target.id}, valid={self.valid}"
f", matched_type={self.matched_type})"
)

0
src/backend/langflow/graph/graph/__init__.py Normal file
View file

149
src/backend/langflow/graph/graph/base.py Normal file
View file

@ -0,0 +1,149 @@
from typing import Dict, List, Type, Union
from langflow.graph.edge.base import Edge
from langflow.graph.graph.constants import NODE_TYPE_MAP
from langflow.graph.node.base import Node
from langflow.graph.node.types import (
FileToolNode,
LLMNode,
ToolkitNode,
)
from langflow.interface.tools.constants import FILE_TOOLS
from langflow.utils import payload
class Graph:
"""A class representing a graph of nodes and edges."""
def __init__(
self,
nodes: List[Dict[str, Union[str, Dict[str, Union[str, List[str]]]]]],
edges: List[Dict[str, str]],
) -> None:
self._nodes = nodes
self._edges = edges
self._build_graph()
def _build_graph(self) -> None:
"""Builds the graph from the nodes and edges."""
self.nodes = self._build_nodes()
self.edges = self._build_edges()
for edge in self.edges:
edge.source.add_edge(edge)
edge.target.add_edge(edge)
# This is a hack to make sure that the LLM node is sent to
# the toolkit node
self._build_node_params()
# remove invalid nodes
self._remove_invalid_nodes()
def _build_node_params(self) -> None:
"""Identifies and handles the LLM node within the graph."""
llm_node = None
for node in self.nodes:
node._build_params()
if isinstance(node, LLMNode):
llm_node = node
if llm_node:
for node in self.nodes:
if isinstance(node, ToolkitNode):
node.params["llm"] = llm_node
def _remove_invalid_nodes(self) -> None:
"""Removes invalid nodes from the graph."""
self.nodes = [
node
for node in self.nodes
if self._validate_node(node)
or (len(self.nodes) == 1 and len(self.edges) == 0)
]
def _validate_node(self, node: Node) -> bool:
"""Validates a node."""
# All nodes that do not have edges are invalid
return len(node.edges) > 0
def get_node(self, node_id: str) -> Union[None, Node]:
"""Returns a node by id."""
return next((node for node in self.nodes if node.id == node_id), None)
def get_nodes_with_target(self, node: Node) -> List[Node]:
"""Returns the nodes connected to a node."""
connected_nodes: List[Node] = [
edge.source for edge in self.edges if edge.target == node
]
return connected_nodes
def build(self) -> List[Node]:
"""Builds the graph."""
# Get root node
root_node = payload.get_root_node(self)
if root_node is None:
raise ValueError("No root node found")
return root_node.build()
def get_node_neighbors(self, node: Node) -> Dict[Node, int]:
"""Returns the neighbors of a node."""
neighbors: Dict[Node, int] = {}
for edge in self.edges:
if edge.source == node:
neighbor = edge.target
if neighbor not in neighbors:
neighbors[neighbor] = 0
neighbors[neighbor] += 1
elif edge.target == node:
neighbor = edge.source
if neighbor not in neighbors:
neighbors[neighbor] = 0
neighbors[neighbor] += 1
return neighbors
def _build_edges(self) -> List[Edge]:
"""Builds the edges of the graph."""
# Edge takes two nodes as arguments, so we need to build the nodes first
# and then build the edges
# if we can't find a node, we raise an error
edges: List[Edge] = []
for edge in self._edges:
source = self.get_node(edge["source"])
target = self.get_node(edge["target"])
if source is None:
raise ValueError(f"Source node {edge['source']} not found")
if target is None:
raise ValueError(f"Target node {edge['target']} not found")
edges.append(Edge(source, target))
return edges
def _get_node_class(self, node_type: str, node_lc_type: str) -> Type[Node]:
"""Returns the node class based on the node type."""
if node_type in FILE_TOOLS:
return FileToolNode
if node_type in NODE_TYPE_MAP:
return NODE_TYPE_MAP[node_type]
return NODE_TYPE_MAP[node_lc_type] if node_lc_type in NODE_TYPE_MAP else Node
def _build_nodes(self) -> List[Node]:
"""Builds the nodes of the graph."""
nodes: List[Node] = []
for node in self._nodes:
node_data = node["data"]
node_type: str = node_data["type"] # type: ignore
node_lc_type: str = node_data["node"]["template"]["_type"] # type: ignore
NodeClass = self._get_node_class(node_type, node_lc_type)
nodes.append(NodeClass(node))
return nodes
def get_children_by_node_type(self, node: Node, node_type: str) -> List[Node]:
"""Returns the children of a node based on the node type."""
children = []
node_types = [node.data["type"]]
if "node" in node.data:
node_types += node.data["node"]["base_classes"]
if node_type in node_types:
children.append(node)
return children

49
src/backend/langflow/graph/graph/constants.py Normal file
View file

@ -0,0 +1,49 @@
from langflow.graph.node.base import Node
from langflow.graph.node.types import (
AgentNode,
ChainNode,
DocumentLoaderNode,
EmbeddingNode,
LLMNode,
MemoryNode,
PromptNode,
TextSplitterNode,
ToolNode,
ToolkitNode,
VectorStoreNode,
WrapperNode,
)
from langflow.interface.agents.base import agent_creator
from langflow.interface.chains.base import chain_creator
from langflow.interface.document_loaders.base import documentloader_creator
from langflow.interface.embeddings.base import embedding_creator
from langflow.interface.llms.base import llm_creator
from langflow.interface.memories.base import memory_creator
from langflow.interface.prompts.base import prompt_creator
from langflow.interface.text_splitters.base import textsplitter_creator
from langflow.interface.toolkits.base import toolkits_creator
from langflow.interface.tools.base import tool_creator
from langflow.interface.vector_store.base import vectorstore_creator
from langflow.interface.wrappers.base import wrapper_creator
from typing import Dict, Type
DIRECT_TYPES = ["str", "bool", "code", "int", "float", "Any", "prompt"]
NODE_TYPE_MAP: Dict[str, Type[Node]] = {
**{t: PromptNode for t in prompt_creator.to_list()},
**{t: AgentNode for t in agent_creator.to_list()},
**{t: ChainNode for t in chain_creator.to_list()},
**{t: ToolNode for t in tool_creator.to_list()},
**{t: ToolkitNode for t in toolkits_creator.to_list()},
**{t: WrapperNode for t in wrapper_creator.to_list()},
**{t: LLMNode for t in llm_creator.to_list()},
**{t: MemoryNode for t in memory_creator.to_list()},
**{t: EmbeddingNode for t in embedding_creator.to_list()},
**{t: VectorStoreNode for t in vectorstore_creator.to_list()},
**{t: DocumentLoaderNode for t in documentloader_creator.to_list()},
**{t: TextSplitterNode for t in textsplitter_creator.to_list()},
}

0
src/backend/langflow/graph/node/__init__.py Normal file
View file

229
src/backend/langflow/graph/node/base.py Normal file
View file

@ -0,0 +1,229 @@
from langflow.cache import base as cache_utils
from langflow.graph.node.constants import DIRECT_TYPES
from langflow.interface 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
if TYPE_CHECKING:
from langflow.graph.edge.base import Edge
class Node:
def __init__(self, data: Dict, base_type: Optional[str] = None) -> None:
self.id: str = data["id"]
self._data = data
self.edges: List["Edge"] = []
self.base_type: Optional[str] = base_type
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"]
)
if self.base_type is None:
for base_type, value in ALL_TYPES_DICT.items():
if self.node_type in value:
self.base_type = base_type
break
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.get("type") == "file":
# Load the type in value.get('suffixes') using
# what is inside value.get('content')
# value.get('value') is the file name
file_name = value.get("value")
content = value.get("content")
type_to_load = value.get("suffixes")
file_path = cache_utils.save_binary_file(
content=content, file_name=file_name, accepted_types=type_to_load
)
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.node_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.node_type} is None. ")
if value.get("type") == "int":
with contextlib.suppress(TypeError, ValueError):
new_value = int(new_value) # type: ignore
params[key] = new_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
logger.debug(f"Building {self.node_type}")
# Build each node in the params dict
for key, value in self.params.copy().items():
# Check if Node or list of Nodes and not self
# to avoid recursion
if isinstance(value, Node):
if value == self:
del self.params[key]
continue
result = value.build()
# If the key is "func", then we need to use the run method
if key == "func":
if 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
elif inspect.iscoroutinefunction(result):
self.params["coroutine"] = result
else:
# turn result which is a function into a coroutine
# so that it can be awaited
self.params["coroutine"] = sync_to_async(result)
self.params[key] = result
elif isinstance(value, list) and all(
isinstance(node, Node) for node in value
):
self.params[key] = []
for node in value:
built = node.build()
if isinstance(built, list):
self.params[key].extend(built)
else:
self.params[key].append(built)
# Get the class from LANGCHAIN_TYPES_DICT
# and instantiate it with the params
# and return the instance
try:
self._built_object = loading.instantiate_class(
node_type=self.node_type,
base_type=self.base_type,
params=self.params,
)
except Exception as exc:
raise ValueError(
f"Error building node {self.node_type}: {str(exc)}"
) from exc
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 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)
def _built_object_repr(self):
return repr(self._built_object)

1
src/backend/langflow/graph/node/constants.py Normal file
View file

@ -0,0 +1 @@
DIRECT_TYPES = ["str", "bool", "code", "int", "float", "Any", "prompt"]

193
src/backend/langflow/graph/node/types.py Normal file
View file

@ -0,0 +1,193 @@
from typing import Any, Dict, List, Optional, Union
from langflow.graph.node.base import Node
from langflow.graph.utils import extract_input_variables_from_prompt
class AgentNode(Node):
def __init__(self, data: Dict):
super().__init__(data, base_type="agents")
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()
#! Cannot deepcopy VectorStore, VectorStoreRouter, or SQL agents
if self.node_type in ["VectorStoreAgent", "VectorStoreRouterAgent", "SQLAgent"]:
return self._built_object
return self._built_object
class ToolNode(Node):
def __init__(self, data: Dict):
super().__init__(data, base_type="tools")
class PromptNode(Node):
def __init__(self, data: Dict):
super().__init__(data, base_type="prompts")
def build(
self,
force: bool = False,
tools: Optional[Union[List[Node], List[ToolNode]]] = None,
) -> Any:
if not self._built or force:
if (
"input_variables" not in self.params
or self.params["input_variables"] is None
):
self.params["input_variables"] = []
# Check if it is a ZeroShotPrompt and needs a tool
if "ShotPrompt" in self.node_type:
tools = (
[tool_node.build() for tool_node in tools]
if tools is not None
else []
)
self.params["tools"] = tools
prompt_params = [
key
for key, value in self.params.items()
if isinstance(value, str) and key != "format_instructions"
]
else:
prompt_params = ["template"]
for param in prompt_params:
prompt_text = self.params[param]
variables = extract_input_variables_from_prompt(prompt_text)
self.params["input_variables"].extend(variables)
self.params["input_variables"] = list(set(self.params["input_variables"]))
self._build()
return self._built_object
class ChainNode(Node):
def __init__(self, data: Dict):
super().__init__(data, base_type="chains")
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()
#! Cannot deepcopy SQLDatabaseChain
if self.node_type in ["SQLDatabaseChain"]:
return self._built_object
return self._built_object
class LLMNode(Node):
built_node_type = None
class_built_object = None
def __init__(self, data: Dict):
super().__init__(data, base_type="llms")
def build(self, force: bool = False) -> Any:
# LLM is different because some models might take up too much memory
# or time to load. So we only load them when we need them.ß
if self.node_type == self.built_node_type:
return self.class_built_object
if not self._built or force:
self._build()
self.built_node_type = self.node_type
self.class_built_object = self._built_object
# Avoid deepcopying the LLM
# that are loaded from a file
return self._built_object
class ToolkitNode(Node):
def __init__(self, data: Dict):
super().__init__(data, base_type="toolkits")
class FileToolNode(ToolNode):
def __init__(self, data: Dict):
super().__init__(data)
class WrapperNode(Node):
def __init__(self, data: Dict):
super().__init__(data, base_type="wrappers")
def build(self, force: bool = False) -> Any:
if not self._built or force:
if "headers" in self.params:
self.params["headers"] = eval(self.params["headers"])
self._build()
return self._built_object
class DocumentLoaderNode(Node):
def __init__(self, data: Dict):
super().__init__(data, base_type="documentloaders")
def _built_object_repr(self):
# This built_object is a list of documents. Maybe we should
# show how many documents are in the list?
if self._built_object:
return f"""{self.node_type}({len(self._built_object)} documents)
Documents: {self._built_object[:3]}..."""
return f"{self.node_type}()"
class EmbeddingNode(Node):
def __init__(self, data: Dict):
super().__init__(data, base_type="embeddings")
class VectorStoreNode(Node):
def __init__(self, data: Dict):
super().__init__(data, base_type="vectorstores")
def _built_object_repr(self):
return "Vector stores can take time to build. It will build on the first query."
class MemoryNode(Node):
def __init__(self, data: Dict):
super().__init__(data, base_type="memory")
class TextSplitterNode(Node):
def __init__(self, data: Dict):
super().__init__(data, base_type="textsplitters")
def _built_object_repr(self):
# This built_object is a list of documents. Maybe we should
# show how many documents are in the list?
if self._built_object:
return f"""{self.node_type}({len(self._built_object)} documents)\nDocuments: {self._built_object[:3]}..."""
return f"{self.node_type}()"

2
src/backend/langflow/interface/loading.py
View file

@ -12,6 +12,7 @@ from langchain.agents.load_tools import (
_LLM_TOOLS,
)
from langchain.agents.loading import load_agent_from_config
from langflow.graph import Graph
from langchain.agents.tools import Tool
from langchain.base_language import BaseLanguageModel
from langchain.callbacks.base import BaseCallbackManager
@ -164,7 +165,6 @@ def instantiate_utility(node_type, class_object, params):
def load_flow_from_json(path: str, build=True):
"""Load flow from json file"""
# This is done to avoid circular imports
from langflow.graph import Graph
with open(path, "r", encoding="utf-8") as f:
flow_graph = json.load(f)

2
src/backend/langflow/interface/run.py
View file

@ -6,7 +6,7 @@ from langchain.schema import AgentAction
from langflow.api.callback import AsyncStreamingLLMCallbackHandler, StreamingLLMCallbackHandler # type: ignore
from langflow.cache.base import compute_dict_hash, load_cache, memoize_dict
from langflow.graph.graph import Graph
from langflow.graph import Graph
from langflow.utils.logger import logger

2
tests/conftest.py
View file

@ -2,6 +2,7 @@ import json
from pathlib import Path
from typing import AsyncGenerator
from langflow.graph.graph.base import Graph
import pytest
from fastapi.testclient import TestClient
from httpx import AsyncClient
@ -46,7 +47,6 @@ def client():
def get_graph(_type="basic"):
"""Get a graph from a json file"""
from langflow.graph.graph import Graph
if _type == "basic":
path = pytest.BASIC_EXAMPLE_PATH

6
tests/test_graph.py
View file

@ -1,10 +1,12 @@
from typing import Type, Union
from langflow.graph.edge.base import Edge
from langflow.graph.node.base import Node
import pytest
from langchain.chains.base import Chain
from langchain.llms.fake import FakeListLLM
from langflow.graph import Edge, Graph, Node
from langflow.graph.nodes import (
from langflow.graph import Graph
from langflow.graph.node.types import (
AgentNode,
ChainNode,
FileToolNode,