Thread persistence with a checkpointer

Without memory, every turn is a cold start. The supervisor has no history, the specialists have no context, and the user has to repeat themselves. LangGraph checkpointers fix this by snapshotting state under a thread_id after every run. The next turn loads that snapshot and the graph picks up where it left off.

Checkpointer lifecycle across turns

How state gets saved and reloaded between chat requests under a thread_id.

agent_graph.py
python
from langgraph.checkpoint.memory import MemorySaver


def build_graph():
    graph = StateGraph(ClaimState)
    # ... add_node and add_edge calls for supervisor and specialists ...
    return graph.compile(checkpointer=MemorySaver())

One line turns the graph from stateless into memory-aware: checkpointer=MemorySaver(). Every time the graph runs, LangGraph writes the final state to the checkpointer, keyed on the thread_id passed in config.

service.py
python
import uuid


def _new_thread_id() -> str:
    return f'thread-{uuid.uuid4().hex[:10]}'


def _thread_config(thread_id: str) -> Dict[str, Any]:
    return {'configurable': {'thread_id': thread_id}}


async def run_chat(
    message: str,
    thread_id: Optional[str] = None,
    customer_id: Optional[str] = None,
) -> Dict[str, Any]:
    thread_id = thread_id or _new_thread_id()
    graph = get_graph()

    snapshot = graph.get_state(_thread_config(thread_id))
    history = (snapshot.values or {}).get('messages', []) if snapshot else []
    history = history + [{'role': 'user', 'content': message}]

    initial = {
        'messages': history,
        'user_input': message,
        'customer_id': customer_id,
    }
    # ... run astream, capture final_state ...

The pattern: generate a thread_id on the first request, return it in the response, expect the client to pass it back on every subsequent request. get_state loads the prior snapshot, the new message gets appended to history, and the graph runs with the combined context.

Quiz: Quiz

Loading practice…