Graph

graphorchestrator.graph.graph

Graph

Represents a directed graph composed of nodes and edges.

Attributes:

Name	Type	Description
nodes	Dict[str, Node]	A dictionary mapping node IDs to Node objects.
concrete_edges	List[ConcreteEdge]	A list of concrete edges in the graph.
conditional_edges	List[ConditionalEdge]	A list of conditional edges in the graph.
start_node	Node	The starting node of the graph.
end_node	Node	The ending node of the graph.

Source code in graphorchestrator\graph\graph.py

class Graph:
    """
    Represents a directed graph composed of nodes and edges.

    Attributes:
        nodes (Dict[str, Node]): A dictionary mapping node IDs to Node objects.
        concrete_edges (List[ConcreteEdge]): A list of concrete edges in the graph.
        conditional_edges (List[ConditionalEdge]): A list of conditional edges in the graph.
        start_node (Node): The starting node of the graph.
        end_node (Node): The ending node of the graph.
    """

    def __init__(
        self, start_node: Node, end_node: Node, name: Optional[str] = "graph"
    ) -> None:
        """
        Initializes a Graph object.

        Args:
            start_node (Node): The starting node of the graph.
            end_node (Node): The ending node of the graph.
            name (Optional[str]): An optional name for the graph (default: "graph").

        Raises:
            TypeError: If start_node or end_node is not of type Node.

        Returns:
            None
        """
        self.nodes: Dict[str, Node] = {}
        if not isinstance(start_node, Node) or not isinstance(end_node, Node):
            raise TypeError("start_node and end_node must be of type Node")

        self.concrete_edges: List[ConcreteEdge] = []
        self.conditional_edges: List[ConditionalEdge] = []
        self.start_node = start_node
        self.end_node = end_node

        GraphLogger.get().info(
            **wrap_constants(
                message="Graph initialized",
                **{
                    LC.EVENT_TYPE: "graph",
                    LC.ACTION: "graph_initialized",
                    LC.CUSTOM: {
                        "start_node_id": start_node.node_id,
                        "end_node_id": end_node.node_id,
                    },
                }
            )
        )

__init__(start_node, end_node, name='graph')

Initializes a Graph object.

Parameters:

Name	Type	Description	Default
start_node	Node	The starting node of the graph.	required
end_node	Node	The ending node of the graph.	required
name	Optional[str]	An optional name for the graph (default: "graph").	'graph'

Raises:

Type	Description
TypeError	If start_node or end_node is not of type Node.

Returns:

Type	Description
None	None

Source code in graphorchestrator\graph\graph.py

def __init__(
    self, start_node: Node, end_node: Node, name: Optional[str] = "graph"
) -> None:
    """
    Initializes a Graph object.

    Args:
        start_node (Node): The starting node of the graph.
        end_node (Node): The ending node of the graph.
        name (Optional[str]): An optional name for the graph (default: "graph").

    Raises:
        TypeError: If start_node or end_node is not of type Node.

    Returns:
        None
    """
    self.nodes: Dict[str, Node] = {}
    if not isinstance(start_node, Node) or not isinstance(end_node, Node):
        raise TypeError("start_node and end_node must be of type Node")

    self.concrete_edges: List[ConcreteEdge] = []
    self.conditional_edges: List[ConditionalEdge] = []
    self.start_node = start_node
    self.end_node = end_node

    GraphLogger.get().info(
        **wrap_constants(
            message="Graph initialized",
            **{
                LC.EVENT_TYPE: "graph",
                LC.ACTION: "graph_initialized",
                LC.CUSTOM: {
                    "start_node_id": start_node.node_id,
                    "end_node_id": end_node.node_id,
                },
            }
        )
    )