Lesson 3.1: The Router (Conditional Edges)

by

A standard edge (add_edge) is like a train track; the train must go there. A conditional edge (add_conditional_edges) is like a junction. The graph pauses, looks at the State, and decides which track to take.

To implement this, we need a special Routing Function. This function doesn’t update the state; it simply returns the name of the next node.

The Scenario

We are building a Customer Support bot.

  • If the user asks about “Refunds”, send them to the Refund_Agent.
  • If they ask about “Tech”, send them to the Tech_Support.

The Code

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

# --- 1. State ---
class SupportState(TypedDict):
    query: str

# --- 2. Nodes ---
def refund_agent(state):
    print("--- REFUND AGENT: Processing refund request ---")
    return {}

def tech_support(state):
    print("--- TECH AGENT: Have you tried turning it off and on? ---")
    return {}

# --- 3. The Router Logic ---
# This function analyzes the state and returns the NAME of the next node.
def route_query(state) -> Literal["refund_node", "tech_node"]:
    user_query = state["query"].lower()
    if "money" in user_query:
        return "refund_node"
    else:
        return "tech_node"

# --- 4. Build Graph ---
builder = StateGraph(SupportState)

builder.add_node("refund_node", refund_agent)
builder.add_node("tech_node", tech_support)

# Start immediately with the decision
builder.set_entry_point("refund_node") # Just for initialization, see below

# WAIT! We usually need a "Start" node to trigger the router.
# Let's add a dummy start node for clarity.
def classify_node(state):
    print(f"--- CLASSIFYING: {state['query']} ---")
    return {}

builder.add_node("classifier", classify_node)
builder.set_entry_point("classifier")

# ADD CONDITIONAL EDGE
# Syntax: add_conditional_edges(source_node, decision_function, path_map)
builder.add_conditional_edges(
    "classifier",      # After this node runs...
    route_query,       # Run this logic...
    {                  # Map the logic's output to actual Nodes
        "refund_node": "refund_node",
        "tech_node": "tech_node"
    }
)

builder.add_edge("refund_node", END)
builder.add_edge("tech_node", END)

app = builder.compile()

# --- 5. Run ---
print("Run 1:")
app.invoke({"query": "I want my money back"})

print("\nRun 2:")
app.invoke({"query": "My screen is broken"})

Output

Run 1:
--- CLASSIFYING: I want my money back ---
--- REFUND AGENT: Processing refund request ---

Run 2:
--- CLASSIFYING: My screen is broken ---
--- TECH AGENT: Have you tried turning it off and on? ---

Leave a Comment