LangGraph Conditional Edges and Control Flow

This is the third post in our LangGraph fundamentals series, building on why graphs and your first stateful agent.

The previous post used a conditional edge for one specific, common pattern: retry-or-finish. Conditional edges support a genuinely broader set of control-flow patterns — real branching based on input classification, parallel paths that later merge, and dynamic routing to different subgraphs entirely. This post covers those patterns directly.

Branching based on classifying the input first

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
 
class SupportState(TypedDict):
    message: str
    category: str
    response: str
 
def classify(state: SupportState) -> SupportState:
    result = model.invoke(
        f"Classify this message as 'billing', 'technical', or 'general': {state['message']}"
    )
    state["category"] = result.content.strip().lower()
    return state
 
def handle_billing(state: SupportState) -> SupportState:
    state["response"] = "Routing to billing: " + model.invoke(
        f"Answer this billing question: {state['message']}"
    ).content
    return state
 
def handle_technical(state: SupportState) -> SupportState:
    state["response"] = "Routing to technical: " + model.invoke(
        f"Answer this technical question: {state['message']}"
    ).content
    return state
 
def handle_general(state: SupportState) -> SupportState:
    state["response"] = model.invoke(f"Answer generally: {state['message']}").content
    return state
 
def route_by_category(state: SupportState) -> str:
    return state["category"]
 
graph = StateGraph(SupportState)
graph.add_node("classify", classify)
graph.add_node("billing", handle_billing)
graph.add_node("technical", handle_technical)
graph.add_node("general", handle_general)
 
graph.set_entry_point("classify")
graph.add_conditional_edges(
    "classify",
    route_by_category,
    {"billing": "billing", "technical": "technical", "general": "general"},
)
graph.add_edge("billing", END)
graph.add_edge("technical", END)
graph.add_edge("general", END)
 
app = graph.compile()

This is a genuinely different use of conditional edges than the previous post's retry loop: classify runs once, and its result routes execution to exactly one of three entirely different downstream nodes — each potentially doing meaningfully different work (a different prompt, a different tool, a different downstream system) — rather than looping back to repeat the same step. The mapping dictionary ({"billing": "billing", ...}) is what makes the routing explicit and inspectable: looking at the graph definition tells you every possible path a message can take, without having to trace through nested conditional logic buried inside a single function.

Parallel paths that later merge

class ResearchState(TypedDict):
    question: str
    web_results: str
    doc_results: str
    final_answer: str
 
def search_web(state: ResearchState) -> ResearchState:
    state["web_results"] = "Web search results here..."
    return state
 
def search_docs(state: ResearchState) -> ResearchState:
    state["doc_results"] = "Internal doc results here..."
    return state
 
def combine(state: ResearchState) -> ResearchState:
    combined_context = f"{state['web_results']}\n{state['doc_results']}"
    state["final_answer"] = model.invoke(
        f"Answer using this context: {combined_context}\n\nQuestion: {state['question']}"
    ).content
    return state
 
graph = StateGraph(ResearchState)
graph.add_node("search_web", search_web)
graph.add_node("search_docs", search_docs)
graph.add_node("combine", combine)
 
graph.set_entry_point("search_web")
graph.add_edge("search_web", "search_docs")   # sequential in this simple example
graph.add_edge("search_docs", "combine")
graph.add_edge("combine", END)

(LangGraph also supports genuinely concurrent execution of independent branches via fan-out/fan-in patterns for cases where search_web and search_docs have no dependency on each other at all — the sequential version shown here keeps the example focused on the state-merging concept itself, which is the part worth understanding first.) The key idea: combine waits until both upstream results are available in state before it runs, merging two independent pieces of context into one final answer — a pattern with no clean equivalent in a single linear chain, which has no natural way to represent "these two things happen independently, then get merged."

Dynamic routing to an entirely different subgraph

def needs_escalation(state: SupportState) -> str:
    if "urgent" in state["message"].lower() or "cancel" in state["message"].lower():
        return "escalate"
    return "normal"
 
graph.add_conditional_edges(
    "classify",
    needs_escalation,
    {"escalate": "human_escalation_node", "normal": "billing"},
)

Nothing restricts a conditional edge's routing function to only reason about the classification itself — here, it also checks for signals warranting escalation to a completely different path (routing to a human, in this simplified example, covered in depth in the next post in this series), independent of what category the message was classified into. This is the actual generality conditional edges provide: any function of the current state can decide the next node, not just a fixed classification result.

What to actually remember from this post

• Conditional edges support genuine branching to different downstream nodes, not just the retry-loop pattern from the previous post — a routing function's return value can send execution anywhere in the graph.
• A node can depend on multiple upstream results merging into shared state before it runs — a pattern with no natural expression in a single linear chain.
• A routing function can reason about anything in state, not just a single classification field — including signals for escalating to an entirely different path.
• The explicit mapping in add_conditional_edges makes every possible path through the graph visible directly in the graph definition, rather than buried in nested conditional logic.

Next in the series: Human-in-the-Loop Workflows with LangGraph, where we cover a graph that genuinely pauses — sometimes for hours or days — waiting on a real person's decision before continuing.