Well, this is looking tougher than I expected. My search of the web has not been horribly productive.

I was planning to use Langgraph’s SQLite checkpointer related methods. But, API docs, at least for me, don’t really explain how to use them. Found an example I will try to implement, but it uses OpenAI. I couldn’t find one using MistralAI, so not too sure what I need to change. And, of course, it didn’t use Streamlit. So…

I did find one other one that used custom SQLite code along with Streamlit’s session_state.

I will attempt to code the Langgraph style to persist long-term history. If I can’t get that to work with my Streamlit module, I will take a look at coding the second approach.

Apparently, I need to make sure that the SQLite instance/connection is closed before I exit the Streamlit app. If not the app will likely hang. Or so I read.

But, first I will see if I can get the simpler memory checkpointer to work. If I can, that will likey help with the SQLite method.

In-memory Chat History

Here’s my first kick at the can. I am showing the code from last time plus my first attempt to get an in-memory history working. I mostly just added the Langgraph history persistence related code from the earlier command line chat module to the current version of the Streamlit app module. The necessary imports were already present.

# chatbot_2.py: simple chat bot
# ver 0.1.0: code an interactive chat app that can run in web browser
# ver 0.2.0: add in-memory chat history
from typing import Annotated
from typing_extensions import TypedDict

import os, time
from dotenv import load_dotenv
from pathlib import Path

from langchain_mistralai.chat_models import ChatMistralAI

from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START

import streamlit as st

load_dotenv()

api_ky = os.getenv("MISTRAL_API_KEY")
model = ChatMistralAI(model="mistral-large-latest", api_key=api_ky)

st_tst = True


def get_file_as_string(f_pth):
  with open(f_pth, encoding="utf-8") as f_md:
    return f_md.read()


# define the function that processes the node, in our case calls the model
# the input parameter state is type annotated as MessagesState
def call_model(state: MessagesState):
    # get llm response to current chat history
    response = model.invoke(state["messages"])
    # update message history with response
    return {"messages": response}


# Test streamlit
if st_tst:
  do_slct = False
  do_rd = True
  # Title of the app
  app_ttl = "# Langchain &amp; Streamlit Demo:<br>LLM Chatbot"

  # Render the readme as markdown using st.markdown.
  f_pth = Path("./app_intro.md")
  readme_text = st.markdown(get_file_as_string(f_pth), unsafe_allow_html=True)

  # define a new graph
  workflow = StateGraph(state_schema=MessagesState)

  # add our single node and an edge to the graph
  # the node is labelled as "model" and call_model is added as the callback function
  workflow.add_node("model", call_model)
  # there are no other nodes, but we need an edge from the special START node to get things going
  # i.e. define the entry point to the graph
  workflow.add_edge(START, "model")

  # compile graph specifying MemorySaver as the checkpointer
  mem = MemorySaver()
  app = workflow.compile(checkpointer=mem)

  # use configuration obj to specify conversation id
  cnfg = {"configurable": {"thread_id": "bark1"}}

  # Once we have the dependencies, add a selector for the app mode on the sidebar.
  st.sidebar.title("What to do")
  if do_slct:
    app_mode = st.sidebar.selectbox("Choose the app mode",
      ["Show instructions", "Show the source code", "Run the app"])
  elif do_rd:
    app_mode = st.sidebar.radio("Choose the app mode",
      ["Show instructions", "Show the source code", "Run the app"])

  if app_mode == "Show instructions":
    st.title("")
    st.sidebar.success('To continue select "Run the app".')
  elif app_mode == "Show the source code":
    st.title("Streamlit App with Sidebar")
    readme_text.empty()
    st.code(get_file_as_string("chatbot_2.py"))
  elif app_mode == "Run the app":
    st.markdown(app_ttl, unsafe_allow_html=True)
    readme_text.empty()
    with st.form("my_form", clear_on_submit=True):
      u_inp = st.text_area(
          "Enter text:",
            placeholder="Say something to the LLM or ask a question",
            key="llm_ask",
      )
      submitted = st.form_submit_button("Submit")
      if submitted:
        in_msgs = [HumanMessage(u_inp)]
        llm_said = app.invoke({"messages": in_msgs}, cnfg)
        st.write(llm_said["messages"][-1].content)

And, here’s a test run. The first input was “My pet’s name is Francine.” The second was “Do you recall my pet’s name.”

Test Run of Streamlit App Using Langgraph to Persist Chat History
First chat query and AI response Second chat query and AI response

Needless to say, that did not work. It took me quite some time to sort out why?

Each time a user interacts with the app — whether by changing a widget value (like a slider or button), uploading a file, or adjusting parameters — Streamlit automatically triggers a rerun of the entire script. This design ensures the app remains responsive and dynamic by continuously updating the interface in response to user input.

LangGraph with Streamlit Intersection, Shivanshu Gupta

Guess I should have paid more attention to the Streamlit documentation.

But that also means, that all of the variables in the Python code get reset as well. So each time I sent a query to the Langgraph app, Streamlit reloaded the script and my MemorySaver was re-initialized. The solution was to put the MemorySaver initalization in a function and decorate the function with @st.cache_resource to ensure Streamlit was maintaining the checkpointer object between requests to the AI.

Here’s the refactored code. No changes elsewhere.

  # compile graph specifying MemorySaver as the checkpointer
  @st.cache_resource
  def init_mem():
    mem = MemorySaver()
    return mem
  app = workflow.compile(checkpointer=init_mem())

Well, I did in fact refactor some other code. But, that was not required to make things work. I just wanted to see the whole conversation after each request. That saves me from loading multiple images as I did above.

... ...
        llm_said = app.invoke({"messages": in_msgs}, cnfg)
        for i in range(len(llm_said["messages"])):
          if i % 2 == 0:
            st.write(f'Me: {llm_said["messages"][i].content}')
          else:
            st.write(f'AI: {llm_said["messages"][i].content}')

And here’s the result (image) of that refactoring.

Successful Session of Streamlit App Using Langgraph to Persist Chat History
successful chat session with persisted chat history

And as you can see the conversation history was being maintained by Streamlit.

At this point, I used the menu to force Streamlit to Rerun the app. The display reverted to its initial state—i.e. the state immediately after Run the app was selected the first time. But, when I submitted a new bit of text, the whole history was faithfully displayed. As you can see below.

A Further Successful Test of Persisting Chat History
successful chat session with persisted chat history

This One Finished

This simple modification and test has taken me much longer than I expected. So, I think I am going to consider this post done. Mainly because I expect the attempt to use a database is going to take me a fair bit longer and probably require considerably more discovery and learning. As such, it likely deserves a post of its own.

Until next time, may your coding of chatbots take less effort than my current attempt is requiring of me.

Resources