Update chatbot.py

chatbot.py

@@ -1,75 +1,108 @@
 import os
 import time
-import pandas as pd
-from huggingface_hub import hf_hub_download
+import json
+import logging
 from groq import Groq
 from langchain.memory import ConversationBufferMemory
+from langchain_openai import ChatOpenAI
+from langchain_community.document_loaders import CSVLoader
 from langchain_community.vectorstores import FAISS
 from deep_translator import GoogleTranslator
 
+# Set up logging
+logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
 
 class Comsatsbot:
-    def __init__(self, hf_space_repo, llm, api_keys, chats_collection,index_path: str = "faiss_kb" ):
+    def __init__(self, hf, llm, api_keys, chats_collection, paths, index_path='faiss_kb'):
+        logger.info("Initializing Comsatsbot...")
         self.llm = llm
         self.api_keys = api_keys
         self.client = None
         self.models = [
+           # "llama3-groq-70b-8192-tool-use-preview",
             "llama-3.3-70b-versatile",
             "llama3-70b-8192"
         ]
         self.memory = ConversationBufferMemory(llm=self.llm, max_token_limit=3000)
         self.chats_collection = chats_collection
         self.index_path = index_path
-        self.hf_space_repo = hf_space_repo
+        self.hf = hf
         self.faiss_index = None
         self.faiss_retriever = None
+        self.paths = paths
         self.initialize_faiss_index()
 
-    def load_data_from_hf_space(self):
-        files = ["english_data.csv", "urdu_data.csv", "FYP Supervisor Feedback.csv"]
+    def load_data(self, paths):
+        logger.info(f"Loading data from paths: {paths}")
         documents = []
-
-        for file in files:
-            local_path = hf_hub_download(repo_id=self.hf_space_repo, filename=file)
-            df = pd.read_csv(local_path)
-            docs = df.astype(str).to_dict(orient="records")
-            documents.extend(docs)
-
+        for path in paths:
+            loader = CSVLoader(file_path=path)
+            data = loader.load()
+            documents.extend(data)
+        logger.debug(f"Loaded {len(documents)} documents.")
         return documents
 
     def initialize_faiss_index(self):
+        logger.info("Initializing FAISS index...")
         if os.path.exists(self.index_path):
-            self.faiss_index = FAISS.load_local(self.index_path)
+            logger.info(f"FAISS index found at {self.index_path}. Loading...")
+            self.faiss_index = FAISS.load_local(self.index_path, self.hf, allow_dangerous_deserialization=True)
         else:
-            documents = self.load_data_from_hf_space()
-            self.faiss_index = FAISS.from_documents(documents)
+            logger.info(f"FAISS index not found. Creating a new one...")
+            documents = self.load_data(self.paths)
+            self.faiss_index = FAISS.from_documents(documents, self.hf)
             self.faiss_index.save_local(self.index_path)
-
         self.faiss_retriever = self.faiss_index.as_retriever(search_kwargs={"k": 5})
+        logger.info("FAISS index initialized successfully.")
 
     def retrieve_answer(self, query):
+        logger.info(f"Retrieving answer for query: {query}")
         if self.faiss_retriever:
-            return self.faiss_retriever.invoke(query)
+            result = self.faiss_retriever.invoke(query)
+            logger.debug(f"Retrieved answer: {result}")
+            return result
+        logger.warning("FAISS retriever is not initialized.")
         return None
 
     def create_chat_record(self, chat_id):
+        logger.info(f"Creating new chat record for chat_id: {chat_id}")
         self.chats_collection.insert_one({
             "_id": chat_id,
             "history": []
         })
 
     def update_chat(self, chat_id, question, answer):
+        logger.info(f"Updating chat history for chat_id: {chat_id}")
         self.chats_collection.update_one(
             {"_id": chat_id},
             {"$push": {"history": {"question": question, "answer": answer}}}
         )
 
     def load_chat(self, chat_id):
+        logger.info(f"Loading chat history for chat_id: {chat_id}")
         chat_record = self.chats_collection.find_one({"_id": chat_id})
         if not chat_record:
+            logger.error(f"Chat ID {chat_id} does not exist.")
             raise KeyError(f"Chat ID {chat_id} does not exist.")
         return chat_record.get('history', [])
 
+    def new_chat(self, chat_id):
+        logger.info(f"Creating new chat with ID: {chat_id}")
+        if self.chats_collection.find_one({"_id": chat_id}):
+            logger.error(f"Chat ID {chat_id} already exists.")
+            raise KeyError(f"Chat ID {chat_id} exists already.")
+        self.create_chat_record(chat_id)
+        return "success"
+
+    def delete_chat(self, chat_id):
+        logger.info(f"Deleting chat record for chat_id: {chat_id}")
+        if not self.chats_collection.find_one({"_id": chat_id}):
+            logger.error(f"Chat ID {chat_id} does not exist.")
+            raise KeyError(f"Chat ID {chat_id} does not exist.")
+        self.chats_collection.delete_one({"_id": chat_id})
+        return "success"
+
     def get_system_prompt(self):
         return """
 You are a comsats assistant to help the user with comsats university-related queries. Your response should be concise, direct, and to the point. Avoid any unnecessary explanations. Always consider the provided context and chat history to generate the answer. 
@@ -78,11 +111,14 @@ Use emojis only when required based on the user's tone and emotions. Do not over
 - **Sad emotions**: Use 😔 when the user is asking about something disappointing or negative.
 - **Surprise**: Use 😯 when the user expresses surprise.
 - **Anger or frustration**: Use 😡 when the user expresses frustration or dissatisfaction.
+If the user asks the same question repeatedly or asks an illogical question, feel free to use emojis to subtly convey frustration, confusion, or amusement.
+
 If the user writes question in urdu, give answer in urdu.
-If the user writes question in English, give answer in English.
+If the user writes question in English, give answer in English .
 please provide the personalized answer and provide answer quickly
 please answer from the dataset i provided to you in csv files. And donot write in every answer that i donot know the exact answer.and refer website only where it is necessary.
-Do not include the phrase "According to the provided context" or "Based on the chat history". Simply generate the answer like a human would, without referencing where the information comes from.
+
+Do not include the phrase "According to the provided context" or "Based on the chat history". Simply generate the answer like a human would, without referencing where the information comes from. 
 If the question requires a URL, format it like this:
 [Click here to visit COMSATS](https://comsats.edu.pk).
 Your task is to help students at COMSATS University, Attock campus, with their university-related queries. The following are key details about the university:
@@ -101,25 +137,118 @@ Context ends here. Now, answer the following question:
         """
 
     def generate_response(self, question, history, context):
+        logger.info(f"Generating response for question: {question}")
         prompt = self.get_system_prompt().format(question=question, history=history, context=context)
 
+        while True:
+            for api_key in self.api_keys:
+                self.client = Groq(api_key=api_key)
+                for model in self.models:
+                    try:
+                        logger.info(f"Calling model {model} for response...")
+                        chat_completion = self.client.chat.completions.create(
+                            messages=[ 
+                                {"role": "system", "content": prompt},
+                                {"role": "user", "content": f"Answer the following question: {question}"}
+                            ],
+                            model=model,
+                            max_tokens=1024,
+                        )
+                        response = chat_completion.choices[0].message.content
+                        logger.debug(f"Received response: {response}")
+                        return response
+                    except Exception as e:
+                        logger.error(f"Error with model {model}: {e}")
+                        time.sleep(2)
+                        continue
+        logger.warning("Unable to generate a response.")
+        return "Sorry, unable to provide an answer at this time."
+
+    def detect_language(self, question):
+        logger.info(f"Detecting language for question: {question}")
         for api_key in self.api_keys:
             self.client = Groq(api_key=api_key)
             for model in self.models:
                 try:
                     chat_completion = self.client.chat.completions.create(
                         messages=[
-                            {"role": "system", "content": prompt},
-                            {"role": "user", "content": f"Answer the following question: {question}"}
+                            {
+                                "role": "system",
+                                "content": """
+                                    You are an expert agent, and your task is to detect the language. 
+                                    Return a JSON: {'detected_language': 'urdu' or 'english'}
+                                """
+                            },
+                            {
+                                "role": "user",
+                                "content": f"Detect the language for: {question}"
+                            }
                         ],
                         model=model,
-                        max_tokens=1024,
+                        max_tokens=256,
+                        response_format={"type": "json_object"},
                     )
-                    return chat_completion.choices[0].message.content
-                except Exception:
+                    response = json.loads(chat_completion.choices[0].message.content)
+                    detected_language = response['detected_language'].lower()
+                    logger.debug(f"Detected language: {detected_language}")
+                    return detected_language
+                except Exception as e:
+                    logger.error(f"Error detecting language: {e}")
                     time.sleep(2)
                     continue
-        return "Sorry, unable to provide an answer at this time."
+        logger.warning("Unable to detect language.")
+        return "english"
 
+    def translate_urdu(self, text):
+        logger.info(f"Translating text to Urdu: {text}")
+        for api_key in self.api_keys:
+            self.client = Groq(api_key=api_key)
+            for model in self.models:
+                try:
+                    chat_completion = self.client.chat.completions.create(
+                        messages=[ 
+                            {
+                                "role": "system",
+                                "content": """
+                                    Translate the following text into proper Urdu. Return a JSON: 
+                                    {'text': 'translated urdu text'}
+                                """
+                            },
+                            {
+                                "role": "user",
+                                "content": f"Translate this: {text}"
+                            }
+                        ],
+                        model=model,
+                        max_tokens=512,
+                        response_format={"type": "json_object"},
+                    )
+                    response = json.loads(chat_completion.choices[0].message.content)
+                    translated_text = response['text']
+                    logger.debug(f"Translated text: {translated_text}")
+                    return translated_text
+                except Exception as e:
+                    logger.error(f"Error translating text: {e}")
+                    time.sleep(2)
+                    continue
+        return text
+
+    def response(self, question, chat_id):
+        logger.info(f"Processing response for question: {question} (chat_id: {chat_id})")
+        chat_history = self.load_chat(chat_id)
+
+        for entry in chat_history:
+            self.memory.save_context({"input": entry["question"]}, {"output": entry["answer"]})
+
+        language = self.detect_language(question)
+
+        if language == 'urdu':
+            question_translation = GoogleTranslator(source='ur', target='en').translate(question)
+            context = self.faiss_retriever.invoke(question_translation)
+        else:
+            context = self.faiss_retriever.invoke(question)
 
+        answer = self.generate_response(question, chat_history, context)
 
+        self.update_chat(chat_id, question, answer)
+        return answer