LangGraph, state and checkpoints

How it works: LangGraph

1. Define the state: a typed dictionary of everything the run needs (messages, the plan, retrieved docs, flags).
2. Write nodes: plain functions that receive the state and return the fields they want to update. A node can call an LLM, a tool, or plain code.
3. Connect them with edges: START → write, write → END.
4. Add conditional edges: a function looks at the state and returns the next node's name, creating branches and loops (retry until good enough, route to a tool node, finish).
5. Compile the graph (optionally with a checkpointer) and invoke it with an initial state.
6. Prebuilt pieces exist for the common agent pattern (an LLM node plus a tool node in a loop), and multi-agent graphs (subgraphs, supervisors).
7. Install: pip install langgraph (version 1.2 as of November 2026). It works with any model provider.

How it works: state and checkpoints

1. State is the single source of truth: every node reads it and returns updates; LangGraph merges them (with "reducers" for fields like message lists that should append rather than overwrite).
2. A checkpointer saves a checkpoint of the state after every step. InMemorySaver for development; database-backed savers (Postgres, SQLite, Redis) in production.
3. Threads: each run or conversation has a thread_id. Invoking again with the same ID continues from the saved state; that's also how per-conversation memory works.
4. What checkpoints unlock:
   • Resume after failure: restart from the last good step instead of from scratch.
   • Human in the loop: pause at a step, wait for approval, resume later (next lesson).
   • Time travel: inspect every past state (get_state_history) and replay from any point to debug.
5. Keep state small and serialisable: IDs and summaries rather than huge blobs.

from typing import TypedDict
 
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
 
class State(TypedDict):
    draft: str
    attempts: int
 
def write(state: State) -> dict:                       # a node: reads state, returns an update
    n = state["attempts"] + 1
    return {"draft": "short" if n < 2 else "a longer, complete answer", "attempts": n}
 
def good_enough(state: State) -> str:                  # a conditional edge: decides where to go next
    return "done" if len(state["draft"]) > 10 or state["attempts"] >= 3 else "retry"
 
g = StateGraph(State)
g.add_node("write", write)
g.add_edge(START, "write")
g.add_conditional_edges("write", good_enough, {"retry": "write", "done": END})   # a loop
app = g.compile(checkpointer=InMemorySaver())          # saves state after every step
 
config = {"configurable": {"thread_id": "user-7"}}
print("result :", app.invoke({"draft": "", "attempts": 0}, config))
history = list(app.get_state_history(config))          # every saved checkpoint, newest first
print("checkpoints saved:", len(history))
print("oldest -> newest :", [h.values.get("attempts") for h in reversed(history)])

result : {'draft': 'a longer, complete answer', 'attempts': 2}
checkpoints saved: 4
oldest -> newest : [None, 0, 1, 2]