LangGraph workflow orchestration: Build a state machine

To orchestrate a workflow with LangGraph, describe it as a state machine. You define one typed state, write each step as a node that returns updates to that state, and connect the nodes with edges. A conditional edge reads the state and picks the next node, which is how you get loops: write, test, and go back to writing until the tests pass or the attempts run out. Compile the graph with a checkpointer and every step is saved, so a run can pause for a human and resume later.

This guide builds that loop end to end with LangGraph 1.x: a coder node, a tester node, a router, a retry cap and a human checkpoint. The full graph is about 40 lines.

Key takeaways

• Use LangGraph when your flow has loops, retries or branches. A straight pipeline does not need a graph.
• Keep all data in one typed state. Nodes return only the keys they change, and LangGraph merges them.
• A conditional edge is a plain Python function that reads the state and returns the name of the next node.
• Put a retry counter in the state and check it in the router, so a loop always has a way out.
• Compile with a checkpointer to save every step. That gives you pause, inspect, resume and human approval.
• LangGraph also stops a runaway graph with a recursion limit, but treat that as a backstop, not your exit rule.

Why does an agent need a state machine instead of a chain?

Because an agent loops and a chain does not. A chain is a straight line: input, prompt, output. That is a DAG (a directed acyclic graph), and by definition it cannot go back to an earlier step.

Real agent work goes back all the time. A draft gets critiqued and rewritten. A tool call fails and the agent fixes its arguments and tries again. Code fails its tests and goes back to the coder. Each of those is a cycle, and you need a graph that allows cycles and a clear rule for when each cycle ends.

A state machine gives you both. Every step reads and writes the same state, and the edges decide where to go next based on what is in that state.

How do you define state in LangGraph?

Define the state as a TypedDict. Each node receives the whole state and returns a dict with only the keys it changed. By default a returned value overwrites the old one. If you want a key to accumulate instead, attach a reducer with Annotated, for example operator.add to append to a list.

# filename: state.py
# description: The one shared state every node reads and updates.
import operator
from typing import Annotated, TypedDict


class CodeState(TypedDict):
    task: str                                   # what the user asked for
    code: str                                   # the latest attempt
    errors: str | None                          # test output from the last run, None if it passed
    attempts: int                               # how many times the coder has run
    history: Annotated[list[str], operator.add] # one line per step, appended, never overwritten

Keep the state small and explicit. If a value matters to a routing decision, it belongs in the state, not in a local variable inside a node. That is what makes the run inspectable later.

How do you build a loop with nodes and conditional edges?

Write each step as a function, add them as nodes, and add a conditional edge whose function returns the next node’s name. This graph writes code, tests it, and loops back with the error until it passes or hits three attempts.

# filename: graph.py
# description: A self-correcting coder loop in LangGraph 1.x.
import subprocess
import tempfile

from langgraph.graph import END, START, StateGraph

from state import CodeState

MAX_ATTEMPTS = 3


def coder(state: CodeState) -> dict:
    # Swap in your model call. The prompt includes the last error on a retry.
    code = write_code(state['task'], previous=state['code'], error=state['errors'])
    return {
        'code': code,
        'attempts': state['attempts'] + 1,
        'history': [f"attempt {state['attempts'] + 1}"],
    }


def tester(state: CodeState) -> dict:
    # Run the code in a separate process with a timeout. Never exec() model output.
    with tempfile.NamedTemporaryFile('w', suffix='.py', delete=False) as f:
        f.write(state['code'])
    run = subprocess.run(['python', f.name], capture_output=True, text=True, timeout=30)
    errors = None if run.returncode == 0 else run.stderr[-2000:]
    return {'errors': errors, 'history': ['passed' if errors is None else 'failed']}


def route(state: CodeState) -> str:
    if state['errors'] is None:
        return 'done'
    if state['attempts'] >= MAX_ATTEMPTS:
        return 'give_up'
    return 'retry'


builder = StateGraph(CodeState)
builder.add_node('coder', coder)
builder.add_node('tester', tester)
builder.add_edge(START, 'coder')
builder.add_edge('coder', 'tester')
builder.add_conditional_edges(
    'tester', route, {'retry': 'coder', 'done': END, 'give_up': END}
)
graph = builder.compile()

result = graph.invoke(
    {'task': 'parse ISO dates', 'code': '', 'errors': None, 'attempts': 0, 'history': []}
)

Read the router first. It is the whole control flow in one function: passed means stop, out of attempts means stop, anything else means loop. The coder gets smarter on each pass only because the state now carries errors, so its prompt can say “here is your last code and the error it caused”.

Older tutorials use set_entry_point('coder'). In current docs the entry is an edge from the virtual START node, as above.

How do you stop a LangGraph loop from running forever?

Give every loop its own exit in the state, like the attempts counter above. Your router should always have a branch that ends the run.

LangGraph adds a backstop. Each run has a recursion limit on the number of steps, and the run raises an error when it hits it. The docs say the default is 1000 steps since version 1.0.6. You can set it per run:

graph.invoke(inputs, config={'recursion_limit': 25})

Note that recursion_limit is a top-level config key, not inside configurable. Do not rely on it as your main exit. An error at step 1000 means the run already burned 1000 steps of model calls. A counter in the state fails early and on purpose.

How do checkpoints let you pause, inspect and resume a graph?

Compile the graph with a checkpointer. LangGraph then saves the state after every step under a thread id. You can read the saved history, resume a failed run from its last good step, or pause the run to ask a human.

To pause, call interrupt() inside a node. The run stops and waits. You resume it by invoking the graph again with Command(resume=...) on the same thread, and that value becomes the return value of interrupt().

# filename: graph.py (changes)
# description: Ask a human before giving up, then resume on the same thread.
# This replaces the tester edge and the compile line from the first version.
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command, interrupt


def ask_human(state: CodeState) -> dict:
    answer = interrupt(f"Failed {state['attempts']} times. Retry? (yes/no)")
    if answer == 'yes':
        return {'attempts': 0, 'history': ['human: retry']}
    return {'history': ['human: stop']}


def after_human(state: CodeState) -> str:
    return 'coder' if state['attempts'] == 0 else 'end'


builder.add_node('ask_human', ask_human)
builder.add_conditional_edges(
    'tester', route, {'retry': 'coder', 'done': END, 'give_up': 'ask_human'}
)
builder.add_conditional_edges('ask_human', after_human, {'coder': 'coder', 'end': END})
graph = builder.compile(checkpointer=InMemorySaver())

config = {'configurable': {'thread_id': 'ticket-42'}}
graph.invoke(inputs, config)                 # runs until the interrupt
graph.invoke(Command(resume='yes'), config)  # the human said retry

InMemorySaver is for local runs and tests. In production use a database checkpointer such as the Postgres one, so a restart does not lose the thread. The LangGraph persistence guide covers thread models and choosing a checkpointer in depth.

When should you not use LangGraph?

When the flow is a straight line. If every request goes input, retrieve, prompt, answer, a few function calls are simpler to read, test and debug than a graph.

Reach for a graph when at least one of these is true:

1. The flow loops: refine, retry, or test and fix.
2. The next step depends on what happened, not on a fixed order.
3. A run must survive a restart or wait hours for a human.
4. You need to replay or inspect each step of a past run.

If you are serving the graph behind an API, the FastAPI and LangGraph chatbot guide shows the wiring.

What to do next

1. Write down your current workflow as boxes and arrows. Mark every arrow that goes backwards. If there are none, keep your chain.
2. Turn the data each step needs into one TypedDict. Add a counter for every loop.
3. Write the router as a plain function and unit test it with hand-made states, without calling a model.
4. Compile with InMemorySaver, run it with a thread_id, and read the saved steps back.
5. Add interrupt() at the one point where a wrong move is expensive, then switch to a database checkpointer before you ship.
6. Design the roles, loops and budgets of a full agent system in the Design Agentic AI Systems workshop.

Frequently asked questions

What is a state machine in LangGraph?

It is a graph of nodes that all read and update one shared, typed state. Nodes do the work, such as a model call or a tool call. Edges decide what runs next. A conditional edge reads the state and returns the next node’s name, so the graph can branch and loop instead of running in a fixed order.

How do I build a retry loop in LangGraph?

Add a conditional edge after the step that can fail. Its function checks the state: on success it returns a name that maps to END, on failure it returns the name of the node to retry. Keep an attempts counter in the state and stop when it reaches your limit, so the loop always ends.

What is the default recursion limit in LangGraph?

The LangGraph docs say the default is 1000 steps starting in version 1.0.6. When a run reaches it, LangGraph raises an error. You can lower it per run with config={'recursion_limit': 25}. Use it as a safety net and give each loop its own counter in the state.

How do I add human approval to a LangGraph workflow?

Compile the graph with a checkpointer and call interrupt() in the node that needs approval. The run pauses and saves its state under the thread id. Resume it by invoking the graph with Command(resume=value) on the same thread. The value you pass becomes the return value of interrupt().

When should I use LangGraph instead of a simple chain?

Use LangGraph when the workflow has loops, branches that depend on results, long waits for a human, or runs you need to replay. Use a chain or plain function calls when every request follows the same straight path. A graph adds structure, and that structure only pays off when the flow is not a straight line.

Sources

• Graph API overview, LangChain docs, October 2026: TypedDict state and reducers, START and conditional edges, compiling with a checkpointer, the 1000-step default recursion limit since 1.0.6, Command and interrupt.
• langgraph on PyPI, Python Package Index, October 2026: release history, 1.0.0 on October 17, 2025 and 1.x releases since.
• LangGraph overview, LangChain docs, October 2026: what LangGraph is for and its persistence model.