app.py

from smolagents import CodeAgent,DuckDuckGoSearchTool, HfApiModel,load_tool,tool
import datetime
import requests
import pytz
import yaml
from tools.final_answer import FinalAnswerTool
import json
import time
from requests_sse import EventSource
from Gradio_UI import GradioUI
from dateutil.relativedelta import relativedelta

# Below is an example of a tool that does nothing. Amaze us with your creativity !
@tool
def get_wikipedia_news(topic: str, limit: int = 5, delay: float = 0.0) -> list:
    """A tool that fetches connects to a live Wikipedia edit stream and looks for edits where the title or comment contains the given topic. If there is a match, it saves details like the page title, user, comment, and timestamp.
    Args:
        topic: The word or phrase to search for in Wikipedia edits.
        limit: The number of edits to collect (default: 5).
        delay: Time in seconds to wait between edits (default: 0.0).
    Returns:
        list: A list of edits, each containing:
              - title (str): The page that was edited.
              - user (str): The person who made the edit.
              - comment (str): The edit summary.
              - timestamp (int): The time the edit was made.
    """
    url = 'https://stream.wikimedia.org/v2/stream/recentchange'
    event_count = 0
    edits = []

    with EventSource(url) as stream:
        for event in stream:
            if event.type == 'message':
                try:
                    change = json.loads(event.data)
                except ValueError:
                    continue  

                # Check if the topic is in the title or comment
                if topic.lower() in change["title"].lower() or topic.lower() in change["comment"].lower():
                    edit_data = {
                        "title": change["title"],  # Page that was edited
                        "user": change["user"],  # Editor's username or IP
                        "comment": change["comment"],  # Summary of the edit
                        "timestamp": change["timestamp"]  # When the edit happened
                    }
                    edits.append(edit_data)
                    event_count += 1

                    if delay > 0:
                        time.sleep(delay)  # Wait before processing the next edit

                    if event_count >= limit:
                        break  # Stop when we have collected enough edits

    return edits

@tool
def get_current_time_in_timezone(timezone: str) -> str:
    """A tool that fetches the current local time in a specified timezone.
    Args:
        timezone: A string representing a valid timezone (e.g., 'America/New_York').
    """
    try:
        # Create timezone object
        tz = pytz.timezone(timezone)
        # Get current time in that timezone
        local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S")
        return f"The current local time in {timezone} is: {local_time}"
    except Exception as e:
        return f"Error fetching time for timezone '{timezone}': {str(e)}"


# @tool
# def get_time_difference(time1: str, timezone1: str, time2: str, timezone2: str) -> str:
#     """A tool that calculates the time difference between two timestamps in different time zones.
#     Args:
#         time1: First timestamp in "YYYY-MM-DD HH:MM:SS" format.
#         timezone1: Time zone of the first timestamp (e.g., "America/New_York").
#         time2: Second timestamp in "YYYY-MM-DD HH:MM:SS" format.
#         timezone2: Time zone of the second timestamp (e.g., "Asia/Kolkata").
#     Returns:
#         str: Time difference as a human-readable string (e.g., "0 years, 0 months, 3 days, 4 hours, 2 minutes, 15 seconds").
#     """
#     try:
#         # Convert input strings to timezone-aware datetime objects
#         tz1 = pytz.timezone(timezone1)
#         tz2 = pytz.timezone(timezone2)

#         dt1 = tz1.localize(datetime.datetime.strptime(time1, "%Y-%m-%d %H:%M:%S"))
#         dt2 = tz2.localize(datetime.datetime.strptime(time2, "%Y-%m-%d %H:%M:%S"))

#         # Calculate absolute difference
#         if dt1 > dt2:
#             dt1, dt2 = dt2, dt1  # Ensure dt1 is earlier for proper calculations

#         diff = relativedelta(dt2, dt1)

#         # Convert difference to human-readable format
#         result = []
#         if diff.years: result.append(f"{diff.years} years")
#         if diff.months: result.append(f"{diff.months} months")
#         if diff.days: result.append(f"{diff.days} days")
#         if diff.hours: result.append(f"{diff.hours} hours")
#         if diff.minutes: result.append(f"{diff.minutes} minutes")
#         if diff.seconds: result.append(f"{diff.seconds} seconds")

#         return ", ".join(result) if result else "0 seconds"

#     except Exception as e:
#         return f"Error calculating time difference: {str(e)}"



final_answer = FinalAnswerTool()

# If the agent does not answer, the model is overloaded, please use another model or the following Hugging Face Endpoint that also contains qwen2.5 coder:
# model_id='https://pflgm2locj2t89co.us-east-1.aws.endpoints.huggingface.cloud' 

model = HfApiModel(
max_tokens=2096,
temperature=0.5,
model_id='Qwen/Qwen2.5-Coder-32B-Instruct',# it is possible that this model may be overloaded
custom_role_conversions=None,
)


# Import tool from Hub
image_generation_tool = load_tool("agents-course/text-to-image", trust_remote_code=True)

with open("prompts.yaml", 'r') as stream:
    prompt_templates = yaml.safe_load(stream)
    
agent = CodeAgent(
    model=model,
    tools=[final_answer], ## add your tools here (don't remove final answer)
    max_steps=6,
    verbosity_level=1,
    grammar=None,
    planning_interval=None,
    name=None,
    description=None,
    prompt_templates=prompt_templates
)


GradioUI(agent).launch()