import os from groq import Groq from langchain.memory import ConversationSummaryBufferMemory from langdetect import detect from deep_translator import GoogleTranslator from langchain_community.document_loaders.csv_loader import CSVLoader from langchain_community.vectorstores import FAISS class Comsatsbot: def __init__(self, hf, llm, api_key, chats_collection, paths, index_path='faiss_kb'): self.llm = llm self.client = Groq(api_key=api_key) # Initialize memory buffer and MongoDB connection self.memory = ConversationSummaryBufferMemory(llm=self.llm, max_token_limit=3000) self.chats_collection = chats_collection self.index_path = index_path self.hf = hf self.faiss_index = None self.faiss_retriever = None self.paths = paths self.initialize_faiss_index() def load_data(self, paths): documents = [] for path in paths: loader = CSVLoader(file_path=path) data = loader.load() documents.extend(data) return documents def initialize_faiss_index(self): if os.path.exists(self.index_path): self.faiss_index = FAISS.load_local(self.index_path, self.hf, allow_dangerous_deserialization=True) self.faiss_retriever = self.faiss_index.as_retriever(search_kwargs={"k": 5}) else: 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}) def retrieve_answer(self, query): if self.faiss_retriever: return self.faiss_retriever.invoke(query) else: print("FAISS retriever is not initialized. Please create or load an index.") return None def create_chat_record(self, chat_id): self.chats_collection.insert_one({ "_id": chat_id, "history": [] }) def update_chat(self, chat_id, question, answer): self.chats_collection.update_one( {"_id": chat_id}, {"$push": {"history": {"question": question, "answer": answer}}} ) def load_chat(self, chat_id): chat_record = self.chats_collection.find_one({"_id": chat_id}) if not chat_record: raise KeyError(f"Chat ID {chat_id} does not exist.") return chat_record.get('history', []) def new_chat(self, chat_id): # Check if chat ID already exists if self.chats_collection.find_one({"_id": chat_id}): raise KeyError(f"Chat ID {chat_id} exist already.") # Create a new chat record if it doesn't exist self.create_chat_record(chat_id) return "success" def delete_chat(self, chat_id): # Check if the chat ID exists if not self.chats_collection.find_one({"_id": chat_id}): raise KeyError(f"Chat ID {chat_id} does not exist.") # Delete the chat if it exists self.chats_collection.delete_one({"_id": chat_id}) return "success" def generate_response(self, question, history, context): chat_completion = self.client.chat.completions.create( messages=[ { "role": "user", "content": f''' You are a conversational and helpfull agent to help the comsats university attock campus students, and your task is to provide concise and direct answers to the questions asked. IF question is about to urdu and then kindly please answer in correct and to the point answer. Dont need to explain irrelevant thinga nd use the correct urdu in response. Please follow these guidelines when answering: DOnt need to explain and repeat the prompt in response. 1. If the question is in **English**, answer in **English**. 2. If the question is in **Urdu**, answer in **Urdu**. 3. Always respond in a **human-like tone** and keep your answers concise, to the point, and friendly. 4. If the question is **conversational** (like greetings, need any converstaion etc), respond in a warm, conversational tone and use appropriate **emojis**. 5. Always consider the provided **context** and **chat history** to formulate your response. 6. If you don’t know the answer to the question or you did not find the answer from the context, kindly respond with "I don't know the answer to this." without adding irrelevant explanations. 7. KIndly generate a perfect and to the point answer with the proper use of emojis. DOnt use any irrelevant text explanation and i want full concise and to the oint answer. 8. Kindly answer in **English** if question ask in **English** and generate answer in **Urdu** if question ask in **Urdu**. **Question:** {question} **Use the following context to answer:** Comsats Attock Campus Provide BSomputerScience, BSSoftwareEngineer BSArtificialIntelligence BSEnglish BSmath BSElectricalEngineering BSComputerEngineering BSBBA Has three departments CS(CS, AI, SE), Math(math, BBA, english) and EE(EE, CE). It has cricket ground and football ground and two canteens. First near math and ee department and second near cs department. There is also mosque near cs department. CS department has threater liker rooms lt and total 9 theaters called lt and math has classroom cr and ee has labs. They accept the nts test for admission and provide the cgpa for 4 on 85 percent and 3.66 between 79 to 84 and many more. {context} **Consider the following chat history for additional context to answer the question:** {history} ''' } ], model="llama3-70b-8192", ) response_content = chat_completion.choices[0].message.content return response_content def response(self, question, chat_id): chat_history = self.load_chat(chat_id) # Load the previous conversation into memory for entry in chat_history: self.memory.save_context({"input": entry["question"]}, {"output": entry["answer"]}) language = detect(question) if language == 'ur': question_translation = GoogleTranslator(source='ur', target='en').translate(question) context = self.faiss_retriever.invoke(question_translation) else: context = self.faiss_retriever.invoke(question) all_content = '' for document in context: page_content = document.page_content all_content += page_content + '\n' # Add a line break between contents for readability answer = self.generate_response(question, self.memory.load_memory_variables({})['history'], all_content) self.update_chat(chat_id, question, answer) return answer