Using LangGraph Conditional Edges to Enable Automatic AI Decision Routing

1. Conditional Edge vs Normal Edge

Normal edges perform a fixed jump and are suitable for simple linear processes, while conditional edges enable dynamic routing for branching decisions. The article illustrates the difference with a diagram where a conditional edge splits from node B to either D or E based on runtime conditions.

2. Minimal Example: Router Function

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

class RoutingState(TypedDict):
    input: str
    route: str

def router(state: RoutingState) -> Literal["path_a", "path_b", "__end__"]:
    """Decide routing based on input"""
    if "紧急" in state["input"]:
        return "path_a"
    elif "普通" in state["input"]:
        return "path_b"
    return END

def path_a(state: RoutingState) -> dict:
    return {"route": "urgent_handler"}

def path_b(state: RoutingState) -> dict:
    return {"route": "normal_handler"}

3. add_conditional_edges Three Elements

graph.add_conditional_edges(
    source,          # 1. source node
    router,          # 2. routing function
    path_map         # 3. mapping of results to target nodes
)

4. Real‑world: RAG Retrieval Routing

def route_question(state: RAGState) -> Literal["vectorstore", "web_search", "llm_fallback"]:
    """Choose retrieval method based on question type"""
    question = state["question"]
    if any(kw in question for kw in ["最新", "今天", "新闻"]):
        return "web_search"   # needs internet
    elif len(question) < 10:
        return "llm_fallback"  # simple, answer directly
    return "vectorstore"      # normal retrieval

workflow.add_conditional_edges(
    START,
    route_question,
    {
        "web_search": "web_search",
        "vectorstore": "retrieve",
        "llm_fallback": "llm_fallback",
    }
)

5. Error Handling and Retry

Simple Retry Mode

class RetryState(TypedDict):
    value: int
    attempts: int

def might_fail(state: RetryState) -> dict:
    """Node that may fail"""
    if state["value"] < 0:
        raise ValueError("值不能为负数")
    return {"value": state["value"] * 2}

def retry_handler(state: RetryState) -> dict:
    """Retry handler"""
    return {"attempts": state["attempts"] + 1}

builder.add_edge(START, "action")
app = builder.compile()

Conditional Edge with Timeout

def decide_next(state: AgentState) -> str:
    if state["finished"]:
        return END
    if state["attempts"] >= 3:
        return "fallback"
    return "action"

6. Multi‑Level Conditional Nesting

# First layer: decide operation type
def route_type(state: State) -> Literal["read", "write", "delete"]:
    return state["operation"]
# Second layer: permission check
def route_permission(state: State) -> Literal["execute", "denied"]:
    if state.get("has_permission"):
        return "execute"
    return "denied"

builder.add_conditional_edges("operation", route_type, {...})
builder.add_conditional_edges("permission_check", route_permission, {...})

7. Day Recap

router function returns target node name or END

add_conditional_edges requires source, router, and path_map

Literal type restricts return values to allowed node names

END node marks workflow termination

Related Links

Official docs: https://langchain-ai.github.io/langgraph/concepts/low_level/#conditional-edges

GitHub repository: https://github.com/langchain-ai/langgraph