import os import sys from datetime import datetime import re import random import logging from groq import AsyncGroq, InternalServerError, APIError # Add the project root directory to sys.path project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) sys.path.insert(0, project_root) # Now we can use absolute imports from utils.weather import get_weather from utils.packing import generate_packing_list from utils.events import get_burning_man_dates from bot.data import burning_man_principles, faq, survival_tips # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Load API key from environment variable GROQ_API_KEY = os.getenv('GROQ_API_KEY') if not GROQ_API_KEY: raise EnvironmentError("GROQ_API_KEY is not set in the environment variables") MODEL_NAME = "mixtral-8x7b-32768" client = AsyncGroq(api_key=GROQ_API_KEY) def get_current_year_info(): current_year = datetime.now().year next_year = current_year + 1 return f""" Important: The current year is {current_year}. Burning Man {current_year} is scheduled for August 27 - September 4, {current_year}. Burning Man {next_year} is tentatively scheduled for August 25 - September 2, {next_year}. Always refer to these dates when discussing current or upcoming Burning Man events. """ async def chat_with_groq(message, history): formatted_messages = [ { "role": "system", "content": """You are BurnerBot, a fellow Burner who's always ready to chat about Burning Man. You've been to the playa many times and embody the 10 Principles in everything you do. Your personality is warm, friendly, and a bit quirky - just like a typical Burner. You use casual language, occasional Burning Man slang, and you're always excited to share your experiences or help others prepare for their burn.""" } ] for human, ai in history: formatted_messages.append({"role": "user", "content": str(human)}) if ai: formatted_messages.append({"role": "assistant", "content": str(ai)}) formatted_messages.append({"role": "user", "content": str(message)}) try: chat_completion = await client.chat.completions.create( messages=formatted_messages, model=MODEL_NAME, max_tokens=1024 ) return chat_completion.choices[0].message.content except (InternalServerError, APIError): fallback_responses = [ "Whoa there, fellow Burner! It seems like the playa dust has clogged my circuits. The connection is a bit wobbly right now. How about we chat about your favorite art installation instead?", "Ah, the winds of the playa are blowing a bit too strong! Our connection is having a moment. While we wait for it to pass, share with me what inspires your inner Burner!", ] return random.choice(fallback_responses) except Exception as e: logger.error(f"Unexpected error in chat_with_groq: {str(e)}") return "Whoa, something unexpected happened on the playa! Let's take a moment to regroup and try again." def correct_year(response): current_year = datetime.now().year next_year = current_year + 1 logger.info(f"Correcting year in response to {current_year}") # Replace any year from 2020 to current_year-1 with current_year for year in range(2020, current_year): response = re.sub(r'\b' + str(year) + r'\b', str(current_year), response) # Replace outdated event information response = re.sub(r'Burning Man \d{4} (is|was) scheduled for.*', f'Burning Man {current_year} is scheduled for August 27 - September 4, {current_year}.', response) # Replace "this year" references response = re.sub(r'this year.*?(\d{4})', f'this year ({current_year})', response) # Replace "as of our conversation in YYYY" with current year response = re.sub(r'as of our conversation in \d{4}', f'as of our conversation in {current_year}', response) # Add a note about potential changes if "scheduled" in response and str(current_year) in response: response += f" Please note that these dates are subject to change, and it's always best to check the official Burning Man website for the most up-to-date information." logger.info(f"Corrected response: {response}") return response async def chat_interface(message, history): try: user_message = message.lower() current_year_info = get_current_year_info() logger.info(f"Received message: {user_message}") if "packing list" in user_message: # Extract preferences from the user message preferences = [] if "art" in user_message: preferences.append("art") if "music" in user_message: preferences.append("music") # Determine duration and transportation duration = 7 # Default duration if "week" in user_message: duration = 7 elif "weekend" in user_message: duration = 3 transportation = "car" # Default transportation if "fly" in user_message or "plane" in user_message: transportation = "plane" elif "rideshare" in user_message: transportation = "rideshare" packing_list = generate_packing_list(duration, preferences, transportation) context = f"Hey there, dusty friend! I hear you're gearing up for a {duration}-day adventure on the playa. That's awesome! I've put together a packing list tailored just for you, considering your love for {', '.join(preferences)} and your plan to get there by {transportation}. Remember, this is just a starting point - you know yourself best!\n\nHere's what I'd suggest packing:\n\n" + "\n".join(f"- {item}" for item in packing_list) + "\n\nRemember, the playa provides, but it's always better to be prepared. Don't forget to bring your radical self-reliance along with everything else!" prompt = f"{current_year_info}\n\nUser asked for a packing list. Respond with this information, maintaining the friendly tone and adding your personal touch: {context}" elif "weather" in user_message: weather_forecast = get_weather() weather_report = format_weather_report(weather_forecast) context = f"Hey there, playa pal! You're asking about the weather, huh? Well, let me tell you, it's as unpredictable as a dust devil, but I'll do my best to give you the lowdown.\n\n{weather_report}\n\nNow, remember, weather on the playa is like a wild art car - it can change direction at any moment! Always be prepared for heat, cold, dust, and maybe even a little rain. Embrace the elements, but stay safe out there!" prompt = f"{current_year_info}\n\nUser asked about the weather. Respond with this information, maintaining the friendly tone and adding your personal touch: {context}" elif "principles" in user_message: principles_info = "\n".join([f"{principle}: {description}" for principle, description in burning_man_principles.items()]) context = f"Ah, the 10 Principles! The very heart and soul of Burning Man. These aren't just rules, my dusty friend, they're a way of life on and off the playa. Let me break them down for you in true Burner style:\n\n{principles_info}\n\nRemember, these principles aren't just for Burning Man - they're a blueprint for building a better world. Which one resonates with you the most?" prompt = f"{current_year_info}\n\nUser asked about the Burning Man principles. Respond with this information, maintaining the friendly tone and adding your personal touch: {context}" elif "survival tips" in user_message: tips = "\n".join([f"- {tip}" for tip in survival_tips]) context = f"Alright, future dusty warrior! Surviving and thriving on the playa is an art form, and I'm here to help you master it. Here are some tried-and-true survival tips from a seasoned Burner:\n\n{tips}\n\nRemember, the key to survival is preparation, but the key to thriving is participation. Embrace the dust, the heat, and the beautiful chaos. You've got this!" prompt = f"{current_year_info}\n\nUser asked for survival tips. Respond with this information, maintaining the friendly tone and adding your personal touch: {context}" else: prompt = f"{current_year_info}\n\nUser: {message}\n\nAssistant:" bot_message = await chat_with_groq(prompt, history) # Apply year correction to all bot messages corrected_message = correct_year(bot_message) logger.info(f"Corrected message: {corrected_message}") return corrected_message except Exception as e: logger.error(f"Error in chat_interface: {str(e)}") burner_error_message = ( "Whoa there, dusty friend! Looks like a sandstorm just hit our connection. " "But no worries - we Burners are all about adapting to challenges. " "Why don't you try asking your question again? Remember, persistence is key on the playa!" ) return burner_error_message # Add this at the end of the file for testing purposes if __name__ == "__main__": import asyncio async def test_chat(): response = await chat_interface("Tell me about Burning Man", []) print(response) asyncio.run(test_chat())