LangChain vs LangGraph: When Your LLM Feature Needs a State Machine

The short answer

LangChain's expression language (LCEL) composes a fixed pipeline: prompt, model, parser, with retries and fallbacks wrapped around it. LangGraph runs a state machine: named nodes, a typed state object, edges that can branch, and a checkpointer that saves the state after each step so a run can pause, fail, and resume.

They are not competitors. A LangGraph node is usually an LCEL chain. The real question is whether the feature has control flow and a lifetime longer than one HTTP request. If it does not, a graph is overhead. If it does, hand-rolling that control flow around a chain is how you end up with a half-built workflow engine hidden in a service layer.

To make the trade-off concrete I build one feature twice: a refund-request assistant for a support team. Version 1 extracts a structured refund request from an email. Version 2 grows into what the business actually asked for: look up the order, apply a policy, send large refunds to a human, and survive a deploy while waiting. Everything below comes from a small reference implementation that I ran against fake chat models and a local Postgres; no real LLM was called.

Version 1: a linear LCEL chain

The first version has one job: turn a customer email into a validated RefundRequest. That is a linear transformation, and LCEL is the right size for it.

# ... imports
class RefundRequest(BaseModel):
    """A refund request extracted from one customer email."""

order_id: str | None = Field(description="Order reference such as ORD-1042, or null")
    amount: float | None = Field(description="Amount requested in GBP, or null")
    reason: Literal["damaged", "late", "wrong_item", "changed_mind", "other"]
    summary: str = Field(description="One neutral sentence for the support agent")

# ... PROMPT = ChatPromptTemplate.from_messages([...])

# Transport failures are worth retrying. A schema violation usually is not:
# the same prompt at temperature 0 tends to fail the same way again.
TRANSIENT = (openai.APITimeoutError, openai.APIConnectionError,
             openai.RateLimitError, openai.InternalServerError)
BAD_OUTPUT = (OutputParserException, ValidationError)

def build_extraction_chain(
    model: BaseChatModel,
    fallback: BaseChatModel | None = None,
    *,
    attempts: int = 3,
    initial_wait: float = 1.0,
) -> Runnable[dict, RefundRequest]:
    def structured(m: BaseChatModel) -> Runnable:
        return m.with_structured_output(RefundRequest).with_retry(
            retry_if_exception_type=TRANSIENT,
            stop_after_attempt=attempts,
            exponential_jitter_params={"initial": initial_wait, "max": 10.0},
        )

chain = PROMPT | structured(model)
    if fallback is None:
        return chain
    return chain.with_fallbacks(
        [PROMPT | structured(fallback)],
        exceptions_to_handle=TRANSIENT + BAD_OUTPUT,
    )

# ... refunds/wiring.py: constructed in a test, never invoked
primary = ChatOpenAI(model="gpt-4.1-mini", temperature=0, timeout=20, max_retries=0, api_key=api_key)

Three decisions in there are worth defending.

Retry only what can succeed on retry. with_retry defaults to retrying on any Exception. That turns a Pydantic validation failure into three identical paid calls. I restrict retries to transport errors, and the test suite asserts that a schema violation reaches the primary model exactly once before the fallback takes over.

Fall back on bad output, not just outages. with_fallbacks also defaults to catching everything, which hides programming errors behind a second model. I name the exceptions: transport failures plus OutputParserException and ValidationError, so a bug in my own code still fails loudly.

Retries live in one layer. The OpenAI client retries twice by default. Stack that under with_retry(stop_after_attempt=3) and a graph RetryPolicy(max_attempts=3) and one bad minute at the provider becomes up to 27 requests per item. I set max_retries=0 on the client and own the policy in the chain. The same thinking applies to queues and idempotency keys, which I cover in reliable LLM workflows.

This chain is 55 lines including the prompt. It runs inside a request handler, it is trivially testable, and it has nothing to persist. If the feature stopped here, I would ship this and stop. The structured-output side of it, with a FastAPI service around it, is in ticket triage with structured output.

What forces the move to a graph

Then the requirements arrive:

• Look up the order in an internal API that sometimes returns 503.
• Auto-approve refunds up to £50 when the order exists, the amount is within the order total and delivery was within 30 days.
• Decline anything outside policy, with a drafted reply.
• Send anything over £50 to a human, who may answer hours later, possibly after a deploy.
• If the email has no order reference, ask the customer instead of guessing.
• Keep an audit trail of every decision and who made it.

Each of these could be an if statement around the chain. The problem is the fourth one. The moment a run has to wait for a person, you need to store "where we were" somewhere, restore it in a different process, and make sure you do not re-run the expensive or dangerous steps. That is a state machine with persistence, whether you call it one or not.

Version 2: state, reducers and conditional edges

The state is a TypedDict. Each node returns a partial update, and LangGraph merges it into the state using the reducer declared on each key. Keys without a reducer are overwritten.

@dataclass(frozen=True)
class Deps:
    """Run-scoped dependencies. Passed as context=, never stored in checkpoints."""
    extractor: Runnable   # the version 1 LCEL chain, unchanged
    replier: Runnable     # REPLY_PROMPT | chat model
    orders: OrderClient
    auto_approve_limit: float = 50.0

class RefundState(TypedDict, total=False):
    email: str
    request: RefundRequest
    order: dict | None
    decision: Decision
    reviewer: str
    messages: Annotated[list[AnyMessage], add_messages]  # append, dedupe by id
    audit: Annotated[list[str], operator.add]            # append-only trail

audit uses operator.add, so every node appends and nothing can silently overwrite an earlier entry. messages uses add_messages, which appends new messages and replaces one that has the same id, so a later node can revise a draft in place instead of stacking a second one. A failed node attempt writes nothing, so retries cannot duplicate entries either.

Dependencies go in Deps, declared with context_schema and read through runtime.context. I keep clients and chains out of the state for two reasons: the checkpointer would try to serialise them, and on resume I want the current deployment's clients, not whatever was alive when the run paused. The flip side, which a test pins down, is that you must pass context= again on every resume call; forget it and the resumed node fails with runtime.context set to None.

The nodes are plain functions from (state, runtime) to a partial update. The business rule is a pure function with no LangGraph imports at all.

def decide(request: RefundRequest, order: dict | None, limit: float) -> Decision:
    """The business rule, kept free of LangGraph so it is trivial to test."""
    if order is None:
        return "ask_for_info"
    if request.amount > order["total"] or order["days_since_delivery"] > 30:
        return "decline"
    return "approve" if request.amount <= limit else "needs_approval"

def lookup_order(state: RefundState, runtime: Runtime[Deps]) -> dict:
    order_id = state["request"].order_id
    runtime.stream_writer({"stage": "order_lookup", "order_id": order_id})
    order = runtime.context.orders.get(order_id)  # may raise OrderServiceUnavailable
    return {"order": order, "audit": [f"order {order_id} found={order is not None}"]}

# ... extract, apply_policy, draft_reply, ask_for_info follow the same shape

def route_decision(state: RefundState) -> Literal["draft_reply", "human_approval", "ask_for_info"]:
    return {"approve": "draft_reply", "decline": "draft_reply",
            "needs_approval": "human_approval", "ask_for_info": "ask_for_info"}[state["decision"]]

def human_approval(state: RefundState) -> dict:
    # On resume this node restarts from the top, so nothing above interrupt()
    # may have side effects (no emails, no ledger writes, no LLM calls).
    r = state["request"]
    answer = interrupt({"order_id": r.order_id, "amount": r.amount,
                        "reason": r.reason, "summary": r.summary})
    decision = "approve" if answer["approved"] else "decline"
    return {"decision": decision, "reviewer": answer["reviewer"],
            "audit": [f"{answer['reviewer']} -> {decision}"]}

The router's Literal return type is not decoration: LangGraph reads it to know the possible destinations, which is what makes get_graph().draw_mermaid() show the dashed branches without a separate path map.

Wiring the graph is where per-node retry policy lives:

LOOKUP_RETRY = RetryPolicy(max_attempts=3, initial_interval=0.5,
                           backoff_factor=2.0, retry_on=OrderServiceUnavailable)

def build_graph(checkpointer: BaseCheckpointSaver | None = None,
                lookup_retry: RetryPolicy = LOOKUP_RETRY):
    g = StateGraph(RefundState, context_schema=Deps)
    # No retry_policy on extract: the chain inside already retries transport errors.
    g.add_node("extract", nodes.extract)
    g.add_node("lookup_order", nodes.lookup_order, retry_policy=lookup_retry)
    g.add_node("apply_policy", nodes.apply_policy)
    g.add_node("human_approval", nodes.human_approval)
    g.add_node("draft_reply", nodes.draft_reply)
    g.add_node("ask_for_info", nodes.ask_for_info)

g.add_edge(START, "extract")
    g.add_conditional_edges("extract", nodes.route_after_extract)
    g.add_edge("lookup_order", "apply_policy")
    g.add_conditional_edges("apply_policy", nodes.route_decision)
    g.add_edge("human_approval", "draft_reply")
    g.add_edge("draft_reply", END)
    g.add_edge("ask_for_info", END)
    return g.compile(checkpointer=checkpointer)

retry_on=OrderServiceUnavailable matters. The default retry_on in LangGraph 1.2 skips common programming errors (ValueError, TypeError, LookupError and friends) but retries almost everything else, including an openai.BadRequestError for a prompt that is too long. A test pins that down. I would rather name the one exception that is worth retrying. The order lookup is a read, so retrying it is safe; a node that issues the refund would not get a retry policy at all until the call carries an idempotency key.

Checkpoints, thread_id and human approval

A checkpointer saves the state after each step under a thread_id. In tests I use InMemorySaver. In production I use PostgresSaver from langgraph-checkpoint-postgres (3.1.2 here), sharing the Postgres instance the service already runs.

def make_postgres_checkpointer(dsn: str, *, max_size: int = 10) -> tuple[PostgresSaver, ConnectionPool]:
    # PostgresSaver needs autocommit and dict rows; prepare_threshold=0 keeps it
    # working behind PgBouncer in transaction mode.
    pool = ConnectionPool(dsn, max_size=max_size, open=True, kwargs={
        "autocommit": True, "prepare_threshold": 0, "row_factory": dict_row})
    saver = PostgresSaver(pool)
    saver.setup()  # idempotent migrations; run at deploy time in real services
    return saver, pool

def test_interrupt_survives_a_process_restart(dsn, make_deps):
    config = {"configurable": {"thread_id": f"refund:{uuid.uuid4()}"}}

# Process A: runs until the approval interrupt, then goes away.
    saver, pool = make_postgres_checkpointer(dsn)
    deps_a, _, reply_a = make_deps(extraction(amount=80.0))
    out = build_graph(saver).invoke({"email": "..."}, config, context=deps_a,
                                    durability="sync", version="v2")
    assert out.interrupts[0].value["amount"] == 80.0
    pool.close()

# Process B: fresh pool, fresh graph object, fresh dependencies.
    saver, pool = make_postgres_checkpointer(dsn)
    deps_b, extractor_b, reply_b = make_deps(extraction(amount=999.0))
    graph = build_graph(saver)
    assert graph.get_state(config).next == ("human_approval",)
    out = graph.invoke(Command(resume={"approved": False, "reviewer": "ade"}),
                       config, context=deps_b, version="v2")
    pool.close()

assert out.value["request"].amount == 80.0   # restored from Postgres, not re-extracted
    assert extractor_b.calls == 0
    assert out.value["decision"] == "decline"
    # ...

Process B's extractor is scripted to return a different amount, and it is never called. The £80 comes back from the checkpoint. That is the property a hand-rolled "save a status column and re-run" design usually gets wrong.

Some details I learned by reading the 1.2 source rather than older blog posts:

• invoke(..., version="v2") returns a GraphOutput with .value and .interrupts. Dict-style access such as result["__interrupt__"] still works on it but emits a deprecation warning.
• durability defaults to "async", meaning a step's checkpoint is written while the next step runs. For approval flows I pass "sync", so each step is persisted before the next one starts.
• interrupt() does not freeze the Python frame. On resume the whole node runs again from the top, and interrupt() returns the resume value the second time. I keep a test that puts a fake "notify reviewer" call before interrupt() and asserts it runs twice. Notifications belong in the caller, driven by out.interrupts, or in their own node before the approval node, which completes and is checkpointed once.
• compile() derives a msgpack allowlist from the state schema, so the Pydantic RefundRequest in state round-trips through Postgres even with LANGGRAPH_STRICT_MSGPACK=true. I ran the Postgres test both ways.

The thread_id is the durable identity of the run, so I derive it from the business key (refund:<ticket id>), not a fresh UUID. A redelivered webhook then lands on the existing thread instead of starting a second approval for the same refund. The test above uses a UUID only to isolate test runs.

The cost is that the state is now a stored schema. Renaming RefundRequest or changing a field type breaks threads that paused under the old code. I treat state changes like database migrations: additive fields with defaults, and drain or migrate paused threads before removing anything. The support agent with human handoff has the same problem at larger scale.

I also do not treat graph state as the system of record. In the MedReclaim AI design (implementation-ready, not shipped), the LangGraph gateway runs the conversation while a separate FastAPI tool server checks identity and permissions and keeps the audit records before any sensitive action runs. The refund example follows the same split in miniature: the audit list is for operators reading a thread, and the ledger entry for an issued refund would live in the payments service.

Streaming what the user and the operator need

A chain streams one thing: tokens. A graph can stream several views of the same run, and version="v2" gives each part a uniform {"type", "ns", "data"} shape.

for part in graph.stream({"email": "..."}, config, context=deps,
                         stream_mode=["updates", "custom", "messages"], version="v2"):
    if part["type"] == "updates":
        node_order.extend(part["data"].keys())       # which node just finished
    elif part["type"] == "custom":
        progress.append(part["data"])                 # runtime.stream_writer(...)
    elif part["type"] == "messages":
        chunk, meta = part["data"]
        if meta["langgraph_node"] == "draft_reply":   # only customer-facing text
            tokens.append(chunk.content)

updates drives an operator timeline. custom carries progress events that nodes emit through runtime.stream_writer. messages streams tokens from any chat model called inside a node, including the extractor's tool-call chunks, which is why I filter on langgraph_node. Without that filter, the extraction step's chunks, which carry the structured payload including the order reference, are mixed into the stream going to the browser. The test asserts the reply arrives as several chunks that join back to the full text.

Testing nodes as functions and the graph with a fake model

Two layers of tests, both fast. The whole suite of 23 tests, including the Postgres one, runs in about a second on my laptop against a local Postgres 16.

Nodes are functions, so I call them directly. Runtime is a plain dataclass, and its stream_writer can be list.append. The chat model is a subclass of GenericFakeChatModel that replays a script of messages or exceptions and makes bind_tools a no-op, so the production code path through with_structured_output runs unchanged.

class ScriptedChatModel(GenericFakeChatModel):
    script: list[Any]
    messages: Any = None  # GenericFakeChatModel's iterator, unused here
    calls: int = 0

def bind_tools(self, tools, *, tool_choice=None, **kwargs):
        return self
    # ... _generate replays self.script; _stream yields text words or tool_call_chunks

def test_lookup_order_node_returns_partial_update_and_progress_event():
    events = []
    orders = FakeOrders({"ORD-1042": {"total": 120.0, "days_since_delivery": 3}})
    runtime = Runtime(context=Deps(extractor=None, replier=None, orders=orders),
                      stream_writer=events.append)
    update = nodes.lookup_order({"request": req()}, runtime)
    assert update == {"order": {"total": 120.0, "days_since_delivery": 3},
                      "audit": ["order ORD-1042 found=True"]}
    assert events == [{"stage": "order_lookup", "order_id": "ORD-1042"}]

def test_failed_run_resumes_from_last_checkpoint_without_repaying_the_llm(make_deps):
    orders = FakeOrders(ORDERS, fail_first=5)
    deps, extractor_model, _ = make_deps(extraction(), orders=orders)
    graph = build_graph(InMemorySaver(), lookup_retry=FAST_RETRY)
    with pytest.raises(OrderServiceUnavailable):
        graph.invoke({"email": "..."}, cfg("T-5"), context=deps)
    assert graph.get_state(cfg("T-5")).next == ("lookup_order",)

orders.fail_first = 0  # the order service recovers
    out = graph.invoke(None, cfg("T-5"), context=deps, version="v2")
    assert out.value["decision"] == "approve"
    assert extractor_model.calls == 1  # extraction was checkpointed, not re-run

The graph-level tests cover each route: auto-approve, interrupt then resume, missing order id (asserting the order service is never called), a flaky lookup that succeeds on the third attempt, and the resume-after-failure case above. invoke(None, config) continues a thread from its last checkpoint, so an outage in a downstream API costs one retry of the failed node, not a fresh extraction.

The fake needed one fix. GenericFakeChatModel streams text but yields nothing for a message that only has tool calls, so the first streaming test failed with "No generations found in stream". Real providers stream tool calls as tool_call_chunks, so the fake does the same. Fakes that behave differently from providers in streaming mode are a common source of green tests and broken UIs. For regression-testing model behaviour itself rather than plumbing, I use eval gates, covered in shadow mode and eval gates.

Choosing, and the cost of a graph too early

My rule: a chain until the feature must wait, branch on data it fetched, or survive a restart mid-run. Any one of those tips it. Branching alone on data the model already returned is often a Python if after the chain.

Choosing a graph too early has real costs. The easiest one to count is code. The chain version is 55 lines. The graph version adds about 160 lines across state, nodes, wiring and the checkpointer, plus four Postgres tables to migrate and back up. The less visible costs matter more:

• Every state field becomes a persisted schema. Refactors that would be free in a chain now need a plan for paused threads.
• Resume semantics leak into node design. It is easy to put a side effect above interrupt() and send two emails.
• Retry layers multiply unless someone owns the whole stack of policies.
• A three-node graph with no branches and no pauses is a chain with more ceremony, and its traces are harder to read.

The migration path is cheap if the chain is written well. Version 2's extract node calls the version 1 chain unchanged, and its tests still pass. Start with the chain, keep business rules in pure functions, and the graph becomes a new outer layer rather than a rewrite.

Checklist

• Start with an LCEL chain; move to a graph when the feature must pause, branch on fetched data, or survive a restart.
• Restrict with_retry to transport errors and name the exceptions in with_fallbacks.
• Set client max_retries=0 and own retries in exactly one layer per call.
• Keep business rules in pure functions with no LangGraph imports.
• Put clients and chains in context_schema, not state; pass context= again on every resume.
• Use reducers (operator.add, add_messages) for anything appended across nodes.
• Give RetryPolicy an explicit retry_on, and only on idempotent nodes.
• Derive thread_id from the business key so redeliveries join the existing run.
• Nothing with side effects above interrupt(); the node re-runs on resume.
• Use durability="sync" for runs that are about to wait on a human.
• Use PostgresSaver with autocommit, dict_row and prepare_threshold=0; run setup() at deploy.
• Treat state changes like database migrations.
• Test nodes directly with a Runtime, and the graph with a scripted fake model that streams like a real provider.
• Filter messages stream parts by langgraph_node before showing tokens to a customer.

Tested with Python 3.12.9, langgraph 1.2.12, langgraph-checkpoint 4.2.0, langgraph-checkpoint-postgres 3.1.2, langchain-core 1.6.5, langchain-openai 1.6.6, psycopg 3.3.6 and PostgreSQL 16.2 on 2026-09-27.