Skip to content

Conditional Edge

graphorchestrator.edges.conditional

ConditionalEdge

Bases: Edge

Represents a conditional edge in a graph.

A ConditionalEdge directs the flow of execution to one of several sink nodes based on the result of a routing function.

Source code in graphorchestrator\edges\conditional.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
class ConditionalEdge(Edge):
    """
    Represents a conditional edge in a graph.

    A ConditionalEdge directs the flow of execution to one of several sink nodes
    based on the result of a routing function.
    """

    def __init__(
        self, source: Node, sinks: List[Node], router: Callable[[State], str]
    ) -> None:
        """
        Initializes a ConditionalEdge.

        Args:
            source (Node): The source node of the edge.
            sinks (List[Node]): A list of sink nodes.
            router (Callable[[State], str]): A routing function that takes a State object and returns
                the ID of the sink node to which the edge should route.

        Raises:
            RoutingFunctionNotDecoratedError: If the router function is not decorated with
                @routing_function.

        """

        self.source = source
        self.sinks = sinks

        if not getattr(router, "is_routing_function", False):
            raise RoutingFunctionNotDecoratedError(router)

        self.routing_function = router
        sink_ids = [s.node_id for s in sinks]

        GraphLogger.get().info(
            **wrap_constants(
                message="Conditional edge created",
                **{
                    LC.EVENT_TYPE: "edge",
                    LC.ACTION: "edge_created",
                    LC.EDGE_TYPE: "conditional",
                    LC.SOURCE_NODE: self.source.node_id,
                    LC.SINK_NODE: sink_ids,  # Using SINK_NODE for consistency; optional to split as LC.SINK_NODES
                    LC.ROUTER_FUNC: router.__name__,
                }
            )
        )

__init__(source, sinks, router)

Initializes a ConditionalEdge.

Parameters:

Name Type Description Default
source Node

The source node of the edge.

 required
sinks List[Node]

A list of sink nodes.

 required
router Callable[[State], str]

A routing function that takes a State object and returns the ID of the sink node to which the edge should route.

 required

Raises:

Type Description
RoutingFunctionNotDecoratedError

If the router function is not decorated with @routing_function.
Source code in graphorchestrator\edges\conditional.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def __init__(
    self, source: Node, sinks: List[Node], router: Callable[[State], str]
) -> None:
    """
    Initializes a ConditionalEdge.

    Args:
        source (Node): The source node of the edge.
        sinks (List[Node]): A list of sink nodes.
        router (Callable[[State], str]): A routing function that takes a State object and returns
            the ID of the sink node to which the edge should route.

    Raises:
        RoutingFunctionNotDecoratedError: If the router function is not decorated with
            @routing_function.

    """

    self.source = source
    self.sinks = sinks

    if not getattr(router, "is_routing_function", False):
        raise RoutingFunctionNotDecoratedError(router)

    self.routing_function = router
    sink_ids = [s.node_id for s in sinks]

    GraphLogger.get().info(
        **wrap_constants(
            message="Conditional edge created",
            **{
                LC.EVENT_TYPE: "edge",
                LC.ACTION: "edge_created",
                LC.EDGE_TYPE: "conditional",
                LC.SOURCE_NODE: self.source.node_id,
                LC.SINK_NODE: sink_ids,  # Using SINK_NODE for consistency; optional to split as LC.SINK_NODES
                LC.ROUTER_FUNC: router.__name__,
            }
        )
    )