reginaldCog/reginald.py

@@ -5,12 +5,25 @@ import asyncio
 import datetime
 import re
 import traceback
+import json
 from collections import Counter
 from redbot.core import Config, commands
 from openai import OpenAIError
 from .permissions import PermissionsMixin
 from .blacklist import BlacklistMixin
 from .memory import MemoryMixin
+from .weather import time_now, get_current_weather, get_weather_forecast
+from .tools_description import TOOLS
+
+
+CALLABLE_FUNCTIONS = {
+    # Dictionary with functions to call.
+    # You can use globals()[func_name](**args) instead, but that's too implicit.
+    'time_now': time_now,
+    'get_current_weather': get_current_weather,
+    'get_weather_forecast': get_weather_forecast,
+}
+
 
 class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
     def __init__(self, bot):
@@ -170,14 +183,37 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
         model = await self.config.openai_model()
         try:
             client = openai.AsyncClient(api_key=api_key)
-            response = await client.chat.completions.create(
+            completion_args = {
-                model=model,
+                'model': model,
-                messages=messages,
+                'messages': messages,
-                max_tokens=2048,
+                'max_tokens': 2048,
-                temperature=0.7,
+                'temperature': 0.7,
-                presence_penalty=0.5,
+                'presence_penalty': 0.5,
-                frequency_penalty=0.5
+                'frequency_penalty': 0.5,
-            )
+                'tools': TOOLS,
+                'tool_choice': 'auto',
+            }
+            response = await client.chat.completions.create(**completion_args)
+            # Checking for function calls
+            tool_calls = response.choices[0].message.tool_calls
+            if tool_calls:
+                for i_call in tool_calls:
+                    # Calling for necessary functions
+                    func_name = i_call.function.name
+                    func_args = json.loads(i_call.function.arguments)
+                    tool_call_id = i_call.id
+                    # Getting function result and putting it into messages
+                    func_result = CALLABLE_FUNCTIONS[func_name](**func_args)
+                    messages.append({
+                        'role': 'tool',
+                        'content': func_result,
+                        'tool_calls': tool_calls,
+                        'tool_call_id': tool_call_id,
+                    })
+                # Second completion required if functions has been called to interpret the result into user-friendly
+                # chat message.
+                response = await client.chat.completions.create(**completion_args)
+
             response_text = response.choices[0].message.content.strip()
             if response_text.startswith("Reginald:"):
                 response_text = response_text[len("Reginald:"):].strip()

reginaldCog/tools_description.py

@@ -0,0 +1,72 @@
+TOOLS = [
+    {
+        'type': 'function',
+        'function': {
+            'name': 'time_now',
+            'description': 'Get current date and time in UTC timezone.',
+        }
+    },
+    {
+        'type': 'function',
+        'function': {
+            'name': 'get_current_weather',
+            'description': '''
+                Gets current weather for specified location.
+            ''',
+            'parameters': {
+                'type': 'object',
+                'properties': {
+                    'location': {
+                        'type': 'string',
+                        'description': '''
+                            Location in human readable format.
+                            e.g: "Copenhagen", or "Copenhagen, Louisiana, US", if needed specifying.
+                        '''
+                    }
+                },
+                'required': [
+                    'location',
+                ],
+                'additionalProperties': False
+            },
+            'strict': True
+        }
+    },
+    {
+        'type': 'function',
+        'function': {
+            'name': 'get_weather_forecast',
+            'description': '''
+                Forecast weather API method returns, depending upon your price plan level, upto next 14 day weather
+                forecast and weather alert as json. The data is returned as a Forecast Object.
+                Forecast object contains astronomy data, day weather forecast and hourly interval weather information
+                for a given city.
+                With a free weather API subscription, only up to three days of forecast can be requested.
+            ''',
+            'parameters': {
+                'type': 'object',
+                'properties': {
+                    'location': {
+                        'type': 'string',
+                        'description': '''
+                            Location in human readable format.
+                            e.g: "Copenhagen", or "Copenhagen, Louisiana, US", if needed specifying.
+                        '''
+                    },
+                    'dt': {
+                        'type': 'string',
+                        'description': '''
+                            The date up until to request the forecast in YYYY-MM-DD format.
+                            Check the **now** function first if you unsure which date it is.
+                        '''
+                    },
+                },
+                'required': [
+                    'location', 'dt'
+                ],
+                'additionalProperties': False
+            },
+            'strict': True
+        }
+    }
+]

reginaldCog/weather.py

@@ -0,0 +1,59 @@
+from datetime import datetime, timezone
+from os import environ
+import requests
+import json
+
+WEATHER_API_KEY = environ.get('WEATHER_API_KEY')
+URL = 'http://api.weatherapi.com/v1'
+
+
+def time_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def get_current_weather(location: str) -> str:
+    weather = Weather(location=location)
+    return json.dumps(weather.realtime())
+
+
+def get_weather_forecast(location: str, days: int = 14, dt: str = '2025-03-24') -> str:
+    weather = Weather(location=location)
+    return json.dumps(weather.forecast(days=days, dt=dt))
+
+
+class Weather:
+    def __init__(self, location: str):
+        self.__location = location
+
+    @property
+    def location(self) -> str:
+        return self.__location
+
+    @staticmethod
+    def make_request(method: str, params: dict) -> dict:
+        response = requests.get(url=f'{URL}{method}', params=params)
+        return response.json()
+
+    def realtime(self):
+        method = '/current.json'
+        params = {
+            'key': WEATHER_API_KEY,
+            'q': self.location,
+        }
+        return self.make_request(method=method, params=params)
+
+    def forecast(self, days: int = 14, dt: str = '2025-03-24'):
+        method = '/forecast.json'
+        params = {
+            'key': WEATHER_API_KEY,
+            'q': self.location,
+            'days': days,
+            'dt': dt,
+        }
+        return self.make_request(method=method, params=params)
+
+
+if __name__ == '__main__':
+    test_weather = Weather('Aqtobe')
+    result = json.dumps(test_weather.forecast(days=14, dt='2025-03-24'), indent=2)
+    print(result)