LangGraph Chapter 5 — Conditional Routing & Graph Control Flow

Senior Architect Interview Series — LangGraph & Agentic AI

Navigation

← Chapter 4 — Tool Calling | Chapter 6 — Memory →

5.0 What This Chapter Covers

Routing is what makes agents intelligent — the ability to dynamically decide what to do next based on context. This chapter covers:

1. How add_conditional_edges works in depth
2. Binary routing vs. multi-way routing
3. The supervisor routing pattern
4. route_question() in your project
5. Building complex routing graphs
6. Common routing patterns and anti-patterns

5.1 Why Routing Matters

In a simple linear pipeline, execution always follows the same fixed path:

Step 1 → Step 2 → Step 3 → Done

In an agent, the path depends on the situation:

What should happen next?
    ├── User asked a factual question → RAG tool → answer
    ├── User wants to analyze data → SQL tool → answer
    ├── User asks a general question → direct LLM answer
    └── LLM needs more context → search → loop back

Routing is the code that makes these decisions. In LangGraph, routing is implemented through conditional edges — edges whose destination is determined at runtime by a routing function.

5.2 add_conditional_edges — Full API

graph.add_conditional_edges(
    source,          # string: which node's output triggers this routing
    path,            # callable: routing function (state → str or END)
    path_map=None    # optional dict: explicit mapping from return values to node names
)

Simple Form (No Map)

graph.add_conditional_edges("call_llm", should_call_tools)

def should_call_tools(state: AgentState) -> str:
    if state["messages"][-1].tool_calls:
        return "call_tools"   # string must be a registered node name
    return END                # or END to stop

LangGraph uses the return value directly as the node name to route to. The return value must be:

• A node name registered with add_node()
• Or END

Explicit Map Form

graph.add_conditional_edges(
    "supervisor",
    route_to_agent,
    {
        "rag":     "rag_agent",     # return value → node name
        "data":    "data_agent",
        "general": "general_agent",
        "end":     END
    }
)

def route_to_agent(state: SupervisorState) -> str:
    return state["route"].lower()   # returns "rag", "data", "general", or "end"

The map is useful when:

• Your routing function returns values that aren't valid node names (e.g., uppercase strings)
• You want to make the routing logic explicit and auditable at graph definition time
• You're using Enum values for safety

5.3 Your Project's Routing — Deep Dive

Your project has two levels of routing:

Level 1 — Tool Loop Routing (in agent.py)

def should_call_tools(state: AgentState) -> str:
    """Binary router: should we call tools or are we done?"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "call_tools"
    return END

graph.add_conditional_edges("call_llm", should_call_tools)

Decision logic:

• LLM emitted tool_calls → execute tools → loop
• LLM emitted plain text → done → return answer

Graph shape:

call_llm ──── should_call_tools() ──── "call_tools" node
                                   └── END

Level 2 — Supervisor Routing (in supervisor.py)

def route_question(question: str) -> str:
    """Three-way router: which specialized agent should handle this?"""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    prompt = f"""Route the following question to the correct agent.
    
    RAG: Questions about Agent Factory features, architecture, documentation
    DATA: Questions requiring database queries, metrics, statistics
    GENERAL: All other conversational questions
    
    Respond with exactly one word: RAG, DATA, or GENERAL
    
    Question: {question}"""
    
    response = llm.invoke([HumanMessage(content=prompt)])
    return response.content.strip().upper()   # "RAG", "DATA", or "GENERAL"

This uses an LLM to classify the question's intent and route it to the appropriate specialized agent.

5.4 Routing Function Taxonomy

There are three types of routing functions based on what reads to make the decision:

Type 1 — State-Based Routing (Most Common)

Reads a field directly from state — fast, deterministic, no LLM call:

def route_from_state(state: AgentState) -> str:
    route = state["route"]   # e.g., set by a previous supervisor node
    if route == "RAG":
        return "rag_agent"
    elif route == "DATA":
        return "data_agent"
    return END

Type 2 — Message-Based Routing

Inspects the last message to decide next step:

def should_call_tools(state: AgentState) -> str:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "call_tools"
    return END

Type 3 — LLM-Based Routing

Uses a separate LLM call to classify and route — more flexible but adds latency and cost:

def llm_router(state: AgentState) -> str:
    question = state["messages"][-1].content
    classification = route_question(question)  # LLM call
    return classification.lower()   # "rag", "data", "general"

5.5 Building a Multi-Agent Routing Graph

Here's how to wire the supervisor routing from supervisor.py into a full LangGraph multi-agent system:

from typing import TypedDict, Annotated
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage
from langgraph.graph import StateGraph, END

class SupervisorState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    route: str   # set by supervisor node

def supervisor_node(state: SupervisorState) -> dict:
    """Classify the question and store routing decision in state."""
    question = state["messages"][-1].content
    route = route_question(question)   # from supervisor.py → "RAG", "DATA", "GENERAL"
    return {"route": route}           # stores routing decision in state

def rag_agent_node(state: SupervisorState) -> dict:
    """Handle RAG questions."""
    answer = run_rag(state["messages"])   # your agent.py
    return {"messages": [AIMessage(content=answer)]}

def data_agent_node(state: SupervisorState) -> dict:
    """Handle data/SQL questions."""
    answer = run_data_agent(state["messages"])
    return {"messages": [AIMessage(content=answer)]}

def general_agent_node(state: SupervisorState) -> dict:
    """Handle general questions."""
    answer = llm.invoke(state["messages"]).content
    return {"messages": [AIMessage(content=answer)]}

def route_after_supervisor(state: SupervisorState) -> str:
    """Route to the correct agent based on supervisor's decision."""
    route = state["route"]
    if route == "RAG":
        return "rag_agent"
    elif route == "DATA":
        return "data_agent"
    else:
        return "general_agent"

# Build the graph
graph = StateGraph(SupervisorState)

graph.add_node("supervisor",    supervisor_node)
graph.add_node("rag_agent",     rag_agent_node)
graph.add_node("data_agent",    data_agent_node)
graph.add_node("general_agent", general_agent_node)

graph.set_entry_point("supervisor")
graph.add_conditional_edges("supervisor", route_after_supervisor)

# All agents go to END after answering
graph.add_edge("rag_agent",     END)
graph.add_edge("data_agent",    END)
graph.add_edge("general_agent", END)

multi_agent = graph.compile()

Graph shape:

        START
          │
          ▼
     ┌──────────┐
     │supervisor│
     └────┬─────┘
          │ route_after_supervisor()
   ┌──────┼──────────┐
   │      │          │
   ▼      ▼          ▼
rag_    data_    general_
agent   agent    agent
   │      │          │
   └──────┴──────────┘
                │
               END

5.6 Fan-Out and Fan-In Routing

LangGraph supports parallel fan-out — routing to multiple nodes simultaneously and collecting results:

# Fan-out: route to multiple research agents in parallel
class ResearchState(TypedDict):
    query: str
    messages: Annotated[list[BaseMessage], add_messages]
    results: Annotated[list[str], operator.add]

def web_search_node(state): ...
def rag_search_node(state): ...
def db_search_node(state): ...

def synthesize_node(state):
    """Combine results from all three sources."""
    ...

graph.add_node("coordinator", coordinator_node)
graph.add_node("web_search",  web_search_node)
graph.add_node("rag_search",  rag_search_node)
graph.add_node("db_search",   db_search_node)
graph.add_node("synthesize",  synthesize_node)

# Fan-out: coordinator routes to ALL three in parallel
graph.add_conditional_edges(
    "coordinator",
    lambda state: ["web_search", "rag_search", "db_search"]  # returns a list!
)

# Fan-in: all three feed into synthesize
graph.add_edge("web_search", "synthesize")
graph.add_edge("rag_search", "synthesize")
graph.add_edge("db_search",  "synthesize")

5.7 Dynamic Routing Based on Confidence

A common production pattern: route to a more expensive model if confidence is low:

def route_by_confidence(state: AgentState) -> str:
    last = state["messages"][-1]
    confidence = state.get("confidence_score", 1.0)
    
    if confidence < 0.4:
        return "expensive_model"   # GPT-4 for hard questions
    elif confidence < 0.7:
        return "medium_model"      # GPT-4o-mini with RAG
    else:
        return "fast_model"        # simple direct answer

5.8 Routing Anti-Patterns

Anti-Pattern 1 — Routing Function Side Effects

# BAD — routing function modifies database
def bad_router(state) -> str:
    db.log_route(state["route"])   # side effect in router!
    return state["route"]

# GOOD — routing functions are PURE functions (no side effects)
def good_router(state) -> str:
    return state["route"]   # pure routing logic only

Routing functions should be pure — they read state and return a string. Side effects belong in nodes.

Anti-Pattern 2 — Non-Deterministic LLM Routing

# RISKY — LLM routing with temperature > 0 can produce inconsistent routes
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)  # BAD for routing

# BETTER — use temperature=0 for classification tasks
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)    # deterministic

Always use temperature=0 for routing/classification LLM calls.

Anti-Pattern 3 — Missing END Path

# BAD — no path to END from conditional_edges
def router(state) -> str:
    if state["needs_tools"]:
        return "tools"
    return "refine"   # could loop forever!

# GOOD — always have a path to END
def router(state) -> str:
    if state["needs_tools"] and state["attempt_count"] < 5:
        return "tools"
    return END

Always ensure conditional routing has at least one path to END.

Anti-Pattern 4 — Overly Complex Routing Functions

# BAD — routing function is too complex
def router(state) -> str:
    if state["a"] and not state["b"]:
        if state["c"] > 5:
            return "x"
        return "y"
    elif state["b"] and state["c"]:
        return "z"
    # ... 20 more conditions

# GOOD — push complexity into the nodes, keep routing simple
def router(state) -> str:
    return state["route"]   # node already computed the route

5.9 Routing with Enum Types

For production code, use Enum values instead of raw strings to avoid typos:

from enum import Enum

class Route(str, Enum):
    RAG     = "rag_agent"
    DATA    = "data_agent"
    GENERAL = "general_agent"

def route_after_supervisor(state: SupervisorState) -> str:
    route = state["route"]
    mapping = {
        "RAG":     Route.RAG,
        "DATA":    Route.DATA,
        "GENERAL": Route.GENERAL
    }
    return mapping.get(route, Route.GENERAL).value

5.10 Testing Routing Logic

Routing functions are pure functions — they're easy to test:

import pytest
from agent.agent import should_call_tools, AgentState
from langgraph.graph import END

def test_routes_to_tools_when_tool_calls_present():
    state = AgentState(messages=[
        AIMessage(content="", tool_calls=[{"id": "1", "name": "rag_search", "args": {}}])
    ])
    assert should_call_tools(state) == "call_tools"

def test_routes_to_end_when_no_tool_calls():
    state = AgentState(messages=[
        AIMessage(content="The answer is 42")
    ])
    assert should_call_tools(state) == END

def test_supervisor_routes_correctly():
    # Mock the LLM if you want pure unit tests
    route = route_question("What is Agent Factory's architecture?")
    assert route == "RAG"

5.11 Interview Q&A

Q: Explain add_conditional_edges and how it differs from add_edge.

add_edge(A, B) is an unconditional connection that always routes to B after A with no decision logic. add_conditional_edges(A, routing_fn) calls routing_fn(current_state) after A completes, and routes to whatever node the function returns. The routing function is a pure Python function — it can read any part of the state to decide. In our agent, we use both: call_tools → call_llm is a fixed edge (always loop back), while call_llm → should_call_tools is conditional (end or continue based on whether the LLM used tools).

Q: How does the supervisor agent decide which sub-agent to call?

Our supervisor.py's route_question() function sends the user's question to gpt-4o-mini with a classification prompt that describes the three categories: RAG (knowledge base questions), DATA (analytical/SQL questions), and GENERAL (conversational). The LLM responds with exactly one word — "RAG", "DATA", or "GENERAL". In a full LangGraph multi-agent graph, the supervisor node stores this route in state, and the routing function reads state["route"] to direct execution to the appropriate sub-agent. We use temperature=0 to ensure deterministic routing.

Q: What are the risks of using an LLM for routing, and how do you mitigate them?

Three risks: (1) Non-determinism — LLM may route differently for identical inputs. Mitigate with temperature=0 and clear, distinct category descriptions. (2) Latency — routing adds an extra LLM call. Mitigate with faster/cheaper models for routing (e.g., gpt-4o-mini instead of gpt-4o). (3) Cost — every question incurs a classification call. Mitigate by caching routes for identical questions or using a fine-tuned smaller model. Alternatively, use a simple keyword/regex classifier for clear-cut cases before falling back to LLM routing.

Q: How would you add a fallback route when the LLM returns an unexpected value?

In the routing function, use a dict.get() with a default: mapping.get(state["route"], "general_agent"). Or use an explicit else clause that routes to a safe default. Then add an alert/log in the default case so you can detect unexpected LLM outputs in production. The routing function should never raise — it should always return a valid node name. This is the defense-in-depth principle for agent routing.

Q: Can a LangGraph routing function route to multiple nodes simultaneously?

Yes — returning a list from the routing function triggers parallel fan-out execution. LangGraph executes all listed nodes concurrently and then fans back in (typically to an aggregator node). This is used for research tasks where multiple sources (web search, RAG, database) should be queried simultaneously. The results are merged using reducers (operator.add for lists). The tradeoff is higher parallelism vs. more complex state management.

5.12 Key One-Liners to Memorize

"add_conditional_edges: routing function reads state, returns next node name or END."

"Route by state (fast) > route by message (medium) > route by LLM (flexible but slow)."

"Routing functions must be PURE — no side effects, just read state and return a string."

"Always use temperature=0 for LLM-based routing — determinism is critical for control flow."

"Every conditional edge MUST have a path to END — no infinite loops in production."

"The optional path_map dict makes routing auditable at graph definition time."

Next: Chapter 6 — Memory: In-Context, Session & Long-Term