import streamlit as st from PyPDF2 import PdfReader from langchain.text_splitter import RecursiveCharacterTextSplitter import os from langchain_google_genai import GoogleGenerativeAIEmbeddings import google.generativeai as genai from langchain_community.vectorstores import FAISS from langchain_google_genai import ChatGoogleGenerativeAI from langchain.chains.question_answering import load_qa_chain from langchain.prompts import PromptTemplate from dotenv import load_dotenv from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware import json # Initialize FastAPI app api_app = FastAPI() # CORS settings to allow cross-origin requests api_app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Load environment variables load_dotenv() # Check if the API key is loaded correctly GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY") if not GOOGLE_API_KEY: raise ValueError("API key not found in environment variables. Please check '.env' file.") # Configure the API key genai.configure(api_key=GOOGLE_API_KEY) def get_pdf_text(pdf_path): """Extract text from a PDF document.""" text = "" try: with open(pdf_path, "rb") as pdf_file: pdf_reader = PdfReader(pdf_file) for page in pdf_reader.pages: extracted_text = page.extract_text() if extracted_text: text += extracted_text except Exception as e: st.error(f"Error reading PDF file: {e}") return text def get_text_chunks(text): """Split text into manageable chunks for FAISS indexing.""" text_splitter = RecursiveCharacterTextSplitter(chunk_size=10000, chunk_overlap=1000) chunks = text_splitter.split_text(text) return chunks def get_vector_store(text_chunks): """Create and save a FAISS vector store from text chunks.""" embeddings = GoogleGenerativeAIEmbeddings(model="models/embedding-001", google_api_key=GOOGLE_API_KEY) vector_store = FAISS.from_texts(text_chunks, embedding=embeddings) vector_store.save_local("faiss_index") def get_conversational_chain(): """Define the prompt template and load the QA chain with the Generative AI model.""" prompt_template = """ {context} You are an AI assistant for Robi Axiata Ltd., specializing in providing personalized internet package recommendations and customer support. Your role is to: 1. Recommend the best internet package based on the user's current task and its duration. 2. Suggest suitable packages based on the user's weekly internet usage patterns. 3. Provide solutions to common internet-related problems. ## Task: {question} Provide a detailed response with: 1. **Recommended Package:** Include package name, data volume, validity, and price. 2. **Alternative Packages:** List alternative options sorted from cheapest to most expensive. 3. **Solution to Issues:** If applicable, provide solutions to any mentioned problems. Begin your response now. """ model = ChatGoogleGenerativeAI(model="gemini-pro", temperature=0.3) prompt = PromptTemplate(template=prompt_template, input_variables=["context", "question"]) chain = load_qa_chain(model, chain_type="stuff", prompt=prompt) return chain def user_input(user_question): """Process user input, perform similarity search, and generate response.""" embeddings = GoogleGenerativeAIEmbeddings(model="models/embedding-001", google_api_key=GOOGLE_API_KEY) # Check if FAISS index exists before loading faiss_index_path = "faiss_index/index.faiss" if not os.path.exists(faiss_index_path): return {"error": "FAISS index not found. Please ensure the documents are processed first."} new_db = FAISS.load_local("faiss_index", embeddings, allow_dangerous_deserialization=True) docs = new_db.similarity_search(user_question) chain = get_conversational_chain() response = chain({"input_documents": docs, "question": user_question}, return_only_outputs=True) return {"response": response["output_text"]} # FastAPI endpoint to expose the API @api_app.post("/api/ask") async def ask_question(request: Request): """API endpoint to handle user questions.""" req_data = await request.json() user_question = req_data.get("question", "") if not user_question: return {"error": "No question provided."} # Call the function that retrieves the answer result = user_input(user_question) return result # Streamlit UI def main(): """Main function to run the Streamlit app.""" st.set_page_config(page_title="Robi AI Customer Service", page_icon="📱") # Display Logo if os.path.exists("robi_logo.png"): st.image("robi_logo.png", width=200) # Ensure you have a 'robi_logo.png' in your project directory else: st.warning("Logo image 'robi_logo.png' not found. Please add it to the project directory.") # Title and Description st.title("📶 Robi AI Customer Service & Recommendation System") st.markdown(""" Welcome to Robi Axiata Ltd.'s AI-powered Customer Service and Recommendation System. Get personalized internet package recommendations based on your usage and tasks. """) # Input field for user's task-based question st.header("🔍 Get Package Recommendations Based on Your Current Task") user_task = st.text_input( "Describe the task you're performing and its estimated duration:", placeholder="e.g., Streaming a movie for 3 hours" ) # Show an example input below the input field st.caption("Example: 'I am downloading large files for 2 hours.'") if user_task: # Format the user question for the AI model user_question = f"Before my current internet package runs out, I am {user_task}. Please recommend the best package to buy next to complete this task, including how much internet is needed and the best options sorted from cheapest to most expensive." result = user_input(user_question) if "error" in result: st.error(result["error"]) else: st.markdown("### 📄 Recommended Package(s):") st.write(result["response"]) st.markdown("---") # Input field for user's weekly usage st.header("📅 Get Package Recommendations Based on Weekly Usage") weekly_usage = st.text_input( "Describe your weekly internet usage patterns and the types of content you consume:", placeholder="e.g., I use social media and stream videos frequently during the week." ) # Show an example input below the input field st.caption("Example: 'I browse the web, stream videos, and play online games every day.'") if weekly_usage: # Format the user question for the AI model user_question = f"I use the internet {weekly_usage}. Please suggest the best Robi internet packages that suit my weekly usage, sorted from cheapest to most expensive." result = user_input(user_question) if "error" in result: st.error(result["error"]) else: st.markdown("### 📄 Suggested Package(s):") st.write(result["response"]) st.markdown("---") # Input field for common problems st.header("🛠️ Need Help? Common Problems and Solutions") problem_description = st.text_input( "Describe any internet-related issues you're facing:", placeholder="e.g., My internet speed is slow during evenings." ) # Show an example input below the input field st.caption("Example: 'I experience low network connection at night.'") if problem_description: # Format the user question for the AI model user_question = f"I am experiencing the following issue: {problem_description}. Please provide solutions to resolve this problem." result = user_input(user_question) if "error" in result: st.error(result["error"]) else: st.markdown("### 🛠️ Solutions:") st.write(result["response"]) st.markdown("---") with st.sidebar: st.header("📄 Documents:") # Check if the FAISS index exists if not os.path.exists("faiss_index/index.faiss"): pdf_path = "customerservicebot.pdf" # Default PDF file if os.path.exists(pdf_path): with st.spinner(f"Processing '{pdf_path}'..."): raw_text = get_pdf_text(pdf_path) if raw_text: text_chunks = get_text_chunks(raw_text) get_vector_store(text_chunks) st.success(f"'{pdf_path}' processed and FAISS index created.") else: st.error(f"No text extracted from '{pdf_path}'. Please check the PDF content.") else: st.error(f"Default PDF '{pdf_path}' not found. Please add it to the project directory.") else: st.info("FAISS index loaded from 'faiss_index'.") if __name__ == "__main__": main()