Short-Term Memory & Checkpointers

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(model, tools=[], checkpointer=InMemorySaver())   # model: ChatOllama in production

alice = {"configurable": {"thread_id": "alice-2026-09-29"}}
bob = {"configurable": {"thread_id": "bob-2026-09-29"}}

agent.invoke({"messages": [{"role": "user", "content": "Hi, my name is Alice."}]}, alice)
agent.invoke({"messages": [{"role": "user", "content": "Hi, my name is Bob."}]}, bob)

for label, cfg in [("alice", alice), ("bob", bob)]:
    reply = agent.invoke({"messages": [{"role": "user", "content": "What is my name?"}]}, cfg)
    print(label, "->", reply["messages"][-1].content, f"({len(reply['messages'])} messages)")

fresh = {"configurable": {"thread_id": "carol-2026-09-29"}}
print("carol ->", agent.invoke({"messages": [{"role": "user", "content": "What is my name?"}]}, fresh)["messages"][-1].content)
agent.invoke({"messages": [{"role": "user", "content": "hi"}]})    # no thread_id

alice -> Your name is Alice. (4 messages)
bob -> Your name is Bob. (4 messages)
carol -> I do not know your name.
ValueError Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id

from langchain.agents import create_agent
from langgraph.checkpoint.sqlite import SqliteSaver

path = "docsassistant-checkpoints.sqlite"
cfg = {"configurable": {"thread_id": "alice-2026-09-29"}}

with SqliteSaver.from_conn_string(path) as saver:          # first "process"
    agent = create_agent(model, tools=[], checkpointer=saver)
    agent.invoke({"messages": [{"role": "user", "content": "Hi, my name is Alice."}]}, cfg)

with SqliteSaver.from_conn_string(path) as saver:          # after a "restart": new saver, new agent, same file
    agent = create_agent(model, tools=[], checkpointer=saver)
    reply = agent.invoke({"messages": [{"role": "user", "content": "What is my name?"}]}, cfg)
    print(reply["messages"][-1].content, f"({len(reply['messages'])} messages)")

import os

from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver

DB_URI = os.environ["DOCSASSISTANT_DB_URL"]          # e.g. postgresql://user@host:5432/docsassistant
cfg = {"configurable": {"thread_id": "alice-2026-09-29"}}

with PostgresSaver.from_conn_string(DB_URI) as saver:
    saver.setup()                                     # creates the checkpoint tables (idempotent)
    agent = create_agent(model, tools=[], checkpointer=saver)
    agent.invoke({"messages": [{"role": "user", "content": "Hi, my name is Alice."}]}, cfg)

with PostgresSaver.from_conn_string(DB_URI) as saver:   # a different worker, later
    agent = create_agent(model, tools=[], checkpointer=saver)
    reply = agent.invoke({"messages": [{"role": "user", "content": "What is my name?"}]}, cfg)
    print(reply["messages"][-1].content)

import sqlite3

from langgraph.checkpoint.serde.encrypted import EncryptedSerializer
from langgraph.checkpoint.sqlite import SqliteSaver

serde = EncryptedSerializer.from_pycryptodome_aes()          # reads LANGGRAPH_AES_KEY from the environment
saver = SqliteSaver(sqlite3.connect("encrypted.sqlite", check_same_thread=False), serde=serde)

agent = create_agent(model, tools=[], checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "state-demo"}}
agent.invoke({"messages": [{"role": "user", "content": "Hi, my name is Alice."}]}, cfg)

snap = agent.get_state(cfg)
print("next:", snap.next, "| step:", snap.metadata["step"], "| source:", snap.metadata["source"])
print("messages:", [(m.type, m.content) for m in snap.values["messages"]])

agent.update_state(cfg, {"messages": [{"role": "user", "content": "Correction: my name is Alicia."}]})
snap2 = agent.get_state(cfg)
print("after update_state:", [(m.type, m.content) for m in snap2.values["messages"]])
print("source:", snap2.metadata["source"])

reply = agent.invoke({"messages": [{"role": "user", "content": "What is my name?"}]}, cfg)
print(reply["messages"][-1].content)

for s in agent.get_state_history(cfg):
    print(" step", s.metadata["step"], s.metadata["source"], "next=", s.next, "msgs=", len(s.values.get("messages", [])))

next: () | step: 1 | source: loop
messages: [('human', 'Hi, my name is Alice.'), ('ai', 'Noted (1 messages in view).')]
after update_state: [('human', 'Hi, my name is Alice.'), ('ai', 'Noted (1 messages in view).'), ('human', 'Correction: my name is Alicia.')]
source: update
Your name is Alicia.
 step 5 loop next= () msgs= 5
 step 4 loop next= ('model',) msgs= 4
 step 3 input next= ('__start__',) msgs= 3
 step 2 update next= () msgs= 3
 step 1 loop next= () msgs= 2
 step 0 loop next= ('model',) msgs= 1
 step -1 input next= ('__start__',) msgs= 0

from langchain.agents import create_agent, AgentState


class DocsState(AgentState):
    user_id: str
    preferences: dict


agent = create_agent(model, tools=[], state_schema=DocsState, checkpointer=InMemorySaver())
agent.invoke(
    {"messages": [{"role": "user", "content": "Hello"}], "user_id": "user_123", "preferences": {"theme": "dark"}},
    {"configurable": {"thread_id": "1"}},
)

graph.checkpointer.delete_thread("answer-42")
print(graph.get_state(cfg).values, list(graph.get_state_history(cfg)))     # -> {} []

import operator
import os
import tempfile
from collections import Counter
from typing import Annotated

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict

calls = Counter()
network_up = False          # the link checker's upstream is down on the first run


class State(TypedDict):
    question: str
    notes: Annotated[list[str], operator.add]


def retrieve(state: State):
    calls["retrieve"] += 1
    return {"notes": ["retrieved 5 chunks"]}


def draft_answer(state: State):
    calls["draft_answer"] += 1                    # stands in for an expensive LLM call
    return {"notes": ["drafted answer"]}


def check_links(state: State):
    calls["check_links"] += 1
    if not network_up:
        raise ConnectionError("link checker unreachable")
    return {"notes": ["links ok"]}


def publish(state: State):
    calls["publish"] += 1
    return {"notes": ["published"]}


builder = StateGraph(State)
for fn in (retrieve, draft_answer, check_links, publish):
    builder.add_node(fn)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "draft_answer")     # draft_answer and check_links run in the same super-step
builder.add_edge("retrieve", "check_links")
builder.add_edge("draft_answer", "publish")
builder.add_edge("check_links", "publish")
builder.add_edge("publish", END)

path = os.path.join(tempfile.mkdtemp(), "durable.sqlite")
cfg = {"configurable": {"thread_id": "answer-42"}}

with SqliteSaver.from_conn_string(path) as saver:
    graph = builder.compile(checkpointer=saver)
    try:
        graph.invoke({"question": "How do I tune HNSW?", "notes": []}, cfg, durability="sync")
    except ConnectionError as err:
        print("run 1 failed:", err)
    snap = graph.get_state(cfg)
    print("next after failure:", sorted(snap.next), "| notes:", snap.values["notes"])
    print("calls so far:", dict(calls))

    network_up = True                              # the dependency recovers
    result = graph.invoke(None, cfg, durability="sync")   # None = resume this thread from its last checkpoint
    print("resumed notes:", result["notes"])
    print("calls in total:", dict(calls))

run 1 failed: link checker unreachable
next after failure: ['check_links'] | notes: ['retrieved 5 chunks', 'drafted answer']
calls so far: {'retrieve': 1, 'check_links': 1, 'draft_answer': 1}
resumed notes: ['retrieved 5 chunks', 'links ok', 'drafted answer', 'published']
calls in total: {'retrieve': 1, 'check_links': 2, 'draft_answer': 1, 'publish': 1}

import operator
import os
import sys
from typing import Annotated

from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict

DB, MODE, PHASE = sys.argv[1], sys.argv[2], sys.argv[3]      # PHASE: crash | resume


class State(TypedDict):
    notes: Annotated[list[str], operator.add]


def retrieve(state: State):
    return {"notes": ["retrieved"]}


def draft_answer(state: State):
    print("  draft_answer runs (expensive)")
    return {"notes": ["drafted"]}


def publish(state: State):
    if PHASE == "crash":
        os._exit(1)                                   # simulate the process being killed mid-run
    return {"notes": ["published"]}


builder = StateGraph(State)
for fn in (retrieve, draft_answer, publish):
    builder.add_node(fn)
builder.add_edge(START, "retrieve")
builder.add_edge("retrieve", "draft_answer")
builder.add_edge("draft_answer", "publish")
builder.add_edge("publish", END)

cfg = {"configurable": {"thread_id": "answer-42"}}
with SqliteSaver.from_conn_string(DB) as saver:
    graph = builder.compile(checkpointer=saver)
    snap = graph.get_state(cfg)
    print(f"[{PHASE}/{MODE}] on start: next={snap.next} notes={snap.values.get('notes')}")
    if PHASE == "crash":
        graph.invoke({"notes": []}, cfg, durability=MODE)
    else:
        print("  result:", graph.invoke(None if snap.next else {"notes": []}, cfg, durability=MODE)["notes"])

[crash/sync] on start: next=() notes=None
  draft_answer runs (expensive)
  exit code 1
[resume/sync] on start: next=('publish',) notes=['retrieved', 'drafted']
  result: ['retrieved', 'drafted', 'published']
[crash/exit] on start: next=() notes=None
  draft_answer runs (expensive)
  exit code 1
[resume/exit] on start: next=() notes=None
  draft_answer runs (expensive)
  result: ['retrieved', 'drafted', 'published']