01Real-world example
Conditional routing
A router function reads the state and returns the name of the next node. Decisions live in that one function.
02The graph
03How the code does it
route_by_status()is a plain function: it readsstate["status"]and returns the name of the next node.add_conditional_edges("read_result", route_by_status, [...])lists every possible destination, which also lets LangGraph draw the fork.- Anything that is not passed or failed falls through to
quarantine, because of the.get(..., "quarantine")default.
Key points
- The router returns a node's name
- The diamond is a decision, not a node
- This is where an agent makes choices
04Run it
terminal
cd chapter_18_LangGraph/src/chapters
python 003_Conditional_Edge.pyoutput
cart_total passed -> logged as green
checkout_pay failed -> Jira bug filed for checkout_pay
login_redirect flaky -> login_redirect moved to quarantine, rerun nightlyDDrills for lesson 3
Work in chapter_18_LangGraph/src/chapters. Change one thing, run it, and compare with the expected result.
- Route an unknown status
Invoke the app with
status="skipped". Where does it go, and why?Expected result
x moved to quarantine, rerun nightly: the router's default sends anything unknown to quarantine. - Give skipped tests their own branch
Add a
log_skipnode, route"skipped"to it, and add it to the destinations list.Hint
Three edits: the router dict,add_nodeplusadd_edge(..., END), and the destinations list.Expected result
Skipped tests reachlog_skip; quarantine now only catches statuses you never planned for. - Return a node that does not exist
Make the router return
"nope". What happens at run time?Expected result
KeyError: 'nope'. The router can only return declared destinations.
SThe full script
The exact file from the course repo, plus the output it printed when this page was written.
003_Conditional_Edge.py
"""003 - A fork in the road: route a test result to the right next step.
add_conditional_edges(source, router, destinations)
The router is a normal function that reads state and returns the NAME of the
next node. That one function is where an "agent" makes decisions.
"""
from typing import Literal, TypedDict
from langgraph.graph import END, START, StateGraph
class State(TypedDict, total=False):
test_name: str
status: str # passed | failed | flaky
action: str
def read_result(state: State) -> dict:
return {} # in real life: parse a Playwright report here
def route_by_status(state: State) -> Literal["log_pass", "file_bug", "quarantine"]:
return {"passed": "log_pass", "failed": "file_bug"}.get(state["status"], "quarantine")
def log_pass(state: State) -> dict:
return {"action": "logged as green"}
def file_bug(state: State) -> dict:
return {"action": f"Jira bug filed for {state['test_name']}"}
def quarantine(state: State) -> dict:
return {"action": f"{state['test_name']} moved to quarantine, rerun nightly"}
builder = StateGraph(State)
builder.add_node("read_result", read_result)
builder.add_node("log_pass", log_pass)
builder.add_node("file_bug", file_bug)
builder.add_node("quarantine", quarantine)
builder.add_edge(START, "read_result")
builder.add_conditional_edges("read_result", route_by_status,
["log_pass", "file_bug", "quarantine"])
for leaf in ("log_pass", "file_bug", "quarantine"):
builder.add_edge(leaf, END)
app = builder.compile()
if __name__ == "__main__":
for name, status in [("cart_total", "passed"),
("checkout_pay", "failed"),
("login_redirect", "flaky")]:
out = app.invoke({"test_name": name, "status": status})
print(f"{name:<15} {status:<7} -> {out['action']}")
print("\n" + app.get_graph().draw_mermaid())Expected output
output
cart_total passed -> logged as green
checkout_pay failed -> Jira bug filed for checkout_pay
login_redirect flaky -> login_redirect moved to quarantine, rerun nightly