Kaushik's Notes

3. Chatbot with memory

4 min read

  • When you build a basic chatbot using LangGraph, you run into an immediate limitation: by default, your chatbot has amnesia.
  • Every time a user starts a conversation, the bot has no recollection of previous interactions.
  • This happens because without memory management, each invocation of your graph is completely independent.
  • This is where the concept of checkpointers in LangGraph come into the picture.

Checkpointer

  • A checkpointer in LangGraph is essentially a way to save the state of your agent or workflow at specific points during execution.

  • Think of it like saving your progress in a video game. When you reach a checkpoint:

    • The current state of everything is saved
    • If something goes wrong later, you can return to this saved point.

  • In the context of LangGraph nodes and workflows:

    • Nodes are the individual steps or components in your workflow
    • Checkpoints save the complete sate after a node finishes its work if an error occurs in a later node, you can resume from the last checkpoint rather than starting the entire workflow again.

  • This is particularly useful for complex workflows where:

    • Processing takes significant time or resources.
    • You want to implement retry mechanisms.
    • You need persistance across sessions or server restarts.

Thread ID

  • A thread ID is simply a unique identifier for each specific conversation or workflow execution. Think of it like:
    • A unique session ID for a user.
    • A conversation ID that groups related messages together
  • The thread ID is necessary because:
    • You might have multiple conversations/workflows running simultaneously.
    • Each needs its own separate saved state.
    • The thread ID helps the system know which saved state belongs to which conversation
  • Without thread IDs, all your conversations would share the same state, which would cause confusion and errors.

MemorySaver (InMemorySaver)

 
from typing import TypedDict, Annotated
from langgraph.graph import add_messages, StateGraph, END
from langchain_groq import ChatGroq
from langchain_core.messages import AIMessage, HumanMessage
from dotenv import load_dotenv
from langgraph.checkpoint.memory import MemorySaver
 
load_dotenv()
 
memory = MemorySaver()
 
llm = ChatGroq(model="llama-3.1-8b-instant")
 
class BasicChatState(TypedDict):
    messages: Annotated[list, add_messages]
 
 
def chatbot(state: BasicChatState):
    return {
        "messages": [llm.invoke(state["messages"])],
    }
 
graph = StateGraph(BasicChatState)
 
graph.add_node("chatbot", chatbot)
graph.set_entry_point("chatbot")
graph.add_edge("chatbot", END)
 
app = graph.compile(checkpointer=memory)
 
# The config is used to set the thread_id for the in-memory checkpointing
# To store and use the in-memory checkpoint, you need to set the thread_id same for all the conversations
config = {"configurable": {
        "thread_id": 1,
    }
}
 
while True:
    user_input = input("User: ")
    if(user_input == "exit"):
        break
 
    result =  app.invoke({
        "messages": [HumanMessage(content=user_input)]
    }, config=config)
    print("AI: ", result["messages"][-1].content)
 
# The in-memory checkpointing will store the state of the conversations
# You can retrieve the state of the conversations using the get_state methos
print(app.get_state(config=config))

SqliteSaver

from typing import TypedDict, Annotated
from langgraph.graph import add_messages, StateGraph, END
from langchain_groq import ChatGroq
from langchain_core.messages import AIMessage, HumanMessage
from dotenv import load_dotenv
# from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3
 
load_dotenv()
 
# LangGraph works with multiple threads
# Sqlite only supports one thread at a time
# So we need to create a connection to the sqlite database explicitly stating to not check for the same thread
sqlite_conn = sqlite3.connect("checkpoint.sqlite", check_same_thread=False)
 
memory = SqliteSaver(sqlite_conn)
 
llm = ChatGroq(model="llama-3.1-8b-instant")
 
class BasicChatState(TypedDict):
    messages: Annotated[list, add_messages]
 
 
def chatbot(state: BasicChatState):
    return {
        "messages": [llm.invoke(state["messages"])],
    }
 
graph = StateGraph(BasicChatState)
 
graph.add_node("chatbot", chatbot)
graph.set_entry_point("chatbot")
graph.add_edge("chatbot", END)
 
app = graph.compile(checkpointer=memory)
 
# The config is used to set the thread_id for the in-memory checkpointing
# To store and use the in-memory checkpoint, you need to set the thread_id same for all the conversations
config = {"configurable": {
        "thread_id": 1,
    }
}
 
while True:
    user_input = input("User: ")
    if(user_input == "exit"):
        break
 
    result =  app.invoke({
        "messages": [HumanMessage(content=user_input)]
    }, config=config)
    print("AI: ", result["messages"][-1].content)
 
# The in-memory checkpointing will store the state of the conversations
# You can retrieve the state of the conversations using the get_state methos
print(app.get_state(config=config))

Graph View

Backlinks