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."

True fan-out/fan-in: genuinely concurrent branches

For branches that truly have no dependency on each other, LangGraph supports adding multiple edges out of a single node, which run concurrently rather than sequentially:

graph.set_entry_point("start")
graph.add_edge("start", "search_web")
graph.add_edge("start", "search_docs")
graph.add_edge("search_web", "combine")
graph.add_edge("search_docs", "combine")

Both search_web and search_docs here run from the same entry point concurrently, and combine — having two incoming edges — automatically waits for both upstream nodes to complete before it runs, regardless of which one happens to finish first. This is a genuinely different execution shape than the sequential version above: real concurrency, not just two steps written next to each other that happen to run one after another. For any two operations that are actually independent (a web search and a database lookup that don't depend on each other's result), this fan-out/fan-in shape reduces total latency compared to running them one after another for no structural reason.

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.

Composing an entire graph as a single node

A graph can itself be compiled and then used as a single node inside a larger graph — genuinely useful once a specific sub-workflow (say, the retry loop from the previous post) needs to be reused as one step within a larger process:

retry_subgraph_app = retry_graph.compile()
 
def draft_with_retry(state: OuterState) -> OuterState:
    sub_result = retry_subgraph_app.invoke({
        "question": state["question"], "draft_answer": "", "quality_score": 0, "attempts": 0
    })
    state["draft"] = sub_result["draft_answer"]
    return state
 
outer_graph.add_node("draft_with_retry", draft_with_retry)

This is the same composability principle covered for Terraform modules calling other modules, and reusable GitHub Actions workflows calling composite actions, elsewhere in this blog: a well-defined, self-contained graph is itself just another callable unit, usable as a single node wherever a larger graph needs exactly that sub-workflow's behavior.

Conditional edges with more than two meaningful branches

The classification example earlier in this post routes to exactly one of three fixed nodes. A routing function isn't limited to a small, fixed set of branches — it's equally valid for a graph to route to a dynamically-selected node chosen from a larger set, useful when the set of possible handlers is itself data-driven rather than hardcoded:

def route_to_handler(state: SupportState) -> str:
    available_handlers = get_registered_handlers()  # e.g., loaded from configuration
    category = state["category"]
    return category if category in available_handlers else "general"
 
graph.add_conditional_edges("classify", route_to_handler, {h: h for h in get_registered_handlers()})

This is worth reaching for specifically when the set of valid destinations genuinely varies at runtime or across deployments (different customer support categories per client, say) rather than being a small, fixed set known entirely at graph-definition time — the mapping dictionary passed to add_conditional_edges can be constructed dynamically, not just written out by hand for a fixed handful of cases.

Default fallback edges

A routing function occasionally returns a value that doesn't cleanly match any of the intended branches — a classification the model got wrong, or a genuinely ambiguous case. Building in an explicit default path prevents this from becoming an unhandled error:

def route_by_category(state: SupportState) -> str:
    category = state["category"]
    if category not in {"billing", "technical", "general"}:
        return "general"  # explicit fallback for anything unexpected
    return category

Treating "the classifier returned something unexpected" as an expected case with a defined fallback, rather than letting it surface as a runtime KeyError against the conditional edge mapping, is the same defensive habit as a Terraform validation block catching a bad variable value immediately and clearly, rather than letting it fail confusingly several steps later.

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.
• True fan-out/fan-in (multiple edges from one node, converging on another) runs branches genuinely concurrently, reducing latency for independent operations, not just organizing sequential steps differently.
• A routing function can reason about anything in state, not just a single classification field — including signals for escalating to an entirely different path.
• A compiled graph can itself be used as a node inside a larger graph — the same composability principle as Terraform modules or reusable GitHub Actions workflows, applied to agentic workflows.

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.