Graph IV: Conditional Graph

In this section, we will learn how to direct the flow of execution dynamically based on user input or state properties using conditional edges.

Objectives

Implement conditional logic to route execution.
Use START and END nodes explicitly in graph routing.
Design multiple nodes for different operations (e.g., addition and subtraction).
Create a router function to handle decision-making and control graph flow.

Goal

Build a graph that dynamically routes execution based on the operation symbol (+ or -).

Sample Input

{"number1": 10, "operation": "-", "number2": 5}

Sample Output

{"number1": 10, "operation": "-", "number2": 5, "finalNumber": 5}

Plan

Define the AgentState schema containing number1, operation, number2, and finalNumber.
Create the node functions (adder, subtractor) and the routing function (decide_next_node).
Build the graph and connect the conditional edges.
Compile and invoke the graph.

Code Implementation

1. Define the State Schema

We define the numbers, the operation indicator, and the final computation field:

from typing import TypedDict

class AgentState(TypedDict):
    number1: int
    operation: str
    number2: int
    finalNumber: int

2. Define the Nodes and Router Function

We write:

An adder node
A subtractor node
A routing function decide_next_node that returns a string mapping to the target node.

def adder(state: AgentState) -> AgentState:
    state["finalNumber"] = state["number1"] + state["number2"]
    return state

def subtractor(state: AgentState) -> AgentState:
    state["finalNumber"] = state["number1"] - state["number2"]
    return state

def decide_next_node(state: AgentState) -> str:
    if state["operation"] == "+":
        return "addition_path"
    elif state["operation"] == "-":
        return "subtraction_path"

3. Create Graph with Conditional Edges

We add the nodes and define the routing connection using add_conditional_edges():

from langgraph.graph import StateGraph, START, END

graph = StateGraph(AgentState)

# Add nodes
graph.add_node("adder_node", adder)
graph.add_node("subtractor_node", subtractor)
graph.add_node("router", lambda state: state) # Passthrough router node

# Connect flow
graph.add_edge(START, "router")

# Define conditional routing
graph.add_conditional_edges(
    "router",
    decide_next_node,
    {
        "addition_path": "adder_node",
        "subtraction_path": "subtractor_node"
    }
)

# Connect calculation nodes to END
graph.add_edge("adder_node", END)
graph.add_edge("subtractor_node", END)

app = graph.compile()

4. Invoke the Graph

result = app.invoke({"number1": 10, "operation": "-", "number2": 5})
print(result["finalNumber"])
# Output: 5

Exercise: Double Routing Calculator 🛹

Goal

Build a double routing calculator that performs two consecutive calculations dynamically using conditional routing gates.

Sample Input

{
    "number1": 10,
    "operation": "-",
    "number2": 5,
    "number3": 7,
    "number4": 2,
    "operation2": "+",
    "finalNumber": 0,
    "finalNumber2": 0
}

Sample Output

"Result 1: 5, Result 2: 9"

Plan

Create AgentState schema to track numbers, operations, and results.
Write mathematical operation nodes and two router functions to evaluate operation symbols.
Connect both stages with conditional edges using a StateGraph, compile and invoke.

Solution

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# 1. State Schema
class AgentState(TypedDict):
    number1: int
    operation: str
    number2: int
    number3: int
    number4: int
    operation2: str
    finalNumber: int
    finalNumber2: int

# 2. Node Functions
def adder1(state: AgentState) -> AgentState:
    state["finalNumber"] = state["number1"] + state["number2"]
    return state

def subtractor1(state: AgentState) -> AgentState:
    state["finalNumber"] = state["number1"] - state["number2"]
    return state

def adder2(state: AgentState) -> AgentState:
    state["finalNumber2"] = state["number3"] + state["number4"]
    return state

def subtractor2(state: AgentState) -> AgentState:
    state["finalNumber2"] = state["number3"] - state["number4"]
    return state

# Routing Functions
def route1(state: AgentState) -> str:
    return "add1" if state["operation"] == "+" else "sub1"

def route2(state: AgentState) -> str:
    return "add2" if state["operation2"] == "+" else "sub2"

# 3. Create Graph
graph = StateGraph(AgentState)

graph.add_node("adder1", adder1)
graph.add_node("subtractor1", subtractor1)
graph.add_node("adder2", adder2)
graph.add_node("subtractor2", subtractor2)
graph.add_node("router1", lambda state: state)
graph.add_node("router2", lambda state: state)

# Connections
graph.add_edge(START, "router1")
graph.add_conditional_edges(
    "router1",
    route1,
    {
        "add1": "adder1",
        "sub1": "subtractor1"
    }
)

graph.add_edge("adder1", "router2")
graph.add_edge("subtractor1", "router2")

graph.add_conditional_edges(
    "router2",
    route2,
    {
        "add2": "adder2",
        "sub2": "subtractor2"
    }
)

graph.add_edge("adder2", END)
graph.add_edge("subtractor2", END)

# 4. Compile and Run
app = graph.compile()
output = app.invoke({
    "number1": 10,
    "operation": "-",
    "number2": 5,
    "number3": 7,
    "number4": 2,
    "operation2": "+",
    "finalNumber": 0,
    "finalNumber2": 0
})
print(f"Result 1: {output['finalNumber']}, Result 2: {output['finalNumber2']}")

Practice & Exercises

To reinforce what you’ve learned in this section (conditional routing and dynamic paths), practice with the interactive notebook:

Practice & Exercises

Practice implementing routing functions, configuring conditional edges, and building a multi-path routing agent.💻 VS Code | 🚀 Colab | 📥 Download

LangGraph Basics

LangGraph Agents

Graph IV: Conditional Graph

Objectives