Skip to content

Conditional Workflows

1. Why this matters

Most useful workflows need to decide: which response to send, which tool to call, whether to retry, when to stop. Conditional edges are the language LangGraph uses to express those decisions declaratively — visible in the graph, traceable in LangSmith.

2. Mental model

A conditional edge is a fork:

flowchart LR
    A[Node A<br/>updates state] --> R{router_fn state}
    R -->|returns positive| P[Positive node]
    R -->|returns negative| N[Negative node]
    R -->|returns neutral| U[Neutral node]

The router function is pure: it reads state, returns a name. It does NOT modify state.

3. Architecture / Flow

flowchart TD
    S((START)) --> CLS[classify_review]
    CLS --> R{sentiment?}
    R -->|positive| POS[positive_response]
    R -->|negative| NEG[run_diagnosis]
    NEG --> N2[negative_response]
    POS --> E((END))
    N2 --> E

The classify_review node writes sentiment into state. The router reads it and picks the next branch.

4. Core concepts

  • add_conditional_edges(src, router, path_map=None)
  • src — name of the node whose output triggers routing.
  • router(state) -> str | list[str] — what to do next. Return a string (one branch) or a list (parallel fan-out, see Send API).
  • path_map — optional {router_output: actual_node_name}. Skipping this means router returns node names directly.
  • Literal[...] return type — best practice; communicates allowed values.
  • Returning END — totally legal — a router can decide to terminate.
  • Multiple conditional edges from the same node — also legal; they're evaluated in order.
  • Conditional edges have NO automatic fall-through — every router output must map to a valid node (or END).

5. Code — minimal working example

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

class S(TypedDict):
    n: int
    parity: str

def classify(state: S):
    return {"parity": "even" if state["n"] % 2 == 0 else "odd"}

def even_branch(state: S):  return {"parity": "EVEN!"}
def odd_branch(state: S):   return {"parity": "ODD!"}

def route(state: S) -> Literal["even_branch", "odd_branch"]:
    return "even_branch" if state["parity"] == "even" else "odd_branch"

b = StateGraph(S)
b.add_node("classify", classify)
b.add_node("even_branch", even_branch)
b.add_node("odd_branch", odd_branch)

b.add_edge(START, "classify")
b.add_conditional_edges("classify", route)
b.add_edge("even_branch", END)
b.add_edge("odd_branch", END)

graph = b.compile()
print(graph.invoke({"n": 4, "parity": ""}))   # parity → "EVEN!"

6. Code — real-world pattern

Sentiment-based customer-review handler with an LLM classifier + structured output:

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from pydantic import BaseModel, Field

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class Sentiment(BaseModel):
    label: Literal["positive", "negative"] = Field(description="sentiment")

class ReviewState(TypedDict):
    review: str
    sentiment: str
    diagnosis: str
    response: str

def find_sentiment(state: ReviewState):
    cls_llm = llm.with_structured_output(Sentiment)
    s = cls_llm.invoke(state["review"])
    return {"sentiment": s.label}

def positive_response(state: ReviewState):
    r = llm.invoke([HumanMessage(f"Write a short thank-you reply to: {state['review']}")])
    return {"response": r.content}

def run_diagnosis(state: ReviewState):
    r = llm.invoke([HumanMessage(
        f"Identify the main issue in 1 sentence: {state['review']}")])
    return {"diagnosis": r.content}

def negative_response(state: ReviewState):
    r = llm.invoke([HumanMessage(
        f"Write a sympathetic apology + offer help. Issue: {state['diagnosis']}\nReview: {state['review']}")])
    return {"response": r.content}

def route_sentiment(state: ReviewState) -> Literal["positive_response", "run_diagnosis"]:
    return "positive_response" if state["sentiment"] == "positive" else "run_diagnosis"

b = StateGraph(ReviewState)
b.add_node("find_sentiment", find_sentiment)
b.add_node("positive_response", positive_response)
b.add_node("run_diagnosis", run_diagnosis)
b.add_node("negative_response", negative_response)

b.add_edge(START, "find_sentiment")
b.add_conditional_edges("find_sentiment", route_sentiment)
b.add_edge("run_diagnosis", "negative_response")
b.add_edge("positive_response", END)
b.add_edge("negative_response", END)

graph = b.compile()
print(graph.invoke({
    "review": "Battery died in 2 days. Avoid.",
    "sentiment": "", "diagnosis": "", "response": "",
})["response"])

Router using path_map when router output ≠ node name:

def route(state) -> str:
    return "p" if state["score"] > 0.7 else "n"   # short codes

builder.add_conditional_edges(
    "score_node",
    route,
    path_map={"p": "positive_node", "n": "negative_node"},
)

Router can return END to terminate early:

def route(state) -> Literal["continue", "stop"]:
    if state["budget_used"] > 1000:
        return "stop"
    return "continue"

builder.add_conditional_edges(
    "step",
    route,
    {"continue": "step", "stop": END},   # loop back or end
)

7. Common pitfalls

  • Router function mutating state. It must be pure — only inspects. State changes belong in the upstream node.
  • Router output not in mapping. If you provide path_map, all possible return values must be keys. Mis-spelled key → KeyError at runtime.
  • Multiple conditional edges from same source written carelessly. They're evaluated in order; the first matching branch is taken. If you really need parallel fan-out, use a list return (Send API, Chapter 7).
  • Forgetting to add a path to END. Some branch must eventually lead to END, or your graph runs forever.
  • No Literal annotation on the router. Works fine, but loses IDE autocomplete and lets typos slip through.

8. When to use vs not use

Use conditional edges when Use static edges when
Routing depends on data in state The next step is always the same
Implementing classification → branching Step ordering is fixed
Quality gates ("if eval fails, retry") No decision needed
Fallback paths

9. Cheatsheet

from typing import Literal

# Simple: router returns the destination node name
def route(state) -> Literal["a", "b"]:
    return "a" if state["cond"] else "b"

builder.add_conditional_edges("src", route)

# With a path mapping (when router returns codes, not node names)
builder.add_conditional_edges(
    "src",
    router_returning_code,
    path_map={"code1": "node_a", "code2": "node_b", "stop": END},
)

# Router can return END to terminate the run
def route(state) -> Literal["next", "done"]:
    return "done" if state["complete"] else "next"

builder.add_conditional_edges("src", route,
                              {"next": "src", "done": END})   # loop or end

10. Q&A — recall test

  • Q: Should the router function mutate state? A: No. Routers are pure — they read state and return a name. Mutations belong in nodes.

  • Q: When do you need a path_map? A: When the router returns short codes / booleans / non-node-name strings. The mapping translates them to actual node names.

  • Q: Can a router return END? A: Yes — a perfectly valid way to terminate a branch early.

  • Q: Why annotate the router with Literal[...]? A: Communicates allowed values to readers and IDEs, helps tooling validate the path mapping.

  • Q: Difference between two add_conditional_edges on the same source vs one that returns a list? A: Two separate conditional edges are checked one-by-one. A list return triggers parallel fan-out (Send API) — multiple branches run concurrently.

Practice

What does this print?

Expected: high

# A router function decides which node to go to next based on state
def route(state):
    return "high" if state["score"] > 50 else "low"
print(route({"score": 75}))

The router must return a string that matches a valid node name (or END)

Expected: True

def route(state):
    return True if state["score"] > 50 else False    # bug: must return string node name, not bool
result = route({"score": 75})
print(isinstance(result, str))

Quiz — Quick check

What you remember

Q1. What does a conditional edge function return?

  • A string matching a node name (or END) — tells the graph where to go next
  • A boolean
  • The next state
  • A new graph

Why: The router function inspects state and decides the next node. Its return value must match one of the keys in the routing dict (or be END).

Q2. What happens with two add_conditional_edges from the same source node?

  • Error — only one allowed
  • Both run; the second condition's target node also gets visited
  • Only the first runs
  • Random selection

Why: Multiple conditional edges from one source can fan out the workflow. Each router function fires independently and adds its target to the next steps.

Q3. What's the recommended pattern for routing in a complex graph?

  • Encapsulate routing logic in dedicated router functions; nodes do the work, routers decide flow
  • Put routing logic inside nodes
  • Use global state
  • Avoid conditionals

Why: Separation of concerns. Nodes focus on producing state changes; routers focus on flow control. Makes graphs easier to read, test, and debug.

Common doubts

How is conditional routing different from an if/else inside a node?

Inside a node, an if/else just produces different state. Conditional routing changes which node runs next. The right tool depends on whether you need different downstream work (route) or just different field values (in-node logic).

Can a conditional edge route to multiple targets?

Yes — return a list of node names instead of a string. LangGraph fans out execution: all listed nodes run (potentially in parallel). Used in map-reduce style patterns.

Should I use conditional edges for error handling?

Yes — common pattern. Have an error field in state; a router checks it and either continues normal flow or jumps to an error-handling node. Combine with retries for robust workflows.