Conditional Branching
Learn how to route execution dynamically using conditional edges, routing functions, and add_conditional_edges in LangGraph.
Conditional Branching
Not all graphs are linear. Conditional branching lets your graph make decisions at runtime, routing execution to different paths based on the current state.
What is Conditional Branching?
Conditional branching means the next node to execute depends on the current state, not a fixed topology.
The classifier node analyzes the input and decides which path to take.
add_conditional_edges
add_conditional_edges is the method that enables conditional routing:
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
class State(TypedDict):
input_text: str
category: str
def classifier(state: State) -> dict:
# Simplified classification — in practice, use an LLM
text = state["input_text"].lower()
if "bill" in text or "payment" in text:
return {"category": "billing"}
elif "bug" in text or "error" in text:
return {"category": "technical"}
else:
return {"category": "general"}
# Routing function: receives state, returns a node name
def route_by_category(state: State) -> str:
return state["category"]
builder = StateGraph(State)
builder.add_node("classifier", classifier)
builder.add_node("tech_support", lambda s: {"tech_support": True})
builder.add_node("billing_support", lambda s: {"billing_support": True})
builder.add_node("general_support", lambda s: {"general_support": True})
builder.add_edge(START, "classifier")
builder.add_conditional_edges(
"classifier", # Source node
route_by_category, # Router function
{ # Mapping: return value → target node
"technical": "tech_support",
"billing": "billing_support",
"general": "general_support"
}
)
builder.add_edge("tech_support", END)
builder.add_edge("billing_support", END)
builder.add_edge("general_support", END)
app = builder.compile()The routing function receives the full state dict and returns a string. That string is used as a key in the mapping dict to look up the target node name.
Router Function Patterns
Simple String Return
def router(state: State) -> str:
if state["is_complete"]:
return "end"
return "continue"Dictionary Mapping
builder.add_conditional_edges(
"analyze",
router,
{
"end": END, # Map to END constant
"continue": "process" # Map to another node
}
)Default Route with Fallback
If the router returns a value not in the mapping, an error is thrown. Always cover all possible return values:
def sentiment_router(state: State) -> str:
sentiment = state["sentiment"]
if sentiment == "positive":
return "positive"
elif sentiment == "negative":
return "negative"
return "neutral" # Default fallback
builder.add_conditional_edges(
"analyze",
sentiment_router,
{
"positive": "handle_positive",
"negative": "handle_negative",
"neutral": "handle_neutral"
}
)LLM-Powered Routing
Use an LLM to classify input and determine routing:
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
llm = ChatOpenAI(model="gpt-4o-mini")
class RouteState(TypedDict):
query: str
route: str
def classify_route(state: RouteState) -> dict:
prompt = ChatPromptTemplate.from_messages([
("system", "Classify the query into one category: 'technical', 'billing', or 'general'. "
"Respond with only the category name."),
("human", "{query}")
])
chain = prompt | llm | StrOutputParser()
route = chain.invoke({"query": state["query"]}).strip().lower()
return {"route": route}
def dynamic_router(state: RouteState) -> str:
return state["route"]
builder.add_conditional_edges(
"classify",
dynamic_router,
{
"technical": "tech_handler",
"billing": "billing_handler",
"general": "general_handler"
}
)When using LLMs for routing, add a post-processing step to clean and validate the route value. Set a default fallback for unexpected outputs.
Conditional Loops
The most common use of conditional edges is looping — repeating a node until a condition is met:
class LoopState(TypedDict):
input: str
result: str
attempts: int
is_valid: bool
def process(state: LoopState) -> dict:
# Attempt to process the input
result = attempt_processing(state["input"])
is_valid = validate_result(result)
return {
"result": result,
"is_valid": is_valid,
"attempts": state["attempts"] + 1
}
def loop_router(state: LoopState) -> str:
if state["is_valid"]:
return "valid"
if state["attempts"] >= 3:
return "max_retries"
return "retry"
builder.add_conditional_edges(
"process",
loop_router,
{
"valid": "format_output", # Success — move forward
"retry": "process", # Retry — loop back
"max_retries": "error_handler" # Give up
}
)Always include a maximum retry count in loops. Without it, a persistent failure causes an infinite loop that hits the recursion limit.
Multi-Condition Routing
Route to different nodes based on multiple state fields:
def complex_router(state: State) -> str:
if state.get("error"):
return "error"
if state["requires_tools"] and state["has_tool_results"]:
return "synthesize"
if state["requires_tools"] and not state["has_tool_results"]:
return "execute_tools"
return "generate"
builder.add_conditional_edges(
"analyze",
complex_router,
{
"error": "error_handler",
"synthesize": "synthesizer",
"execute_tools": "tool_executor",
"generate": "generator"
}
)Conditional Edges from Multiple Nodes
You can add conditional edges from any node, not just a classifier:
builder.add_conditional_edges("validate", validate_router, {...})
builder.add_conditional_edges("search", search_router, {...})
builder.add_conditional_edges("generate", quality_check_router, {...})Ternary Router Pattern
A simple binary decision:
def is_complete(state: State) -> str:
return "done" if state.get("finished") else "continue"
builder.add_conditional_edges(
"worker",
is_complete,
{
"done": END,
"continue": "worker" # Loop back
}
)Complete Example: Intelligent Query Router
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
from typing import Annotated, List
from operator import add
llm = ChatOpenAI(model="gpt-4o-mini")
# Tools
@tool
def search_web(query: str) -> str:
"""Search the web for information."""
return f"Web results for: {query}"
@tool
def calculate(expr: str) -> str:
"""Calculate math expressions."""
return str(eval(expr, {"__builtins__": {}}, {}))
class RouterState(TypedDict):
messages: Annotated[List, add]
query: str
route: str
response: str
def classify_node(state: RouterState) -> dict:
prompt = ChatPromptTemplate.from_messages([
("system", "Route the query to: 'web_search', 'calculator', or 'chat'. "
"Respond with only the route name."),
("human", "{query}")
])
chain = prompt | llm | StrOutputParser()
route = chain.invoke({"query": state["query"]}).strip().lower()
return {"route": route}
def router_fn(state: RouterState) -> str:
return state["route"]
def web_search_node(state: RouterState) -> dict:
result = search_web.invoke({"query": state["query"]})
return {"response": result, "messages": [f"[Web] {result}"]}
def calculator_node(state: RouterState) -> dict:
result = calculate.invoke({"expr": state["query"]})
return {"response": result, "messages": [f"[Calc] {result}"]}
def chat_node(state: RouterState) -> dict:
response = llm.invoke(f"Answer this: {state['query']}")
return {"response": response.content, "messages": [f"[Chat] {response.content}"]}
builder = StateGraph(RouterState)
builder.add_node("classify", classify_node)
builder.add_node("web_search", web_search_node)
builder.add_node("calculator", calculator_node)
builder.add_node("chat", chat_node)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", router_fn, {
"web_search": "web_search",
"calculator": "calculator",
"chat": "chat"
})
builder.add_edge("web_search", END)
builder.add_edge("calculator", END)
builder.add_edge("chat", END)
app = builder.compile()
# Test
result = app.invoke({
"messages": [],
"query": "What is 15 * 7?",
"route": "",
"response": ""
})
print(result["response"]) # 105
result = app.invoke({
"messages": [],
"query": "Who invented Python?",
"route": "",
"response": ""
})
print(result["response"]) # Chat or web search resultThis pattern — classify → route → execute specialized handler — is the foundation of all intelligent routing agents.
Routing Best Practices
- Always handle all possible route values — missing a mapping raises an error
- Add a default/fallback route for unexpected router outputs
- Validate the router output when using LLMs for routing
- Use descriptive route names that match node names for clarity
- Limit routing depth — deeply nested conditional chains are hard to debug
- Log the route decision for debugging and observability
Practice Questions
What method is used for conditional branching in LangGraph?
What does a router function receive and return?
What happens if a router returns a value not in the mapping dict?
What is the most common use of conditional edges in agents?
How can you create a simple binary (yes/no) conditional edge?
What should you always include in a loop with conditional edges?
Can you use an LLM as a router in LangGraph?
What is the advantage of using add_conditional_edges over multiple add_edge calls?
How do you make a conditional edge loop back to the source node?
Which constant can be used as a target in the conditional edge mapping?
Key Takeaways
add_conditional_edges(source, router, mapping)enables dynamic routing- Router functions receive state and return a string key
- The mapping dict translates router output into target node names
- Loops are created by mapping a route back to a previously executed node
- LLM-powered routing uses an LLM to classify and write the route to state
- Always cover all possible router outputs in the mapping dict
- Include termination conditions in loops to prevent infinite execution
- END can be a target in conditional edge mappings
- Log route decisions for debugging and observability