LangChain Conversation Memory -- Checkpointer

By default, each agent.invoke() is independent, and the Agent does not remember what was said before.

Checkpointer (checkpoint saver) allows the Agent to remember conversation history, enabling true multi-turn conversations.

Problems without Checkpointer

First, let's look at the situation without Checkpointer:

Example

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    system_prompt="You are the assistant of EXAMPLE Tutorial.",
)

# Round 1
result1 = agent.invoke({
    "messages": [HumanMessage(content=My name is Xiaoming)]
})
print(fRound 1: {result1['messages'][-1].content})

# Round 2 — Agent doesn't remember the content of the first round!
result2 = agent.invoke({
    "messages": [HumanMessage(content=What's my name?)]
})
print(fRound 2: {result2['messages'][-1].content})

Output:

Round 1: Hello Xiaoming! Nice to meet you.
Round 2: Sorry, I don't have your information, and I don't know what your name is.

Using Checkpointer to Remember Conversations

After adding Checkpointer, conversations under the same thread_id are automatically linked:

Example

from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

# Create an in-memory Checkpointer
checkpointer = InMemorySaver()

model = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)
agent = create_agent(
    model=model,
    checkpointer=checkpointer,  # Pass in the Checkpointer
    system_prompt="You are the assistant of EXAMPLE Tutorial.",
)

# Use thread_id to identify the conversation thread
config = {"configurable": {"thread_id": "user-001"}}

# Round 1
result1 = agent.invoke(
    {"messages": [HumanMessage(content="My name is Xiaoming, I am learning Python")]},
    config=config,
)
print(fRound 1: {result1['messages'][-1].content})

# Second round — using the same thread_id, the Agent remembered!
result2 = agent.invoke(
    {"messages": [HumanMessage(content="What is my name? What am I learning?")]},
    config=config,
)
print(fRound 2: {result2['messages'][-1].content})

Output:

Round 1: Hello Xiaoming! Python is a great language to start with. How can I help you?
Round 2: Your name is Xiaoming and you are learning Python. Is there anything specific I can help with?

thread_id is the key. Conversations under the same thread_id are continuous, while conversations under different thread_id values are completely isolated. This lets you serve multiple users with a single Agent instance.

How Checkpointer Works

Checkpointer automatically saves a state snapshot (checkpoint) after each Agent execution. The next time the same thread_id is used, the state is automatically restored from the most recent checkpoint.

Detailed workflow:

  1. Call agent.invoke() and pass in config (containing thread_id)
  2. The Agent checks whether a checkpoint exists for that thread_id
  3. If it exists, load historical messages, append new messages, and continue
  4. After execution, automatically save a new checkpoint

Example

from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

checkpointer = InMemorySaver()
agent = create_agent(
    model=init_chat_model("deepseek:deepseek-v4-flash", temperature=0),
    checkpointer=checkpointer,
    system_prompt="You are the assistant of EXAMPLE Tutorial.",
)

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

# Simulate multi-turn conversation
questions = [
    My name is Xiaoming,
    "I am learning Python",
    "Help me summarize information about me",
]

for i, q in enumerate(questions, 1):
    result = agent.invoke(
        {"messages": [HumanMessage(content=q)]},
        config=config,
    )

    # Check checkpoint status
    state = agent.get_state(config)
    print(f"\nAfter round {i}:")
    print(f" Message count: {len(state.values.get('messages', []))}")
    print(f" Next step: {state.next}")
    print(f" Reply: {result['messages'][-1].content[:80]}...")

Output:

After round 1:
  Message count: 2
  Next step: ()
  Reply: Hello Xiaoming! Nice to meet you.

After round 2:
  Message count: 4
  Next step: ()
  Reply: Python is a very popular programming language, simple and easy to learn.

After round 3:
  Message count: 6
  Next step: ()
  Reply: Based on our conversation, here is your info: your name is Xiaoming, and you are learning Python.

Checkpointer Types

Type Storage Location Persistence Installation Applicable Scenarios
InMemorySaver Memory No (lost after program exits) Built-in Development debugging, unit testing
SqliteSaver SQLite database Yes langgraph-checkpoint-sqlite Standalone deployment, small applications
PostgresSaver PostgreSQL Yes langgraph-checkpoint-postgres Production environment, multi-instance sharing

If the Agent usesainvoke()asynchronous calls, you need to switch to the corresponding async versionAsyncSqliteSaver / AsyncPostgresSaver, the usage is similar to the sync version, except it needs to be paired withasync withto use.

SqliteSaver Example

InMemorySaverOnly saved in memory, data is lost after the program exits. If you want conversations to be persisted, you can useSqliteSaver。

First-time use requires installing SQLite Checkpointer:

pip install langgraph-checkpoint-sqlite

Example

from langgraph.checkpoint.sqlite import SqliteSaver
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.messages import HumanMessage

model = init_chat_model(
    "deepseek:deepseek-v4-flash",
    temperature=0,
)

# Must use the with statement to enter; the database connection will be closed automatically upon exit
with SqliteSaver.from_conn_string("conversations.db") as checkpointer:
    agent = create_agent(
        model=model,
        checkpointer=checkpointer,
        system_prompt="You are the assistant of EXAMPLE Tutorial.",
    )

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

    result = agent.invoke(
        {
            "messages": [
                HumanMessage(content=Hello)
            ]
        },
        config=config,
    )

    print(result["messages"][-1].content)

SqliteSaver saves checkpoints generated by each conversation in the SQLite database. Even if the program exits or restarts, as long as you continue using the same database file andthread_id, the Agent can restore the previously saved conversation state. Note:withThe database connection will close after the code block ends, so logic involving multiple Agent calls should be placedwithinside the block; for long-running services (such as web applications), it is recommended to hold the connection during the application lifetime, or use the async version ofAsyncSqliteSaver。

Managing Conversation Threads

Viewing Conversation State

Example

# View conversation status
state = agent.get_state(config)
print(f"Next step: {state.next}")  # () indicates idle
print(f"Message count: {len(state.values.get('messages', []))}")

# View conversation history
for msg in state.values.get("messages", []):
    print(f"  [{msg.type}] {str(msg.content)[:60]}")

Creating a New Thread

Example

# Different thread_id = different independent conversations
config_alice = {"configurable": {"thread_id": "alice"}}
config_bob = {"configurable": {"thread_id": "bob"}}

# Alice's conversation
agent.invoke(
    {"messages": [HumanMessage(content="I am Alice")]},
    config=config_alice,
)

# Bob's conversation—completely independent, doesn't know what Alice said
agent.invoke(
    {"messages": [HumanMessage(content=# Verify isolation)]},
    config=config_bob,
)

"Alice's conversation message count: {len(alice_state.values['messages'])}"
alice_state = agent.get_state(config_alice)
bob_state = agent.get_state(config_bob)
print(f"Bob's conversation message count: {len(bob_state.values['messages'])}")
print(fUpdating State — Manually Modifying Conversations)

Updating State — Manually Modifying Conversations

Example

Example

from langchain.messages import SystemMessage

"(User upgraded to VIP membership)"
agent.update_state(
    config,
    {
        "messages": [
            SystemMessage(content=# Subsequent conversations will include this inserted message)
        ]
    }
)

The parameters of update_state() are processed through the add_messages reducer (for the messages field), so new messages are appended rather than overwritten. If you want to clear history and start over, the simplest way is to use a new

; if you really need to delete some historical messages, you can pass the correspondingthread_id(fromRemoveMessage) to precisely remove the specified messages.langchain.messagesOther Extensions

AI thinking...