Conditional Edges and Routing

Conditional edges are what separate a graph from a pipeline. They let the agent make decisions at runtime — inspecting the current state and choosing which node to visit next. Without conditional edges, you have a linear sequence. With them, you have a true agent that can branch, route, and loop.

add_conditional_edges

The method signature:

Python

builder.add_conditional_edges(
    source_node,        # str: the node this edge leaves from
    routing_function,   # callable: takes state, returns a string key
    path_map,           # dict (optional): maps return values to node names
)

routing_function — a plain Python function that:

Receives the full state
Returns a string that identifies the next node

path_map — a dictionary mapping the router's return values to node names. If omitted, the return value is used directly as the node name.

A Simple Router

Python

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

class QueryState(TypedDict):
    query: str
    query_type: Optional[str]
    answer: Optional[str]

def classify(state: QueryState) -> dict:
    """Classify the query type."""
    q = state["query"].lower()
    if any(word in q for word in ["symptom", "drug", "dose", "medication", "treatment"]):
        return {"query_type": "medical"}
    elif any(word in q for word in ["law", "legal", "regulation", "contract"]):
        return {"query_type": "legal"}
    else:
        return {"query_type": "general"}

def route_by_type(state: QueryState) -> str:
    """Router function — returns the name of the next node."""
    qtype = state.get("query_type", "general")
    if qtype == "medical":
        return "medical_handler"
    elif qtype == "legal":
        return "legal_handler"
    else:
        return "general_handler"

def medical_handler(state: QueryState) -> dict:
    return {"answer": f"[Medical] Consulting clinical knowledge base for: {state['query']}"}

def legal_handler(state: QueryState) -> dict:
    return {"answer": f"[Legal] Reviewing regulatory context for: {state['query']}"}

def general_handler(state: QueryState) -> dict:
    return {"answer": f"[General] Here is what I know about: {state['query']}"}

builder = StateGraph(QueryState)
builder.add_node("classify", classify)
builder.add_node("medical_handler", medical_handler)
builder.add_node("legal_handler", legal_handler)
builder.add_node("general_handler", general_handler)

builder.add_edge(START, "classify")

builder.add_conditional_edges(
    "classify",
    route_by_type,
    {
        "medical_handler": "medical_handler",
        "legal_handler": "legal_handler",
        "general_handler": "general_handler",
    }
)

builder.add_edge("medical_handler", END)
builder.add_edge("legal_handler", END)
builder.add_edge("general_handler", END)

app = builder.compile()

result = app.invoke({"query": "What is the typical dose of aspirin?", "query_type": None, "answer": None})
print(result["answer"])
# [Medical] Consulting clinical knowledge base for: What is the typical dose of aspirin?

Omitting the path_map

If your router returns node names directly (not aliases), you can omit the path_map:

Python

def route_by_type(state: QueryState) -> str:
    qtype = state.get("query_type", "general")
    return f"{qtype}_handler"   # Returns "medical_handler", "legal_handler", etc.

# No path_map needed — return value IS the node name
builder.add_conditional_edges("classify", route_by_type)

Use path_map when:

You want to decouple router logic from node naming
Your router returns short codes like "A", "B" that map to verbose node names
You route to END (which requires special handling, shown next)

Routing to END

To conditionally terminate execution, import END and include it in your path_map:

Python

from langgraph.graph import END

def route_or_finish(state: QueryState) -> str:
    if state.get("answer"):
        return "done"        # maps to END
    return "retry"           # maps back to generate

builder.add_conditional_edges(
    "generate",
    route_or_finish,
    {
        "done": END,         # "done" maps to END sentinel
        "retry": "generate", # "retry" loops back
    }
)

The END sentinel tells LangGraph to stop execution. You can route to END from any node using a conditional edge — you are not limited to terminal nodes.

Full Example: Medical Query Router

This example builds a realistic router that sends medical queries to a specialist node and general queries to a default handler, with validation and retry logic:

Python

import os
from typing import TypedDict, Optional, Annotated
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, SystemMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

os.environ["OPENAI_API_KEY"] = "sk-..."

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

class MedicalRouterState(TypedDict):
    query: str
    query_category: Optional[str]       # "drug_interaction", "dosage", "symptoms", "general"
    specialist_required: bool
    retrieved_context: Optional[str]
    answer: Optional[str]
    confidence: Optional[str]           # "high", "medium", "low"
    needs_retry: bool
    retry_count: int
    messages: Annotated[list[BaseMessage], add_messages]

# ── Node: Classify the query ──────────────────────────────────────────────────

def classify_medical_query(state: MedicalRouterState) -> dict:
    """Use an LLM to classify the query into a medical category."""
    system = (
        "Classify the medical query into exactly one of these categories:\n"
        "- drug_interaction: about two or more drugs taken together\n"
        "- dosage: about how much of a drug to take\n"
        "- symptoms: about medical symptoms or conditions\n"
        "- general: anything else\n"
        "Reply with only the category name."
    )
    response = llm.invoke([
        SystemMessage(content=system),
        HumanMessage(content=state["query"]),
    ])
    category = response.content.strip().lower()
    if category not in ("drug_interaction", "dosage", "symptoms", "general"):
        category = "general"

    specialist_required = category in ("drug_interaction", "dosage")

    print(f"[classify] Category: {category} | Specialist: {specialist_required}")

    return {
        "query_category": category,
        "specialist_required": specialist_required,
        "messages": [HumanMessage(content=state["query"])],
    }

# ── Router function ───────────────────────────────────────────────────────────

def route_after_classify(state: MedicalRouterState) -> str:
    """Route to specialist or general handler based on classification."""
    if state.get("specialist_required"):
        return "specialist_node"
    return "general_node"

# ── Node: General medical handler ─────────────────────────────────────────────

def general_node(state: MedicalRouterState) -> dict:
    """Handle general medical queries with a standard LLM response."""
    response = llm.invoke([
        SystemMessage(content=(
            "You are a helpful medical information assistant. Provide clear, accurate information. "
            "Always recommend consulting a healthcare provider for medical decisions."
        )),
        HumanMessage(content=state["query"]),
    ])
    answer = response.content

    print(f"[general] Answer: {len(answer)} chars")
    return {
        "answer": answer,
        "confidence": "medium",
        "messages": [AIMessage(content=answer)],
    }

# ── Node: Specialist handler ──────────────────────────────────────────────────

def specialist_node(state: MedicalRouterState) -> dict:
    """Handle drug interactions and dosage queries with enhanced precision."""
    category = state.get("query_category", "general")

    if category == "drug_interaction":
        system = (
            "You are a clinical pharmacologist. Analyze drug interactions with precision. "
            "Describe the mechanism, clinical significance (minor/moderate/major), and management. "
            "Include a safety disclaimer."
        )
    else:  # dosage
        system = (
            "You are a clinical pharmacist. Provide accurate dosing information. "
            "Include: standard adult dose, dosing frequency, max daily dose, "
            "and renal/hepatic adjustment notes if relevant. Include safety disclaimer."
        )

    response = llm.invoke([
        SystemMessage(content=system),
        HumanMessage(content=state["query"]),
    ])
    answer = response.content

    print(f"[specialist] Category: {category} | Answer: {len(answer)} chars")
    return {
        "answer": answer,
        "confidence": "high",
        "messages": [AIMessage(content=answer)],
    }

# ── Node: Validate answer quality ────────────────────────────────────────────

def validate_answer(state: MedicalRouterState) -> dict:
    """Check if the answer is adequate or needs retry."""
    answer = state.get("answer", "")

    # Simple heuristic: short answers likely indicate a problem
    is_adequate = len(answer) > 100 and "I don't know" not in answer.lower()

    needs_retry = not is_adequate and state.get("retry_count", 0) < 2

    print(f"[validate] Answer length: {len(answer)} | Adequate: {is_adequate} | Retry: {needs_retry}")

    return {
        "needs_retry": needs_retry,
        "retry_count": state.get("retry_count", 0) + (1 if needs_retry else 0),
    }

def route_after_validate(state: MedicalRouterState) -> str:
    """Route to retry classification or finish."""
    if state.get("needs_retry"):
        return "classify"      # retry from the top
    return END

# ── Build the graph ───────────────────────────────────────────────────────────

builder = StateGraph(MedicalRouterState)

builder.add_node("classify", classify_medical_query)
builder.add_node("general_node", general_node)
builder.add_node("specialist_node", specialist_node)
builder.add_node("validate", validate_answer)

# Entry
builder.add_edge(START, "classify")

# Routing after classify
builder.add_conditional_edges(
    "classify",
    route_after_classify,
    {
        "general_node": "general_node",
        "specialist_node": "specialist_node",
    }
)

# Both handlers converge to validate
builder.add_edge("general_node", "validate")
builder.add_edge("specialist_node", "validate")

# After validate, finish or retry
builder.add_conditional_edges(
    "validate",
    route_after_validate,
    {
        "classify": "classify",
        END: END,
    }
)

app = builder.compile()

# ── Test it ───────────────────────────────────────────────────────────────────

def run_query(query: str):
    result = app.invoke({
        "query": query,
        "query_category": None,
        "specialist_required": False,
        "retrieved_context": None,
        "answer": None,
        "confidence": None,
        "needs_retry": False,
        "retry_count": 0,
        "messages": [],
    })
    print(f"\nQuery: {query}")
    print(f"Category: {result['query_category']}")
    print(f"Confidence: {result['confidence']}")
    print(f"Answer:\n{result['answer']}\n")
    print("=" * 60)

run_query("Can I take ibuprofen and aspirin together?")
run_query("What are the symptoms of type 2 diabetes?")
run_query("What is the maximum daily dose of metformin?")

Multiple Conditional Edges from the Same Node

A node can have multiple conditional edges in different directions — but in practice each node has exactly one add_conditional_edges call (since the router handles all branches). The pattern for multiple exit conditions is always the same:

Python

def complex_router(state: MyState) -> str:
    if condition_a(state):
        return "node_a"
    elif condition_b(state):
        return "node_b"
    elif some_error(state):
        return "error_node"
    return END

builder.add_conditional_edges(
    "source",
    complex_router,
    {
        "node_a": "node_a",
        "node_b": "node_b",
        "error_node": "error_node",
        END: END,
    }
)

All routing logic lives in the router function. The path_map is just a translation table.

Summary

| Concept | Code | |---|---| | Add conditional edge | add_conditional_edges(from, router_fn, path_map) | | Router function | def router(state) -> str | | Route to END | Include END: END in path_map | | Omit path_map | When router returns node names directly | | Multiple branches | All handled inside a single router function |

Conditional edges are what make LangGraph agents agents. They give your graph the ability to make decisions — routing based on LLM output, state flags, scores, or any Python logic you can write.

Conditional Edges and Routing

Conditional Edges and Routing

add_conditional_edges

A Simple Router

Omitting the path_map

Routing to END

Full Example: Medical Query Router

Multiple Conditional Edges from the Same Node

Summary

Enjoyed this article?

Leave a comment