Conditional Edges: Letting State Drive the Path

Recall Module 6’s preview — add_conditional_edges briefly appeared, routing START itself. This module gives conditional routing the full, proper treatment it deserves, because it’s genuinely the mechanism that turns a fixed sequence into a real, responsive workflow. Every scenario below is a real, common shape you’ll actually build.

The mechanism, stated precisely

builder.add_conditional_edges(
    "source_node",      # which node's output triggers this decision
    routing_function,    # a function taking state, returning a node name (or list of names)
    ["node_a", "node_b"] # every node name the routing function is allowed to return
)

The routing function reads state — nothing more exotic than that — and returns the name of whichever node should run next. Let’s build five genuinely real versions of this.

Scenario 1: model selection based on complexity

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model

class State(TypedDict):
    query: str
    answer: str

fast_model = init_chat_model("openai:gpt-4o-mini")
strong_model = init_chat_model("openai:gpt-4o")

def answer_fast(state: State) -> dict:
    return {"answer": fast_model.invoke(state["query"]).content}

def answer_strong(state: State) -> dict:
    return {"answer": strong_model.invoke(state["query"]).content}

def route_by_complexity(state: State) -> str:
    return "answer_strong" if len(state["query"]) > 150 else "answer_fast"

builder = StateGraph(State)
builder.add_node("answer_fast", answer_fast)
builder.add_node("answer_strong", answer_strong)
builder.add_conditional_edges(START, route_by_complexity, ["answer_fast", "answer_strong"])
builder.add_edge("answer_fast", END)
builder.add_edge("answer_strong", END)

graph = builder.compile()
print(graph.invoke({"query": "What's 2+2?", "answer": ""}))

A short, simple question genuinely never touches the more expensive model at all — a real, deliberate cost decision, made explicitly in code, not left to chance.

Scenario 2: support routing by category

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

class State(TypedDict):
    query: str
    category: Literal["billing", "technical", "general"]
    response: str

def classify(state: State) -> dict:
    q = state["query"].lower()
    if "charge" in q or "refund" in q:
        return {"category": "billing"}
    if "error" in q or "not working" in q:
        return {"category": "technical"}
    return {"category": "general"}

def billing_response(state: State) -> dict:
    return {"response": "Routed to billing support."}

def technical_response(state: State) -> dict:
    return {"response": "Routed to technical support."}

def general_response(state: State) -> dict:
    return {"response": "Routed to general support."}

def route_by_category(state: State) -> str:
    return f"{state['category']}_response"

builder = StateGraph(State)
builder.add_node("classify", classify)
builder.add_node("billing_response", billing_response)
builder.add_node("technical_response", technical_response)
builder.add_node("general_response", general_response)

builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route_by_category, ["billing_response", "technical_response", "general_response"])
builder.add_edge("billing_response", END)
builder.add_edge("technical_response", END)
builder.add_edge("general_response", END)

graph = builder.compile()
print(graph.invoke({"query": "I was charged twice", "category": "general", "response": ""}))

flowchart TD
    START --> classify
    classify -->|billing| billing_response --> END1[END]
    classify -->|technical| technical_response --> END2[END]
    classify -->|general| general_response --> END3[END]

Notice route_by_category reads state["category"] — a field a previous node already computed — rather than recomputing the classification itself. This is a genuinely important, real pattern: classification and routing are two separate, distinct responsibilities, even though they’re closely related.

Scenario 3: RAG versus a direct answer

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

class State(TypedDict):
    query: str
    needs_retrieval: bool
    answer: str

def check_if_retrieval_needed(state: State) -> dict:
    keywords = ["policy", "documentation", "how do i", "what is our"]
    return {"needs_retrieval": any(k in state["query"].lower() for k in keywords)}

def retrieve_and_answer(state: State) -> dict:
    return {"answer": "Based on our policy documents: ..."}

def answer_directly(state: State) -> dict:
    return {"answer": "Direct answer, no lookup needed."}

def route(state: State) -> str:
    return "retrieve_and_answer" if state["needs_retrieval"] else "answer_directly"

builder = StateGraph(State)
builder.add_node("check_if_retrieval_needed", check_if_retrieval_needed)
builder.add_node("retrieve_and_answer", retrieve_and_answer)
builder.add_node("answer_directly", answer_directly)

builder.add_edge(START, "check_if_retrieval_needed")
builder.add_conditional_edges("check_if_retrieval_needed", route, ["retrieve_and_answer", "answer_directly"])
builder.add_edge("retrieve_and_answer", END)
builder.add_edge("answer_directly", END)

graph = builder.compile()
print(graph.invoke({"query": "What's our return policy?", "needs_retrieval": False, "answer": ""}))

Recall this exact decision from your RAG course — retrieval genuinely costs real time and real tokens, and a question that doesn’t need it shouldn’t pay that cost. Here, that decision is a real, explicit, testable routing function, not an implicit assumption buried inside a single, do-everything node.

Scenario 4: tool selection

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

class State(TypedDict):
    intent: str
    result: str

def retry_payment_node(state: State) -> dict:
    return {"result": "Payment retried."}

def issue_refund_node(state: State) -> dict:
    return {"result": "Refund issued."}

def escalate_node(state: State) -> dict:
    return {"result": "Escalated to a human."}

def route_by_intent(state: State) -> str:
    mapping = {"retry": "retry_payment_node", "refund": "issue_refund_node"}
    return mapping.get(state["intent"], "escalate_node")

builder = StateGraph(State)
builder.add_node("retry_payment_node", retry_payment_node)
builder.add_node("issue_refund_node", issue_refund_node)
builder.add_node("escalate_node", escalate_node)

builder.add_conditional_edges(START, route_by_intent, ["retry_payment_node", "issue_refund_node", "escalate_node"])
builder.add_edge("retry_payment_node", END)
builder.add_edge("issue_refund_node", END)
builder.add_edge("escalate_node", END)

graph = builder.compile()
print(graph.invoke({"intent": "refund", "result": ""}))
print(graph.invoke({"intent": "unknown_thing", "result": ""}))

Notice mapping.get(state["intent"], "escalate_node") — a genuine, deliberate fallback. An intent the routing function doesn’t recognize doesn’t crash the graph; it safely, explicitly falls through to human escalation instead.

Scenario 5: human escalation based on risk

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

class State(TypedDict):
    refund_amount: float
    customer_tenure_years: float
    path: str

def assess_risk(state: State) -> dict:
    return {}

def route_by_risk(state: State) -> str:
    high_risk = state["refund_amount"] > 500 or state["customer_tenure_years"] < 0.5
    return "human_review" if high_risk else "auto_process"

def auto_process(state: State) -> dict:
    return {"path": "Processed automatically."}

def human_review(state: State) -> dict:
    return {"path": "Sent to a human for review."}

builder = StateGraph(State)
builder.add_node("assess_risk", assess_risk)
builder.add_node("auto_process", auto_process)
builder.add_node("human_review", human_review)

builder.add_edge(START, "assess_risk")
builder.add_conditional_edges("assess_risk", route_by_risk, ["auto_process", "human_review"])
builder.add_edge("auto_process", END)
builder.add_edge("human_review", END)

graph = builder.compile()
print(graph.invoke({"refund_amount": 600.0, "customer_tenure_years": 3.0, "path": ""}))

Notice the routing function combines two genuinely different real risk signals — dollar amount and how new the customer is — into one decision. This is the real, honest shape of a genuine escalation policy: not one simple threshold, but a combination of factors, expressed directly and readably in code.

Where routing logic actually belongs

This is worth being explicit about, since there’s genuinely more than one valid place to put it, and beginners often assume there’s only one correct answer.

flowchart LR
    A["Inside a node\n(e.g. classify sets state['category'])"] --> D[Where should logic live?]
    B["Inside the routing function\n(reads state, returns a node name)"] --> D
    C["Inside state itself\n(a pre-computed flag, like needs_retrieval)"] --> D

A genuinely good rule: computation belongs in a node; the decision about where to go belongs in the routing function. Recall Scenario 2 — classify computes the category; route_by_category merely reads it. Keeping these separated makes both pieces independently testable, exactly the discipline your future testing module will build on directly.

The real, documented impact of getting routing right

It’s worth grounding all five scenarios above in something concrete, because “route based on state” can sound like a small, tidy engineering detail rather than something with genuine, measurable business consequences. Airbnb’s own customer support AI assistant, which now resolves over 40 percent of customer inquiries without any human agent involved, is publicly credited specifically with contributing to a real, measured 10 percent year-over-year reduction in cost per booking. The documented architecture behind that result is explicitly described as a triage-deflection-escalation workflow — not one single, undifferentiated chatbot handling every request the same way, but exactly the pattern this module has spent five scenarios building: classify first, then route each real case down a genuinely different path based on what it actually is. Separately, industry-wide data on manual, human-driven ticket routing has found it misroutes as much as 35 percent of tickets, directly contributing to missed service commitments and real, duplicated rework. The routing functions you just wrote in this module — imperfect as any first version will be — are solving a genuinely real, well-documented, and expensive problem, not a hypothetical one.

Common mistakes worth avoiding

Doing real computation inside a routing function. A routing function calling a model, or running a database query, blurs the line this module just drew — recall Scenario 2’s clean separation. Routing functions should read state and decide; they shouldn’t do the work of producing what they’re reading.

Forgetting to list every possible return value in the edges list. Recall Module 6’s own warning — if route_by_intent could theoretically return a node name not included in add_conditional_edges’s third argument, that specific path fails only when it’s actually taken, often invisible until a real, live input triggers it.

Not providing a genuine fallback for unrecognized input. Recall Scenario 4’s mapping.get(..., "escalate_node") — a routing function that assumes its input will always match one of a few expected cases will eventually meet an input it didn’t anticipate. Deciding in advance what should happen then is a real, deliberate design choice, not an afterthought.

What you should take away from this module

• add_conditional_edges(node, routing_function, [possible_targets]) is the real mechanism behind every one of this module’s five scenarios.
• Model selection, support routing, RAG-vs-direct, tool selection, and risk-based escalation are all genuinely the same underlying pattern, applied to five different real problems.
• Computation belongs in nodes; decisions belong in routing functions — keeping these separated makes both independently testable.
• A routing function needs an explicit, deliberate fallback for input it wasn’t specifically designed to handle.

Where this goes next

The next module covers Command — a genuinely different mechanism for the cases where a node needs to both update state and decide where to go, in one single step, rather than splitting that work across a node and a separate routing function.