YouTip LogoYouTip

Langchain Checkpointer

Title: LangChain Conversation Memory β€” Checkpointer | Tutorial\\n\\nBy default, each agent.invoke() is independent β€” the Agent doesn't remember previous conversations. Checkpointer allows the Agent to remember conversation history, enabling true multi-turn dialogue.\\n\\n* * *\\n\\n## Problems without Checkpointer\\n\\nLet's first see what happens without Checkpointer:\\n\\n## Example\\n\\nfrom dotenv import load_dotenv\\n\\n load_dotenv()\\n\\nfrom langchain.agents import create_agent\\n\\nfrom langchain.chat_models import init_chat_model\\n\\nfrom langchain.messages import HumanMessage\\n\\nmodel = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)\\n\\n agent = create_agent(\\n\\n model=model,\\n\\n system_prompt="You areTutorial TUTORIAL assistant。",\\n\\n)\\n\\n# First round\\n\\n result1 = agent.invoke({\\n\\n"messages": [HumanMessage(content="My name is Xiaoming")]\\n\\n})\\n\\nprint(f"First Round: {result1['messages'].content}")\\n\\n# Second round β€” Agent doesn't remember the first round!\\n\\n result2 = agent.invoke({\\n\\n"messages": [HumanMessage(content="What is my name?")]\\n\\n})\\n\\nprint(f"Second Round: {result2['messages'].content}")\\n\\nOutput:\\n\\nFirst round: Hello Xiao Ming! Nice to meet you.\\nSecond round: Sorry, I don't have your information, so I don't know your name.\\n\\n* * *\\n\\n## Using Checkpointer to Remember Conversations\\n\\nAfter adding Checkpointer, conversations under the same thread_id will be automatically associated:\\n\\n## Example\\n\\nfrom langgraph.checkpoint.memory import InMemorySaver\\n\\nfrom langchain.agents import create_agent\\n\\nfrom langchain.chat_models import init_chat_model\\n\\nfrom langchain.messages import HumanMessage\\n\\n# Create an in-memory Checkpointer\\n\\n checkpointer = InMemorySaver()\\n\\nmodel = init_chat_model("deepseek:deepseek-v4-flash", temperature=0)\\n\\n agent = create_agent(\\n\\n model=model,\\n\\n checkpointer=checkpointer,# Pass in Checkpointer\\n\\n system_prompt="You areTutorial TUTORIAL assistant。",\\n\\n)\\n\\n# Use thread_id to identify the conversation thread\\n\\n config ={"configurable": {"thread_id": "user-001"}}\\n\\n# First round\\n\\n result1 = agent.invoke(\\n\\n{"messages": [HumanMessage(content="My name is Xiaoming,I am learning Python")]},\\n\\n config=config,\\n\\n)\\n\\nprint(f"First round: {result1['messages'].content}")\\n\\n# Second round β€” using the same thread_id, Agent remembers!\\n\\n result2 = agent.invoke(\\n\\n{"messages": [HumanMessage(content="What is my name?What am I learning?")]},\\n\\n config=config,\\n\\n)\\n\\nprint(f"Second round: {result2['messages'].content}")\\n\\nOutput:\\n\\nFirst round: Hello Xiao Ming! Python is a greatBeginner Language, is there anything I can help you with?\\nSecond round: Your name is Xiao Ming, you are learning Python. Do you have any specific questions I can help you with?\\n\\n> thread_id is the key. Conversations under the same thread_id are continuous, while conversations between different thread_ids are completely isolated. This allows you to use a single Agent instance to serve multiple users simultaneously.\\n\\n* * *\\n\\n## How Checkpointer Works\\n\\nCheckpointer automatically saves state snapshots (checkpoints) after each Agent execution. When called next time with the same thread_id, it automatically restores from the most recent checkpoint.\\n\\nSpecific workflow:\\n\\n1. Call agent.invoke(), passing in config (containing thread_id)\\n2. Agent checks if there's a checkpoint for this thread_id\\n3. If yes, load historical messages, append new messages, and continue\\n4. After execution completes, automatically save new checkpoint\\n\\n## Example\\n\\nfrom langgraph.checkpoint.memory import InMemorySaver\\n\\nfrom langchain.agents import create_agent\\n\\nfrom langchain.chat_models import init_chat_model\\n\\nfrom langchain.messages import HumanMessage\\n\\ncheckpointer = InMemorySaver()\\n\\n agent = create_agent(\\n\\n model=init_chat_model("deepseek:deepseek-v4-flash", temperature=0),\\n\\n checkpointer=checkpointer,\\n\\n system_prompt="You areTutorial TUTORIAL assistant。",\\n\\n)\\n\\nconfig ={"configurable": {"thread_id": "demo-001"}}\\n\\n# Simulate multi-turn conversation\\n\\n questions =[\\n\\n"My name is Xiaoming",\\n\\n"I am learning Python",\\n\\n"Help me summarize information about me",\\n\\n]\\n\\nfor i, q in enumerate(questions,1):\\n\\n result = agent.invoke(\\n\\n{"messages": [HumanMessage(content=q)]},\\n\\n config=config,\\n\\n)\\n\\n# View checkpoint state\\n\\n state = agent.get_state(config)\\n\\nprint(f"\\\\n After round {i}:")\\n\\nprint(f" Message count: {len(state.values.get('messages', []))}")\\n\\nprint(f" Next step: {state.next}")\\n\\nprint(f" Reply: {result['messages'].content[:80]}...")\\n\\nOutput:\\n\\nAfter round 1: Message count: 2 Next step: () Reply: Hello Xiao Ming! Nice to meet you.\\nAfter round 2: Message count: 4 Next step: () Reply: Python is a very popular programming language, easy to learn.\\nAfter round 3: Message count: 6 Next step: () Reply: Based on our conversation, your information is as follows: Your name is Xiao Ming, learning Python.\\n\\n* * *\\n\\n## Checkpointer Types\\n\\n| Type | Storage Location | Persistence | Use Case |\\n| --- | --- | --- | --- |\\n| InMemorySaver | Memory | No (lost on restart) | Development debugging, testing |\\n| SqliteSaver | SQLite file | Yes | Single-machine deployment, small-scale applications |\\n| PostgresSaver | PostgreSQL | Yes | Production environment, multi-instance sharing |\\n\\n### SqliteSaver Example\\n\\n## Example\\n\\nfrom langgraph.checkpoint.sqlite import SqliteSaver\\n\\n# SQLite persistence β€” data persists after restart\\n\\n checkpointer = SqliteSaver.from_conn_string("conversations.db")\\n\\nagent = create_agent(\\n\\n model="deepseek:deepseek-v4-flash",\\n\\n checkpointer=checkpointer,\\n\\n)\\n\\n# Usage is exactly the same as InMemorySaver\\n\\n config ={"configurable": {"thread_id": "user-001"}}\\n\\n result = agent.invoke(\\n\\n{"messages": [{"role": "user","content": "Hello"}]},\\n\\n config=config,\\n\\n)\\n\\n# After restarting the program, conversations with the same thread_id still exist\\n\\n* * *\\n\\n## Managing Conversation Threads\\n\\n### Viewing Conversation State\\n\\n## Example\\n\\n# View conversation state\\n\\n state = agent.get_state(config)\\n\\nprint(f"Next step: {state.next}")# () means idle\\n\\nprint(f"Message count: {len(state.values.get('messages', []))}")\\n\\n# View conversation history\\n\\nfor msg in state.values.get("messages",[]):\\n\\nprint(f" [{msg.type}] {str(msg.content)[:60]}")\\n\\n### Creating New Threads\\n\\n## Example\\n\\n# Different thread_id = different independent conversation\\n\\n config_alice ={"configurable": {"thread_id": "alice"}}\\n\\n config_bob ={"configurable": {"thread_id": "bob"}}\\n\\n# Alice's conversation\\n\\n agent.invoke(\\n\\n{"messages": [HumanMessage(content="I am Alice")]},\\n\\n config=config_alice,\\n\\n)\\n\\n# Bob's conversation β€” completely independent, doesn't know what Alice said\\n\\n agent.invoke(\\n\\n{"messages": [HumanMessage(content="I am Bob")]},\\n\\n config=config_bob,\\n\\n)\\n\\n# Verify isolation\\n\\n alice_state = agent.get_state(config_alice)\\n\\n bob_state = agent.get_state(config_b
← Langchain Human In The LoopLangchain Wrap Tool Call β†’