5. Conditional Routing for Agents

Overview

In the previous capsule you implemented an agent's cycle: reason → tools → reason, with conditional edges that decide "keep going or stop." That's binary routing — two possible paths. But a real agent doesn't make binary decisions. After running a tool, it might need to research more, synthesize what it has, handle an error, or escalate to a human. Those decisions are multi-destination routing, and they're what define your agent's real behavior.

Conditional routing is the agent's decision layer. If the nodes are "what the agent does", the conditional edges are "what it decides to do next". In M3 (capsule 03) you worked on tool routing: deciding which tools to give the model based on context. Here it's different — you're deciding which graph node to go to after each step. It isn't "which tools it has available" but "which path the execution takes." Tool routing controls the model's capabilities. Flow routing controls the agent's behavior.

The difference matters because the routing functions you design here are what make your agent behave intelligently without depending 100% on the LLM for every flow decision. The LLM reasons about content. Your routing functions reason about process.

Routing as the Agent's Decision Points

From "continue/stop" to real decisions

In capsule 04, your conditional edge asked a single question: do I keep iterating or finish?

def should_continue(state: AgentState) -> str:
    if not state["messages"][-1].tool_calls:
        return "end"
    return "tools"

That's an if/else. But an agent with planning, research, analysis and synthesis nodes needs richer decisions:

                    ┌─────────────────────┐
                    │     AFTER ACTING    │
                    │   What do I do now? │
                    └──────────┬──────────┘
                               │
              ┌────────────────┼────────────────┐
              │                │                │
         "research"      "synthesize"       "error"
              │                │                │
              ▼                ▼                ▼
        ┌──────────┐   ┌──────────────┐  ┌──────────────┐
        │ RESEARCH │   │  SYNTHESIZE  │  │ERROR_HANDLER │
        └──────────┘   └──────────────┘  └──────────────┘

Every arrow is a decision point. The routing function evaluates the current state and picks the path. It isn't the LLM deciding — it's your code, based on concrete signals from the state.

Why not let the LLM decide everything?

You could ask the LLM to return {"next_step": "synthesize"} and route based on that. Sometimes that's appropriate. But there are three reasons to prefer deterministic routing when you can:

1. Cost. Every LLM decision burns tokens. Deterministic routing is free.
2. Predictability. if iteration_count >= 5: return "synthesize" always behaves the same way.
3. Speed. A Python function evaluates in microseconds. An LLM call takes seconds.

The rule: use deterministic routing when the decision can be expressed with state data. Use LLM routing when the decision requires understanding the content.

Routing Based on Tool Results

After your tools run, the results determine what the agent should do. A successful result with rich data deserves analysis. An error needs recovery. A "no results" needs rephrasing.

Setup and routing function

We use the same setup from previous capsules (AgentState with messages, iteration_count, max_iterations, error_count; search_web and analyze_data tools). The tools return prefixes like "DETAILED:", "ERROR:", "NO_RESULTS:" so the routing function can classify them.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langchain_core.messages import ToolMessage, SystemMessage, HumanMessage

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    iteration_count: int
    max_iterations: int
    error_count: int

@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    if "quantum" in query.lower():
        return "DETAILED: Quantum computing uses qubits in superposition..."
    if "xyz_nonexistent" in query.lower():
        return "NO_RESULTS: No relevant information found."
    return f"General results for: {query}"

@tool
def analyze_data(text: str) -> str:
    """Analyze data and extract insights."""
    if len(text) < 20:
        return "ERROR: Not enough text to analyze."
    return f"ANALYSIS: {len(text.split())} words processed."

tools = [search_web, analyze_data]
tools_by_name = {t.name: t for t in tools}
model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)

def route_after_tools(state: AgentState) -> str:
    """Decide the next step based on what the tools returned."""
    last_messages = []
    for msg in reversed(state["messages"]):
        if isinstance(msg, ToolMessage):
            last_messages.append(msg)
        else:
            break
    
    has_error = any("ERROR:" in m.content for m in last_messages)
    has_no_results = any("NO_RESULTS:" in m.content for m in last_messages)
    has_detailed = any("DETAILED:" in m.content for m in last_messages)
    
    if has_error and state.get("error_count", 0) >= 3:
        return "synthesize"
    if has_error:
        return "retry"
    if has_no_results:
        return "reformulate"
    if has_detailed:
        return "analyze"
    
    return "reason"

Each return value is a different destination in the graph:

• "retry" → run the tools again with adjusted parameters
• "reformulate" → the LLM rephrases the search
• "analyze" → an analysis node processes the rich data
• "synthesize" → generate a final answer with what's available
• "reason" → the standard cycle (keep reasoning)

Wiring it into the graph

The nodes (reason_node, tools_node, reformulate_node, analyze_node, synthesize_node) follow the patterns from previous capsules. What's new is how they connect:

graph = StateGraph(AgentState)
graph.add_node("reason", reason_node)
graph.add_node("tools", tools_node)
graph.add_node("reformulate", reformulate_node)
graph.add_node("analyze", analyze_node)
graph.add_node("synthesize", synthesize_node)

graph.add_edge(START, "reason")
graph.add_conditional_edges("reason", should_continue, {"tools": "tools", "end": END})
graph.add_conditional_edges("tools", route_after_tools, {
    "retry": "reason",
    "reformulate": "reformulate",
    "analyze": "analyze",
    "synthesize": "synthesize",
    "reason": "reason",
})
graph.add_edge("reformulate", "reason")
graph.add_edge("analyze", "reason")
graph.add_edge("synthesize", END)

agent = graph.compile()

Visualize it to verify — the diagram should show tools with 5 outgoing arrows, each one a decision point:

from IPython.display import Image, display
display(Image(agent.get_graph().draw_mermaid_png()))

Routing Based on Confidence

If the model has little information, it should research more before synthesizing. You simulate confidence with structured output:

from pydantic import BaseModel, Field

class ReasoningOutput(BaseModel):
    thinking: str = Field(description="Your internal reasoning")
    confidence: float = Field(description="How confident you are (0.0 to 1.0)")
    needs_more_info: bool = Field(description="Whether you need to search for more information")

model_structured = model.with_structured_output(ReasoningOutput)

Node and routing with confidence

class ConfidenceState(TypedDict):
    messages: Annotated[list, add_messages]
    iteration_count: int
    max_iterations: int
    confidence: float

def reason_with_confidence(state: ConfidenceState) -> dict:
    assessment = model_structured.invoke(state["messages"])
    return {
        "confidence": assessment.confidence,
        "iteration_count": state.get("iteration_count", 0) + 1,
    }

def route_by_confidence(state: ConfidenceState) -> str:
    if state.get("iteration_count", 0) >= state.get("max_iterations", 10):
        return "synthesize"
    
    confidence = state.get("confidence", 0.0)
    if confidence >= 0.8:
        return "synthesize"
    elif confidence >= 0.5:
        return "analyze"
    else:
        return "research"

Wiring

graph = StateGraph(ConfidenceState)
graph.add_node("assess", reason_with_confidence)
graph.add_node("research", research_node)
graph.add_node("analyze", analyze_node)
graph.add_node("synthesize", synthesize_node)

graph.add_edge(START, "assess")
graph.add_conditional_edges("assess", route_by_confidence, {
    "research": "research",
    "analyze": "analyze",
    "synthesize": "synthesize",
})
graph.add_edge("research", "assess")
graph.add_edge("analyze", "assess")
graph.add_edge("synthesize", END)

The flow: assess evaluates confidence → low confidence sends it to research → back to assess → eventually confidence rises → synthesize → END.

Routing Based on Task State

When your agent follows a plan (a preview of M5), the task's progress determines the routing:

class PlanState(TypedDict):
    messages: Annotated[list, add_messages]
    plan: list[dict]
    current_step: int
    iteration_count: int
    max_iterations: int

def route_by_plan(state: PlanState) -> str:
    if state.get("iteration_count", 0) >= state.get("max_iterations", 15):
        return "synthesize"
    
    plan = state.get("plan", [])
    pending = [step for step in plan if step["status"] == "pending"]
    
    if not pending:
        return "synthesize"
    if len(pending) <= 1:
        return "final_research"
    return "research"

The research_node takes the next sub-question from the plan, researches it, marks it "completed", and increments current_step. That way the routing reflects real progress: while there are sub-questions, it researches. When only one is left, it digs deeper. When none are left, it synthesizes.

Multi-destination Routing

Pattern: Router with N destinations

def multi_router(state: AgentState) -> str:
    if state.get("iteration_count", 0) >= state.get("max_iterations", 10):
        return "synthesize"
    if state.get("error_count", 0) >= 3:
        return "fallback"
    last = state["messages"][-1]
    if not hasattr(last, "tool_calls") or not last.tool_calls:
        return "synthesize"
    
    tool_names = {tc["name"] for tc in last.tool_calls}
    if "search_web" in tool_names: return "web_tools"
    if "query_database" in tool_names: return "db_tools"
    if "analyze_data" in tool_names: return "analysis_tools"
    return "general_tools"

This router has 6 destinations: three safety nets (synthesize, fallback, synthesize) and three by tool type. The key is that the safety nets are evaluated first.

Pattern: LLM as the router

When the decision requires semantic understanding, the LLM picks the destination. The key: the LLM's logic goes in a node, not in the routing function:

class RouterDecision(BaseModel):
    next_step: Literal["research", "analyze", "synthesize", "clarify"] = Field(
        description="The next step based on the progress so far"
    )
    reasoning: str = Field(description="Why you chose this step")

router_model = model.with_structured_output(RouterDecision)

def llm_router_node(state: AgentState) -> dict:
    """Node that asks the LLM to decide."""
    decision = router_model.invoke(
        state["messages"] + [
            SystemMessage(content="Decide: 'research' if information is missing, 'analyze' if there's "
                          "data to process, 'synthesize' if you can answer.")
        ]
    )
    return {"router_decision": decision.next_step}

def route_from_llm(state: AgentState) -> str:
    """Routing function that only reads the decision from the state."""
    return state.get("router_decision", "research")

When to use each type

The hybrid pattern is the most common in production: check state conditions first (free), and only consult the LLM if the situation is ambiguous.

Routing Functions: Patterns and Best Practices

Pattern 1: Pure routing functions

A routing function should be pure: read the state, return a string. No side effects.

# ✅ Pure — only reads state
def route_pure(state: AgentState) -> str:
    if state["iteration_count"] >= state["max_iterations"]:
        return "synthesize"
    return "research"

# ❌ Impure — makes an API call inside the router
def route_impure(state: AgentState) -> str:
    result = expensive_api_call(state["messages"])
    if result.score > 0.8:
        return "synthesize"
    return "research"

If you need complex logic (like an LLM router), put it in a node that writes to the state, and make the routing function only read that result.

Pattern 2: Logging decisions

In production you need to know why the agent took each path. Logging is the only acceptable side effect in a routing function:

import logging
logger = logging.getLogger("agent.routing")

def route_with_logging(state: AgentState) -> str:
    iteration = state.get("iteration_count", 0)
    confidence = state.get("confidence", 0.0)
    
    if iteration >= state.get("max_iterations", 10):
        decision, reason = "synthesize", f"max_iterations ({iteration})"
    elif confidence >= 0.8:
        decision, reason = "synthesize", f"confidence OK ({confidence:.2f})"
    else:
        decision, reason = "research", f"low confidence ({confidence:.2f})"
    
    logger.info(f"[Iter {iteration}] Route: {decision} | {reason}")
    return decision

Pattern 3: Handle edge cases and composition

Always include a default. For complex routing, split the conditions into testable functions that you compose in priority order (safety nets first):

def is_budget_exhausted(state): return state.get("budget_remaining", 1.0) <= 0
def is_max_iterations(state): return state.get("iteration_count", 0) >= state.get("max_iterations", 10)
def has_sufficient_data(state): return state.get("confidence", 0.0) >= 0.8

def route_composed(state: AgentState) -> str:
    if not state["messages"]: return "research"
    if is_budget_exhausted(state): return "synthesize"
    if is_max_iterations(state): return "synthesize"
    if has_sufficient_data(state): return "synthesize"
    return "research"

Comparison: Routing in Agents vs Routing in Workflows

In the LangChain guide (#9) you used conditional edges for workflows — classifying emails, routing queries. Routing for agents is fundamentally different:

The fundamental difference: in a workflow, routing happens at the start to classify. In an agent, routing happens on every iteration to adapt the behavior. Routing functions are the agent's adaptation mechanism.

Connection to the Project

In this module's project (capsule 08, Research Agent State Machine):

• Your Research Agent has planning, research, analysis, and synthesis nodes. The conditional edges between them implement this capsule's patterns: routing by tool results, by plan state, and by iterations.
• Routing between research and analysis is based on result quality. Routing to synthesis combines a complete plan + sufficient confidence, or max iterations.

In later modules:

• M5 (Planning): Routing by quality score after reflection. Low quality → research. High → synthesis.
• M6 (Memory): Routing can consider persisted history — "did we already research this?"
• M8 (Multi-Agent): The supervisor uses multi-destination routing where the destinations are agents.

Troubleshooting

Problem 1: The routing function returns a destination that doesn't exist

Symptom: ValueError: Expected one of ['research', 'synthesize', 'end'].

Solution: Every string your function returns must be in the add_conditional_edges mapping. Use Literal type hints to catch it early:

def my_router(state) -> Literal["research", "fallback"]:
    ...

graph.add_conditional_edges("reason", my_router, {
    "research": "research",
    "fallback": "fallback",
})

Problem 2: The agent always takes the same path

Cause: The conditions are in the wrong order, or the state field isn't being updated.

Solution: Add logging and check the values. The most common cause: the node that should update confidence doesn't return it in its dict, so the state never changes.

def route_debug(state: AgentState) -> str:
    print(f"[DEBUG] confidence={state.get('confidence')}, "
          f"iterations={state.get('iteration_count')}")
    # ... routing logic

Problem 3: Routing creates an infinite loop between two nodes

Cause: No condition leads to a terminal node. The nodes point at each other with no exit.

Solution: The safety nets (max iterations, budget) should always be evaluated first:

def route_safe(state: AgentState) -> str:
    if state["iteration_count"] >= state["max_iterations"]:
        return "synthesize"  # There's always an escape
    # ... rest of the logic

Problem 4: The LLM router returns unexpected values

Solution: Validate and apply a default — the model doesn't always respect the schema:

VALID_ROUTES = {"research", "analyze", "synthesize"}

def route_from_llm(state: AgentState) -> str:
    decision = state.get("router_decision", "research")
    return decision if decision in VALID_ROUTES else "research"

Problem 5: Conditional edges don't show up in draw_mermaid_png

Cause: You used add_edge (singular, direct) instead of add_conditional_edges (conditional).

# ❌ Direct edge
graph.add_edge("tools", "reason")

# ✅ Conditional edges
graph.add_conditional_edges("tools", route_fn, {
    "reason": "reason", "analyze": "analyze", "synthesize": "synthesize",
})

Exercises

Exercise 1: Identify the routing destinations (Easy)

List every possible destination and the condition that triggers each one:

def route_agent(state):
    if state["iteration_count"] >= 10:
        return "synthesize"
    if state.get("error_count", 0) >= 3:
        return "error_recovery"
    last = state["messages"][-1]
    if not last.tool_calls:
        return "synthesize"
    if any(tc["name"] == "search" for tc in last.tool_calls):
        return "search_tools"
    return "general_tools"

Exercise 2: Implement routing by confidence (Medium)

Implement a StateGraph with 3 nodes (assess, research, synthesize). assess generates a random confidence. Routing: < 0.6 → research, >= 0.6 → synthesize. research goes back to assess. Include max_iterations.

import random
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import SystemMessage, HumanMessage

class ConfState(TypedDict):
    messages: Annotated[list, add_messages]
    confidence: float
    iteration_count: int
    max_iterations: int

def assess_node(state: ConfState) -> dict:
    confidence = random.uniform(0.3, 0.95)
    iteration = state.get("iteration_count", 0) + 1
    print(f"[Assess] Iter {iteration} | Confidence: {confidence:.2f}")
    return {"confidence": confidence, "iteration_count": iteration}

def research_node(state: ConfState) -> dict:
    return {"messages": [SystemMessage(content="Research complete.")]}

def synthesize_node(state: ConfState) -> dict:
    return {"messages": [SystemMessage(content="Answer synthesized.")]}

def route_confidence(state: ConfState) -> str:
    if state.get("iteration_count", 0) >= state.get("max_iterations", 8):
        return "synthesize"
    if state.get("confidence", 0.0) >= 0.6:
        return "synthesize"
    return "research"

graph = StateGraph(ConfState)
graph.add_node("assess", assess_node)
graph.add_node("research", research_node)
graph.add_node("synthesize", synthesize_node)
graph.add_edge(START, "assess")
graph.add_conditional_edges("assess", route_confidence, {
    "research": "research", "synthesize": "synthesize",
})
graph.add_edge("research", "assess")
graph.add_edge("synthesize", END)

agent = graph.compile()
result = agent.invoke({
    "messages": [HumanMessage(content="Research LangGraph")],
    "confidence": 0.0, "iteration_count": 0, "max_iterations": 8,
})
print(f"Iterations: {result['iteration_count']}, confidence: {result['confidence']:.2f}")

Exercise 3: Routing by plan status (Medium)

Implement a routing function for a plan with 3 sub-questions: all completed → "synthesize", 2+ pending → "parallel_research", 1 pending → "deep_research", max_iterations override → "synthesize".

def route_by_plan_status(state) -> str:
    if state.get("iteration_count", 0) >= state.get("max_iterations", 10):
        return "synthesize"
    
    pending = [s for s in state.get("plan", []) if s["status"] == "pending"]
    
    if len(pending) == 0:
        return "synthesize"
    elif len(pending) >= 2:
        return "parallel_research"
    else:
        return "deep_research"

# Test
cases = [
    ([], 2, "synthesize"),
    ([{"status": "pending"}, {"status": "pending"}], 2, "parallel_research"),
    ([{"status": "pending"}], 2, "deep_research"),
    ([{"status": "pending"}, {"status": "pending"}], 10, "synthesize"),
]
for pending_items, iters, expected in cases:
    state = {"plan": pending_items, "iteration_count": iters, "max_iterations": 10, "messages": []}
    assert route_by_plan_status(state) == expected
    print(f"{len(pending_items)} pending, iter {iters} → {expected} ✓")

Exercise 4: Graph with multi-destination routing (Hard)

Implement a StateGraph with: reason, web_search, db_query, analyze, synthesize, error_handler. The routing function routes to different destinations based on the requested tool call. Include return edges and draw_mermaid_png.

from langgraph.graph import StateGraph, START, END

class MultiState(TypedDict):
    messages: Annotated[list, add_messages]
    iteration_count: int
    max_iterations: int
    error_count: int

def route_after_reason(state: MultiState) -> str:
    if state.get("iteration_count", 0) >= state.get("max_iterations", 8):
        return "synthesize"
    if state.get("error_count", 0) >= 3:
        return "error_handler"
    last = state["messages"][-1]
    if not hasattr(last, "tool_calls") or not last.tool_calls:
        return "synthesize"
    tool_names = {tc["name"] for tc in last.tool_calls}
    if "search_web" in tool_names: return "web_search"
    if "query_database" in tool_names: return "db_query"
    if "analyze_data" in tool_names: return "analyze"
    return "synthesize"

graph = StateGraph(MultiState)
graph.add_node("reason", reason)
graph.add_node("web_search", web_search_node)
graph.add_node("db_query", db_query_node)
graph.add_node("analyze", analyze_node)
graph.add_node("synthesize", synthesize)
graph.add_node("error_handler", error_handler)

graph.add_edge(START, "reason")
graph.add_conditional_edges("reason", route_after_reason, {
    "web_search": "web_search", "db_query": "db_query",
    "analyze": "analyze", "synthesize": "synthesize",
    "error_handler": "error_handler",
})
graph.add_edge("web_search", "reason")
graph.add_edge("db_query", "reason")
graph.add_edge("analyze", "reason")
graph.add_edge("error_handler", "synthesize")
graph.add_edge("synthesize", END)

agent = graph.compile()
display(Image(agent.get_graph().draw_mermaid_png()))

Exercise 5: Hybrid routing — deterministic + LLM (Hard)

Implement routing where the safety nets are deterministic, but in the "ambiguous zone" (iterations 3-7, no errors) an LLM decides between "research", "analyze", or "synthesize" using with_structured_output.

from pydantic import BaseModel, Field

class RouteDecision(BaseModel):
    next_step: Literal["research", "analyze", "synthesize"] = Field(
        description="research=search for more, analyze=go deeper, synthesize=answer"
    )

router_model = model.with_structured_output(RouteDecision)

def llm_router_node(state) -> dict:
    decision = router_model.invoke(
        state["messages"] + [
            SystemMessage(content=f"Iteration {state['iteration_count']}. Decide the next step.")
        ]
    )
    return {"llm_route": decision.next_step}

def hybrid_route(state) -> str:
    if state.get("iteration_count", 0) >= state.get("max_iterations", 10):
        return "synthesize"
    if state.get("error_count", 0) >= 3:
        return "synthesize"
    last = state["messages"][-1]
    if not hasattr(last, "tool_calls") or not last.tool_calls:
        return "synthesize"
    if state.get("iteration_count", 0) <= 2:
        return "research"
    return "llm_decide"

def route_from_llm(state) -> str:
    valid = {"research", "analyze", "synthesize"}
    decision = state.get("llm_route", "research")
    return decision if decision in valid else "research"

graph = StateGraph(HybridState)
graph.add_node("reason", reason_node)
graph.add_node("llm_router", llm_router_node)
graph.add_node("research", research_node)
graph.add_node("analyze", analyze_node)
graph.add_node("synthesize", synthesize_node)

graph.add_edge(START, "reason")
graph.add_conditional_edges("reason", hybrid_route, {
    "research": "research", "synthesize": "synthesize", "llm_decide": "llm_router",
})
graph.add_conditional_edges("llm_router", route_from_llm, {
    "research": "research", "analyze": "analyze", "synthesize": "synthesize",
})
graph.add_edge("research", "reason")
graph.add_edge("analyze", "reason")
graph.add_edge("synthesize", END)

Summary

In this capsule you learned:

• Conditional routing is the agent's decision layer. The nodes are "what it does" — the conditional edges are "what it decides to do next." Routing ≠ tool routing (M3). Here you decide which graph node the execution goes to.
• Routing by tool results: After running tools, route by the quality of the result — rich data → analyze, error → retry, no results → rephrase.
• Routing by confidence: The model assesses its own confidence with structured output. Low → research more. High → synthesize.
• Routing by task state: The routing reflects the plan's progress — pending sub-questions determine the next step.
• Multi-destination routing: 3+ possible destinations. Deterministic (if/else), based on tool calls, LLM-driven, or hybrid.
• Best practices: Pure routing functions, decision logging, explicit edge cases, composed conditions, safety nets always first.
• Agent vs workflow: In workflows, routing classifies once. In agents, it routes on every iteration — it's the adaptation mechanism.

Next capsule: Functional API for Agents — implementing the same routing and cycle patterns using @entrypoint and @task, with Python's if/else and while instead of add_conditional_edges.

Additional Resources

1. LangGraph Conditional Edges — Official documentation — Complete add_conditional_edges reference
2. LangGraph Routing — How-To Guide — Branching and routing patterns
3. Pydantic Structured Output — with_structured_output for LLM routing
4. LangGraph Visualization — Verify routing with draw_mermaid_png
5. ReAct Agent Routing — Routing patterns for agents