Implementing save points in llm

Table of contents

Intro

LLM are stateless and input in goes and out, how do we maintain state and retrieve it from a particular state? We achieve it using checkpointer in langgraph that can save state and retrieve and continue from the state. It’s pretty useful for the use case like long running agents and HITL based agents where the agent can be put too sleep and continue when the external request to start is triggered.

We will see this in action for a simple check pointer chat application using postgres.

Packages required

uv add langgraph langgraph-checkpoint-postgres psycopg[binary] psycopg-pool langchain-openai

Implementation

We are using postgres for the save / check points for this application. Ensure you have your postgres instance up and running for this activity.

• Defining the graph state

# 1. Graph state schema
class State(TypedDict):
    messages: Annotated[list, add_messages]

• LLM and the chatbot node

# 2. LLM + node
llm = ChatOpenAI(model="gpt-4o-mini")


def chatbot(state: State):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

• Building the graph

# 3. Build graph
builder = StateGraph(State)
builder.add_node("chatbot", chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)

• Postgres checkpoint and invoking the graph

We first setup the postgres PostgresSaver to invoke the setup() method which creates the checkpoint tables if not available. These tables are used by the langgraph to implement the checkpoint saving functionality.

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # idempotent; fine to call every run, but only needed once
    graph = builder.compile(checkpointer=checkpointer)

    # --- Thread A: Alice ---
    run_turn(graph, "alice_thread", "Hi! My name is Alice and my favorite color is teal.")
    run_turn(graph, "alice_thread", "What is my name and favorite color?")

    # --- Thread B: a completely separate conversation ---
    run_turn(graph, "bob_thread", "What is my name and favorite color?")
    # Bob's thread has no prior state, so the model should have no idea

    # --- Back to Thread A: state persists independently per thread ---
    run_turn(graph, "alice_thread", "I live in Wonderland. Where do I live?")

    # --- Inspect stored state directly ---
    state = graph.get_state({"configurable": {"thread_id": "alice_thread"}})
    print(f"Alice's thread has {len(state.values['messages'])} messages stored.")

def run_turn(graph, thread_id: str, user_text: str):
    """Send one message on a given thread and print the reply."""
    config = {"configurable": {"thread_id": thread_id}}
    result = graph.invoke({"messages": [HumanMessage(content=user_text)]}, config)
    print(f"[{thread_id}] User: {user_text}")
    print(f"[{thread_id}] Assistant: {result['messages'][-1].content}\n")

The run_turn method invokes the graph from the thread_id to resume from the chat from the checkpoint. Below is the run results.

Conclusion

This application demonstrates an simple checkpoint using postgres and langgraph, in the upcoming articles we can see long running agent with HITL implementation. If you want to see the source code of the above example refer here.