Spaces:
Running
Running
| import os | |
| import time | |
| 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, 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 = hf | |
| self.faiss_index = None | |
| self.faiss_retriever = None | |
| self.paths = paths | |
| self.initialize_faiss_index() | |
| def load_data(self, paths): | |
| logger.info(f"Loading data from paths: {paths}") | |
| documents = [] | |
| 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): | |
| 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: | |
| 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: | |
| 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. | |
| Use emojis only when required based on the user's tone and emotions. Do not overuse them in every answer. provide emojis only when it is context based required. Here's when you should use emojis: | |
| - **Happy emotions**: Use 😊 or 😄 when the user expresses satisfaction or asks for something positive. | |
| - **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. | |
| you have to answer the all questions asked from you in relevant context. Do not answer out of the context. | |
| If the user writes question in urdu, give answer in urdu. | |
| 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. | |
| 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: | |
| - Departments: CS, AI, SE, Math, BBA, EE, CE, English. | |
| - Facilities: Cricket ground, football ground, two canteens (near CS and Math/EE), mosque near CS department, LT rooms in CS, classrooms in Math, and labs in EE. | |
| - Admission: Accepts NTS test, CGPA requirements: 85% for CGPA 4.0, 79-84% for CGPA 3.66. | |
| - Available degrees: BS Computer Science, BS Software Engineering, BS Artificial Intelligence, BS English, BS Math, BS Electrical Engineering, BS Computer Engineering, BS BBA. | |
| Consider the following chat history for additional context to answer the question: | |
| {history} | |
| When answering: | |
| - Answer in a conversational and friendly tone. | |
| - Be concise and to the point, while still being helpful. | |
| - If you don’t know the answer from the context or chat history, simply write the generalized answer from the model.. | |
| Context ends here. Now, answer the following question: | |
| {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": """ | |
| 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=256, | |
| response_format={"type": "json_object"}, | |
| ) | |
| 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 | |
| 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 |