WhiteRabbit, the cat scheduler

[jacopopalumbo01]

Description

Following @pieroit suggestions, I added an initial scheduling system. I've thought to make a sort of wrapper of a really interesting scheduling module (https://apscheduler.readthedocs.io/en/3.x/index.html). This module also permits to save scheduled jobs in a DB (in a future it can be a nice feature to have persistence for some types of jobs).

I provide an example of how it is already usable at the actual state: PLUGIN:

from cat.mad_hatter.decorators import hook
from cat.looking_glass.white_rabbit import WhiteRabbit

@hook
def before_cat_sends_message(final_output, cat):    
    sched = WhiteRabbit()
    sched.schedule_chat_message(cat=cat, content="White Rabbit: Oh dear! Oh dear! I shall be too late!", minutes=1)
    return final_output

And we obtain:

Related to issue #493

Type of change

• [ ] Bug fix (non-breaking change which fixes an issue)
• [x] New feature (non-breaking change which adds functionality)
• [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
• [x] This change requires a documentation update

Checklist:

• [x] My code follows the style guidelines of this project
• [x] I have performed a self-review of my own code
• [x] I have commented my code, particularly in hard-to-understand areas

[zAlweNy26]

The idea of the WhiteRabbit class: 🔥 🔥 🔥 ❤️

[valentimarco]

Super!

[pieroit]

Enter the White Rabbit :heart_decoration: :rabbit: Thanks @jacopopalumbo01 great contribution

[pieroit]

PS.: consider having .white_rabbit as attribute for StrayCat So inside a hook or a tool, one can just do: cat.white_rabbit.schedule_chat_message(content="White Rabbit: Oh dear! Oh dear! I shall be too late!", minutes=1)

Let's iterate a little before documenting this.

[zAlweNy26]

Agree with @pieroit . I also would suggest to be able to schedule a generic function, not only the send_ws_message. What do you think?

[lucagobbi]

This is so cool guys! @jacopopalumbo01 kudos to you. IMHO we should surely extend the WhiteRabbit methods to include date tasks, interval tasks and cron tasks. With those you can accept whatver task passed by the user in a plugin an do some cool automation stuff like making the llm summarize web pages every once in a while, update documents on a daily basis...

I think something like this should work:

def schedule_date_task(self, task_func, days=0, hours=0, minutes=0, seconds=0, milliseconds=0, microseconds=0, **kwargs):
        """
        Schedule a task to run at a specific date and time

        Parameters
        ----------
        task_func: function
            The task function to be scheduled.
        days: int
            Days to wait.
        hours: int
            Hours to wait.
        minutes: int
            Minutes to wait.
        seconds: int
            Seconds to wait.
        milliseconds: int
            Milliseconds to wait.
        microseconds: int
            Microseconds to wait.
        """
        schedule = datetime.today() + timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds, milliseconds=milliseconds, microseconds=microseconds)

        self.scheduler.add_job(task_func, 'date', run_date=schedule, kwargs=kwargs)

def schedule_interval_task(self, task_func, days=0, hours=0, minutes=0, seconds=0, start_time=None, **kwargs):
        """
        Schedule a task to run at a specified interval

        Parameters
        ----------
        task_func: function
            The task function to be scheduled.
        days: int
            Interval in days.
        hours: int
            Interval in hours.
        minutes: int
            Interval in minutes.
        seconds: int
            Interval in seconds.
        start_time: datetime, optional
            The start time of the task. If not provided, the task starts immediately.
        """
        if start_time is None:
            start_time = datetime.now()

        self.scheduler.add_job(task_func, 'interval', days=days, hours=hours, minutes=minutes, seconds=seconds,
                               start_date=start_time, kwargs=kwargs)

def schedule_cron_task(self, task_func, year=None, month=None, day=None, week=None, day_of_week=None, hour=None,
                           minute=None, second=None, start_time=None, **kwargs):
        """
        Schedule a task using a cron expression

        Parameters
        ----------
        task_func: function
            The task function to be scheduled.
        year: int or str, optional
            4-digit year.
        month: int or str, optional
            Month (1-12).
        day: int or str, optional
            Day of the month (1-31).
        week: int or str, optional
            ISO week (1-53).
        day_of_week: int or str, optional
            Number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun).
        hour: int or str, optional
            Hour (0-23).
        minute: int or str, optional
            Minute (0-59).
        second: int or str, optional
            Second (0-59).
        start_time: datetime, optional
            The start time of the task. If not provided, the task starts immediately.
        """
        if start_time is None:
            start_time = datetime.now()

        self.scheduler.add_job(
            task_func,
            'cron',
            year=year,
            month=month,
            day=day,
            week=week,
            day_of_week=day_of_week,
            hour=hour,
            minute=minute,
            second=second,
            start_date=start_time,
            kwargs=kwargs
        )

[valentimarco]

Does python have some sort of Date object to incapusulate all the time stuff? IMHO Too much args for each function

[lucagobbi]

Does python have some sort of Date object to incapusulate all the time stuff? IMHO Too much args for each function

You're right, but since they have defaults when you call it you can only specify the one you need. Also when we are talking about date task I think it is pretty straightforward to merge it in a datetime object like the one you compute in the function, for the others idk, especially cronjob i think it has to stay like a cron-like expression

[jacopopalumbo01]

PS.: consider having .white_rabbit as attribute for StrayCat So inside a hook or a tool, one can just do: cat.white_rabbit.schedule_chat_message(content="White Rabbit: Oh dear! Oh dear! I shall be too late!", minutes=1)

Let's iterate a little before documenting this.

Thank you, I will add it in the next pr!

[jacopopalumbo01]

Agree with @pieroit . I also would suggest to be able to schedule a generic function, not only the send_ws_message. What do you think?

Yes, the white rabbit at the actual state is only a start point. Surely it needs more functionalities. Maybe at the next dev meeting we could discuss what to add :)

[jacopopalumbo01]

This is so cool guys! @jacopopalumbo01 kudos to you. IMHO we should surely extend the WhiteRabbit methods to include date tasks, interval tasks and cron tasks. With those you can accept whatver task passed by the user in a plugin an do some cool automation stuff like making the llm summarize web pages every once in a while, update documents on a daily basis...

I think something like this should work:

def schedule_date_task(self, task_func, days=0, hours=0, minutes=0, seconds=0, milliseconds=0, microseconds=0, kwargs=None):
        """
        Schedule a task to run at a specific date and time

        Parameters
        ----------
        task_func: function
            The task function to be scheduled.
        days: int
            Days to wait.
        hours: int
            Hours to wait.
        minutes: int
            Minutes to wait.
        seconds: int
            Seconds to wait.
        milliseconds: int
            Milliseconds to wait.
        microseconds: int
            Microseconds to wait.
        kwargs: dict, optional
            Additional keyword arguments to pass to the task function.
        """
        schedule = datetime.today() + timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds, milliseconds=milliseconds, microseconds=microseconds)

        self.scheduler.add_job(task_func, 'date', run_date=schedule, kwargs=kwargs)

def schedule_interval_task(self, task_func, days=0, hours=0, minutes=0, seconds=0, start_time=None, kwargs=None):
        """
        Schedule a task to run at a specified interval

        Parameters
        ----------
        task_func: function
            The task function to be scheduled.
        days: int
            Interval in days.
        hours: int
            Interval in hours.
        minutes: int
            Interval in minutes.
        seconds: int
            Interval in seconds.
        start_time: datetime, optional
            The start time of the task. If not provided, the task starts immediately.
        kwargs: dict, optional
            Additional keyword arguments to pass to the task function.
        """
        if start_time is None:
            start_time = datetime.now()

        self.scheduler.add_job(task_func, 'interval', days=days, hours=hours, minutes=minutes, seconds=seconds,
                               start_date=start_time, kwargs=kwargs)

def schedule_cron_task(self, task_func, year=None, month=None, day=None, week=None, day_of_week=None, hour=None,
                           minute=None, second=None, start_time=None, kwargs=None):
        """
        Schedule a task using a cron expression

        Parameters
        ----------
        task_func: function
            The task function to be scheduled.
        year: int or str, optional
            4-digit year.
        month: int or str, optional
            Month (1-12).
        day: int or str, optional
            Day of the month (1-31).
        week: int or str, optional
            ISO week (1-53).
        day_of_week: int or str, optional
            Number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun).
        hour: int or str, optional
            Hour (0-23).
        minute: int or str, optional
            Minute (0-59).
        second: int or str, optional
            Second (0-59).
        start_time: datetime, optional
            The start time of the task. If not provided, the task starts immediately.
        kwargs: dict, optional
            Additional keyword arguments to pass to the task function.
        """
        if start_time is None:
            start_time = datetime.now()

        self.scheduler.add_job(
            task_func,
            'cron',
            year=year,
            month=month,
            day=day,
            week=week,
            day_of_week=day_of_week,
            hour=hour,
            minute=minute,
            second=second,
            start_date=start_time,
            kwargs=kwargs
        )

Definitely yes! Another feature that could be a nice to have is having automatic scheduled tools. A plugin dev could develop a tool as usual and the agent handles the tool as follows: (Example with the "Make me a coffe tool", without any change to the plugin implementation):

• "Human: Make me a coffe in 5 mins"
• The agent recognizes the tool and time scheduling. It adds the tool as a job executing in five minutes
• After 5 minutes the coffe machine starts