Create gradio_app.py

gradio_app.py

@@ -0,0 +1,321 @@
+"""
+Gradio interface for the RAG-based response system.
+This will be deployed to your Hugging Face Space.
+"""
+
+import gradio as gr
+import os
+import tempfile
+import json
+from pathlib import Path
+import torch
+from transformers import AutoTokenizer, AutoModel, pipeline
+from langchain_community.document_loaders import PyPDFLoader, WebBaseLoader
+from langchain_community.vectorstores import FAISS
+from langchain_community.embeddings import HuggingFaceEmbeddings
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+
+# Define paths
+DOCUMENTS_DIR = Path("documents")
+DOCUMENTS_DIR.mkdir(exist_ok=True)
+VECTOR_DB_PATH = Path("vector_db")
+
+# Initialize models
+model_name = "sentence-transformers/all-MiniLM-L6-v2"
+embeddings = HuggingFaceEmbeddings(model_name=model_name)
+
+# Initialize vector store
+if VECTOR_DB_PATH.exists():
+    try:
+        vector_db = FAISS.load_local(str(VECTOR_DB_PATH), embeddings)
+        print("Loaded existing vector database.")
+    except Exception as e:
+        print(f"Error loading vector database: {e}")
+        vector_db = None
+else:
+    vector_db = None
+
+# Define possible intents
+POSSIBLE_INTENTS = [
+    "product_inquiry", 
+    "technical_support", 
+    "billing_question", 
+    "general_information",
+    "appointment_scheduling",
+    "complaint",
+    "other"
+]
+
+# Default responses for when RAG fails or no documents are available
+DEFAULT_RESPONSES = {
+    "product_inquiry": "Thank you for your interest in our products. I'll gather the information and have someone contact you with more details.",
+    "technical_support": "I understand you're experiencing technical issues. Let me find the right person to help you resolve this.",
+    "billing_question": "Thank you for your billing inquiry. I'll connect you with our billing department for assistance.",
+    "general_information": "Thank you for reaching out. I'll make sure you get the information you need.",
+    "appointment_scheduling": "I'd be happy to help schedule an appointment for you. Let me find the next available slot.",
+    "complaint": "I'm sorry to hear about your experience. Your feedback is important to us, and we'll address this promptly.",
+    "other": "Thank you for your call. I'll make sure your message gets to the right person."
+}
+
+def load_pdf(file):
+    """Load a PDF document into the vector store"""
+    global vector_db
+    
+    try:
+        # Save the uploaded file temporarily
+        temp_dir = tempfile.mkdtemp()
+        temp_path = os.path.join(temp_dir, file.name)
+        
+        with open(temp_path, "wb") as f:
+            f.write(file.read())
+        
+        # Save a copy to the documents directory
+        target_path = os.path.join(DOCUMENTS_DIR, file.name)
+        with open(target_path, "wb") as f:
+            with open(temp_path, "rb") as src:
+                f.write(src.read())
+        
+        # Load and process the PDF
+        loader = PyPDFLoader(temp_path)
+        documents = loader.load()
+        
+        # Split the documents
+        text_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=1000,
+            chunk_overlap=200
+        )
+        chunks = text_splitter.split_documents(documents)
+        
+        # Update or create vector store
+        if vector_db is None:
+            vector_db = FAISS.from_documents(chunks, embeddings)
+            vector_db.save_local(str(VECTOR_DB_PATH))
+        else:
+            vector_db.add_documents(chunks)
+            vector_db.save_local(str(VECTOR_DB_PATH))
+        
+        return f"Successfully added {file.name} to the knowledge base with {len(chunks)} chunks."
+    
+    except Exception as e:
+        return f"Error processing PDF: {str(e)}"
+
+def load_website(url):
+    """Load a website into the vector store"""
+    global vector_db
+    
+    try:
+        # Load content from website
+        loader = WebBaseLoader(url)
+        documents = loader.load()
+        
+        # Save the URL reference
+        with open(os.path.join(DOCUMENTS_DIR, "websites.txt"), "a") as f:
+            f.write(f"{url}\n")
+        
+        # Split the documents
+        text_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=1000,
+            chunk_overlap=200
+        )
+        chunks = text_splitter.split_documents(documents)
+        
+        # Update or create vector store
+        if vector_db is None:
+            vector_db = FAISS.from_documents(chunks, embeddings)
+            vector_db.save_local(str(VECTOR_DB_PATH))
+        else:
+            vector_db.add_documents(chunks)
+            vector_db.save_local(str(VECTOR_DB_PATH))
+        
+        return f"Successfully added {url} to the knowledge base with {len(chunks)} chunks."
+    
+    except Exception as e:
+        return f"Error processing website: {str(e)}"
+
+def generate_response(query, intent=None):
+    """Generate a response based on the query and intent"""
+    global vector_db
+    
+    # If no intent provided, use a default
+    if not intent or intent not in POSSIBLE_INTENTS:
+        intent = "general_information"
+    
+    # If no vector database, return default response
+    if vector_db is None:
+        return DEFAULT_RESPONSES.get(intent, DEFAULT_RESPONSES["other"])
+    
+    try:
+        # Query the vector database
+        retrieved_docs = vector_db.similarity_search(query, k=3)
+        
+        if not retrieved_docs:
+            return DEFAULT_RESPONSES.get(intent, DEFAULT_RESPONSES["other"])
+        
+        # Combine retrieved document chunks
+        context = "\n\n".join([doc.page_content for doc in retrieved_docs])
+        
+        # Create prompt for the model
+        prompt = f"""
+        Based on the following context information and the user's query,
+        provide a helpful, professional, and concise response.
+        
+        Context:
+        {context}
+        
+        User query: {query}
+        
+        Intent: {intent}
+        
+        Response:
+        """
+        
+        # Generate response using a pre-trained model
+        tokenizer = AutoTokenizer.from_pretrained("google/flan-t5-base")
+        model = AutoModel.from_pretrained("google/flan-t5-base")
+        
+        inputs = tokenizer(prompt, return_tensors="pt", max_length=512, truncation=True)
+        outputs = model.generate(
+            inputs["input_ids"],
+            max_length=200,
+            num_beams=4,
+            early_stopping=True
+        )
+        
+        response_text = tokenizer.decode(outputs[0], skip_special_tokens=True)
+        
+        # If model output is empty or too short, use retrieved text with a prefix
+        if len(response_text) < 20:
+            response_text = f"Based on the information I have: {context[:200]}..."
+        
+        return response_text
+        
+    except Exception as e:
+        print(f"Error generating response: {e}")
+        return DEFAULT_RESPONSES.get(intent, DEFAULT_RESPONSES["other"])
+
+def list_documents():
+    """List all documents in the knowledge base"""
+    files = list(DOCUMENTS_DIR.glob("*.pdf"))
+    
+    # Add websites if available
+    website_file = DOCUMENTS_DIR / "websites.txt"
+    websites = []
+    if website_file.exists():
+        with open(website_file, "r") as f:
+            websites = [line.strip() for line in f if line.strip()]
+    
+    return {
+        "PDFs": [f.name for f in files],
+        "Websites": websites
+    }
+
+# Create the Gradio interface
+with gr.Blocks(title="Call Assistant RAG System") as demo:
+    gr.Markdown("# Call Assistant RAG System")
+    gr.Markdown("Add documents and websites to the knowledge base, and test the response generation.")
+    
+    with gr.Tab("Add Knowledge"):
+        with gr.Row():
+            with gr.Column():
+                pdf_input = gr.File(label="Upload PDF Document")
+                pdf_button = gr.Button("Add PDF to Knowledge Base")
+                pdf_output = gr.Textbox(label="PDF Upload Status")
+                
+                pdf_button.click(
+                    load_pdf,
+                    inputs=[pdf_input],
+                    outputs=[pdf_output]
+                )
+            
+            with gr.Column():
+                url_input = gr.Textbox(label="Website URL")
+                url_button = gr.Button("Add Website to Knowledge Base")
+                url_output = gr.Textbox(label="Website Status")
+                
+                url_button.click(
+                    load_website,
+                    inputs=[url_input],
+                    outputs=[url_output]
+                )
+    
+    with gr.Tab("Knowledge Base"):
+        list_button = gr.Button("List Documents in Knowledge Base")
+        knowledge_output = gr.JSON(label="Documents")
+        
+        list_button.click(
+            list_documents,
+            inputs=[],
+            outputs=[knowledge_output]
+        )
+    
+    with gr.Tab("Test Response Generation"):
+        with gr.Row():
+            query_input = gr.Textbox(label="Query / Transcription")
+            intent_input = gr.Dropdown(
+                choices=POSSIBLE_INTENTS,
+                label="Intent",
+                value="general_information"
+            )
+        
+        test_button = gr.Button("Generate Response")
+        response_output = gr.Textbox(label="Generated Response")
+        
+        test_button.click(
+            generate_response,
+            inputs=[query_input, intent_input],
+            outputs=[response_output]
+        )
+    
+    with gr.Tab("API"):
+        gr.Markdown("""
+        ## API Documentation
+        
+        This Gradio app exposes an API endpoint that can be used by your Twilio application.
+        
+        ### Endpoint: `/api/predict`
+        
+        **Method:** POST
+        
+        **Input:**
+        ```json
+        {
+            "data": [
+                "user's transcribed query",
+                "detected intent"
+            ]
+        }
+        ```
+        
+        **Output:**
+        ```json
+        {
+            "data": [
+                "generated response"
+            ],
+            "duration": 1.2345
+        }
+        ```
+        
+        **Example Python code to call the API:**
+        ```python
+        import requests
+        
+        response = requests.post(
+            "https://huggingface.co/spaces/iajitpanday/vBot-1.5/api/predict", 
+            json={"data": ["How do I reset my password?", "technical_support"]}
+        )
+        
+        result = response.json()
+        generated_response = result["data"][0]
+        ```
+        """)
+
+# Define API function for Gradio Spaces
+def api_response(query, intent=None):
+    """API function for Gradio Spaces"""
+    response = generate_response(query, intent)
+    return [response]
+
+# Define the API
+demo.queue()
+demo.launch()