Building an Orchestrator with LangGraph

What an orchestrator does

Once an AI system has more than a couple of steps, some part of it has to answer several questions. What runs next? What information does it need? Should it retry? Is a person's approval needed? Is the work finished? That part is the orchestrator.

LangGraph is a framework for expressing this logic as a graph. Instead of burying control flow inside prompts or nested conditionals, you describe the steps and the transitions between them explicitly, which makes the workflow easier to inspect and test.

This guide concentrates on concepts and design. The code is deliberately plain Python, intended to show the shape of each part. Exact LangGraph method names and import paths can change between releases, so check the current documentation for the version you install.

The core concepts

State is the shared data for a run. Each node reads from it and returns updates to it.
Nodes are the units of work. A node might call an LLM, run a tool or execute ordinary Python.
Edges define what follows a node. A normal edge always leads to the same next node.
Conditional routing uses a function that inspects the state and picks the next node.
Checkpointing saves state as a graph runs, which supports resuming, multi-turn memory and pausing for review.

Graphs can include cycles, and that is one of the main reasons to use this model. A loop in which an agent acts, a result is checked and the agent tries again can be expressed directly.

If you are still deciding whether your project needs a graph at all, see LangChain vs LangGraph: when to use each.

A worked example: a support question router

Suppose an assistant handles incoming questions. Some need an answer from product documentation, others relate to a customer's account, and a few should go to a person. Answers should be checked before they are sent, with a limited number of retries.

Before writing any code, sketch the flow:

classify --> search_docs ----+
classify --> lookup_account -+--> draft --> check
classify --> human_review

check --(passed)--> finish
check --(failed, retries left)--> draft
check --(failed, no retries)--> human_review

This sketch is a useful design artefact in itself. It exposes the decisions, the loop and the exits, and it can be reviewed by someone who never reads the code.

Step 1: design the state

State is the contract among all of the nodes, so it deserves careful thought. What should last beyond a single run is a separate question, covered in agent memory. Here is the state for this example:

from typing import TypedDict

class SupportState(TypedDict):
    question: str
    category: str       # "docs", "account" or "human"
    context: list[str]  # material used for the draft
    draft: str
    check_passed: bool
    attempts: int

A few principles help here:

Store facts rather than whole transcripts. Named, specific fields are easier to validate and route on than a long message history.
Define how updates merge. Most fields can simply be overwritten, but a list that accumulates, or one that parallel nodes write to, needs a defined merge rule. LangGraph supports this through reducers attached to state fields.
Track control data. Fields such as an attempt count make loops bounded and testable.
Keep secrets out of state. Checkpoints may persist state to storage, so think about what is saved there, including personal data.

Step 2: write focused nodes

A node takes the state and returns only the fields it wants to change:

def classify(state):
    label = classify_question(state["question"])
    if label not in {"docs", "account", "human"}:
        label = "human"   # safe default
    return {"category": label}

def draft(state):
    text = write_answer(state["question"], state["context"])
    return {"draft": text, "attempts": state["attempts"] + 1}

def check(state):
    passed = is_supported(state["draft"], state["context"])
    return {"check_passed": passed}

Here classify_question, write_answer and is_supported are your own functions. Note the validation in classify: model output is checked against the allowed values, and anything unexpected goes to the safest route. Because nodes are ordinary functions, each one can be tested by passing in a state and inspecting the returned update, without running the graph.

Not every node needs a model. Retrieval, account lookups and validation rules are often better as plain code, which is cheaper and more predictable.

Step 3: add conditional routing

Routing functions read the state and return the name of the next step:

MAX_ATTEMPTS = 2

def route_by_category(state):
    return {
        "docs": "search_docs",
        "account": "lookup_account",
    }.get(state["category"], "human_review")

def route_after_check(state):
    if state["check_passed"]:
        return "finish"
    if state["attempts"] < MAX_ATTEMPTS:
        return "draft"
    return "human_review"

Keep routing functions simple and free of side effects. If a decision needs a model's judgement, make that judgement in a node, write the validated result into the state, then route on the field. This separates the fuzzy decision from the deterministic control flow, and it makes each path straightforward to unit test: construct a state, call the function and assert on the result.

Step 4: connect the graph

With the parts defined, you assemble the graph by doing the following with LangGraph's graph builder:

Create a graph based on your state definition.
Register each function as a named node.
Set classify as the entry point.
Add conditional edges from classify and check using the routing functions.
Add normal edges for the fixed transitions, such as from search_docs to draft.
Connect the terminal nodes to the graph's end.
Compile the graph, attaching a checkpointer if you need persistence.

The result can then be invoked with an initial state, which here must set attempts to 0 because draft reads it. Runs should be traced, so that you can see the path taken and the state at each step.

Step 5: use checkpointing and human review

With a checkpointer attached, state is saved as the graph runs, typically associated with an identifier for the conversation or job. This enables three important behaviours:

Recovery. A run interrupted by an error can continue from saved state rather than starting over.
Memory across turns. A later message in a conversation can continue with the state from earlier ones.
Human-in-the-loop. A run can pause before a sensitive step, wait for a person, then resume with their decision. The guide on human-in-the-loop approval for AI agents covers what to gate and how to make approval reliable.

Choose a durable storage backend for production, since an in-memory store is lost on restart. Also remember that resuming may re-execute a step. Any node with side effects, such as sending an email or issuing a refund, should be designed so that running it twice does no harm, for example by using an idempotency key.

Common pitfalls

Unbounded loops. Always combine an attempt counter or step limit with a defined fallback path.
Overgrown state. If every node reads and writes everything, dependencies become invisible. Keep fields purposeful.
Too many nodes. Extra small nodes can make a graph harder to read. Create a node when a step needs separate routing, retries, tracing or testing.
Trusting model output in routing. Validate labels and provide a safe default, as the example does.
Parallel write conflicts. Nodes running in parallel and updating the same field need an appropriate merge rule.
Using a graph where a function would do. A linear sequence of two or three steps rarely needs an orchestrator.

Testing the orchestrator

Test at three levels. Unit test routing functions and non-LLM nodes with ordinary assertions. Run the graph with model calls replaced by stubs to verify that each path, including the fallback, behaves as designed. Finally, evaluate complete runs with real models against a set of realistic inputs, reviewing both the final output and the path taken.

To decide whether you need several agents inside this structure, read single agent vs multi-agent systems. For implementation support, see AI orchestration with LangGraph.

Frequently asked questions

What is LangGraph used for?

LangGraph is used to build stateful AI workflows and agents as graphs. It is particularly useful when a workflow needs branching, loops, persistence or human review, rather than a simple linear sequence.

What is the difference between an edge and a conditional edge?

A normal edge always leads from one node to the same next node. A conditional edge calls a routing function, which examines the current state and selects the next node.

Does every LangGraph node have to call an LLM?

No. A node can run any Python logic, including database queries and validation. Using plain code for deterministic steps usually makes a workflow cheaper and more reliable.

How do you stop a graph from looping forever?

Track attempts in the state and have the routing function exit to a fallback once a limit is reached. A framework-level step limit is a useful additional safeguard, not a replacement for a designed exit path.

Building an orchestrator with LangGraph