Chapter 18 . LangGraph . Lesson 4 of 11

Retries with a limit

An edge can point back to an earlier node. That makes a loop, and every loop needs a cap.

Next: Parallel branches and reducers All lessons Script on GitHub

01Real-world example

Illustration: Like Playwright's retries: rerun a failing test up to three times. A pass after a failure marks the test as flaky. — Like Playwright's retries: rerun a failing test up to three times. A pass after a failure marks the test as flaky.

02The graph

Retries with a limit as a graph — should_retry() sends a failed test back to run_test, three times at most.

03How the code does it

should_retry() returns "run_test" to loop back or "report" to stop: an edge that points backwards.
MAX_ATTEMPTS is the cap. Without it a broken test loops until the recursion limit, which is 10007 steps by default on langgraph 1.2.12.
scripted_outcomes makes the run repeatable: the same input gives the same verdict every time, so the demo can show flakiness on purpose.

Key points

A loop is an edge that points back
Keep the counter in the state
Set recursion_limit as a backstop

04Run it

terminal

cd chapter_18_LangGraph/src/chapters
python 004_Loop_Retry.py

output

login_redirect (flaky)
   attempt 1: failed
   attempt 2: failed
   attempt 3: passed
   => FLAKY: passed on attempt 3 after 2 failure(s)

DDrills for lesson 4

Work in chapter_18_LangGraph/src/chapters. Change one thing, run it, and compare with the expected result.

Lower the cap
Set MAX_ATTEMPTS = 2 and run the three scenarios again. Which verdict changes?

Expected result
login_redirect becomes REAL FAILURE: failed all 2 attempts; it never gets its third, passing attempt.
Remove the cap
Make the router always retry on failure, give checkout_pay fifty failures, and invoke with {"recursion_limit": 10}.

Expected result
GraphRecursionError: Recursion limit of 10 reached without hitting a stop condition. Always cap loops yourself.
Test the graph with pytest
Write three tests: flaky passes on attempt 3, broken stops at the cap, stable runs once.

Hint
Import the script with importlib.import_module("004_Loop_Retry"); module names can start with digits that way.

Expected result
See the Solution tab on the chapter page: three tests, all passing.

SThe full script

The exact file from the course repo, plus the output it printed when this page was written.

004_Loop_Retry.py

"""004 - Loops: an edge that points backwards. This is what LangFlow can't do easily.

A retry loop for a test, with a hard cap. The router either sends us back to
run_test again or forward to the report. The "attempt outcomes" are scripted so
the demo is repeatable: a flaky test and a genuinely broken one.
"""
from typing import Literal, TypedDict

from langgraph.graph import END, START, StateGraph

MAX_ATTEMPTS = 3


class State(TypedDict):
    test_name: str
    scripted_outcomes: list[str]   # what each attempt will return (simulation)
    attempts: int
    history: list[str]
    verdict: str


def run_test(state: State) -> dict:
    n = state["attempts"]
    outcome = state["scripted_outcomes"][n]
    print(f"   attempt {n + 1}: {outcome}")
    return {"attempts": n + 1, "history": state["history"] + [outcome]}


def should_retry(state: State) -> Literal["run_test", "report"]:
    if state["history"][-1] == "passed":
        return "report"
    if state["attempts"] < MAX_ATTEMPTS:
        return "run_test"          # <- the loop
    return "report"


def report(state: State) -> dict:
    h = state["history"]
    if h[-1] == "passed" and "failed" in h:
        verdict = f"FLAKY: passed on attempt {len(h)} after {h.count('failed')} failure(s)"
    elif h[-1] == "passed":
        verdict = "STABLE PASS"
    else:
        verdict = f"REAL FAILURE: failed all {len(h)} attempts"
    return {"verdict": verdict}


builder = StateGraph(State)
builder.add_node("run_test", run_test)
builder.add_node("report", report)
builder.add_edge(START, "run_test")
builder.add_conditional_edges("run_test", should_retry, ["run_test", "report"])
builder.add_edge("report", END)

app = builder.compile()

if __name__ == "__main__":
    scenarios = {
        "login_redirect (flaky)": ["failed", "failed", "passed"],
        "checkout_pay (broken)": ["failed", "failed", "failed"],
        "cart_total (stable)": ["passed"],
    }
    for name, outcomes in scenarios.items():
        print(name)
        out = app.invoke({"test_name": name, "scripted_outcomes": outcomes,
                          "attempts": 0, "history": [], "verdict": ""})
        print("   =>", out["verdict"], "\n")

Expected output

output

login_redirect (flaky)
   attempt 1: failed
   attempt 2: failed
   attempt 3: passed
   => FLAKY: passed on attempt 3 after 2 failure(s)