06. Conditional Edges
Conditional edges route execution based on the current state rather than running the same next node every time. They require two things: a routing function that receives the state and returns a string (node name), and a mapping dict from possible return values to destination nodes.
from typing import Literal
def route_based_on_confidence(state: AgentState) -> Literal["high_confidence_node", "low_confidence_node"]:
confidence = state.get("confidence", 0.5)
if confidence >= 0.7:
return "high_confidence_node"
return "low_confidence_node"
builder.add_conditional_edges(
"evaluator",
route_based_on_confidence,
{
"high_confidence_node": "high_confidence_node",
"low_confidence_node": "low_confidence_node",
}
)
The routing function can return any string—it does not have to match a defined node name, but if it returns a name with no node attached, compile() raises InvalidNodeError. Multiple conditional edges can originate from the same node, each with its own routing function—this is how you implement a dispatcher that looks at one field and routes to many possible workers.
A common pattern is to combine conditional edges with a routing function that returns END to conditionally terminate:
def route_or_stop(state: AgentState) -> Literal["continue", "__end__"]:
if len(state["messages"]) > 10:
return "__end__"
return "continue"
A failure mode: conditional edge routing functions must be pure—reading from the state is fine, but writing to it (mutating the state dict) is not and causes non-deterministic behavior. Also, routing functions that raise exceptions leave the graph in an undefined state. Always return a valid node name or __end__.
Local verification checkpoint
Run the smallest example from this chapter in a local workspace and record the package version, runtime, data path, and observed output. If the result depends on model size, vector count, CPU/GPU backend, or available memory, note that constraint beside the exercise so the lesson remains reproducible.
Build a three-step pipeline: start → conditional branch to path_a or path_b (based on whether a mode field is "fast" or "deep") → aggregator → END. Run both branches from a single invoke() call and confirm the correct path executes each time.