State, Memory & Persistence
so the agent stops forgetting

Sahil's research agent works for a single question. But he wants a conversation: ask something, get an answer, then ask a follow-up that refers back to the first. He tries it, and the agent stares blankly, because the second invoke knows nothing about the first. The state from run one was returned to him and then thrown away.

So this module climbs in a deliberate order: first how a node's return value merges into state (reducers), then where that state is saved (checkpointers), then how to partition saved state per user (threads), and finally the two payoffs that persistence unlocks: streaming and time travel.

Back in Module 01 a node returned {"answer": text} and that simply overwrote the old answer. Overwrite is the default reducer: last write wins. But overwrite is wrong for anything that should accumulate. If your findings field overwrote on every research step, the agent would only ever remember its most recent lookup. You want each step to add to the list, not replace it.

A reducer is the rule that combines the old value of a field with the update a node returns. You attach one to a field by annotating its type. The most common reducer is "append to a list", which the standard library gives you as operator.add (or LangGraph's add_messages for chat).

from typing import Annotated
from operator import add
from typing_extensions import TypedDict

class ResearchState(TypedDict):
    question: str
    findings: Annotated[list, add]   # reducer = add → updates APPEND
    answer: str                      # no reducer → updates OVERWRITE

# A node returning {"findings": ["fact A"]} now ADDS to the list.
# Two parallel nodes can each contribute; their lists are merged, not lost.

This is exactly the machinery behind MessagesState from Module 01. Its single field is declared as Annotated[list, add_messages], and add_messages is a smarter append that also handles updating a message by id. You were already using a reducer; now you can write your own.

By default the state schema is also what callers pass in and what they get back. Often you want the internal state to carry scratch fields that you do not want to expose. LangGraph lets you declare a separate input schema and output schema, so the public surface stays clean while the graph keeps private working fields.

class InputState(TypedDict):    question: str
class OutputState(TypedDict):   answer: str
class InternalState(TypedDict): question: str; scratch: list; answer: str

builder = StateGraph(InternalState, input_schema=InputState, output_schema=OutputState)
# callers send only {question}; they receive only {answer}; scratch stays hidden

Here is the line that solves Sahil's amnesia. A checkpointer is a backend that saves a snapshot of the graph's state after every step. Attach one at compile time and the graph automatically persists its progress. The simplest is InMemorySaver (state held in RAM, good for learning and tests); for production you swap in a database-backed saver such as the SQLite or Postgres checkpointers, with no change to your graph logic.

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()                 # swap for SqliteSaver/PostgresSaver in prod
graph = builder.compile(checkpointer=checkpointer)   # ← the whole trick

That single argument changes the agent's nature. Now, between steps, the runtime writes the current state to the checkpointer. If the process dies after step three of a five-step task, restarting resumes from the saved snapshot instead of starting over. This durability is the foundation that human-in-the-loop (Module 03) is built on: pausing for a human is just "stop, the state is already saved, resume when they reply."

One agent serves many users. Sahil's research bot should not mix Priya's conversation into Ravi's. A thread is a named, independent timeline of state. You pick a thread by passing a thread_id in the config on every call. Same thread id, same continuing memory. New thread id, a fresh blank conversation.

config = {"configurable": {"thread_id": "priya-001"}}

graph.invoke({"messages": [user("What is LangGraph?")]}, config)
# ... later, SAME thread id → the agent remembers the first turn ...
graph.invoke({"messages": [user("And how is it different from LangChain?")]}, config)
# a DIFFERENT thread_id would start clean, with no memory of the above

The checkpointer you just learned is short-term memory: it remembers this conversation, scoped to a thread. But sometimes you want facts that outlive any single conversation: a user's name, their preferences, things learned in a thread last week that should apply today. That is long-term memory, and LangGraph models it separately with a Store, keyed by a namespace you choose (often the user id) rather than by thread.

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
graph = builder.compile(checkpointer=checkpointer, store=store)

# inside a node you receive the store and read/write by (namespace, key):
def remember(state, *, store):
    ns = ("user", "priya")               # namespace, not thread
    store.put(ns, "prefers", {"format": "bullets"})
    saved = store.get(ns, "prefers")      # available in ANY future thread
    ...

Calling invoke waits for the whole graph to finish, then hands you one final result. For a multi-step agent that can take many seconds, and a blank screen feels broken. stream instead yields output as it happens, so you can show progress or token-by-token text. You choose what granularity you want with a stream mode.

for chunk in graph.stream(inputs, config, stream_mode="updates"):
    # chunk = {node_name: {field: new_value}}, one per completed node
    print(chunk)
# Same graph, same state, same memory, just a different way to consume output.

Because the checkpointer saves a snapshot after every step, a thread is not just a "current state", it is a full history of states. That history is a superpower. You can list every checkpoint, inspect what the state looked like at step three, and even resume execution from an earlier checkpoint, optionally after editing it. This is how you debug "why did the agent decide that?" and how you let a user undo a bad turn.

# current snapshot for this thread
snap = graph.get_state(config)
print(snap.values, snap.next)        # state + which node would run next

# every snapshot, newest first
for s in graph.get_state_history(config):
    print(s.config["configurable"]["checkpoint_id"])

# resume from an OLD checkpoint: pass its config back into the graph
graph.invoke(None, an_old_checkpoint_config)   # re-runs forward from there

Upgrade your Module 01 graph into something that holds a conversation. Do it for real, two turns on one thread, then a third turn on a new thread to prove isolation.

from langgraph.graph import StateGraph, START, MessagesState
from langgraph.checkpoint.memory import InMemorySaver

def chat(state: MessagesState):
    # TODO 1: call llm on state["messages"], return {"messages": [reply]}
    ...

builder = StateGraph(MessagesState)
builder.add_node("chat", chat)
builder.add_edge(START, "chat")

# TODO 2: compile with an InMemorySaver()
graph = builder.compile(...)

cfg = {"configurable": {"thread_id": "t1"}}
graph.invoke({"messages": [user("My name is Sahil.")]}, cfg)
out = graph.invoke({"messages": [user("What is my name?")]}, cfg)
# TODO 3: confirm it answers "Sahil". Then repeat with thread_id "t2"
#         and confirm it does NOT know the name.

• Turn two on the same thread correctly recalled a fact from turn one.
• A new thread_id started blank, proving conversations are isolated.
• You added a reducer (Annotated[list, add]) to a custom field and saw it accumulate.
• You called get_state_history and saw more than one checkpoint for a thread.
• You can explain checkpointer vs Store (this-chat vs this-user) without peeking.

More in Module 05. = must-know cold.