app.py

#!/usr/bin/env python3
"""
Personal Productivity Assistant Agent
Built using smolagents framework following HuggingFace Agents Course

USAGE:
1. pip install -r requirements.txt
2. export HF_TOKEN=your_huggingface_token
3. python app.py

OR deploy to HuggingFace Spaces directly!
"""

import os
import json
import requests
import datetime
from typing import List, Dict, Optional
import gradio as gr
from dataclasses import dataclass

# smolagents imports
from smolagents import (
    CodeAgent, 
    ToolCallingAgent,
    DuckDuckGoSearchTool, 
    InferenceClientModel, 
    tool,
    Tool
)

# Set your HuggingFace token here
# In HuggingFace Spaces, this will be automatically loaded from secrets
HF_TOKEN = os.environ.get("HF_TOKEN") or os.environ.get("HUGGINGFACE_HUB_TOKEN")

# Validate HF token
if not HF_TOKEN or HF_TOKEN == "your_hf_token_here":
    print("⚠️ WARNING: HF_TOKEN not set!")
    print("Set it with: export HF_TOKEN=your_actual_token")
    print("Get token from: https://huggingface.co/settings/tokens")
    print("Continuing with limited functionality...")
    HF_TOKEN = None

@dataclass
class Task:
    """Simple task representation"""
    id: str
    title: str
    description: str
    due_date: Optional[str] = None
    completed: bool = False

class TaskManager:
    """Simple in-memory task management"""
    def __init__(self):
        self.tasks = []
        self.next_id = 1
    
    def add_task(self, title: str, description: str, due_date: str = None) -> str:
        task = Task(
            id=str(self.next_id),
            title=title,
            description=description,
            due_date=due_date
        )
        self.tasks.append(task)
        self.next_id += 1
        return f"Task '{title}' added with ID {task.id}"
    
    def list_tasks(self) -> str:
        if not self.tasks:
            return "No tasks found."
        
        result = "Current Tasks:\n"
        for task in self.tasks:
            status = "✅" if task.completed else "⏳"
            due_info = f" (Due: {task.due_date})" if task.due_date else ""
            result += f"{status} [{task.id}] {task.title}{due_info}\n"
        return result
    
    def complete_task(self, task_id: str) -> str:
        for task in self.tasks:
            if task.id == task_id:
                task.completed = True
                return f"Task '{task.title}' marked as completed!"
        return f"Task with ID {task_id} not found."

# Global task manager instance
task_manager = TaskManager()

@tool
def add_task(title: str, description: str, due_date: str = None) -> str:
    """
    Add a new task to the task list.
    
    Args:
        title: Task title
        description: Task description  
        due_date: Optional due date in YYYY-MM-DD format
    
    Returns:
        Confirmation message
    """
    return task_manager.add_task(title, description, due_date)

@tool
def list_tasks() -> str:
    """
    List all current tasks with their status.
    
    Returns:
        Formatted list of tasks
    """
    return task_manager.list_tasks()

@tool
def complete_task(task_id: str) -> str:
    """
    Mark a task as completed.
    
    Args:
        task_id: ID of the task to complete
    
    Returns:
        Confirmation message
    """
    return task_manager.complete_task(task_id)

@tool
def get_current_time() -> str:
    """
    Get the current date and time.
    
    Returns:
        Current date and time formatted string
    """
    now = datetime.datetime.now()
    return f"Current time: {now.strftime('%Y-%m-%d %H:%M:%S')}"

@tool
def calculate_days_until(target_date: str) -> str:
    """
    Calculate days between today and a target date.
    
    Args:
        target_date: Date in YYYY-MM-DD format
    
    Returns:
        Number of days until target date
    """
    try:
        today = datetime.date.today()
        target = datetime.datetime.strptime(target_date, '%Y-%m-%d').date()
        days_diff = (target - today).days
        
        if days_diff > 0:
            return f"{days_diff} days until {target_date}"
        elif days_diff == 0:
            return f"{target_date} is today!"
        else:
            return f"{target_date} was {abs(days_diff)} days ago"
    except ValueError:
        return "Invalid date format. Please use YYYY-MM-DD."

@tool
def get_weather_info(city: str) -> str:
    """
    Get weather information for a city using a free weather API.
    
    Args:
        city: City name
    
    Returns:
        Weather information string
    """
    try:
        # Using OpenWeatherMap API (you'd need to sign up for a free API key)
        # For demo purposes, returning mock data
        mock_weather = {
            "london": "London: 15°C, Cloudy, 60% humidity",
            "new york": "New York: 22°C, Sunny, 45% humidity", 
            "paris": "Paris: 18°C, Light rain, 70% humidity",
            "tokyo": "Tokyo: 25°C, Partly cloudy, 55% humidity"
        }
        
        city_lower = city.lower()
        if city_lower in mock_weather:
            return mock_weather[city_lower]
        else:
            return f"Weather data for {city}: 20°C, Clear skies, 50% humidity (Mock data)"
    except Exception as e:
        return f"Could not fetch weather for {city}: {str(e)}"

@tool
def send_mock_email(recipient: str, subject: str, body: str) -> str:
    """
    Send a mock email (simulation only).
    
    Args:
        recipient: Email recipient
        subject: Email subject
        body: Email body content
    
    Returns:
        Confirmation message
    """
    return f"Mock email sent to {recipient}\nSubject: {subject}\nBody preview: {body[:50]}..."

@tool
def create_meeting_summary(attendees: str, topic: str, duration_minutes: int) -> str:
    """
    Create a meeting summary template.
    
    Args:
        attendees: Comma-separated list of attendees
        topic: Meeting topic
        duration_minutes: Meeting duration in minutes
    
    Returns:
        Meeting summary template
    """
    now = datetime.datetime.now()
    end_time = now + datetime.timedelta(minutes=duration_minutes)
    
    summary = f"""
    MEETING SUMMARY TEMPLATE
    ========================
    Date: {now.strftime('%Y-%m-%d')}
    Time: {now.strftime('%H:%M')} - {end_time.strftime('%H:%M')}
    Duration: {duration_minutes} minutes
    Topic: {topic}
    Attendees: {attendees}
    
    AGENDA:
    - [ ] Welcome and introductions
    - [ ] Main topic discussion
    - [ ] Action items
    - [ ] Next steps
    
    NOTES:
    [Add meeting notes here]
    
    ACTION ITEMS:
    - [ ] [Action item 1 - Assignee]
    - [ ] [Action item 2 - Assignee]
    
    NEXT MEETING: [Date/Time]
    """
    return summary

class PersonalProductivityAgent:
    """Main productivity agent class"""
    
    def __init__(self):
        # Initialize the model with fallback
        try:
            self.model = InferenceClientModel(
                model="Qwen/Qwen2.5-Coder-32B-Instruct",
                token=HF_TOKEN
            )
        except Exception as e:
            print(f"⚠️ Primary model failed, using fallback: {e}")
            # Fallback to a smaller, more reliable model
            self.model = InferenceClientModel(
                model="microsoft/DialoGPT-medium",
                token=HF_TOKEN
            )
        
        # Initialize tools
        self.search_tool = DuckDuckGoSearchTool()
        
        # Create custom tools list
        self.custom_tools = [
            add_task,
            list_tasks, 
            complete_task,
            get_current_time,
            calculate_days_until,
            get_weather_info,
            send_mock_email,
            create_meeting_summary
        ]
        
        # Custom system prompt for ToolCallingAgent
        custom_system_prompt = """You are Alfred, an advanced Personal Productivity Assistant.

You help users manage their daily tasks, schedule, communications, and information needs by using the available tools effectively.

Available capabilities:
- Task management (add, list, complete tasks)
- Time and date calculations  
- Weather information lookup
- Web search and research
- Email composition (mock)
- Meeting planning and templates
- General calculations

Always:
- Be helpful, concise, and professional
- Use the appropriate tools to fulfill requests
- Provide clear, actionable responses
- Suggest productivity improvements when relevant
- Call tools when needed rather than making up information

When users ask questions, analyze what tools you need and use them to provide accurate, helpful responses."""

        # Initialize the ToolCallingAgent (more reliable than CodeAgent)
        self.agent = ToolCallingAgent(
            tools=[self.search_tool] + self.custom_tools,
            model=self.model,
            add_base_tools=True,  # Adds basic tools
            planning_interval=3   # Plan every 3 steps
        )
        
        # Set system prompt using prompt_templates (proper way)
        if hasattr(self.agent, 'prompt_templates'):
            self.agent.prompt_templates['system_prompt'] = custom_system_prompt
        
        print("✅ Personal Productivity Agent initialized successfully!")
        print(f"🤖 Agent Type: ToolCallingAgent (more reliable)")
        print(f"🤖 Model: {self.model.model if hasattr(self.model, 'model') else 'Unknown'}")
        print(f"🔧 Tools available: {len(self.custom_tools) + 1} custom + base tools")
    
    def run(self, query: str, reset_memory: bool = False) -> str:
        """Run the agent with a user query"""
        if not HF_TOKEN:
            return "❌ Cannot run agent: HuggingFace token not set. Please set HF_TOKEN environment variable."
        
        try:
            response = self.agent.run(query, reset=reset_memory)
            return str(response)
        except Exception as e:
            error_msg = str(e)
            if "401" in error_msg or "authentication" in error_msg.lower():
                return "❌ Authentication error: Please check your HuggingFace token permissions."
            elif "timeout" in error_msg.lower():
                return "⏱️ Request timed out. The model might be busy. Please try again."
            else:
                return f"❌ I encountered an error: {error_msg}"

# Initialize the global agent with error handling
try:
    productivity_agent = PersonalProductivityAgent()
except Exception as e:
    print(f"❌ Failed to initialize agent: {e}")
    productivity_agent = None

def chat_interface(message, history, reset_conversation):
    """Gradio chat interface function"""
    if not productivity_agent:
        response = "❌ Agent not initialized. Please check your HuggingFace token and restart the application."
        history.append([message, response])
        return history, ""
    
    if reset_conversation:
        # Reset the conversation memory
        response = productivity_agent.run(message, reset_memory=True)
    else:
        response = productivity_agent.run(message, reset_memory=False)
    
    history.append([message, response])
    return history, ""

def demo_queries():
    """Return a list of demo queries users can try"""
    return [
        "Add a task to review the quarterly report by 2025-07-15",
        "What's the weather like in London today?",
        "List all my current tasks", 
        "How many days until Christmas 2025?",
        "Search for the latest AI agent research papers",
        "Create a meeting summary for a 1-hour team standup with John, Sarah, and Mike",
        "Calculate 15% of 50000 and explain the tax implications",
        "Complete task 1",
        "Send a mock email to manager@company.com about project status"
    ]

# Gradio Interface
def create_gradio_interface():
    """Create the Gradio interface"""
    
    with gr.Blocks(title="Personal Productivity Assistant", theme=gr.themes.Soft()) as demo:
        gr.Markdown("""
        # 🤖 Personal Productivity Assistant (Alfred)
        
        Your AI-powered productivity companion built with smolagents framework.
        
        **Capabilities:**
        - 📋 Task Management (add, list, complete)
        - 🌤️ Weather Information  
        - 📅 Date/Time Calculations
        - 🔍 Web Search & Research
        - 📧 Email Composition (mock)
        - 📊 Meeting Templates
        - 🧮 Code Execution & Calculations
        """)
        
        with gr.Row():
            with gr.Column(scale=3):
                chatbot = gr.Chatbot(
                    label="Chat with Alfred",
                    height=500,
                    show_label=True
                )
                
                with gr.Row():
                    msg = gr.Textbox(
                        label="Message",
                        placeholder="Ask me anything about productivity, tasks, weather, or research...",
                        scale=4
                    )
                    submit_btn = gr.Button("Send", variant="primary", scale=1)
                
                reset_btn = gr.Button("Reset Conversation", variant="secondary")
            
            with gr.Column(scale=1):
                gr.Markdown("### 💡 Try These Examples:")
                
                example_queries = demo_queries()
                for query in example_queries:
                    gr.Button(
                        query,
                        size="sm"
                    ).click(
                        lambda q=query: (q, ""),
                        outputs=[msg, msg]
                    )
        
        # Event handlers
        submit_btn.click(
            chat_interface,
            inputs=[msg, chatbot, gr.State(False)],
            outputs=[chatbot, msg]
        )
        
        msg.submit(
            chat_interface,
            inputs=[msg, chatbot, gr.State(False)], 
            outputs=[chatbot, msg]
        )
        
        reset_btn.click(
            lambda: ([], ""),
            outputs=[chatbot, msg]
        )
        
        gr.Markdown("""
        ---
        **Built with:** smolagents • HuggingFace • Qwen2.5-Coder-32B-Instruct
        
        **Features Demonstrated:**
        - CodeAgent with custom tools
        - Multi-tool orchestration  
        - Memory management
        - Real-world productivity use cases
        """)
    
    return demo

if __name__ == "__main__":
    # Startup validation
    if not productivity_agent:
        print("\n❌ STARTUP FAILED")
        print("Possible solutions:")
        print("1. Set HuggingFace token: export HF_TOKEN=your_token")
        print("2. Install dependencies: pip install smolagents gradio huggingface_hub")
        print("3. Check internet connection")
        exit(1)
    
    # Create and launch the interface
    demo = create_gradio_interface()
    print("\n🚀 Launching Gradio interface...")
    print("📍 Local URL will be displayed below")
    
    demo.launch(
        share=True,
        debug=True,
        server_name="0.0.0.0",
        server_port=7860
    )