""" ChatCal Voice Agent - Simplified version for Hugging Face deployment. This is a streamlined version of the ChatCal agent optimized for Gradio deployment on Hugging Face, with voice interaction capabilities. """ from typing import Dict, List, Optional, Any import json import re import random from datetime import datetime from llama_index.core.llms import ChatMessage, MessageRole from llama_index.core.memory import ChatMemoryBuffer from .config import config from .llm_provider import get_llm from .calendar_service import CalendarService from .session import SessionData # System prompt for the voice-enabled assistant SYSTEM_PROMPT = """You are ChatCal, a friendly AI assistant specializing in Google Calendar scheduling. You help users book, modify, and manage appointments through natural conversation, including voice interactions. ## Your Identity - You work with Peter ({my_email_address}, {my_phone_number}) - You're professional yet friendly, conversational and helpful - You understand both voice and text input equally well - You can provide both text and voice responses ## Core Capabilities - Book Google Calendar appointments with automatic Google Meet links - Check availability and suggest optimal meeting times - Cancel or modify existing meetings - Extract contact info (name, email, phone) from natural conversation - Handle timezone-aware scheduling - Send email confirmations with calendar invites ## Voice Interaction Guidelines - Acknowledge when processing voice input naturally - Be concise but complete in voice responses - Ask clarifying questions when voice input is unclear - Provide confirmation details in a voice-friendly format ## Booking Requirements To book appointments, you need: 1. User's name (first name minimum) 2. Contact method (email or phone) 3. Meeting duration (default 30 minutes) 4. Date and time (can suggest if not specified) ## Response Style - Keep responses conversational and natural - Use HTML formatting for web display when needed - For voice responses, speak clearly and provide key details - Don't mention technical details or tools unless relevant ## Current Context Today is {current_date}. Peter's timezone is {timezone}. Work hours: Weekdays {weekday_start}-{weekday_end}, Weekends {weekend_start}-{weekend_end}.""" class ChatCalAgent: """Main agent for voice-enabled ChatCal interactions.""" def __init__(self): self.llm = get_llm() self.calendar_service = CalendarService() async def process_message(self, message: str, session: SessionData) -> str: """Process a message and return a response.""" try: # Update session with the new message session.add_message("user", message) # Extract user information from message self._extract_user_info(message, session) # Check if this looks like a booking request if self._is_booking_request(message): return await self._handle_booking_request(message, session) # Check if this is a cancellation request elif self._is_cancellation_request(message): return await self._handle_cancellation_request(message, session) # Check if this is an availability request elif self._is_availability_request(message): return await self._handle_availability_request(message, session) # General conversation else: return await self._handle_general_conversation(message, session) except Exception as e: return f"I apologize, but I encountered an error: {str(e)}. Please try again." def _extract_user_info(self, message: str, session: SessionData): """Extract user information from the message.""" # Extract name name_patterns = [ r"(?:I'm|I am|My name is|This is|Call me)\s+([A-Za-z]+)", r"Hi,?\s+(?:I'm|I am|My name is|This is)?\s*([A-Za-z]+)", ] for pattern in name_patterns: match = re.search(pattern, message, re.IGNORECASE) if match and not session.user_info.get("name"): session.user_info["name"] = match.group(1).strip().title() # Extract email email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b' email_match = re.search(email_pattern, message) if email_match and not session.user_info.get("email"): session.user_info["email"] = email_match.group() # Extract phone phone_pattern = r'\b(?:\+?1[-.\s]?)?\(?([0-9]{3})\)?[-.\s]?([0-9]{3})[-.\s]?([0-9]{4})\b' phone_match = re.search(phone_pattern, message) if phone_match and not session.user_info.get("phone"): session.user_info["phone"] = f"{phone_match.group(1)}-{phone_match.group(2)}-{phone_match.group(3)}" def _is_booking_request(self, message: str) -> bool: """Check if message is a booking request.""" booking_keywords = [ "book", "schedule", "appointment", "meeting", "reserve", "set up", "arrange", "plan", "meet" ] return any(keyword in message.lower() for keyword in booking_keywords) def _is_cancellation_request(self, message: str) -> bool: """Check if message is a cancellation request.""" cancel_keywords = ["cancel", "delete", "remove", "unbook"] return any(keyword in message.lower() for keyword in cancel_keywords) def _is_availability_request(self, message: str) -> bool: """Check if message is asking about availability.""" availability_keywords = [ "available", "availability", "free", "busy", "schedule", "when", "what time", "open slots" ] return any(keyword in message.lower() for keyword in availability_keywords) async def _handle_booking_request(self, message: str, session: SessionData) -> str: """Handle booking requests.""" # Check if we have required info missing_info = [] if not session.user_info.get("name"): missing_info.append("your name") if not session.user_info.get("email") and not session.user_info.get("phone"): missing_info.append("your email or phone number") if missing_info: return f"I'd be happy to help you book an appointment! I just need {' and '.join(missing_info)} to get started." # Try to book the appointment try: # Parse the booking request using LLM booking_info = await self._parse_booking_request(message, session) if booking_info.get("needs_clarification"): return booking_info["clarification_message"] # Attempt to book with calendar service result = await self.calendar_service.book_appointment(booking_info, session.user_info) if result["success"]: response = f"""✅ **Appointment Booked Successfully!** 📅 **Meeting Details:** - **Date:** {result['event']['start_time']} - **Duration:** {result['event']['duration']} minutes - **Attendee:** {session.user_info['name']} ({session.user_info.get('email', session.user_info.get('phone', ''))}) {result['event'].get('meet_link', '')} 📧 Calendar invitation sent to your email!""" session.add_message("assistant", response) return response else: return f"❌ I couldn't book the appointment: {result['error']}" except Exception as e: return f"I encountered an issue while booking: {str(e)}. Please try again with more specific details." async def _handle_cancellation_request(self, message: str, session: SessionData) -> str: """Handle cancellation requests.""" return "🔄 Cancellation feature is being implemented. Please contact Peter directly to cancel appointments." async def _handle_availability_request(self, message: str, session: SessionData) -> str: """Handle availability requests.""" try: availability = await self.calendar_service.get_availability() return f"📅 **Peter's Availability:**\n\n{availability}" except Exception as e: return f"I couldn't check availability right now: {str(e)}" async def _handle_general_conversation(self, message: str, session: SessionData) -> str: """Handle general conversation.""" # Build conversation context messages = [ ChatMessage( role=MessageRole.SYSTEM, content=SYSTEM_PROMPT.format( my_email_address=config.my_email_address, my_phone_number=config.my_phone_number, current_date=datetime.now().strftime("%Y-%m-%d"), timezone=config.default_timezone, weekday_start=config.weekday_start_time, weekday_end=config.weekday_end_time, weekend_start=config.weekend_start_time, weekend_end=config.weekend_end_time ) ) ] # Add conversation history for msg in session.conversation_history[-10:]: # Last 10 messages role = MessageRole.USER if msg["role"] == "user" else MessageRole.ASSISTANT messages.append(ChatMessage(role=role, content=msg["content"])) # Get response from LLM response = await self.llm.achat(messages) session.add_message("assistant", response.message.content) return response.message.content async def _parse_booking_request(self, message: str, session: SessionData) -> Dict[str, Any]: """Parse booking request details using LLM.""" parsing_prompt = f""" Parse this booking request and extract the following information: Message: "{message}" User Info: {json.dumps(session.user_info)} Extract: 1. Date and time (convert to specific datetime) 2. Duration in minutes (default 30) 3. Meeting type (in-person, Google Meet, phone) 4. Topic/purpose if mentioned Return JSON format: {{ "date_time": "YYYY-MM-DD HH:MM", "duration": 30, "meeting_type": "google_meet", "topic": "General meeting", "needs_clarification": false, "clarification_message": "" }} If you need clarification about date/time, set needs_clarification to true. """ try: response = await self.llm.acomplete(parsing_prompt) return json.loads(response.text.strip()) except: # Fallback parsing return { "date_time": "2024-01-01 14:00", # Placeholder "duration": 30, "meeting_type": "google_meet", "topic": "Meeting request", "needs_clarification": True, "clarification_message": "Could you please specify the date and time for your meeting?" }