Search code examples
pythonsqlitechatgradiosqlite3-python

How can I update display of chat history upon page refresh?


My chat UI using Gradio:

import sqlite3
import gradio as gr
import time

formatted_history = []
sqlite = None

def loadHistoryFromDB():
    global sqlite,formatted_history

    sql="SELECT role,message from chat_history order by created_at_unix ASC Limit 10";
    cur = sqlite.cursor()
    cur.execute(sql)
    rows = cur.fetchall()
    for row in rows:
        formatted_history.append({
            'role':row[0],
            'content':row[1]
        })

def chatHandler(message,history,generate_image):
    sql = "INSERT INTO chat_history (role, message, created_at_unix) VALUES (?, ?, ?)"
    current_unix_time = int(time.time())  # Get the current Unix timestamp

    response = f"Hello {message}"

    sqlite.execute(sql, ('user', message, int(time.time())))
    time.sleep(3)
    sqlite.execute(sql, ('assistant', response, int(time.time())))
    sqlite.commit()

    yield response

if __name__ == "__main__":

    sqlite = sqlite3.connect("chat_history.db",check_same_thread=False)
    loadHistoryFromDB()

    with gr.Blocks() as demo:
        chatbot = gr.Chatbot(type="messages", value=formatted_history)
        chat = gr.ChatInterface(chatHandler,type="messages",chatbot=chatbot)

    demo.launch()

I use sqlite3 to store chat history. But a page refresh fails to retrieve recently sent messages. I only get the ones already fetched from database.

The bug occurs when:

  1. I launch my python script.
  2. I visit page indicated by Gradio.
  3. I place a message.
  4. I then refresh the page.

How can I show recently sent messages on page refresh and not only history already loaded? I tried:

import sqlite3
import gradio as gr
import time

formatted_history = []
sqlite = None

def loadHistoryFromDB():
    global sqlite,formatted_history

    sql="SELECT role,message from chat_history order by created_at_unix ASC Limit 10";
    cur = sqlite.cursor()
    cur.execute(sql)
    rows = cur.fetchall()
    for row in rows:
        formatted_history.append({
            'role':row[0],
            'content':row[1]
        })

def chatHandler(message,history,generate_image):
    sql = "INSERT INTO chat_history (role, message, created_at_unix) VALUES (?, ?, ?)"
    current_unix_time = int(time.time())  # Get the current Unix timestamp

    response = f"Hello {message}"
    history.append({"role":'user','content':message})
    sqlite.execute(sql, ('user', message, int(time.time())))
    time.sleep(3)
    history.append({"role":'assistant','content':response})
    sqlite.execute(sql, ('assistant', response, int(time.time())))
    sqlite.commit()

    yield response

if __name__ == "__main__":

    sqlite = sqlite3.connect("chat_history.db",check_same_thread=False)
    loadHistoryFromDB()

    with gr.Blocks() as demo:
        chatbot = gr.Chatbot(type="messages", value=formatted_history)
        chat = gr.ChatInterface(chatHandler,type="messages",chatbot=chatbot)

    demo.launch()

But I still fail to load the most recent chat history upon refresh.


Solution

  • The solution is to turn formatted_history into a state. In order to do this is to use demo.load at your code and set as output the chatbot element.

    The final example is:

    import sqlite3
    import gradio as gr
    import time
    
    formatted_history = []
    sqlite = None
    
    
    def loadHistoryFromDB():
        """Fetch the latest 10 messages from the database and populate `formatted_history`."""
        global sqlite, formatted_history
    
        sql = "SELECT role, message FROM chat_history ORDER BY created_at_unix ASC LIMIT 10"
        cur = sqlite.cursor()
        cur.execute(sql)
        rows = cur.fetchall()
        for row in rows:
            formatted_history.append({
                'role': row[0],
                'content': row[1]
            })
    
    
    def chatHandler(message, history):
        global sqlite,formatted_history
        """Handle new messages and update chat history in the database and UI."""
        sql = "INSERT INTO chat_history (role, message, created_at_unix) VALUES (?, ?, ?)"
        current_unix_time = int(time.time())  # Get the current Unix timestamp
    
        # Save user's message to the database
        sqlite.execute(sql, ('user', message, current_unix_time))
        formatted_history.append({"role":"user","content":message})
        # Generate the assistant's response
        response = f"Hello {message}"
    
        # Save assistant's response to the database
        sqlite.execute(sql, ('assistant', response, int(time.time())))
        formatted_history.append({"role":"assistant","content":response})
        sqlite.commit()
    
        yield response
    
    if __name__ == "__main__":
        # Initialize the database connection
        sqlite = sqlite3.connect("chat_history.db", check_same_thread=False)
    
        # Load initial chat history
        loadHistoryFromDB()
    
        # Set up the Gradio interface
        with gr.Blocks() as demo:
            chatbot = gr.Chatbot(type="messages", value=formatted_history)
            chat = gr.ChatInterface(fn=chatHandler, type="messages", chatbot=chatbot)
            demo.load(lambda :formatted_history,None,chatbot)
    
        # Launch the application
        demo.launch()
    

    First and foremost pay attention at:

        with gr.Blocks() as demo:
            chatbot = gr.Chatbot(type="messages", value=formatted_history)
            chat = gr.ChatInterface(fn=chatHandler, type="messages", chatbot=chatbot)
            demo.load(lambda :formatted_history,None,chatbot)
    

    As explained above, I set the chatbot as output of the whatever I store upon formatted_history. Another way you should pay attention is in the function chatHandler In that we append the nessesary messages into formatted_history:

    def chatHandler(message, history):
        global sqlite,formatted_history
        """Handle new messages and update chat history in the database and UI."""
        sql = "INSERT INTO chat_history (role, message, created_at_unix) VALUES (?, ?, ?)"
        current_unix_time = int(time.time())  # Get the current Unix timestamp
    
        sqlite.execute(sql, ('user', message, current_unix_time))
        
        # Append Value upon `formatted_history` in order to update the state
        formatted_history.append({"role":"user","content":message})
    
        # Generate the assistant's response
        response = f"Hello {message}"
    
        sqlite.execute(sql, ('assistant', response, int(time.time())))
    
        # Append Value upon `formatted_history` in order to update the state
        formatted_history.append({"role":"assistant","content":response})
        sqlite.commit()
    
        yield response