Update app.py

app.py

@@ -1,229 +1,433 @@
-"""
-Simplified Flask application for the AI call assistant system that doesn't rely on pipecat.
-"""
-
-from flask import Flask, request
-from twilio.twiml.voice_response import VoiceResponse, Gather
+import gradio as gr
 import os
-import requests
+import tempfile
 import json
-import logging
-from datetime import datetime
-from utils import transcribe_audio, classify_intent, get_rag_response, text_to_speech, get_fallback_response
-
-# Configure logging
-logging.basicConfig(
-    level=logging.INFO,
-    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
-    handlers=[
-        logging.FileHandler("call_assistant.log"),
-        logging.StreamHandler()
-    ]
-)
-logger = logging.getLogger(__name__)
-
-app = Flask(__name__)
-
-# Twilio credentials
-TWILIO_ACCOUNT_SID = "ACacb8e8cc6de0f7d6c6b9c9f252a38c67"
-TWILIO_AUTH_TOKEN = "1b2826409367e9799262553538489e54"
-TWILIO_PHONE_NUMBER = "+19704064410"
-
-# Hugging Face Space URL
-HF_SPACE_URL = "https://huggingface.co/spaces/iajitpanday/vBot-1.5"
-
-# Session storage (in production, replace with a database)
-call_sessions = {}
-
-@app.route("/", methods=['GET'])
-def index():
-    return "AI Call Assistant is running!"
-
-@app.route("/answer", methods=['POST'])
-def answer_call():
-    """Handle incoming Twilio calls"""
-    call_sid = request.values.get('CallSid')
-    caller = request.values.get('From', 'Unknown')
+import requests
+import base64
+from pathlib import Path
+from transformers import 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."
+}
+
+# Create a classifier
+try:
+    classifier = pipeline("zero-shot-classification", model="facebook/bart-large-mnli")
+except Exception as e:
+    print(f"Error loading classifier: {e}")
+    classifier = None
+
+def classify_intent(text):
+    """Classify the intent of the user's message"""
+    if not text or not classifier:
+        return "other", 0.0
     
-    logger.info(f"Received call from {caller} with SID: {call_sid}")
+    try:
+        results = classifier(
+            text, 
+            candidate_labels=POSSIBLE_INTENTS,
+            hypothesis_template="This is a {} request."
+        )
+        
+        top_intent = results["labels"][0]
+        confidence = results["scores"][0]
+        
+        return top_intent, confidence
+    except Exception as e:
+        print(f"Error classifying intent: {e}")
+        return "other", 0.0
+
+def load_pdf(file):
+    """Load a PDF document into the vector store"""
+    global vector_db
     
-    # Initialize session for this call
-    call_sessions[call_sid] = {
-        'caller': caller,
-        'start_time': datetime.now().isoformat(),
-        'conversation': []
-    }
+    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."
     
-    response = VoiceResponse()
+    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
     
-    # Welcome message
-    response.say("Hello! Thank you for calling. How can I help you today?")
+    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."
     
-    # Gather speech input
-    gather = Gather(input='speech', 
-                   action='/process_speech', 
-                   method='POST', 
-                   speechTimeout='auto', 
-                   speechModel='phone_call',
-                   language='en-US')
+    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
     
-    response.append(gather)
+    # If no intent provided, use a default
+    if not intent or intent not in POSSIBLE_INTENTS:
+        intent = "general_information"
     
-    # If no input is received
-    response.say("I didn't hear anything. Please call back when you're ready.")
-    response.hangup()
+    # If no vector database, return default response
+    if vector_db is None:
+        return DEFAULT_RESPONSES.get(intent, DEFAULT_RESPONSES["other"])
     
-    return str(response)
-
-def handle_call(audio_url, transcription):
-    """Process the call audio and transcription directly without pipecat"""
     try:
-        # Step 1: Use transcription or enhance with Whisper if needed
-        if not transcription:
-            transcription = transcribe_audio(audio_url)
+        # Query the vector database
+        retrieved_docs = vector_db.similarity_search(query, k=3)
         
-        # Step 2: Classify intent
-        intent, confidence = classify_intent(transcription)
+        if not retrieved_docs:
+            return DEFAULT_RESPONSES.get(intent, DEFAULT_RESPONSES["other"])
         
-        # Step 3: Generate response
-        if confidence >= 0.6 and transcription:
-            response = get_rag_response(transcription, intent, HF_SPACE_URL)
+        # Combine retrieved document chunks
+        context = "\n\n".join([doc.page_content for doc in retrieved_docs])
+        
+        # Simple response generation by combining context with templates
+        if len(context) > 10:
+            if intent == "product_inquiry":
+                return f"Based on the information I have: {context[:300]}... Would you like to know more specific details?"
+            elif intent == "technical_support":
+                return f"I found some information that might help with your issue: {context[:300]}... Is there a specific part you'd like me to explain further?"
+            elif intent == "billing_question":
+                return f"Regarding your billing question: {context[:300]}... Would you like me to connect you with our billing department for more details?"
+            else:
+                return f"Here's what I found that might help answer your question: {context[:300]}... Is there anything specific you'd like me to clarify?"
         else:
-            response = get_fallback_response()
-        
-        return {
-            "transcription": transcription,
-            "intent": intent,
-            "confidence": confidence,
-            "response": response
-        }
+            return DEFAULT_RESPONSES.get(intent, DEFAULT_RESPONSES["other"])
+        
     except Exception as e:
-        logger.error(f"Error handling call: {e}")
-        return {
-            "transcription": transcription,
-            "intent": "error",
-            "confidence": 0.0,
-            "response": get_fallback_response()
-        }
-
-@app.route("/process_speech", methods=['POST'])
-def process_speech():
-    """Process speech input from the caller"""
-    call_sid = request.values.get('CallSid')
-    speech_result = request.values.get('SpeechResult', '')
-    
-    logger.info(f"Call {call_sid} - Speech input: {speech_result}")
-    
-    if call_sid in call_sessions:
-        call_sessions[call_sid]['conversation'].append({
-            'role': 'user',
-            'content': speech_result,
-            'timestamp': datetime.now().isoformat()
-        })
-    
-    # Get the recording URL if available
-    recording_url = request.values.get('RecordingUrl')
-    
-    # Process the call without pipecat
-    call_result = handle_call(recording_url, speech_result)
-    
-    # Create response
-    response = VoiceResponse()
-    
-    # Use the generated response
-    ai_response = call_result["response"]
-    response.say(ai_response)
-    
-    if call_sid in call_sessions:
-        call_sessions[call_sid]['conversation'].append({
-            'role': 'assistant',
-            'content': ai_response,
-            'timestamp': datetime.now().isoformat()
-        })
-    
-    # Ask if there's anything else
-    response.redirect('/anything_else')
-    
-    return str(response)
+        print(f"Error generating response: {e}")
+        return DEFAULT_RESPONSES.get(intent, DEFAULT_RESPONSES["other"])
 
-@app.route("/anything_else", methods=['POST'])
-def anything_else():
-    """Ask if there's anything else the caller needs help with"""
-    call_sid = request.values.get('CallSid')
-    
-    response = VoiceResponse()
-    
-    response.say("Is there anything else I can help you with today?")
+def list_documents():
+    """List all documents in the knowledge base"""
+    files = list(DOCUMENTS_DIR.glob("*.pdf"))
     
-    # Gather speech input
-    gather = Gather(input='speech', 
-                   action='/handle_followup', 
-                   method='POST', 
-                   speechTimeout='auto', 
-                   speechModel='phone_call')
-    response.append(gather)
+    # 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()]
     
-    # If no input is received
-    response.say("Thank you for calling. Have a great day!")
-    response.hangup()
+    return {
+        "PDFs": [f.name for f in files],
+        "Websites": websites
+    }
+
+# Special handler for Twilio
+def handle_twilio_request(data):
+    """Process Twilio request data"""
+    try:
+        # Extract relevant information from Twilio data
+        if "SpeechResult" in data:
+            # This is a speech transcription
+            query = data.get("SpeechResult", "")
+            intent, _ = classify_intent(query)
+            response = generate_response(query, intent)
+            
+            # Create TwiML response
+            twiml = f"""<?xml version="1.0" encoding="UTF-8"?>
+            <Response>
+                <Say>{response}</Say>
+                <Pause length="1"/>
+                <Say>Is there anything else I can help you with today?</Say>
+                <Gather input="speech" action="https://huggingface.co/spaces/iajitpanday/vBot-1.5/api/twilio/followup" method="POST" speechTimeout="auto" speechModel="phone_call"/>
+                <Say>Thank you for calling. Have a great day!</Say>
+            </Response>
+            """
+            return twiml
+            
+        elif "TranscriptionText" in data:
+            # This is a transcription callback
+            query = data.get("TranscriptionText", "")
+            intent, _ = classify_intent(query)
+            response = generate_response(query, intent)
+            
+            # Create SMS response using Twilio API
+            # Note: This requires Twilio credentials which we're avoiding
+            return f"Response would be sent via SMS: {response}"
+            
+        elif "CallStatus" in data and data.get("CallStatus") == "ringing":
+            # Initial call handling
+            twiml = """<?xml version="1.0" encoding="UTF-8"?>
+            <Response>
+                <Say>Hello! Thank you for calling. How can I help you today?</Say>
+                <Gather input="speech" action="https://huggingface.co/spaces/iajitpanday/vBot-1.5/api/twilio/speech" method="POST" speechTimeout="auto" speechModel="phone_call"/>
+                <Say>I didn't hear anything. Please call back when you're ready.</Say>
+            </Response>
+            """
+            return twiml
+            
+        else:
+            # Follow-up or fallback
+            twiml = """<?xml version="1.0" encoding="UTF-8"?>
+            <Response>
+                <Say>Thank you for your call. I've recorded your message and will process it shortly.</Say>
+            </Response>
+            """
+            return twiml
+            
+    except Exception as e:
+        print(f"Error processing Twilio request: {e}")
+        # Return a generic TwiML response
+        twiml = """<?xml version="1.0" encoding="UTF-8"?>
+        <Response>
+            <Say>I'm sorry, I encountered an error processing your request. Please try again later.</Say>
+        </Response>
+        """
+        return twiml
+
+# API endpoints for Twilio
+def twilio_speech_handler(query):
+    """API endpoint for Twilio speech processing"""
+    # Process the query
+    intent, _ = classify_intent(query)
+    response = generate_response(query, intent)
     
-    return str(response)
+    # Create TwiML response
+    twiml = f"""<?xml version="1.0" encoding="UTF-8"?>
+    <Response>
+        <Say>{response}</Say>
+        <Pause length="1"/>
+        <Say>Is there anything else I can help you with today?</Say>
+        <Gather input="speech" action="https://huggingface.co/spaces/iajitpanday/vBot-1.5/api/twilio/followup" method="POST" speechTimeout="auto" speechModel="phone_call"/>
+        <Say>Thank you for calling. Have a great day!</Say>
+    </Response>
+    """
+    return twiml
 
-@app.route("/handle_followup", methods=['POST'])
-def handle_followup():
-    """Handle the caller's follow-up response"""
-    call_sid = request.values.get('CallSid')
-    speech_result = request.values.get('SpeechResult', '')
+def twilio_followup_handler(query):
+    """API endpoint for Twilio follow-up handling"""
+    if any(word in query.lower() for word in ["yes", "yeah", "sure", "please", "correct"]):
+        twiml = """<?xml version="1.0" encoding="UTF-8"?>
+        <Response>
+            <Say>Great! How else can I help you today?</Say>
+            <Gather input="speech" action="https://huggingface.co/spaces/iajitpanday/vBot-1.5/api/twilio/speech" method="POST" speechTimeout="auto" speechModel="phone_call"/>
+            <Say>I didn't hear anything. Thank you for calling. Goodbye!</Say>
+        </Response>
+        """
+    else:
+        twiml = """<?xml version="1.0" encoding="UTF-8"?>
+        <Response>
+            <Say>Thank you for calling. Have a great day!</Say>
+        </Response>
+        """
+    return twiml
+
+def twilio_call_handler():
+    """API endpoint for initial Twilio call handling"""
+    twiml = """<?xml version="1.0" encoding="UTF-8"?>
+    <Response>
+        <Say>Hello! Thank you for calling. How can I help you today?</Say>
+        <Gather input="speech" action="https://huggingface.co/spaces/iajitpanday/vBot-1.5/api/twilio/speech" method="POST" speechTimeout="auto" speechModel="phone_call"/>
+        <Say>I didn't hear anything. Please call back when you're ready.</Say>
+    </Response>
+    """
+    return twiml
+
+# Create 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.")
     
-    logger.info(f"Call {call_sid} - Follow-up response: {speech_result}")
+    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]
+                )
     
-    # Simple yes/no detection
-    if any(word in speech_result.lower() for word in ["yes", "yeah", "sure", "please", "correct"]):
-        response = VoiceResponse()
-        response.say("Great! How else can I help you today?")
-        
-        # Gather speech input again
-        gather = Gather(input='speech', 
-                       action='/process_speech', 
-                       method='POST', 
-                       speechTimeout='auto', 
-                       speechModel='phone_call')
-        response.append(gather)
-        
-        # If no input is received
-        response.say("I didn't hear anything. Thank you for calling. Goodbye!")
-        response.hangup()
+    with gr.Tab("Knowledge Base"):
+        list_button = gr.Button("List Documents in Knowledge Base")
+        knowledge_output = gr.JSON(label="Documents")
         
-    else:
-        response = VoiceResponse()
-        response.say("Thank you for calling. Have a great day!")
+        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"
+            )
         
-        # Save conversation log before hanging up
-        if call_sid in call_sessions:
-            call_sessions[call_sid]['end_time'] = datetime.now().isoformat()
-            # In production, save this to a database
-            logger.info(f"Call {call_sid} completed - Conversation log saved")
+        test_button = gr.Button("Generate Response")
+        response_output = gr.Textbox(label="Generated Response")
         
-        response.hangup()
+        test_button.click(
+            generate_response,
+            inputs=[query_input, intent_input],
+            outputs=[response_output]
+        )
     
-    return str(response)
+    with gr.Tab("Twilio Integration"):
+        gr.Markdown("""
+        ## Twilio Integration Instructions
+        
+        This Gradio app provides API endpoints for Twilio integration. Follow these steps to set up:
+        
+        1. Log into your Twilio account
+        2. Go to Phone Numbers → Manage → Active numbers
+        3. Select your number (+19704064410)
+        4. For "A Call Comes In", select "Webhook" and enter:
+           - URL: `https://huggingface.co/spaces/iajitpanday/vBot-1.5/api/twilio/call`
+           - Method: HTTP POST
+        
+        The system will automatically:
+        - Answer incoming calls
+        - Process speech input
+        - Generate responses using your knowledge base
+        - Handle follow-up questions
+        """)
+        
+        gr.Markdown("""
+        ## API Documentation
+        
+        This app exposes several API endpoints for Twilio integration:
+        
+        1. `/api/twilio/call` - Initial call handling
+        2. `/api/twilio/speech` - Processes speech input
+        3. `/api/twilio/followup` - Handles follow-up responses
+        
+        All endpoints return TwiML responses that Twilio can understand.
+        """)
 
-@app.route("/call_status", methods=['POST'])
-def call_status():
-    """Handle call status updates from Twilio"""
-    call_sid = request.values.get('CallSid')
-    call_status = request.values.get('CallStatus')
-    
-    logger.info(f"Call {call_sid} status: {call_status}")
-    
-    # If call ended abnormally, save the log
-    if call_status in ['completed', 'busy', 'failed', 'no-answer', 'canceled']:
-        if call_sid in call_sessions:
-            call_sessions[call_sid]['end_time'] = datetime.now().isoformat()
-            call_sessions[call_sid]['final_status'] = call_status
-            # In production, save this to a database
-    
-    return "OK"
+# Define API functions (these are needed for Gradio API endpoints)
+def api_response(query, intent=None):
+    """Standard API function for response generation"""
+    response = generate_response(query, intent)
+    return [response]
+
+def api_twilio_call():
+    """API function for initial Twilio call handling"""
+    return twilio_call_handler()
+
+def api_twilio_speech(speech_result=None):
+    """API function for Twilio speech processing"""
+    return twilio_speech_handler(speech_result)
+
+def api_twilio_followup(speech_result=None):
+    """API function for Twilio follow-up handling"""
+    return twilio_followup_handler(speech_result)
 
-if __name__ == "__main__":
-    app.run(debug=True, host='0.0.0.0', port=5000)
+# Mount these functions as API endpoints
+demo.queue()
+demo.launch()