Conditional edges

Conditional edges allow you to select which node will be executed next, after the current one.

from typing import Literal, TypedDict

from langgraph.types import Command
from langgraph.graph import StateGraph, START, END

Definition

The are two main ways to define the conditional edges:

• You can annotate node outputs as a langgraph.types.Command with a literal that specifies the possible next nodes as a generic argument. Such a node must return a langgraph.types.Command containing a state update and next node to be executed.

• The add_conditional_edges method of the builder, takes the method that must return the names of the possible subsequent nodes.

---

The following cell combines both definitions: node_a is referenced by the add_conditional_edges and itself selects either END or node_b.

class State(TypedDict):
    goto: Literal["a", "__end__"]

def conditional_edge(state: State) -> Literal["a", "__end__"]:
    return state["goto"]

def node_a(state: State) -> Command[Literal["b", "__end__"]]:
    print("a node is invoked")
    return state

def node_b(state: State) -> State:
    return state

(
    StateGraph(State)
    .add_node("a", node_a)
    .add_node("b", node_b)

    .add_conditional_edges(START, conditional_edge)
    .add_edge("a", END)
    .compile()
)

Goto several

To execute several nodes after a given conditional node, return a list of values containing desired nodes.

---

The following cell defines a graph with a conditional node that can invoke three other nodes, but randomly invokes two of them.

from random import sample

class State(TypedDict):
    pass

def node_a(state: State) -> State:
    print("node a")
    return state
def node_b(state: State) -> State:
    print("node b")
    return state
def node_c(state: State) -> State:
    print("node c")
    return state

def conditional(state: State) -> Command[Literal["a", "b", "c"]]:
    return Command(goto=sample(["a", "b", "c"], k=2))

graph = (
    StateGraph(State)
    .add_node("conditional", conditional)

    .add_node("a", node_a)
    .add_node("b", node_b)
    .add_node("c", node_c)

    .add_edge("__start__", "conditional")
    .add_edge("a", "__end__")
    .add_edge("b", "__end__")
    .add_edge("c", "__end__")

    .compile()
)

graph

The following cell invokes the graph for a several times.

for i in range(3):
    print("=" * 80)
    graph.invoke({})

================================================================================
node a
node c
================================================================================
node a
node c
================================================================================
node a
node b

Send to node

The conditional edge that returns langgraph.types.Send sends the specified data to the specified node.

Check official description in Send section of the official documentation.

---

The following cell defines route function that sends {"out": "to target node"} to the target_node.

from langgraph.types import Send

class OverallState(TypedDict):
    ans: str

def route(state: OverallState) -> Send:
    return Send("target_node", {"out": "to target node"})

def target_node(state: dict) -> OverallState:
    print("State for the target node", state)
    return OverallState(ans="This is from target node")

graph = (
    StateGraph(OverallState)
    .add_node("target_node", target_node)

    .add_conditional_edges(START, route, ["target_node"])
    .add_edge("target_node", END)

    .compile()
)

graph

The invocation of the graph shows

graph.invoke(OverallState(ans="input"))

State for the target node {'out': 'to target node'}

{'ans': 'This is from target node'}