The live LangGraph lesson gets you a model node, a tools node, and a conditional edge. This deep dive is what you reach for once that graph needs to survive a restart, pause for a human, or branch in ways a single conditional edge cannot express cleanly.

State: reducers, not just a dict

State updates in LangGraph are not overwrites by default. They go through a reducer you declare per field. The built-in `add_messages` reducer appends new messages instead of replacing the list, which is why a chat history survives across nodes without every node having to re-pass the full log.

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

def keep_max(existing: int, new: int) -> int:
    return max(existing, new)

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    retries: Annotated[int, keep_max]
    plan: str | None

Command: routing and state update in one return

Instead of a node returning a state dict and a separate conditional-edge function deciding where to go next, a node can return a `Command` that does both - update state and pick the next node - which keeps routing logic next to the reasoning that produced it.

from langgraph.types import Command
from typing import Literal

def triage_node(state: AgentState) -> Command[Literal["specialist", "human_review", "__end__"]]:
    severity = classify(state["messages"][-1])
    if severity == "needs_human":
        return Command(goto="human_review", update={"plan": "escalated"})
    if severity == "routine":
        return Command(goto="specialist", update={"plan": "auto-handle"})
    return Command(goto="__end__", update={"plan": "no action needed"})

Subgraphs: compose, don't sprawl

A subgraph is a compiled graph used as a single node inside a parent graph. This is how you keep a multi-agent system readable: each specialist is its own graph with its own state, tested in isolation, then wired into the orchestrator as one node instead of forty inlined nodes.

Persistence: checkpointers are what make HITL real

• `InMemorySaver` - fine for local dev, gone on restart.
• `SqliteSaver` - durable across restarts on a single machine, good default for a capstone.
• Postgres / Redis checkpointers - production-grade, shared across multiple processes or workers.

from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
graph = builder.compile(checkpointer=checkpointer, interrupt_before=["send_email"])

config = {"configurable": {"thread_id": "ticket-T-101"}}
graph.invoke({"messages": [...]}, config)
# ... process pauses before send_email; a human approves ...
graph.invoke(None, config)  # resume from exactly where it stopped

Time travel: replay and fork

Because every step is checkpointed, you can list a thread's history, pick an earlier checkpoint, and re-run from there with different state, invaluable for debugging why an agent went wrong three steps before the failure, without re-running everything from the start.

Streaming modes

• `values` - full state after each step, simplest to reason about.
• `updates` - only what changed at each step, smaller payloads.
• `messages` - token-by-token streaming for the model's own output, what you want for a chat UI.