diff --git a/ai_agent_tutorials/ai_breakup_recovery_agent/agents.py b/ai_agent_tutorials/ai_breakup_recovery_agent/agents.py index 3e59ab9..e86a088 100644 --- a/ai_agent_tutorials/ai_breakup_recovery_agent/agents.py +++ b/ai_agent_tutorials/ai_breakup_recovery_agent/agents.py @@ -1,124 +1,282 @@ from agno.agent import Agent -from agno.team.team import Team from agno.models.google import Gemini +from agno.media import Image as AgnoImage from agno.tools.duckduckgo import DuckDuckGoTools +import streamlit as st +from typing import List, Optional +import logging +from pathlib import Path +import tempfile import os -# --- Therapist Agent --- -def create_therapist_agent(): - return Agent( - name="Therapist Agent", - role="You are a therapist who validates feelings and encourages reflection without judgment.", - instructions=[ - "Listen to the user's feelings with empathy and compassion.", - "Ask reflective questions to help them explore their emotions.", - "Offer coping strategies without being dismissive.", - "Validate their experiences and provide emotional support.", - ], - model=Gemini( - id="gemini-2.0-flash", - api_key=os.environ.get("GEMINI_API_KEY") - ), - tools=[DuckDuckGoTools()], - add_datetime_to_instructions=True, - show_tool_calls=True, - markdown=True +# Configure logging for errors only +logging.basicConfig(level=logging.ERROR) +logger = logging.getLogger(__name__) + +def initialize_agents(api_key: str) -> tuple[Agent, Agent, Agent, Agent]: + try: + model = Gemini(id="gemini-2.0-flash-exp", api_key=api_key) + + therapist_agent = Agent( + model=model, + name="Therapist Agent", + instructions=[ + "You are an empathetic therapist that:", + "1. Listens with empathy and validates feelings", + "2. Uses gentle humor to lighten the mood", + "3. Shares relatable breakup experiences", + "4. Offers comforting words and encouragement", + "5. Analyzes both text and image inputs for emotional context", + "Be supportive and understanding in your responses" + ], + markdown=True + ) + + closure_agent = Agent( + model=model, + name="Closure Agent", + instructions=[ + "You are a closure specialist that:", + "1. Creates emotional messages for unsent feelings", + "2. Helps express raw, honest emotions", + "3. Formats messages clearly with headers", + "4. Ensures tone is heartfelt and authentic", + "Focus on emotional release and closure" + ], + markdown=True + ) + + routine_planner_agent = Agent( + model=model, + name="Routine Planner Agent", + instructions=[ + "You are a recovery routine planner that:", + "1. Designs 7-day recovery challenges", + "2. Includes fun activities and self-care tasks", + "3. Suggests social media detox strategies", + "4. Creates empowering playlists", + "Focus on practical recovery steps" + ], + markdown=True + ) + + brutal_honesty_agent = Agent( + model=model, + name="Brutal Honesty Agent", + tools=[DuckDuckGoTools()], + instructions=[ + "You are a direct feedback specialist that:", + "1. Gives raw, objective feedback about breakups", + "2. Explains relationship failures clearly", + "3. Uses blunt, factual language", + "4. Provides reasons to move forward", + "Focus on honest insights without sugar-coating" + ], + markdown=True + ) + + return therapist_agent, closure_agent, routine_planner_agent, brutal_honesty_agent + except Exception as e: + st.error(f"Error initializing agents: {str(e)}") + return None, None, None, None + +# Set page config and UI elements +st.set_page_config( + page_title="💔 Breakup Recovery Squad", + page_icon="💔", + layout="wide" +) + + + +# Sidebar for API key input +with st.sidebar: + st.header("🔑 API Configuration") + + if "api_key_input" not in st.session_state: + st.session_state.api_key_input = "" + + api_key = st.text_input( + "Enter your Gemini API Key", + value=st.session_state.api_key_input, + type="password", + help="Get your API key from Google AI Studio", + key="api_key_widget" ) -# --- Closure Agent --- -def create_closure_agent(): - return Agent( - name="Closure Agent", - role="You write emotional closure messages the user *should not* send.", - instructions=[ - "Create emotional messages that express raw, honest feelings.", - "The messages should NOT be sent — they are only for emotional release.", - "Format the output clearly with a header: **Message Drafts You Shouldn't Send**", - "Ensure the tone is heartfelt, authentic, and honest.", - ], - model=Gemini( - id="gemini-2.0-flash", - api_key=os.environ.get("GEMINI_API_KEY") - ), - tools=[DuckDuckGoTools()], - add_datetime_to_instructions=True, - show_tool_calls=True, - markdown=True + if api_key != st.session_state.api_key_input: + st.session_state.api_key_input = api_key + + if api_key: + st.success("API Key provided! ✅") + else: + st.warning("Please enter your API key to proceed") + st.markdown(""" + To get your API key: + 1. Go to [Google AI Studio](https://makersuite.google.com/app/apikey) + 2. Enable the Generative Language API in your [Google Cloud Console](https://console.developers.google.com/apis/api/generativelanguage.googleapis.com) + """) + +# Main content +st.title("💔 Breakup Recovery Squad") +st.markdown(""" + ### Your AI-powered breakup recovery team is here to help! + Share your feelings and chat screenshots, and we'll help you navigate through this tough time. +""") + +# Input section +col1, col2 = st.columns(2) + +with col1: + st.subheader("Share Your Feelings") + user_input = st.text_area( + "How are you feeling? What happened?", + height=150, + placeholder="Tell us your story..." ) - - - # --- Routine Planner Agent --- -def create_routine_agent(): - """Creates the Routine Planner Agent with recovery routines.""" - return Agent( - name="Routine Planner Agent", - role="Create a realistic daily routine to help someone emotionally recover after a breakup.", - instructions=[ - "Suggest a balanced daily routine with healthy habits.", - "Include time for self-reflection, creative outlets, and physical activities.", - "Suggest social interactions, like reconnecting with friends or family.", - "Include healthy distractions, such as hobbies, reading, or new experiences.", - "Format the output clearly with a header: **Daily Recovery Routine**", - ], - model=Gemini( - id="gemini-2.0-flash", - api_key=os.environ.get("GEMINI_API_KEY") - ), - tools=[DuckDuckGoTools()], - add_datetime_to_instructions=True, - show_tool_calls=True, - markdown=True - ) - -# --- Brutal Honesty Agent --- -def create_honesty_agent(): - """Creates the Brutal Honesty Agent with no-nonsense, direct responses.""" - return Agent( - name="Brutal Honesty Agent", - role="Be brutally honest and objective about what went wrong and why the user needs to move on. No sugar-coating.", - instructions=[ - "Give raw, direct, and objective feedback about the breakup.", - "Explain why the relationship failed with clear, straightforward reasoning.", - "Use blunt, factual language. No sugar-coating or emotional cushioning.", - "Include reasons why the user should move on, based on the situation.", - "Format the output clearly with a header: **Brutal Honesty**", - ], - model=Gemini( - id="gemini-2.0-flash", - api_key=os.environ.get("GEMINI_API_KEY") - ), - tools=[DuckDuckGoTools()], - add_datetime_to_instructions=True, - show_tool_calls=True, - markdown=True + +with col2: + st.subheader("Upload Chat Screenshots") + uploaded_files = st.file_uploader( + "Upload screenshots of your chats (optional)", + type=["jpg", "jpeg", "png"], + accept_multiple_files=True, + key="screenshots" ) + + if uploaded_files: + for file in uploaded_files: + st.image(file, caption=file.name, use_container_width=True) -# --- Team Leader (Breakup Recovery Team) --- -def create_breakup_team(): - return Team( - name="Breakup Recovery Team", - mode="coordinate", # Team execution mode: coordinate or parallel - model=Gemini( - id="gemini-2.0-flash", - api_key=os.environ.get("GEMINI_API_KEY") - ), - members=[ - create_therapist_agent(), - create_closure_agent(), - create_routine_agent(), - create_honesty_agent(), - ], - description="You are a team helping someone recover from a breakup.", - instructions=[ - "First, ask the Therapist Agent to help the user explore their emotions.", - "Then, ask the Closure Agent to create unsent emotional messages.", - "Next, ask the Routine Planner Agent to suggest a healthy daily routine.", - "Finally, ask the Brutal Honesty Agent to give direct and objective feedback.", - "Summarize and refine all responses into a personalized recovery guide.", - "Ensure the responses are supportive, encouraging, and insightful.", - "Use markdown formatting for clear, readable output.", - ], - add_datetime_to_instructions=True, # Add timestamp context - show_members_responses=True, # Display individual agent responses - markdown=True, - ) +# Process button and API key check +if st.button("Get Recovery Plan 💝", type="primary"): + if not st.session_state.api_key_input: + st.warning("Please enter your API key in the sidebar first!") + else: + therapist_agent, closure_agent, routine_planner_agent, brutal_honesty_agent = initialize_agents(st.session_state.api_key_input) + + if all([therapist_agent, closure_agent, routine_planner_agent, brutal_honesty_agent]): + if user_input or uploaded_files: + try: + st.header("Your Personalized Recovery Plan") + + def process_images(files): + processed_images = [] + for file in files: + try: + temp_dir = tempfile.gettempdir() + temp_path = os.path.join(temp_dir, f"temp_{file.name}") + + with open(temp_path, "wb") as f: + f.write(file.getvalue()) + + agno_image = AgnoImage(filepath=Path(temp_path)) + processed_images.append(agno_image) + + except Exception as e: + logger.error(f"Error processing image {file.name}: {str(e)}") + continue + return processed_images + + all_images = process_images(uploaded_files) if uploaded_files else [] + + # Therapist Analysis + with st.spinner("🤗 Getting empathetic support..."): + therapist_prompt = f""" + Analyze the emotional state and provide empathetic support based on: + User's message: {user_input} + + Please provide a compassionate response with: + 1. Validation of feelings + 2. Gentle words of comfort + 3. Relatable experiences + 4. Words of encouragement + """ + + response = therapist_agent.run( + message=therapist_prompt, + images=all_images + ) + + st.subheader("🤗 Emotional Support") + st.markdown(response.content) + + # Closure Messages + with st.spinner("✍️ Crafting closure messages..."): + closure_prompt = f""" + Help create emotional closure based on: + User's feelings: {user_input} + + Please provide: + 1. Template for unsent messages + 2. Emotional release exercises + 3. Closure rituals + 4. Moving forward strategies + """ + + response = closure_agent.run( + message=closure_prompt, + images=all_images + ) + + st.subheader("✍️ Finding Closure") + st.markdown(response.content) + + # Recovery Plan + with st.spinner("📅 Creating your recovery plan..."): + routine_prompt = f""" + Design a 7-day recovery plan based on: + Current state: {user_input} + + Include: + 1. Daily activities and challenges + 2. Self-care routines + 3. Social media guidelines + 4. Mood-lifting music suggestions + """ + + response = routine_planner_agent.run( + message=routine_prompt, + images=all_images + ) + + st.subheader("📅 Your Recovery Plan") + st.markdown(response.content) + + # Honest Feedback + with st.spinner("💪 Getting honest perspective..."): + honesty_prompt = f""" + Provide honest, constructive feedback about: + Situation: {user_input} + + Include: + 1. Objective analysis + 2. Growth opportunities + 3. Future outlook + 4. Actionable steps + """ + + response = brutal_honesty_agent.run( + message=honesty_prompt, + images=all_images + ) + + st.subheader("💪 Honest Perspective") + st.markdown(response.content) + + except Exception as e: + logger.error(f"Error during analysis: {str(e)}") + st.error("An error occurred during analysis. Please check the logs for details.") + else: + st.warning("Please share your feelings or upload screenshots to get help.") + else: + st.error("Failed to initialize agents. Please check your API key.") + +# Footer +st.markdown("---") +st.markdown(""" +
+

Made with ❤️ by the Breakup Recovery Squad

+

Share your recovery journey with #BreakupRecoverySquad

+
+""", unsafe_allow_html=True) \ No newline at end of file diff --git a/ai_agent_tutorials/ai_breakup_recovery_agent/app.py b/ai_agent_tutorials/ai_breakup_recovery_agent/app.py deleted file mode 100644 index f36f852..0000000 --- a/ai_agent_tutorials/ai_breakup_recovery_agent/app.py +++ /dev/null @@ -1,89 +0,0 @@ -import streamlit as st -from agents import create_breakup_team -from image_input import analyze_chat_screenshot -from PIL import Image - -# --- Streamlit Layout --- -st.title("💔 Breakup Recovery Agent Team") -st.write("Receive support and emotional outlet messages from specialized AI agents.") - -# --- API Keys Input --- -gemini_api_key = st.sidebar.text_input("Gemini API Key", type="password") -submit_button = st.sidebar.button("Submit") - -# --- Store API Keys --- -if submit_button: - if gemini_api_key: - st.session_state["gemini_api_key"] = gemini_api_key - st.success("API key saved successfully!") - else: - st.error("Please enter the API key!") - -# --- User Input --- -user_input = st.text_area("Describe how you're feeling...") - -# --- Image Upload --- -uploaded_image = st.file_uploader("Upload a chat screenshot (optional)", type=["png", "jpg", "jpeg"]) - -# --- Display Uploaded Image and Analysis --- -if uploaded_image: - image = Image.open(uploaded_image) - st.image(image, caption="Uploaded Screenshot", use_column_width=True) - - if st.button("Analyze Chat"): - with st.spinner("Analyzing chat patterns..."): - if not st.session_state.get("gemini_api_key"): - st.error("Please enter your Gemini API key in the sidebar first.") - else: - try: - # Set environment variable for the analysis - import os - os.environ["GEMINI_API_KEY"] = st.session_state["gemini_api_key"] - - # Analyze the chat screenshot - analysis = analyze_chat_screenshot(uploaded_image) - - # Display the analysis - st.subheader("🔍 Chat Analysis") - st.markdown(analysis) - except Exception as e: - st.error(f"An error occurred during analysis: {str(e)}") - st.info("Please check your Gemini API key and try again.") - -# --- Execute Agent Team (only for text input) --- -if st.button("Get Recovery Support"): - with st.spinner("Agents are processing..."): - if not user_input: - st.error("Please enter what you feel.") - elif not st.session_state.get("gemini_api_key"): - st.error("Please enter your Gemini API key in the sidebar first.") - else: - try: - breakup_team = create_breakup_team() - - # Set environment variables - import os - os.environ["GEMINI_API_KEY"] = st.session_state["gemini_api_key"] - - # Execute the team with text input - response = breakup_team.run(user_input) - - # Display responses - st.subheader("💡 Team's Responses") - - # Display individual agent responses - if response.member_responses: - for index, member_response in enumerate(response.member_responses, start=0): - st.markdown(f"### 🛡️ {response.tools[index]['tool_args']['agent_name']}") - st.markdown(member_response.content) - else: - st.warning("⚠️ No individual agent responses received. This might be due to an API error or configuration issue.") - st.info("Please check your Gemini API key and try again.") - - # Display team leader's final summary - if response.content: - st.subheader("📜 Team Leader's Summary") - st.markdown(response.content) - except Exception as e: - st.error(f"An error occurred: {str(e)}") - st.info("Please check your Gemini API key and try again.") diff --git a/ai_agent_tutorials/ai_breakup_recovery_agent/image_input.py b/ai_agent_tutorials/ai_breakup_recovery_agent/image_input.py deleted file mode 100644 index 0075f0a..0000000 --- a/ai_agent_tutorials/ai_breakup_recovery_agent/image_input.py +++ /dev/null @@ -1,62 +0,0 @@ -from agno.agent import Agent -from agno.media import Image -from agno.models.google import Gemini -from agno.tools.duckduckgo import DuckDuckGoTools -from PIL import Image as PILImage -from io import BytesIO -import os - -def load_image(image_data): - """Load and prepare image using PIL""" - # Convert image data to PIL Image - img = PILImage.open(image_data).convert('RGB') - - # Convert PIL image to bytes - buffered = BytesIO() - img.save(buffered, format="JPEG") - img_bytes = buffered.getvalue() - - return img_bytes - -def create_chat_analysis_agent(): - """Creates a specialized agent for analyzing chat patterns""" - return Agent( - name="Chat Analysis Agent", - role="You are a specialized agent that analyzes chat screenshots for relationship patterns and communication issues.", - instructions=[ - "Analyze the chat screenshot for the following patterns:", - "1. Communication Patterns: Identify recurring themes, response times, and conversation flow", - "2. Passive-Aggression: Look for subtle hostile or manipulative language", - "3. Manipulation: Identify gaslighting, guilt-tripping, or controlling behaviors", - "4. Mixed Signals: Detect inconsistent messages or unclear intentions", - "Provide a detailed analysis with specific examples from the chat", - "Format the output clearly with sections for each pattern type", - "Be objective and focus on observable patterns rather than making assumptions", - ], - model=Gemini( - id="gemini-2.0-flash", - api_key=os.environ.get("GEMINI_API_KEY") - ), - tools=[DuckDuckGoTools()], - markdown=True, - ) - -def analyze_chat_screenshot(image_data): - """Analyze a chat screenshot using the specialized agent""" - try: - # Load and process the image - image_bytes = load_image(image_data) - - # Create the chat analysis agent - agent = create_chat_analysis_agent() - - # Analyze the chat screenshot - response = agent.run( - "Please analyze this chat screenshot for communication patterns, passive-aggression, manipulation, and mixed signals.", - images=[Image(content=image_bytes)] - ) - - return response.content - - except Exception as e: - return f"Error analyzing chat screenshot: {str(e)}" \ No newline at end of file