app.py

import os
import gradio as gr #framework
import random
import numpy as np #library
from huggingface_hub import InferenceClient #package
from sentence_transformers import SentenceTransformer #library
import torch #framework

HF_TOKEN = os.environ.get("HF_TOKEN").strip()
#initialized HF client for text gen.
client = InferenceClient(token = HF_TOKEN, model = "Qwen/Qwen2.5-7B-Instruct")
#LOADED EMBEDDING MODEL
embedding_model = SentenceTransformer('all-MiniLM-L6-v2')

with open("knowledgebase.txt", 'r', encoding = 'UTF-8')as file:
    file_to_process = file.read()
def pre_process(text): #creating a function to preprocess text which accepts a text argument
    cleaned_text = text.strip() #strip whole text, then break it down in the next lines into chunks, which are cleaned further
    chunks = cleaned_text.split("\n") #this line needs further explanation
    #the line below needs adequate explanation.
    cleaned_chunks = [chunk.strip() for chunk in chunks if chunk.strip()]
    return cleaned_chunks
def create_embeddings(text_chunks): #embedding creation function in which text_chunks are passed
    if not text_chunks:
        # Defensive block to prevent app from crashing on empty text files
        print("⚠️ WARNING: knowledgebase.txt contains no valid text lines!")
        return torch.empty(0, 384)
    return embedding_model.encode(text_chunks, convert_to_tensor = True) #be careful to take this outside the if statement.
TEXT_CHUNKS = pre_process(file_to_process)
CHUNK_EMBEDDINGS = create_embeddings(TEXT_CHUNKS) #CALLING THE EMBEDDING CREATION FUNCTION TO CREATE EMBEDDINGS FROM THE PRE-PROCESSED TEXT CHUNKS
def get_top_chunks(query, chunk_embeddings, text_chunks, k=3): #are query, chunk embeddings and text chunks just positional args?
    if chunk_embeddings.numel() == 0:
        return ["No knowledge base data found."]
    query_embeddings = embedding_model.encode(query, convert_to_tensor = True)
    query_embeddings_normalized = torch.nn.functional.normalize(query_embeddings, p=2, dim = 0 )
   
    chunk_embeddings_normalized = torch.nn.functional.normalize(chunk_embeddings, p=2, dim=-1)
 # horizontal columns for 384 vector embedding scores
    
    similarities = torch.matmul(chunk_embeddings_normalized, query_embeddings_normalized) #when were they normalized?? 
    #PRINTING TOP INDICES
    actual_K = min(k, len(text_chunks))
    top_indices = torch.topk(similarities, k = actual_K).indices
    top_chunks =  [text_chunks[i.item()]for i in top_indices] #how is this represented in data containers? its difficult to interpret prima facie.
    return top_chunks
    
def respond(message,history): #defining a function, finally for the chatbot to respond
    retrieved_data = get_top_chunks(message, CHUNK_EMBEDDINGS, TEXT_CHUNKS)
    context_str = " ".join(retrieved_data) #joining to a context string to group / aggregate all data
    system_prompt = (
        f"You are an anthropomorphic travel assistant chat bot catered for solo women travellers to ensure their safety comfort and happiness. Prescribe the best locations and answer in ENGLISH and be straightforward and sensitive., and you sustain the chat with friendly role-play mystic conversations based on Asian culture, with the data you use to craft such conversations: {context_str}"
    )
    messages = [{"role":"system", "content": system_prompt}]
    
    for msg in history: #note that history is a list of dicts, so we access values according to keys?
        
        messages.append({"role": msg["role"], "content": msg["content"]})
    
    messages.append({"role":"user", "content": message}) #message argument passed in function to record user message
    response_txt = ""
    
    stream = client.chat_completion( #streaming the bot
    messages=messages,
    max_tokens=500,
    temperature=1,
    stream = True
    )
    for chunk in stream:
        token = chunk.choices[0].delta.content
        if token:
            response_txt+=token
            yield response_txt
mystic_theme = gr.themes.Ocean( #creating a theme for the chatbot ui, with the oceans method in theme
    font = [gr.themes.GoogleFont("Playfair Display"), "ui-sans-serif","sans-serif" ],
    primary_hue = "slate",
    secondary_hue = "emerald" 

)
with gr.Blocks() as chatbot: #write the structured execution and rationale
    with gr.Row():
        pass
        #insert any other selectors
    with gr.Row():
        with gr.Column(scale = 4): #what is the scale for?
            gr.Markdown("🏮VoyagHERS🏮")
            gr.Markdown("A mystic, friendly, cultured companion tailored for solo women travellers in Asia.")
    with gr.Row():
        with gr.Column(scale = 3):
            chatbot_component = gr.Chatbot(
                
                label = "guardian angel conversation",
                height = 500
            )
    #creating placeholder text to reduce UX friction and make the bot more personable. 
    #also adding buttons
    with gr.Row():
        msg_input = gr.Textbox(
            placeholder ="Ask about local customs, safe havens, exploratory strategies...",
            show_label = False,
            scale = 4
        )
        submit_button = gr.Button("Consult 🔮", variant = "primary", scale =1)
    gr.Examples(
        examples = [
            "Where can I find the most trustworthy tour guides in Borneo?",
            "Give me a solo traveler's guide for a safe Taipei night market visit",
            "How to identify suspicious behaviour in local vendors and other tourists?"
        ],
        inputs = msg_input,
        label = "SPEAK WITH THE ULTIMATE VOYAGHER HELPER"
    )    # providing example prompts for the user to choose
    def chat_wrapper(message, history): #created a wrapper to modify/ adapt the functionality of the respond function
        if not message.strip():
            return "", history #unpack to empty str and history
       
        history.append({"role":"user", "content": message})
        yield "", history #why unpack like this
        response_generate = respond(message, history[:-1]) #step of -1 to ensure latest responses get appended to the history
        
        history.append({"role":"assistant", "content":""})
        for partial_text in response_generate:
            history[-1]["content"] = partial_text #latest content
            yield "", history
            
    msg_input.submit(chat_wrapper,
                inputs = [msg_input, chatbot_component],
                outputs = [msg_input, chatbot_component])
    submit_button.click( chat_wrapper,
                    inputs = [msg_input, chatbot_component],
                    outputs = [msg_input, chatbot_component]
    
)

if __name__ == '__main__':
    chatbot.launch(theme = mystic_theme, css = "footer {visibility: hidden}")