import os, logging import glob import logging import re import hashlib import numpy as np import asyncio from typing import List, Optional, Any import shutil import secrets from contextlib import asynccontextmanager from nps_agent import run_nps_task # --- Pydantic & FastAPI --- from pydantic import BaseModel, Field import uvicorn from fastapi import FastAPI, HTTPException, APIRouter, UploadFile, File, Depends from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from fastapi.security import HTTPBasic, HTTPBasicCredentials # --- LangChain & RAG Core --- from PyPDF2 import PdfReader try: from docx import Document as DocxDocument except ImportError: DocxDocument = None from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.vectorstores import FAISS from langchain_classic.retrievers import EnsembleRetriever from langchain_classic.chains.combine_documents import ( create_stuff_documents_chain, ) from langchain_classic.chains import create_retrieval_chain from langchain_huggingface import HuggingFaceEmbeddings from langchain_core.language_models import LLM from langchain_core.callbacks.manager import CallbackManagerForLLMRun from langchain_core.documents import Document from langchain_core.retrievers import BaseRetriever from langchain_core.prompts import PromptTemplate # --- Scikit-Learn for TF-IDF --- from sklearn.feature_extraction.text import TfidfVectorizer from sklearn.metrics.pairwise import cosine_similarity # --- Mistral AI --- from mistralai import Mistral as MistralClient from dotenv import load_dotenv load_dotenv() # --- Global Variables & Configuration --- logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) # Define the original, read-only data folder from the repo REPO_DATA_FOLDER = "data" # Use persistent storage if available (on Hugging Face Spaces) if os.path.exists("/data"): logger.info("Persistent storage '/data' found. Using it for PDFs and cache.") DATA_FOLDER = "/data/pdf_storage" CACHE_DIR = "/data/cache" else: logger.info( "No persistent storage '/data' found. Using local 'data' folder and '/tmp/cache'." ) DATA_FOLDER = REPO_DATA_FOLDER CACHE_DIR = "/tmp/cache" os.makedirs(DATA_FOLDER, exist_ok=True) os.makedirs(CACHE_DIR, exist_ok=True) retrieval_chain = None HF_CACHE = "/tmp/hf" os.environ["HF_HOME"] = HF_CACHE os.environ["HF_DATASETS_CACHE"] = HF_CACHE os.environ["HF_MODULES_CACHE"] = HF_CACHE retrieval_chain = None db_path = "/data/default.sqlite" if os.path.exists("/data") else "default.sqlite" # --- Pydantic Models --- # --- Pydantic Models --- class QuestionRequest(BaseModel): question: str = Field(..., min_length=1) mode: str = Field("chat", description="Mode of operation: 'chat' or 'agent'") language: str = Field("English", description="Language of the question and expected answer") class AgentRequest(BaseModel): task: str = Field(..., min_length=5, description="The task for the browser agent to perform") class AnswerResponse(BaseModel): answer: str sources: List[str] # --- Greeting Handler --- GREETING_RESPONSES = { "hi": "Hello! How can I assist you?", "hello": "Hello! How can I help you today?", "bye": "Bye-bye! See you soon.", "ty": "Your Welcome!", } # --- Enhanced Prompt --- UNIVERSITY_ASSISTANT_PROMPT = PromptTemplate.from_template( """ You are a meticulous Guru Nanak Dev University Academic Assistant. Your task is to provide accurate, specific, and concise answers based ONLY on the provided Context. **Instructions:** 1. **Analyze the Context:** Carefully read the user's question and the provided context to synthesize an answer. 2. **Extract Specifics:** If asked for a fee, number, or date, provide the EXACT information. 3. **Cite Sources:** After your answer, cite the source document and page number: `(Source: Prospectus.pdf, Page: 15)`. 4. **Handle Missing Information:** If the info is not in the context, state: "Based on the provided documents, I could not find specific information about this topic." --- **Context:** {context} --- **Question:** {input} **Answer (with citations):** """ ) # --- Custom Scikit-Learn TF-IDF Retriever --- class ScikitLearnTfidfRetriever(BaseRetriever): docs: List[Document] tfidf_vectorizer: TfidfVectorizer tfidf_matrix: Any k: int = 15 class Config: arbitrary_types_allowed = True def _get_relevant_documents(self, query: str) -> List[Document]: query_vector = self.tfidf_vectorizer.transform([query]) similarity_scores = cosine_similarity(query_vector, self.tfidf_matrix).flatten() top_k_indices = np.argsort(similarity_scores)[-self.k :][::-1] return [self.docs[i] for i in top_k_indices] async def _aget_relevant_documents( self, query: str, *, run_manager=None ) -> List[Document]: return self._get_relevant_documents(query) # --- MistralLLM Class --- class MistralLLM(LLM): client: Any = None model: str = "mistral-small-latest" temperature: float = 0.0 api_key: str = "" def __init__( self, api_key: str, model: str = "mistral-small-latest", temperature: float = 0.0, ): super().__init__() self.api_key = api_key self.model = model self.temperature = temperature self.client = MistralClient(api_key=self.api_key) @property def _llm_type(self) -> str: return "mistral" def _call( self, prompt: str, stop: Optional[List[str]] = None, run_manager: Optional[CallbackManagerForLLMRun] = None, **kwargs: Any, ) -> str: messages = [{"role": "user", "content": prompt}] response = self.client.chat.complete( model=self.model, messages=messages, temperature=self.temperature ) return response.choices[0].message.content.strip() # --- Caching and Data Processing Functions --- def get_data_hash(folder_path=DATA_FOLDER) -> str: """Generates a hash based on the filenames and modification times of supported files.""" hasher = hashlib.md5() if not os.path.exists(folder_path): return "no_data" supported_extensions = [".pdf", ".txt", ".md", ".docx"] files_to_hash = [] for ext in supported_extensions: files_to_hash.extend(glob.glob(os.path.join(folder_path, f"*{ext}"))) for file_path in sorted(files_to_hash): try: stat = os.stat(file_path) file_info = f"{os.path.basename(file_path)}-{stat.st_mtime}-{stat.st_size}" hasher.update(file_info.encode("utf-8")) except FileNotFoundError: continue # File might have been deleted during the process return hasher.hexdigest() def get_documents_from_files(folder_path=DATA_FOLDER) -> List[Document]: docs = [] supported_extensions = [".pdf", ".txt", ".md", ".docx"] for file_path in glob.glob(os.path.join(folder_path, "*")): file_name = os.path.basename(file_path) file_ext = os.path.splitext(file_name)[1].lower() if file_ext not in supported_extensions: continue if file_ext == ".pdf": pdf_reader = PdfReader(file_path) for i, page in enumerate(pdf_reader.pages): text = page.extract_text() or "" docs.append( Document( page_content=text, metadata={"source": file_name, "page": i + 1} ) ) elif file_ext in [".txt", ".md"]: with open(file_path, "r", encoding="utf-8") as f: text = f.read() docs.append( Document( page_content=text, metadata={"source": file_name, "page": 1} ) ) elif file_ext == ".docx": if DocxDocument is None: logger.warning( "Skipping .docx file because 'python-docx' is not installed. Please install it with 'pip install python-docx'." ) continue try: doc = DocxDocument(file_path) text = "\n".join([para.text for para in doc.paragraphs]) docs.append( Document( page_content=text, metadata={"source": file_name, "page": 1} ) ) except Exception as e: logger.error(f"Error processing .docx file {file_name}: {e}") return docs async def initialize_rag_system(): global retrieval_chain logger.info("🚀 Initializing RAG system...") os.makedirs(CACHE_DIR, exist_ok=True) data_hash = get_data_hash() cache_path = os.path.join(CACHE_DIR, f"vectorstore_{data_hash}.faiss") embeddings = HuggingFaceEmbeddings( model_name="BAAI/bge-base-en-v1.5", model_kwargs={"device": "cpu"}, encode_kwargs={"normalize_embeddings": True}, ) if os.path.exists(cache_path): logger.info(f"✅ Loading vector store from cache: {cache_path}") vectorstore = FAISS.load_local( cache_path, embeddings, allow_dangerous_deserialization=True ) documents = get_documents_from_files() text_chunks = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=150 ).split_documents(documents) else: logger.warning( "No cache found or data has changed. Building new vector store..." ) documents = get_documents_from_files() if not documents: logger.warning( "No documents found in /data folder. RAG system will be inactive." ) retrieval_chain = None # Ensure chain is None return text_chunks = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=150 ).split_documents(documents) logger.info( f"Embedding {len(text_chunks)} chunks... (This will be slow on the first run)" ) vectorstore = FAISS.from_documents(documents=text_chunks, embedding=embeddings) logger.info(f"💾 Saving new vector store to cache: {cache_path}") vectorstore.save_local(cache_path) # --- Chain Setup --- mistral_api_key = os.getenv("MISTRAL_API_KEY") if not mistral_api_key: raise ValueError("MISTRAL_API_KEY not found") llm = MistralLLM(api_key=mistral_api_key) question_answer_chain = create_stuff_documents_chain( llm, UNIVERSITY_ASSISTANT_PROMPT ) faiss_retriever = vectorstore.as_retriever( search_type="similarity", search_kwargs={"k": 15} ) tfidf_vectorizer = TfidfVectorizer(stop_words="english") if not text_chunks: logger.warning("No text chunks to process for TF-IDF.") tfidf_retriever = ScikitLearnTfidfRetriever( docs=[], tfidf_vectorizer=tfidf_vectorizer, tfidf_matrix=tfidf_vectorizer.fit_transform([]), k=15, ) else: tfidf_matrix = tfidf_vectorizer.fit_transform( [doc.page_content for doc in text_chunks] ) tfidf_retriever = ScikitLearnTfidfRetriever( docs=text_chunks, tfidf_vectorizer=tfidf_vectorizer, tfidf_matrix=tfidf_matrix, k=15, ) ensemble_retriever = EnsembleRetriever( retrievers=[tfidf_retriever, faiss_retriever], weights=[0.5, 0.5] ) retrieval_chain = create_retrieval_chain(ensemble_retriever, question_answer_chain) logger.info("✅ RAG system initialized successfully.") def extract_sources_from_docs(docs: List[Document]) -> List[str]: """Extracts unique source and page numbers from a list of documents.""" return sorted( list( set( f"({doc.metadata.get('source', 'N/A')}, Page: {doc.metadata.get('page', 'N/A')})" for doc in docs ) ) ) @asynccontextmanager async def lifespan(app: FastAPI): # Startup event # On Hugging Face, copy initial files from the read-only repo folder to persistent storage if DATA_FOLDER != REPO_DATA_FOLDER and os.path.exists(REPO_DATA_FOLDER): supported_extensions = [".pdf", ".txt", ".md", ".docx"] if not any( any(f.endswith(ext) for ext in supported_extensions) for f in os.listdir(DATA_FOLDER) ): logger.info( f"Copying initial files from '{REPO_DATA_FOLDER}' to '{DATA_FOLDER}'..." ) for filename in os.listdir(REPO_DATA_FOLDER): if any(filename.endswith(ext) for ext in supported_extensions): src_path = os.path.join(REPO_DATA_FOLDER, filename) dest_path = os.path.join(DATA_FOLDER, filename) if not os.path.exists(dest_path): shutil.copy(src_path, dest_path) await initialize_rag_system() yield # Shutdown event (if needed in future) # --- FastAPI Application --- app = FastAPI(title="GNDU Assistant (Cached)", version="4.0.0", lifespan=lifespan) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) security = HTTPBasic() def get_current_username(credentials: HTTPBasicCredentials = Depends(security)): correct_username = secrets.compare_digest( credentials.username, os.environ.get("ADMIN_USERNAME", "admin") ) correct_password = secrets.compare_digest( credentials.password, os.environ.get("ADMIN_PASSWORD", "admin") ) if not (correct_username and correct_password): raise HTTPException( status_code=401, detail="Incorrect email or password", headers={"WWW-Authenticate": "Basic"}, ) return credentials.username admin_router = APIRouter() @admin_router.get("/files") async def get_files(_: str = Depends(get_current_username)): """Lists all supported files in the data directory.""" if not os.path.exists(DATA_FOLDER): return [] supported_extensions = [".pdf", ".txt", ".md", ".docx"] files = sorted( [ f for f in os.listdir(DATA_FOLDER) if any(f.endswith(ext) for ext in supported_extensions) ] ) return files @admin_router.post("/files") async def upload_file( file: UploadFile = File(...), _: str = Depends(get_current_username) ): """Uploads a new PDF file.""" file_path = os.path.join(DATA_FOLDER, file.filename) if os.path.exists(file_path): raise HTTPException( status_code=409, detail="File with this name already exists." ) try: with open(file_path, "wb") as buffer: shutil.copyfileobj(file.file, buffer) finally: file.file.close() await initialize_rag_system() return {"filename": file.filename} @admin_router.delete("/files/{filename}") async def delete_file(filename: str, _: str = Depends(get_current_username)): """Deletes a PDF file.""" if ".." in filename: raise HTTPException(status_code=400, detail="Invalid filename.") file_path = os.path.join(DATA_FOLDER, filename) if not os.path.exists(file_path): raise HTTPException(status_code=404, detail="File not found.") os.remove(file_path) await initialize_rag_system() return {"message": f"File '{filename}' deleted successfully."} app.include_router(admin_router, prefix="/admin") async def translate_text(text: str, target_language: str, api_key: str) -> str: """Translates text to the target language using Mistral.""" if target_language.lower() == "english": return text client = MistralClient(api_key=api_key) prompt = f"Translate the following text to {target_language}. Return ONLY the translated text, no unique prefix/suffix.\n\nText: {text}" try: messages = [{"role": "user", "content": prompt}] response = await client.chat.complete_async( model="mistral-small-latest", messages=messages, temperature=0.0 ) return response.choices[0].message.content.strip() except Exception as e: logger.error(f"Translation failed: {e}") return text # Fallback to original text @app.post("/ask", response_model=AnswerResponse) async def ask_question(request: QuestionRequest): normalized_question = request.question.strip().lower() mistral_api_key = os.getenv("MISTRAL_API_KEY") # 1. Translate Query to English if needed english_question = request.question if request.language.lower() != "english": try: # Use a simple synchronous call for now if async is tricky with the client version, # but let's try to stick to the pattern. # IMPORTANT check: MistralClient in this code might be sync or async. # The existing MistralLLM uses sync `client.chat.complete`. # We will use sync here for safety to match MistralLLM usage, or wrap in run_in_executor if needed. # For simplicity in this step, let's use the same sync client pattern as MistralLLM. client = MistralClient(api_key=mistral_api_key) trans_prompt = f"Translate the following text to English. Return ONLY the translated text.\n\nText: {request.question}" trans_resp = client.chat.complete( model="mistral-small-latest", messages=[{"role": "user", "content": trans_prompt}], temperature=0.0 ) english_question = trans_resp.choices[0].message.content.strip() logger.info(f"Translated '{request.question}' ({request.language}) -> '{english_question}' (English)") except Exception as e: logger.error(f"Query translation failed: {e}") # 2. Process Request answer_text = "" sources_list = [] if request.mode == "agent": try: logger.info(f"🚀 Executing Agent Task: {english_question}") answer_text = await run_nps_task(english_question) sources_list = ["Autonomus Agent"] except Exception as e: logger.error(f"Error running agent task: {e}") raise HTTPException(status_code=500, detail=f"Agent Error: {str(e)}") elif normalized_question in GREETING_RESPONSES: answer_text = GREETING_RESPONSES[normalized_question] sources_list = ["Conversational Greeting"] else: # RAG Logic if not retrieval_chain: raise HTTPException( status_code=503, detail="RAG system is not initialized. Please add PDF files to the /data folder and restart.", ) try: # Modify the input to specificy the output language in the prompt if it's RAG # Actually, RAG prompt expects 'input' and 'context'. # We will rely on post-translation for consistency across Agent/Chat result = await retrieval_chain.ainvoke({"input": english_question}) answer_text = result.get("answer", "No answer could be generated.") sources_list = extract_sources_from_docs(result.get("context", [])) except Exception as e: logger.error(f"Error during RAG chain invocation: {e}") raise HTTPException(status_code=500, detail=str(e)) # 3. Translate Answer back to Target Language if request.language.lower() != "english": try: client = MistralClient(api_key=mistral_api_key) back_prompt = f"Translate the following text to {request.language}. Return ONLY the translated text.\n\nText: {answer_text}" back_resp = client.chat.complete( model="mistral-small-latest", messages=[{"role": "user", "content": back_prompt}], temperature=0.0 ) answer_text = back_resp.choices[0].message.content.strip() except Exception as e: logger.error(f"Answer translation failed: {e}") return AnswerResponse(answer=answer_text, sources=sources_list) @app.post("/agent/run") async def run_agent(request: AgentRequest): """ Executes a task using the NPS Nanobot Agent. """ try: logger.info(f"🚀 Received agent task: {request.task}") result = await run_nps_task(request.task) return {"result": result} except Exception as e: logger.error(f"Error running agent task: {e}") raise HTTPException(status_code=500, detail=str(e)) app.mount("/", StaticFiles(directory="frontend", html=True), name="static") if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=7860)