When you build genuinely complex AI agents, older frameworks (LangChain Chains and the like) show their limits: they only support simple one-way linear flows. If the agent needs non-linear loops, has to test its own code, branch on the result of a trial run, or pause for human approval, you need a directed state graph.

The answer is LangGraph — a library for building AI agents as stateful graphs. This lesson digs into how the node-and-edge structure works, how shared state is managed, how the human-in-the-loop interrupt works, and building an agent that writes code, tests it, and fixes itself in a cycle.

✅ What you need before starting
Software: Ollama with a chat model (ollama pull qwen2.5:7b). You do not need to install LangGraph — see the callout immediately below.

Knowledge you need: Lesson 17 for the agent loop and its three safety nets. This lesson replaces that linear for loop with a graph that branches and cycles.
⚡ The project does NOT use the LangGraph library
Stated up front so you do not go hunting for an import langgraph that is not there: the project reimplements the three ideas behind LangGraph using the Python standard library — state with reducers, conditional edges that create a cycle, and an interrupt that returns a checkpoint. The reason is the same as every other project in this series: a library hides exactly what you need to see, and all three ideas fit in under 100 lines.

For production, use LangGraph — it handles checkpoint persistence, parallel branches and streaming for you. But after this lesson you will know precisely what it is doing underneath.

18.1 Why graph-based agents exist

A real AI workflow rarely runs in a straight line. Take writing code: the model writes code ➔ it gets compiled ➔ if it fails, go back and fix it with the error attached ➔ if it succeeds, move on to storage. That is a graph containing a cycle.

LangGraph models an agent as a graph:

  • Nodes: processing steps — Python functions that take the current state, do something (call the LLM, run code) and return an update.
  • Edges: where to go next. Two kinds: fixed edges (always A to B) and conditional edges, which branch on the logical result of the node that just ran.
⚠️ Pitfall: the spaghetti graph
When designing an agent graph it is easy to end up drawing too many cross-linked nodes, producing a tangle that is extremely hard to debug and whose state is impossible to reason about. The clean design rule: split a huge graph into independent sub-graphs, each responsible for exactly one specialised task.

18.2 Managing shared state

The heart of LangGraph is the state object: a shared data structure (a Pydantic model or a TypedDict) threaded through every node in the graph.

  • State synchronisation: each node returns the fields it updated. The system writes those into the shared state, either overwriting or accumulating them via a reducer.
  • Reducer functions: these define how a particular field is merged. For chat history, for instance, an operator.add reducer appends new messages onto the existing list instead of replacing it.
🔢 Working the reducer out by hand
Suppose the history field currently holds ["Xin chào"] and a node has just returned ["Tôi cần giúp đỡ"].
  • With an operator.add reducer the result is ["Xin chào", "Tôi cần giúp đỡ"] — the old history is preserved and the new message is appended.
  • With NO reducer (the default overwrite) the result is only ["Tôi cần giúp đỡ"] — the entire earlier conversation is wiped instantly.
This is why declaring the right reducer per field is mandatory: overwrite is the default, so any field that needs to accumulate over time (chat history, an error log) must declare a reducer explicitly, or important data silently disappears after every node run.

18.3 Human-in-the-loop approval

In a corporate production environment, letting AI act 100% autonomously on consequential actions (sending customer email, paying invoices, writing code straight onto a live server) is extremely dangerous.

The human-in-the-loop mechanism places an interrupt just before the graph reaches a sensitive node. The graph freezes its current state (state persistence), the system pauses and notifies an administrator. Once a human approves — or edits the state directly — the graph resumes from the interrupt.

⚡ Safety boundary: only approve after verification
Human-in-the-loop is the last line of defence against an LLM losing control. A well-built code-writing agent must include a trial run in an isolated sandbox to catch its own errors before asking a human to approve anything.

18.4 Lesson 18 project: an agent that writes, tests and fixes itself

The project builds a stateful graph engine in plain Python, running this flow:

Start ➔ Coder node (writes a Python function) ➔ Tester node (runs it in a separate process) ➔ conditional edge: on failure go back to the coder with the error log (at most 3 times), on success continue to the approval interrupt.

langgraph_agent.py 
"""Lesson 18 project: a stateful agent graph, built on LangGraph's ideas.

Run:  python3 langgraph_agent.py
Needs: Ollama with a chat model (`ollama pull qwen2.5:7b`).

This file does NOT import langgraph. Like every project in this series it uses
the standard library only, and reimplements the three ideas the lesson is about
so you can see them working rather than trust a library:

  * a State object whose fields are merged through per-field reducers,
  * conditional edges that make the graph cyclic (coder -> tester -> coder),
  * a real interrupt: the graph stops and hands back a checkpoint, instead of
    blocking on input() inside a node.

The generated code is executed in a separate process with a timeout, never with
exec() in this one - see the note in run_generated_code.
"""

import json
import operator
import subprocess
import sys
import tempfile
import textwrap
import re
import urllib.error
import urllib.request

OLLAMA = "http://localhost:11434"
CHAT_PREFERRED = ["qwen2.5:14b-instruct", "qwen2.5:7b", "qwen2.5-coder:7b",
                  "llama3.1", "llama3.2"]
MAX_RETRIES = 3


# ---------------------------------------------------------------------------
# 1. State, and the reducers that merge updates into it
# ---------------------------------------------------------------------------

# The whole point of a reducer: it says HOW a field is merged, not just what it
# holds. Fields absent from this map are overwritten, which is the default in
# LangGraph too - and the reason an unreduced history field silently vanishes.
REDUCERS = {
    "history": operator.add,
    "test_log": operator.add,
}


def merge(state, update):
    """Apply one node's return value to the state, field by field."""
    merged = dict(state)
    for key, value in update.items():
        reducer = REDUCERS.get(key)
        merged[key] = reducer(merged.get(key, type(value)()), value) if reducer else value
    return merged


def initial_state(task, expected=None):
    return {"task": task, "code": "", "history": [], "test_log": [],
            "attempts": 0, "approved": None, "expected": expected}


def demo_reducers():
    """Show the difference the reducer map makes, on one field."""
    print("=== What a reducer changes ===")
    state = {"history": ["Xin chào"], "code": "v1"}
    update = {"history": ["Tôi cần giúp đỡ"], "code": "v2"}
    print(f"  state  : {state}")
    print(f"  update : {update}")
    print(f"  merged : {merge(state, update)}")
    print("  'history' has a reducer, so it appends. 'code' has none, so it is")
    print("  overwritten. Forget the reducer and the chat history disappears")
    print("  one node at a time, with no error anywhere.\n")


# ---------------------------------------------------------------------------
# 2. Talking to Ollama - the pattern from Lesson 13
# ---------------------------------------------------------------------------


def installed_models():
    try:
        with urllib.request.urlopen(f"{OLLAMA}/api/tags", timeout=5) as response:
            return [m["name"] for m in json.loads(response.read())["models"]]
    except urllib.error.URLError:
        return []


def pick(names, preferred):
    for wanted in preferred:
        for name in names:
            if name == wanted or name.startswith(wanted):
                return name
    return None


def chat(prompt, model):
    payload = {"model": model, "stream": False, "options": {"temperature": 0.0},
               "messages": [{"role": "user", "content": prompt}]}
    request = urllib.request.Request(
        f"{OLLAMA}/api/chat", data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(request) as response:
            return json.loads(response.read())["message"]["content"]
    except urllib.error.HTTPError as exc:
        detail = json.loads(exc.read() or b"{}").get("error", "no detail")
        raise RuntimeError(f"Ollama answered HTTP {exc.code}: {detail}") from None
    except urllib.error.URLError as exc:
        raise RuntimeError(f"cannot reach Ollama at {OLLAMA} - {exc.reason}") from None


# ---------------------------------------------------------------------------
# 3. Running model-written code without trusting it
# ---------------------------------------------------------------------------


def run_generated_code(code, timeout=5):
    """Execute the model's code in a SEPARATE process, with a time limit.

    Lesson 17 established that you never eval() model output. Running a whole
    generated program is the same problem, larger: exec(code, {}) in this
    process shares the interpreter, the filesystem and the network with the
    agent. A subprocess can at least be killed on timeout and cannot corrupt
    the parent's state. It is still not a security boundary - for untrusted
    input you need a container - but it is the minimum that is honest.
    """
    harness = textwrap.dedent("""
        import json, sys
        {code}
        try:
            print("RESULT:" + json.dumps(solve_problem()))
        except NameError:
            print("FAIL:no function named solve_problem")
        except Exception as exc:
            print("FAIL:" + type(exc).__name__ + ": " + str(exc))
    """).format(code=code)
    with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False,
                                     encoding="utf-8") as handle:
        handle.write(harness)
        path = handle.name
    try:
        finished = subprocess.run([sys.executable, path], capture_output=True,
                                  text=True, timeout=timeout)
    except subprocess.TimeoutExpired:
        return False, f"timed out after {timeout}s"
    output = (finished.stdout or finished.stderr).strip().splitlines()
    last = output[-1] if output else "no output"
    if last.startswith("RESULT:"):
        return True, last[len("RESULT:"):]
    return False, last.removeprefix("FAIL:")


# ---------------------------------------------------------------------------
# 4. The nodes
# ---------------------------------------------------------------------------


def coder_node(state, model):
    print("  [coder] writing solve_problem()")
    prompt = (
        "Bạn là lập trình viên Python. Viết duy nhất một hàm tên solve_problem() "
        "giải quyết yêu cầu sau, và hàm phải RETURN kết quả chứ không print.\n"
        f"Yêu cầu: {state['task']}\n"
        "Chỉ trả về code Python trong khối ```python ... ```, không giải thích."
    )
    if state["test_log"]:
        prompt += f"\nLần chạy trước lỗi: {state['test_log'][-1]}\nHãy sửa lại."
    content = chat(prompt, model)
    match = re.search(r"```(?:python)?(.*?)```", content, re.DOTALL)
    code = (match.group(1) if match else content).strip()
    return {"code": code, "attempts": state["attempts"] + 1,
            "history": [f"coder wrote {len(code)} characters"]}


def tester_node(state):
    """Run the code AND check the answer.

    Checking only that the code ran is the weak test Lesson 10 warned about:
    a function that returns the wrong number still "passes". The expected
    value lives in the state, so the tester compares against it.
    """
    print("  [tester] running it in a subprocess")
    ok, detail = run_generated_code(state["code"])
    if ok and state.get("expected") is not None:
        if detail.strip() != json.dumps(state["expected"]):
            ok, detail = False, f"returned {detail}, expected {state['expected']}"
    print(f"  [tester] {'passed' if ok else 'failed'}: {detail}")
    return {"test_log": [f"{'SUCCESS' if ok else 'FAIL'}: {detail}"],
            "history": [f"tester reported {'success' if ok else 'failure'}"]}


def approval_node(state, decision):
    """The human turn. It takes the decision as an argument, never input()."""
    return {"approved": decision == "y",
            "history": [f"human answered {decision!r}"]}


# ---------------------------------------------------------------------------
# 5. The graph engine: conditional edges, cycles, and one real interrupt
# ---------------------------------------------------------------------------


def route_after_tester(state):
    """A conditional edge. Three outcomes, so the graph is cyclic."""
    if state["test_log"][-1].startswith("SUCCESS"):
        return "approval"
    if state["attempts"] >= MAX_RETRIES:
        return "give_up"
    return "coder"


def demo_retry_cycle():
    """Feed deliberately broken code through tester + router, no model needed.

    This exercises the conditional edge itself: does a failing test really send
    the graph back to the coder, and does the attempt limit really stop it?
    """
    print("=== The conditional edge, driven with code that cannot pass ===")
    state = initial_state("demo", expected=2)
    state = merge(state, {"code": "def solve_problem():\n    return 999"})
    for _ in range(MAX_RETRIES + 1):
        state = merge(state, {"attempts": state["attempts"] + 1})
        state = merge(state, tester_node(state))
        destination = route_after_tester(state)
        print(f"  attempt {state['attempts']} -> route to {destination!r}")
        if destination != "coder":
            break
    print(f"  The edge returned to 'coder' {MAX_RETRIES - 1} times, then gave up")
    print(f"  at the limit of {MAX_RETRIES}. That is the cycle, and its brake.\n")


def run_until_interrupt(state, model):
    """Walk the graph and STOP before the approval node, returning a checkpoint.

    This is what human-in-the-loop means in practice: the process does not sit
    blocked on input(). It saves where it is and returns. A web backend can put
    that checkpoint in a database and resume it days later from a different
    machine - impossible if the node called input().
    """
    node = "coder"
    while True:
        if node == "coder":
            state = merge(state, coder_node(state, model))
            node = "tester"
        elif node == "tester":
            state = merge(state, tester_node(state))
            node = route_after_tester(state)
        elif node == "give_up":
            return state, "give_up"
        elif node == "approval":
            return state, "approval"


def resume(state, decision):
    """Continue from the checkpoint once a human has decided."""
    state = merge(state, approval_node(state, decision))
    return state


def main():
    demo_reducers()

    names = installed_models()
    if not names:
        print("Ollama is not reachable. Start it, then run this again.")
        return
    model = pick(names, CHAT_PREFERRED)
    if not model:
        print("No suitable chat model found. Try: ollama pull qwen2.5:7b")
        return
    print(f"model: {model}\n")

    demo_retry_cycle()

    task = "Trả về phần dư của phép chia 17 cho 5."
    print(f"=== Running the graph: {task} ===")
    state, stopped_at = run_until_interrupt(initial_state(task, expected=2), model)
    print(f"\n  graph paused at: {stopped_at}   after {state['attempts']} attempt(s)")
    print(f"  proposed code:\n{textwrap.indent(state['code'], '    ')}")

    print("\n=== The checkpoint that gets handed to the human ===")
    checkpoint = json.dumps({k: v for k, v in state.items() if k != "code"},
                            ensure_ascii=False)
    print(f"  {checkpoint}")
    print("  Serialisable, so it can go in a database and be resumed later.\n")

    for decision in ("n", "y"):
        final = resume(state, decision)
        verdict = "approved" if final["approved"] else "rejected"
        print(f"=== Resuming the same checkpoint with decision {decision!r} ===")
        print(f"  result: {verdict}")
        print(f"  history: {final['history'][-1]}")
    print("\n  Both runs start from the same checkpoint, so the human's answer")
    print("  is the only thing that differs. The graph itself never re-ran.")


if __name__ == "__main__":
    main()

Reducers: what decides whether history survives

Section 18.2 claims that without a reducer data silently disappears. The program demonstrates that on a single merge, with two fields: history has a reducer, code does not:

Terminal 
=== What a reducer changes ===
  state  : {'history': ['Xin chào'], 'code': 'v1'}
  update : {'history': ['Tôi cần giúp đỡ'], 'code': 'v2'}
  merged : {'history': ['Xin chào', 'Tôi cần giúp đỡ'], 'code': 'v2'}
  'history' has a reducer, so it appends. 'code' has none, so it is
  overwritten. Forget the reducer and the chat history disappears
  one node at a time, with no error anywhere.

This is the worst class of bug in a stateful system: no exception is raised, no warning appears. The graph keeps running, it simply erases its prior history at every node — and you only find out when the agent starts forgetting what it already tried.

Conditional edges: does the graph really cycle?

That question has to be answered by forcing it, not by waiting for the model to happen to write something wrong. The program injects code that is guaranteed to produce the wrong answer into the state and pushes it through the tester and the router:

Terminal 
=== The conditional edge, driven with code that cannot pass ===
  [tester] running it in a subprocess
  [tester] failed: returned 999, expected 2
  attempt 1 -> route to 'coder'
  [tester] running it in a subprocess
  [tester] failed: returned 999, expected 2
  attempt 2 -> route to 'coder'
  [tester] running it in a subprocess
  [tester] failed: returned 999, expected 2
  attempt 3 -> route to 'give_up'
  The edge returned to 'coder' 2 times, then gave up
  at the limit of 3. That is the cycle, and its brake.

The first two times the conditional edge returns 'coder' — the graph loops back to fix. On the third, having hit MAX_RETRIES, it returns 'give_up'. That is the cycle, and that is the cycle's brake. Without the brake the agent would keep fixing forever — precisely the pitfall from Lesson 17.

💡 A tester that only checks "did it run" is an empty test
The first version of this project only checked that solve_problem existed and did not raise. A function returning 999 passed cleanly. That is exactly the near-meaningless test Lesson 10 identified. The new version stores the expected value in the state and the tester compares against it — which is why the line failed: returned 999, expected 2 above means something.

A real run against the model

Terminal 
=== Running the graph: Trả về phần dư của phép chia 17 cho 5. ===
  [coder] writing solve_problem()
  [tester] running it in a subprocess
  [tester] passed: 2

  graph paused at: approval   after 1 attempt(s)
  proposed code:
    def solve_problem():
        return 17 % 5

=== The checkpoint that gets handed to the human ===
  {"task": "Trả về phần dư của phép chia 17 cho 5.", "history": ["coder wrote 38 characters", "tester reported success"], "test_log": ["SUCCESS: 2"], "attempts": 1, "approved": null, "expected": 2}
  Serialisable, so it can go in a database and be resumed later.

The interrupt: why you must not use input()

The first version called input() inside the approval node. Convenient, but it welds the agent to one running process and one keyboard with a person at it. It cannot run as a background job, it cannot run in CI, and if the administrator goes to lunch the process hangs all afternoon.

Real human-in-the-loop works differently: the graph stops and returns a checkpoint — a serialisable structure you can store in a database, push to a web interface, and resume three days later from a different machine. The program proves it by resuming the same checkpoint twice with two different decisions:

Terminal 
=== Resuming the same checkpoint with decision 'n' ===
  result: rejected
  history: human answered 'n'
=== Resuming the same checkpoint with decision 'y' ===
  result: approved
  history: human answered 'y'

  Both runs start from the same checkpoint, so the human's answer
  is the only thing that differs. The graph itself never re-ran.

The graph never re-ran: no second model call, no re-execution of the code. Only the human's decision changed. That is the property that makes checkpoints useful in production — and the thing input() can never give you.

⚠️ The heaviest pitfall: running code the model wrote
Lesson 17 just taught: never eval() model output. And now this lesson asks you to run an entire program the model wrote — the same problem, much larger. The first version used exec(state.code, local_env): the model's code ran in the same process as the agent, with the same filesystem and network access, and no way to stop it if it entered an infinite loop.

The new version writes the code to a temporary file and runs it via subprocess with a timeout. To be precise: this is still not a security boundary — the child process can still read your files and still reach the internet. It guarantees exactly two things: it can be killed when it hangs, and it cannot corrupt the parent's state. For code generated from untrusted input, the honest minimum is a separate container (Docker, gVisor) or a dedicated sandbox service.

How to run this project on your machine

  1. Start Ollama, pull a chat model, then run python3 langgraph_agent.py. The first two parts (reducers and the conditional edge) run even without Ollama.
  2. The sample task is easy enough that the model usually gets it right first time — which is why the cycle demonstration has to force the failure with broken code. To watch the agent genuinely repair itself, change task to something harder with a matching expected.
  3. Then try breaking it three ways:
    • Delete "history": operator.add from REDUCERS. Run again and look at the history field in the checkpoint: only the last line remains, every earlier trace has evaporated. No error is reported.
    • Change MAX_RETRIES to 1. The conditional edge loses its loop-back branch entirely — the graph stops being cyclic and becomes an ordinary linear for loop.
    • In demo_retry_cycle, change the broken code to "def solve_problem():\n while True: pass". The subprocess timeout fires after 5 seconds and returns timed out instead of hanging the whole program — which is exactly why a separate process is used rather than exec.

Lesson summary & what comes next

🔑 What you achieved:
  • Achieved: implementing state with per-field reducers, and seeing a field without one get overwritten into oblivion with no error raised.
  • Achieved: building conditional edges that create a coder → tester → coder cycle, and proving both the cycle and its retry-limit brake by forcing broken code through it.
  • Achieved: implementing a genuine human-in-the-loop interrupt — returning a serialisable checkpoint that can be resumed twice with two decisions, instead of blocking the process on input().
  • Achieved: running model-generated code in a separate process with a timeout, and knowing precisely the limits of that measure.

Bridge to the next lesson: everything so far has been about steering an existing model with prompts, tools and graphs. But some things prompts cannot teach: a company's house voice, a highly specific output format, a narrow knowledge domain. Lesson 19 moves to changing the model's weights themselves, with LoRA.

Download the practice code for this lesson

The Python file langgraph_agent.py — state with reducers, cyclic conditional edges, a checkpoint-based interrupt, and sandboxed execution of the generated code (run python3 langgraph_agent.py, needs Ollama with a chat model):

Download langgraph_agent.py

📖 Further reading

Related lessons in this series

Lesson 17: AI agents & the ReAct loop Lesson 19: Fine-tuning an LLM with LoRA Back to the Practical AI Engineer roadmap

Comments