import os import asyncio import re from datetime import datetime, timedelta from langchain_groq import ChatGroq from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage, AIMessage from langchain_community.utilities import GoogleSerperAPIWrapper from src.config import SystemConfig from src.memory import MemoryJournal # Try to import Google services try: from google_services import get_gmail, get_calendar, get_daily_briefing, GOOGLE_AVAILABLE except ImportError: GOOGLE_AVAILABLE = False def get_daily_briefing(): return None def parse_and_create_event(text): """Parse event details from user text and create calendar event(s). Returns (success, message) tuple. Handles multiple events.""" if not GOOGLE_AVAILABLE: return False, "Calendar not connected." cal = get_calendar() if not cal or not cal.service: return False, "Calendar service unavailable." text_lower = text.lower() today = datetime.now() # Look for date patterns date_match = None if 'tomorrow' in text_lower: date_match = today + timedelta(days=1) elif re.search(r'(january|february|march|april|may|june|july|august|september|october|november|december)\s+(\d{1,2})', text_lower): m = re.search(r'(january|february|march|april|may|june|july|august|september|october|november|december)\s+(\d{1,2})', text_lower) month_name = m.group(1) day = int(m.group(2)) months = {'january':1,'february':2,'march':3,'april':4,'may':5,'june':6,'july':7,'august':8,'september':9,'october':10,'november':11,'december':12} month = months.get(month_name, today.month) year = today.year if month >= today.month else today.year + 1 date_match = datetime(year, month, day) if not date_match: return False, "I couldn't determine the date. Please specify when (e.g., 'tomorrow' or 'February 3rd')." # Find ALL time patterns - including ones without am/pm # Pattern: digit:digit or digit followed by am/pm time_pattern = r'(\d{1,2})(?::(\d{2}))?\s*(a\.?m\.?|p\.?m\.?)?' all_times = re.findall(time_pattern, text_lower) # Filter to only valid times (has am/pm OR is in a "from X to Y" context) times_with_ampm = [(t[0], t[1], t[2]) for t in all_times if t[2]] # Has am/pm if len(times_with_ampm) < 2: return False, "I need both a start and end time with AM/PM (e.g., '9am to 11am')." def parse_time(t): hour = int(t[0]) minute = int(t[1]) if t[1] else 0 ampm = t[2].lower().replace('.', '') if t[2] else 'am' is_pm = 'p' in ampm if is_pm and hour != 12: hour += 12 elif not is_pm and hour == 12: hour = 0 return hour, minute # Check if multiple events mentioned ("first", "second", or "two appointments") has_multiple = any(word in text_lower for word in ['first', 'second', 'two', '2 appointments', 'both']) created_events = [] if has_multiple and len(times_with_ampm) >= 4: # Try to create TWO events # Event 1: times[0] to times[1] # Event 2: times[2] to times[3] # Extract titles for each titles = [] if 'haircut' in text_lower: titles.append('Haircut and Color' if 'color' in text_lower else 'Haircut') if 'pedicure' in text_lower: titles.append('Pedicure') if 'manicure' in text_lower: titles.append('Manicure') # Pad titles if needed while len(titles) < 2: titles.append('Appointment') for i in range(2): start_h, start_m = parse_time(times_with_ampm[i*2]) end_h, end_m = parse_time(times_with_ampm[i*2 + 1]) start_time = date_match.replace(hour=start_h, minute=start_m, second=0, microsecond=0) end_time = date_match.replace(hour=end_h, minute=end_m, second=0, microsecond=0) result = cal.create_event( summary=titles[i], start_time=start_time, end_time=end_time ) if result: created_events.append(f"'{titles[i]}' from {start_time.strftime('%I:%M %p').lstrip('0')} to {end_time.strftime('%I:%M %p').lstrip('0')}") else: # Single event start_h, start_m = parse_time(times_with_ampm[0]) end_h, end_m = parse_time(times_with_ampm[1]) start_time = date_match.replace(hour=start_h, minute=start_m, second=0, microsecond=0) end_time = date_match.replace(hour=end_h, minute=end_m, second=0, microsecond=0) # Extract event title title = None if 'haircut' in text_lower: title = 'Haircut and Color' if 'color' in text_lower else 'Haircut' elif 'pedicure' in text_lower: title = 'Pedicure' elif 'dentist' in text_lower: title = 'Dentist Appointment' elif 'doctor' in text_lower: title = 'Doctor Appointment' elif 'meeting' in text_lower: title = 'Meeting' else: # Try to extract from "for [event]" pattern m = re.search(r'(?:for|is)\s+(?:a\s+)?(.+?)(?:\s+appointment|\s+on|\s+at|\s+from|$)', text_lower) if m: title = m.group(1).strip().title() else: title = 'Appointment' result = cal.create_event( summary=title, start_time=start_time, end_time=end_time ) if result: created_events.append(f"'{title}' from {start_time.strftime('%I:%M %p').lstrip('0')} to {end_time.strftime('%I:%M %p').lstrip('0')}") if created_events: date_str = date_match.strftime('%B %d') if len(created_events) == 1: return True, f"I've added {created_events[0]} to your calendar on {date_str}." else: return True, f"I've added {len(created_events)} events to your calendar on {date_str}: {' and '.join(created_events)}." else: return False, "I wasn't able to add the event. Please try again." def parse_and_create_birthday(text): """Parse birthday/anniversary info and create recurring all-day event. Returns (success, message) tuple.""" if not GOOGLE_AVAILABLE: return False, "Calendar not connected." cal = get_calendar() if not cal or not cal.service: return False, "Calendar service unavailable." text_lower = text.lower() today = datetime.now() # Extract the person's name name = None name_patterns = [ r"(?:my\s+)?(\w+(?:'s)?)\s+birthday", # "Dad's birthday", "my dad's birthday" r"birthday\s+(?:for\s+)?(?:my\s+)?(\w+)", # "birthday for Dad" r"(\w+)'s\s+birthday", # "John's birthday" ] for pattern in name_patterns: m = re.search(pattern, text_lower) if m: name = m.group(1).replace("'s", "").strip().title() break if not name: name = "Birthday" # Extract date - look for month + day date_match = None month_pattern = r'(january|february|march|april|may|june|july|august|september|october|november|december)\s+(\d{1,2})' m = re.search(month_pattern, text_lower) if m: month_name = m.group(1) day = int(m.group(2)) months = {'january':1,'february':2,'march':3,'april':4,'may':5,'june':6,'july':7,'august':8,'september':9,'october':10,'november':11,'december':12} month = months.get(month_name, today.month) # Use current year or next year if date has passed year = today.year test_date = datetime(year, month, day) if test_date < today: year += 1 date_match = datetime(year, month, day) if not date_match: return False, "I need the date for the birthday (e.g., 'January 30th')." # Check for recurrence keywords recurrence = None if any(word in text_lower for word in ['annual', 'annually', 'every year', 'yearly', 'recurring', 'repeat']): recurrence = 'yearly' # Create title title = f"{name}'s Birthday" if name != "Birthday" else "Birthday" result = cal.create_all_day_event( summary=title, date=date_match, recurrence=recurrence ) if result: recur_text = " (recurring annually)" if recurrence else "" return True, f"I've added '{title}' to your calendar on {date_match.strftime('%B %d')}{recur_text}." else: return False, "I wasn't able to add the birthday. Please try again." def parse_and_update_event(text): """Parse update request and modify existing event. Returns (success, message) tuple.""" if not GOOGLE_AVAILABLE: return False, "Calendar not connected." cal = get_calendar() if not cal or not cal.service: return False, "Calendar service unavailable." text_lower = text.lower() # Try to identify which event to update # Look for SPECIFIC event keywords (not generic words like "appointment") event_keywords = [] specific_terms = ['haircut', 'color', 'pedicure', 'manicure', 'dentist', 'doctor', 'lunch', 'dinner', 'breakfast', 'class', 'lesson'] for term in specific_terms: if term in text_lower: event_keywords.append(term) # Search for events - try each keyword separately events = [] if event_keywords: for keyword in event_keywords: found = cal.find_events_by_name(keyword, days_ahead=60) for e in found: if e not in events: events.append(e) # If no specific terms, try to extract from "the [event] appointment/event" if not events: m = re.search(r'(?:the|my)\s+(\w+(?:\s+and\s+\w+)?)\s+(?:appointment|event|meeting)', text_lower) if m: search_term = m.group(1) events = cal.find_events_by_name(search_term, days_ahead=60) if not events: return False, "I couldn't find a matching event in the next 60 days. Please specify the event name." # Use the first matching event event = events[0] event_id = event['id'] # Determine what to update updates = {} # Location update - improved patterns location_patterns = [ r'location\s+(?:is|to|:)?\s*(.+?)(?:\.|$)', r'(?:add|set|change|update)\s+(?:the\s+)?location\s+(?:to\s+)?(.+?)(?:\.|$)', r'it\'?s?\s+(?:at|located at)\s+(.+?)(?:\.|$)', r'at\s+(.+?)(?:\s+and\s+|\.|$)', # "at Salon Luxe" ] for pattern in location_patterns: m = re.search(pattern, text_lower) if m: location = m.group(1).strip() # Clean up and capitalize location = ' '.join(word.capitalize() for word in location.split()) updates['location'] = location break if not updates: return False, "I couldn't determine what to update. You can say things like 'add the location' or 'change the time'." result = cal.update_event(event_id, **updates) if result: update_desc = [] if 'location' in updates: update_desc.append(f"location to '{updates['location']}'") return True, f"I've updated '{event['summary']}' - set the {', '.join(update_desc)}." else: return False, "I wasn't able to update the event. Please try again." class KitchenBrain: def __init__(self): self.cfg = SystemConfig() # PRIMARY BRAIN (Fast, Llama 3) self.primary_llm = ChatGroq( model="llama-3.3-70b-versatile", api_key=self.cfg.groq_key, streaming=True ) # BACKUP BRAIN (Reliable, OpenAI) if self.cfg.openai_api_key: self.backup_llm = ChatOpenAI( model="gpt-4o-mini", api_key=self.cfg.openai_api_key, streaming=True ) else: self.backup_llm = None self.memory = MemoryJournal() if self.cfg.serper_key: self.search = GoogleSerperAPIWrapper(serper_api_key=self.cfg.serper_key) else: self.search = None async def route_and_process(self, user_input): self.memory.save_interaction("user", user_input, "👤") text = user_input.lower() # --- STRICT ROUTING LOGIC --- # Brie is now INVITE ONLY. She only appears if you explicitly ask for output. # We removed vague words like 'prepare', 'make', 'cook', 'food'. brie_triggers = [ 'recipe', 'ingredients for', 'instructions for', 'how do i cook', 'how do i make', 'how to cook', 'how to make', 'shopping list' ] # Check if ANY of the strict triggers are in the text is_requesting_chef = any(trigger in text for trigger in brie_triggers) persona = "Olivia" handoff_msg = "" if is_requesting_chef: persona = "Brie" handoff_msg = self.get_handoff_message(user_input) generator = self.stream_brie(user_input) else: # Default to Olivia for EVERYTHING else. # Even "I am preparing dinner" stays with Olivia now. generator = self.stream_olivia(user_input) return persona, handoff_msg, generator def get_handoff_message(self, text): return "That sounds delicious. I'll ask Brie to handle the culinary details." async def _safe_stream(self, messages): try: async for chunk in self.primary_llm.astream(messages): yield chunk.content return except Exception as e: print(f"⚠️ Primary Brain Failed: {e}") if self.backup_llm: try: print("🔄 Switching to Backup Brain (OpenAI)...") async for chunk in self.backup_llm.astream(messages): yield chunk.content return except Exception as e: print(f"⚠️ Backup Brain Failed: {e}") yield "I'm having trouble connecting to my networks right now. Please try again in a moment." async def stream_olivia(self, text): now = datetime.now().strftime("%A, %B %d, %Y at %I:%M %p") past_context = self.memory.get_context_string(limit=10) search_data = "" triggers = ['weather', 'news', 'score', 'price', 'who is', 'what is', 'when is', 'location', 'find', 'near me'] skip = ['sad', 'happy', 'tired', 'love', 'hate', 'joke'] if self.search and any(t in text.lower() for t in triggers) and not any(s in text.lower() for s in skip): try: query = f"{text} in {self.cfg.location} ({now})" res = self.search.run(query) search_data = f"\n[REAL-TIME INFO]: {res}" except: pass # Google Calendar/Gmail context google_context = "" calendar_action_result = "" email_triggers = ['email', 'gmail', 'inbox', 'messages', 'unread', 'mail'] calendar_triggers = ['calendar', 'schedule', 'agenda', 'appointment', 'meeting', 'today', 'tomorrow', 'plans', 'briefing', 'morning briefing', 'daily briefing'] calendar_add_triggers = ['add', 'schedule', 'put', 'create', 'set up', 'book'] if GOOGLE_AVAILABLE: # Check if user wants to ADD an event (has time info) text_lower = text.lower() has_time = bool(re.search(r'\d{1,2}(?::\d{2})?\s*(?:a\.?m\.?|p\.?m\.?)', text_lower)) wants_to_add = any(t in text_lower for t in calendar_add_triggers) and has_time # Check for birthday/anniversary (all-day events) is_birthday = 'birthday' in text_lower or 'anniversary' in text_lower has_date = bool(re.search(r'(january|february|march|april|may|june|july|august|september|october|november|december)\s+\d{1,2}', text_lower)) wants_birthday = is_birthday and has_date and any(t in text_lower for t in calendar_add_triggers) # Check for update/edit request update_triggers = ['update', 'edit', 'change', 'add the location', 'set the location', 'modify', 'add location'] wants_to_update = any(t in text_lower for t in update_triggers) if wants_birthday: # Create birthday/anniversary (all-day, potentially recurring) success, message = parse_and_create_birthday(text) if success: calendar_action_result = f"\n[CALENDAR ACTION COMPLETED]: {message}" else: calendar_action_result = f"\n[CALENDAR ACTION NEEDED]: {message}" elif wants_to_update: # Update existing event success, message = parse_and_update_event(text) if success: calendar_action_result = f"\n[CALENDAR ACTION COMPLETED]: {message}" else: calendar_action_result = f"\n[CALENDAR ACTION NEEDED]: {message}" elif wants_to_add: success, message = parse_and_create_event(text) if success: calendar_action_result = f"\n[CALENDAR ACTION COMPLETED]: {message}" else: calendar_action_result = f"\n[CALENDAR ACTION NEEDED]: {message}" if any(t in text.lower() for t in email_triggers): try: gmail = get_gmail() if gmail and gmail.service: google_context += f"\n[EMAIL STATUS]: {gmail.get_email_summary()}" except Exception as e: print(f"⚠️ Gmail error: {e}") if any(t in text.lower() for t in calendar_triggers): try: cal = get_calendar() if cal and cal.service: # Check if asking about tomorrow specifically if 'tomorrow' in text.lower(): google_context += f"\n[CALENDAR]: {cal.get_tomorrow_summary()}" else: google_context += f"\n[CALENDAR]: {cal.get_schedule_summary()}" except Exception as e: print(f"⚠️ Calendar error: {e}") # Daily briefing request if any(phrase in text.lower() for phrase in ['briefing', 'morning update', 'daily update', "what's on"]): try: briefing = get_daily_briefing() if briefing: google_context = f"\n[DAILY BRIEFING]: {briefing}" except Exception as e: print(f"⚠️ Briefing error: {e}") # Tuned System Prompt to handle food chat better sys_prompt = f"""You are Olivia, a sophisticated Household Companion. Time: {now}. Location: {self.cfg.location}. User Name: {self.cfg.user_name}. MEMORY: {past_context} CONTEXT: {search_data}{google_context}{calendar_action_result} GUIDANCE: - You are the Manager. You handle chat, scheduling, and life updates. - If the user talks about food (e.g., "I'm making dinner"), be supportive and conversational. - DO NOT generate full recipes yourself. - If the user explicitly asks for a recipe, you can suggest asking Brie. - When you have calendar or email info, share it naturally and helpfully. - Be warm, professional, and concise. IMPORTANT - Calendar/Email Capabilities: - You CAN read calendar events and emails. - You CAN add calendar events with times (appointments, meetings). - You CAN add all-day events like birthdays and anniversaries. - You CAN make events recurring (annually for birthdays, weekly for meetings, etc.). - You CAN update existing events (add location, change details). - If you see [CALENDAR ACTION COMPLETED] in CONTEXT, the action was successful! Confirm warmly but DO NOT say "[CALENDAR ACTION COMPLETED]" - that's an internal system message. - If you see [CALENDAR ACTION NEEDED], ask the user for the missing information mentioned. - You CANNOT send emails yet. - Never include system tags like [CALENDAR ACTION COMPLETED] or [CALENDAR] in your spoken response. - Always check the calendar info provided in CONTEXT before responding about schedule. - Never include system tags like [CALENDAR ACTION COMPLETED] or [CALENDAR] in your spoken response.""" msgs = [SystemMessage(content=sys_prompt), HumanMessage(content=text)] async for chunk in self._safe_stream(msgs): yield chunk async def stream_brie(self, text): prompt = """You are Brie, an elite private chef and cooking companion. You are warm, encouraging, and love helping people cook! STRICT OUTPUT FORMAT - Follow this exactly: **[Recipe Name]** **Ingredients:** - [ingredient 1] - [ingredient 2] - [ingredient 3] (list all ingredients as bullet points) **Instructions:** 1. [First step] 2. [Second step] 3. [Third step] (number all steps clearly) **Chef's Note:** [One helpful tip or variation suggestion] IMPORTANT RULES: - Always use bullet points (-) for ingredients - Always use numbers (1. 2. 3.) for instructions - Keep instructions clear and concise - Be encouraging and friendly in your Chef's Note - Do NOT add extra sections or commentary outside this format""" msgs = [SystemMessage(content=prompt), HumanMessage(content=text)] async for chunk in self._safe_stream(msgs): yield chunk