AI Chatbot with RAG, Memory and a User Interface

In this tutorial, we’ll explore how to build an AI-powered chatbot using RAG, Qdrant as our vector database and an agent to manage memory. We’ll also add a graphical user interface (GUI).

Qdrant VS Chroma DB

Qdrant is also a vector database. ChromaDB is ideal for developers who want to plug vector databases into AI models quickly and easily, while Qdrant is better suited to enterprise solutions that need high performance and scalability.

Prerequisites

Installing the Required Packages

To start, we need to install all the required Python packages. Use the following pip command to install them all at once:

pip install langchain langchain-qdrant qdrant-client azure-identity pyautogen dotenv langchain-openai

Setting Up the .env File

Create a file called .env in the directory with the following content:

AZURE_TENANT_ID = "your-azure-tenant-id"
AZURE_CLIENT_ID = "your-azure-client-id"
AZURE_CLIENT_SECRET = "your-azure-client-secret"

Replace your-azure-tenant-id, your-azure-client-id and your-azure-client-secret with your real Azure credentials.

Create the Vector DB

We create a Qdrant client and define a collection to store our vectors.

from qdrant_client import QdrantClient
from qdrant_client.http.models import Distance, VectorParams

client = QdrantClient(path="c:\\zeroQB\\")

client.create_collection(
    collection_name="zero_collection",
    vectors_config=VectorParams(size=3072, distance=Distance.COSINE),
)

Finally, we generate unique IDs for each document chunk and add it to Qdrant.

from langchain_qdrant import QdrantVectorStore

vector_store = QdrantVectorStore(
    client=client,
    collection_name="zero_collection",
    embedding=embeddings,
)

uuids = [str(uuid4()) for _ in range(len(docs))]

vector_store.add_documents(documents=docs, ids=uuids)

The Code Step by Step

Load Environment Variables

First, we load the environment variables from the .env file:

from dotenv import load_dotenv
load_dotenv()

Setting Up Azure Credentials and the Access Token

We set up the Azure credentials and get the access token:

import os
from azure.identity import ChainedTokenCredential, EnvironmentCredential

credential = ChainedTokenCredential(EnvironmentCredential())
access_token = credential.get_token("https://cognitiveservices.azure.com/.default")

os.environ.update({
    "AZURE_OPENAI_ENDPOINT": "https://zerogap.openai.azure.com/",
    "AZURE_OPENAI_API_KEY": access_token.token,
    "OPENAI_API_TYPE": "azure_ad",
    "OPENAI_DEPLOYMENT": "gpt-4o"
})

Initialising the OpenAI Models

We initialise the OpenAI models:

from langchain_openai import AzureOpenAIEmbeddings, AzureChatOpenAI

llm = AzureChatOpenAI(openai_api_version="2023-07-01-preview", azure_deployment="gpt-4o", temperature=0.5)
embeddings = AzureOpenAIEmbeddings(azure_deployment="embedding", openai_api_version="2023-07-01-preview", chunk_size=1)

Setting Up the AutoGen Agent

We set up the AutoGen agent:

config_list = [{
    "model": "gpt-4o",
    "api_type": "azure",
    "api_key": os.environ['AZURE_OPENAI_API_KEY'],
    "base_url": os.environ["AZURE_OPENAI_ENDPOINT"],
    "api_version": "2024-02-01"
}]

Load the Vector Database

We load the vector database from an existing collection:

from langchain.vectorstores import Qdrant

vector_store = Qdrant.from_existing_collection(embeddings, path="c:\\zeroQB\\", collection_name="zero_collection")
retriever = vector_store.as_retriever()

Format Documents for Display

We format a list of documents into a single string for display:

def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs) if docs else ""

Tool for the Agent

A tool the agent uses to answer questions from the DB:

def get_documents(question: str) -> str:
    context = retriever.invoke(question)
    return format_docs(context) if context else "No information found in the Zerogap Documents."

Chat History

We manage the chat history by using an agent to summarise the conversation. This reduces the number of tokens needed to process later interactions.

chat_history = []

def chat_history_handler():
    if chat_history:
        summary = "Chat history: " + "\n".join(chat_history)
        result = user_proxy.initiate_chat(
            condenser,
            message=summary,
            summary_method="last_msg",
            max_turns=1,
        )
        return "Chat history summary: " + result.summary
    return ""

def chat_history_parser(chat_result):
    for msg in chat_result.chat_history:
        if 'name' in msg:
            chat_history.append(f"{msg['name']}: {msg['content']}")

Run the Agent

We define how the agent runs to process the user’s query:

def run_agent(query):
    chat_history_context = chat_history_handler()
    chat_result = user_proxy.initiate_chat(
        assistant,
        message=f"{query}\n{chat_history_context}",
        summary_method="last_msg",
        max_turns=2,
    )
    chat_history_parser(chat_result)
    return chat_result.summary

Set Up the Graphical Interface

We set up the graphical interface using tkinter:

import threading
import tkinter as tk
import tkinter.ttk as ttk
from tkinter import scrolledtext

def on_ask():
    def run_query():
        question = question_entry.get()
        if question.lower() == "quit":
            root.destroy()
        else:
            loading_var.set("Loading...")
            progress_bar.pack(pady=5)
            progress_bar.start()
            root.update_idletasks()
            result = run_agent(question)
            loading_var.set("")
            progress_bar.stop()
            progress_bar.pack_forget()
            result_text.config(state=tk.NORMAL)
            result_text.insert(tk.END, f"You: {question}\nBot: {result}\n\n")
            result_text.config(state=tk.DISABLED)
            question_entry.delete(0, tk.END)

    threading.Thread(target=run_query).start()

root = tk.Tk()
root.title("ZEROGAP AI QA Bot")
root.configure(bg="#2e2e2e")

style = ttk.Style()
style.theme_use("clam")
style.configure("TLabel", background="#2e2e2e", foreground="#ffffff", font=("Helvetica", 12))
style.configure("TEntry", fieldbackground="#4d4d4d", foreground="#ffffff")
style.configure("TButton", background="#4d4d4d", foreground="#ffffff", font=("Helvetica", 10, "bold"))
style.map("TButton", background=[("active", "#5e5e5e")])

ttk.Label(root, text="Welcome to the ZEROGAP AI Chat Bot!").pack(pady=5)

question_entry = ttk.Entry(root, width=50)
question_entry.pack(pady=10)

ask_button = ttk.Button(root, text="Ask", command=on_ask)
ask_button.pack(pady=5)

result_text = scrolledtext.ScrolledText(root, width=60, height=20, state=tk.DISABLED, bg="#4d4d4d", fg="#ffffff", insertbackground="#ffffff")
result_text.pack(pady=10)

loading_var = tk.StringVar()
loading_label = ttk.Label(root, textvariable=loading_var, background="#2e2e2e", foreground="#ffffff", font=("Helvetica", 12))
loading_label.pack(pady=5)
loading_var.set("")

progress_bar = ttk.Progressbar(root, mode='indeterminate')

root.mainloop()

Complete code

import os
import threading
import tkinter as tk
import tkinter.ttk as ttk
from tkinter import scrolledtext
from azure.identity import ChainedTokenCredential, EnvironmentCredential
from langchain_openai import AzureOpenAIEmbeddings, AzureChatOpenAI
from langchain.vectorstores import Qdrant
from autogen import register_function, ConversableAgent
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

# Instructions for setting up the environment and installing necessary packages

# 1. Install the required Python packages:
#    Use the following pip command to install all necessary packages in one go:
# pip install langchain langchain-qdrant qdrant-client azure-identity pyautogen dotenv langchain-openai

# 2. Set up the .env file:
#    Create a file named `.env` in the `qdrant_agents` directory with the following content:
#    ```
#    AZURE_TENANT_ID = "your-azure-tenant-id"
#    AZURE_CLIENT_ID = "your-azure-client-id"
#    AZURE_CLIENT_SECRET = "your-azure-client-secret"
#    ```
#    Replace `your-azure-tenant-id`, `your-azure-client-id`, and `your-azure-client-secret` with your actual Azure credentials.

# Set up Azure credentials and access token
credential = ChainedTokenCredential(EnvironmentCredential())
access_token = credential.get_token("https://cognitiveservices.azure.com/.default")

# Set OS environment variables
os.environ.update({
    "AZURE_OPENAI_ENDPOINT": "https://zerogap.openai.azure.com/",
    "AZURE_OPENAI_API_KEY": access_token.token,
    "OPENAI_API_TYPE": "azure_ad",
    "OPENAI_DEPLOYMENT": "gpt-4o"
})

# Initialize the OpenAI models
llm = AzureChatOpenAI(openai_api_version="2023-07-01-preview", azure_deployment="gpt-4o", temperature=0.5)
embeddings = AzureOpenAIEmbeddings(azure_deployment="embedding", openai_api_version="2023-07-01-preview", chunk_size=1)

# Configuration for AutoGen Agent
config_list = [{
    "model": "gpt-4o",
    "api_type": "azure",
    "api_key": os.environ['AZURE_OPENAI_API_KEY'],
    "base_url": os.environ["AZURE_OPENAI_ENDPOINT"],
    "api_version": "2024-02-01"
}]

# Load the vector database from existing collection
vector_store = Qdrant.from_existing_collection(embeddings, path="c:\\zeroQB\\", collection_name="zero_collection")
retriever = vector_store.as_retriever()

# Format documents for display
# Formats a list of documents into a single string for display.
# Args:
#     docs (list): List of document objects.
# Returns:
#     str: Formatted string of document contents.
def format_docs(docs):
    return "\n\n".join(doc.page_content for doc in docs) if docs else ""

# Tool for the Agent to answer questions
# Retrieves documents related to the given question from the vector store.
# Args:
#     question (str): The question to query the vector store.
# Returns:
#     str: Formatted string of document contents or a message if no information is found.
def get_documents(question: str) -> str:
    context = retriever.invoke(question)
    return format_docs(context) if context else "No information found in the Zerogap Documents."

# Chat history
chat_history = []

# Set up Conversable Agents
user_proxy = ConversableAgent(
    name="User",
    is_termination_msg=lambda msg: msg.get("content") is not None and "TERMINATE" in msg["content"],
    human_input_mode="NEVER",
    default_auto_reply="TERMINATE",
)

assistant = ConversableAgent(
    name="Assistant",
    system_message="""You are the ZeroGap AI Bot. You help users with their questions. Follow these instructions step by step:
    1. Answer the user's question you can use any tool available or the chat history.
    2. You end every response with 'TERMINATE'.""",
    llm_config={"config_list": config_list, "cache_seed": None},
    human_input_mode="NEVER",
)

condenser = ConversableAgent(
    name="Condenser",
    system_message="""Your job is to summarize the chat history, keeping the most relevant information. ONLY return the summary, without any other text.""",
    llm_config={"config_list": config_list, "cache_seed": None},
    human_input_mode="NEVER",
)

# Register the Document tool
register_function(
    get_documents,
    caller=assistant,
    executor=user_proxy,
    name="get_documents",
    description="Gets information from Zerogap Documents",
)

# Chat history handler
# Handles the chat history by condensing it into a summary.
# Returns:
#     str: Summary of the chat history.
def chat_history_handler():
    if chat_history:
        summary = "Chat history: " + "\n".join(chat_history)
        # Condense the chat history
        result = user_proxy.initiate_chat(
            condenser,
            message=summary,
            summary_method="last_msg",
            max_turns=1,
        )
        return "Chat history summary: " + result.summary
    return ""

# Chat history parser
# Parses the chat result and updates the chat history.
# Args:
#     chat_result (object): The result object from the chat.
def chat_history_parser(chat_result):
    for msg in chat_result.chat_history:
        if 'name' in msg:
            chat_history.append(f"{msg['name']}: {msg['content']}")

# Run the agent
# Runs the agent to process the given query.
# Args:
#     query (str): The user's query.
# Returns:
#     str: The summary of the agent's response.
def run_agent(query):
    chat_history_context = chat_history_handler()
    chat_result = user_proxy.initiate_chat(
        assistant,
        message=f"{query}\n{chat_history_context}",
        summary_method="last_msg",
        max_turns=2,
    )
    # Parse chat history
    chat_history_parser(chat_result)
    return chat_result.summary

# Handle user input and display results
# Handles the 'Ask' button click event, runs the agent query, and updates the GUI with the result.
def on_ask():
    def run_query():
        question = question_entry.get()
        if question.lower() == "quit":
            root.destroy()
        else:
            loading_var.set("Loading...")
            progress_bar.pack(pady=5)
            progress_bar.start()
            root.update_idletasks()
            result = run_agent(question)
            loading_var.set("")
            progress_bar.stop()
            progress_bar.pack_forget()
            result_text.config(state=tk.NORMAL)
            result_text.insert(tk.END, f"You: {question}\nBot: {result}\n\n")
            result_text.config(state=tk.DISABLED)
            question_entry.delete(0, tk.END)

    threading.Thread(target=run_query).start()

# Set up the GUI with a dark theme
root = tk.Tk()
root.title("ZEROGAP AI QA Bot")
root.configure(bg="#2e2e2e")

# Style configuration
style = ttk.Style()
style.theme_use("clam")
style.configure("TLabel", background="#2e2e2e", foreground="#ffffff", font=("Helvetica", 12))
style.configure("TEntry", fieldbackground="#4d4d4d", foreground="#ffffff")
style.configure("TButton", background="#4d4d4d", foreground="#ffffff", font=("Helvetica", 10, "bold"))
style.map("TButton", background=[("active", "#5e5e5e")])

# Welcome labels
ttk.Label(root, text="Welcome to the ZEROGAP AI Chat Bot!").pack(pady=5)

# Question entry
question_entry = ttk.Entry(root, width=50)
question_entry.pack(pady=10)

# Fancy Ask button
ask_button = ttk.Button(root, text="Ask", command=on_ask)
ask_button.pack(pady=5)

# Result text area
result_text = scrolledtext.ScrolledText(root, width=60, height=20, state=tk.DISABLED, bg="#4d4d4d", fg="#ffffff", insertbackground="#ffffff")
result_text.pack(pady=10)

# Loading bar
loading_var = tk.StringVar()
loading_label = ttk.Label(root, textvariable=loading_var, background="#2e2e2e", foreground="#ffffff", font=("Helvetica", 12))
loading_label.pack(pady=5)
loading_var.set("")

# Add a progress bar
progress_bar = ttk.Progressbar(root, mode='indeterminate')

root.mainloop()