import gradio as gr
import os
import tempfile
import requests
from pathlib import Path
# Create necessary directories
DOCUMENTS_DIR = Path("documents")
DOCUMENTS_DIR.mkdir(exist_ok=True)
# Define possible intents
POSSIBLE_INTENTS = [
"product_inquiry",
"technical_support",
"billing_question",
"general_information",
"appointment_scheduling",
"complaint",
"other"
]
# Default responses
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."
}
# Simple document storage (in-memory for this example)
knowledge_base = []
def classify_intent(text):
"""Simple keyword-based intent classification"""
if not text:
return "other", 0.6
text = text.lower()
if any(word in text for word in ["buy", "purchase", "product", "price", "cost"]):
return "product_inquiry", 0.8
elif any(word in text for word in ["help", "issue", "problem", "fix", "broken"]):
return "technical_support", 0.8
elif any(word in text for word in ["bill", "payment", "charge", "invoice"]):
return "billing_question", 0.8
elif any(word in text for word in ["appointment", "schedule", "book", "meeting"]):
return "appointment_scheduling", 0.8
elif any(word in text for word in ["unhappy", "disappointed", "complaint", "bad"]):
return "complaint", 0.8
else:
return "general_information", 0.6
def load_pdf(file):
"""Load a PDF document into the knowledge base"""
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())
# In a real implementation, we would extract text and add to vector store
# For this simplified version, we'll just store the file name
knowledge_base.append({"type": "pdf", "name": file.name, "path": target_path})
return f"Successfully added {file.name} to the knowledge base."
except Exception as e:
return f"Error processing PDF: {str(e)}"
def load_website(url):
"""Load a website into the knowledge base"""
try:
# In a real implementation, we would fetch and process the website
# For this simplified version, we'll just store the URL
knowledge_base.append({"type": "website", "url": url})
# Save the URL reference
with open(os.path.join(DOCUMENTS_DIR, "websites.txt"), "a") as f:
f.write(f"{url}\n")
return f"Successfully added {url} to the knowledge base."
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"""
# If no intent provided, use a default
if not intent or intent not in POSSIBLE_INTENTS:
intent = "general_information"
# For this simplified version, we'll just use the default responses
# In a real implementation, we would search the knowledge base
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
}
# Twilio voice handlers
def twilio_call_handler(data=None):
"""Handle incoming Twilio calls"""
twiml = """
Hello! Thank you for calling. How can I help you today?
I didn't hear anything. Please call back when you're ready.
"""
return twiml
def twilio_speech_handler(data):
"""Process speech from Twilio"""
# Extract speech from Twilio request
speech_result = data.get("SpeechResult", "")
# Process the speech
intent, _ = classify_intent(speech_result)
response = generate_response(speech_result, intent)
twiml = f"""
{response}
Is there anything else I can help you with today?
Thank you for calling. Have a great day!
"""
return twiml
# Create Gradio interface
with gr.Blocks(title="Call Assistant System") as demo:
gr.Markdown("# Call Assistant 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("Twilio Integration"):
gr.Markdown("""
## Twilio Integration Instructions
This app provides an API endpoint for Twilio voice 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", create a TwiML bin with:
```xml
Hello! Thank you for calling. How can I help you today?
I didn't hear anything. Please call back when you're ready.
```
The system will:
- Process speech input through the /api/predict endpoint
- Generate responses using your knowledge base
- Handle follow-up questions
""")
# Define API prediction function for Twilio
def predict_api(data):
"""API endpoint for Twilio integration"""
# Check if this is a speech result from Twilio
if isinstance(data, dict) and "SpeechResult" in data:
return twilio_speech_handler(data)
# If it's a call initial request or other Twilio request
# Just return the call handler response
return twilio_call_handler(data)
# Launch the interface with API
demo.queue()
demo.launch()