Kanium-PUBLIC/KaniumCogs
5
0
Fork 0
Code Issues 6 Pull Requests Packages Projects 1 Releases Wiki Activity

Compare commits

base: Kanium-PUBLIC:dev
Branches Tags
Kanium-PUBLIC:master
Kanium-PUBLIC:dev
..
compare: Kanium-PUBLIC:master
Branches Tags
Kanium-PUBLIC:dev
Kanium-PUBLIC:master

1 Commits
dev ... master

Author SHA1 Message Date
T-BENZIN
074ff8bbb4 Merge pull request 'Made improvements in AI Agent's tool get_forecast_weather' (#8) from dev into master 
Reviewed-on: #8
Reviewed-by: AllfatherHatt <allfatherhatt@noreply.localhost>
 2025-06-18 17:13:22 +02:00
10 changed files with 33 additions and 515 deletions
Show Stats Expand all files Collapse all files

38
reginaldCog/demo_bot.py
View File

@ -1,38 +0,0 @@
import os
import discord
from discord.ext import commands
from reginaldCog.messenger_clients.services import MessageService
TOKEN = os.getenv('SCREAMING_OPOSSUM') # Your Discord bot token goes here
intents = discord.Intents.default()
intents.message_content = True
bot = commands.Bot(command_prefix='!', intents=intents)
@bot.event
async def on_ready():
print(f'Logged in as {bot.user} (ID: {bot.user.id})')
print('------')
try:
synced = await bot.tree.sync()
print(f'Synced {len(synced)} command(s).')
except Exception as e:
print(f'Failed to sync commands: {e}')
@bot.event
async def on_message(message: discord.Message):
if message.author == bot.user:
return
async with message.channel.typing():
message_service = MessageService(message)
response = await message_service.get_llm_response()
await message.channel.send(response)
print(response)
if __name__ == '__main__':
if TOKEN is None:
raise RuntimeError('Discord token is not set')
bot.run(TOKEN)

0
reginaldCog/llm_clients/__init__.py
View File

192
reginaldCog/llm_clients/llm_client.py
View File

@ -1,192 +0,0 @@
from abc import ABC, abstractmethod
from enum import Enum
from openai import OpenAI
from discord import Message
from reginaldCog.messenger_clients.messenger_client import ClientMessage, DiscordMessageAdapter
class ILLMContent(ABC):
pass
class OpenAIContent(ILLMContent):
def __init__(self):
self._content_items: list[dict[str, str]] = []
@property
def content_items(self) -> list[dict]:
return self._content_items
@content_items.setter
def content_items(self, value: list[dict[str, str]]):
self._content_items = value
class ILLMContentBuilder(ABC):
pass
class OpenAIContentBuilder(ILLMContentBuilder):
def __init__(self, content: OpenAIContent):
self.content = content
def add_output_text(self, text: str):
item = {"type": "output_text", "text": text}
self.content.content_items.append(item)
return self
def add_input_text(self, text: str):
item = {"type": "input_text", "text": text}
self.content.content_items.append(item)
return self
def add_input_image(self, image_url: str):
item = {"type": "input_image", "image_url": image_url}
self.content.content_items.append(item)
return self
def add_from_dict(self, item: dict):
self.content.content_items.append(item)
return self
class ILLMMessage(ABC):
pass
class OpenAIMessage(ILLMMessage):
def __init__(self):
self.content = OpenAIContent()
self.role = ""
@property
def to_dict(self):
return {"role": self.role, "content": self.content.content_items}
class ILLMMessageBuilder(ABC):
pass
class OpenAIMessageBuilder(ILLMMessageBuilder):
def __init__(self, message: OpenAIMessage):
self.message = message
def set_role(self, role: str):
self.message.role = role
return self
def set_content(self, content: OpenAIContent):
self.message.content = content
return self
class ILLMPrompt(ABC):
pass
class OpenAIPrompt(ILLMPrompt):
def __init__(self):
self.messages = []
@property
def to_list(self):
return [i_message.to_dict for i_message in self.messages]
class ILLMPromptBuilder(ABC):
@abstractmethod
def __init__(self, prompt: ILLMPrompt):
pass
@abstractmethod
def add_message(self, message: ILLMMessage):
pass
class OpenAIPromptBuilder(ILLMPromptBuilder):
def __init__(self, prompt: OpenAIPrompt):
self.prompt = prompt
def add_message(self, message: OpenAIMessage):
self.prompt.messages.append(message)
class ILLMClient(ABC):
@abstractmethod
def get_response(self, prompt: ILLMPrompt):
pass
class OpenAIClient(ILLMClient):
content_class = OpenAIContent
content_builder_class = OpenAIContentBuilder
message_class = OpenAIMessage
message_builder_class = OpenAIMessageBuilder
prompt_class = OpenAIPrompt
prompt_builder_class = OpenAIPromptBuilder
def __init__(self):
self.model = 'gpt-4.1-mini'
self.client = OpenAI()
def get_response(self, prompt: OpenAIPrompt):
response_input = {"model": self.model, "input": prompt.to_list}
return self.client.responses.create(**response_input)
class IMessageAdapter(ABC):
@abstractmethod
def to_message(self) -> ILLMMessage:
pass
class OpenAIResponseAdapter(IMessageAdapter):
def __init__(self, response):
self.response = response
self.response_output = response.output[0]
def to_message(self) -> OpenAIMessage:
content = OpenAIContent()
content_builder = OpenAIContentBuilder(content)
message = OpenAIMessage()
message_builder = OpenAIMessageBuilder(message)
message_builder.set_role(self.response_output.role)\
.set_content(content)
for i_content_item in self.response_output.content:
item = i_content_item.to_dict()
content_builder.add_from_dict(item)
return message
class LMMClientType(Enum):
OPENAI = OpenAIClient
class MessengerClientMessageAdapter(IMessageAdapter):
def __init__(self, message: ClientMessage, llm_client: LMMClientType):
self.message = message
self.llm_client = llm_client
def to_message(self) -> ILLMMessage:
content = self.llm_client.value.content_class()
content_builder = self.llm_client.value.content_builder_class(content)
message = self.llm_client.value.message_class()
message_builder = self.llm_client.value.message_builder_class(message)
message_builder.set_role("user")\
.set_content(content)
if self.message:
content_builder.add_input_text(self.message.content)
for i_image_url in self.message.image_urls:
content_builder.add_input_image(i_image_url)
return message
if __name__ == "__main__":
pass

71
reginaldCog/llm_clients/openai_data_models.py
View File

@ -1,71 +0,0 @@
from dataclasses import dataclass, field
from abc import ABC
# region Content classes
@dataclass(frozen=True)
class Content(ABC):
type: str = field(init=False, default='')
@dataclass(frozen=True)
class InputText(Content):
type: str = field(init=False, default='input_text')
text: str
@dataclass(frozen=True)
class OutputText(Content):
type: str = field(init=False, default='output_text')
text: str
@dataclass(frozen=True)
class InputImage(Content):
type: str = field(init=False, default='input_image')
image_url: str | None = field(default=None)
file_id: str | None = field(default=None)
def __post_init__(self):
if self.image_url is None and self.file_id is None:
raise ValueError('Either `image_url` or `file_id` must be provided.')
@dataclass(frozen=True)
class UrlCitation(Content):
type: str = field(init=False, default='url_citation')
# To be done
@dataclass(frozen=True)
class FunctionCall(Content):
type: str = field(init=False, default='function_call')
id: str
call_id: str
name: str
arguments: dict
@dataclass(frozen=True)
class FunctionCallOutput(Content):
type: str = field(init=False, default='function_call_output')
call_id: str
output: str
# endregion Content classes
@dataclass(frozen=True)
class Message:
role: str
content: list[Content]
@dataclass(frozen=True)
class Prompt:
model: str
input: list[Message]
@dataclass(frozen=True)
class Response:
output: list[Message]

0
reginaldCog/memory_clients/__init__.py
View File

0
reginaldCog/messenger_clients/__init__.py
View File

65
reginaldCog/messenger_clients/discord_data_models.py
View File

@ -1,65 +0,0 @@
from dataclasses import dataclass, field
from datetime import datetime
# region dataclasses
@dataclass(frozen=True)
class Attachment:
content_type: str
filename: str
id: int
size: int
url: str
ephemeral: bool = False
@dataclass(frozen=True)
class Channel:
created_at: datetime
id: int
name: str
def __str__(self) -> str:
return self.name
@dataclass(frozen=True)
class Role:
id: int
name: str
def __str__(self) -> str:
return self.name
@dataclass(frozen=True)
class Member:
bot: bool
created_at: datetime
display_name: str # For regular users this is just their global name or their username, but if they have a guild specific nickname then that is returned instead.
global_name: str # The user's global nickname, taking precedence over the username in display.
id: int
joined_at: datetime
mention: str
name: str # The user's username.
nick: str # The guild specific nickname of the user. Takes precedence over the global name.
roles: list[Role] = field(default_factory=list)
def __str__(self) -> str:
return self.name
@dataclass(frozen=True)
class Message:
author: Member
channel: Channel
created_at: datetime
id: int
attachments: list[Attachment] = field(default_factory=list)
channel_mentions: list[Channel] = field(default_factory=list)
content: str = ''
mentions: list[Member] = field(default_factory=list)
if __name__ == '__main__':
pass

77
reginaldCog/messenger_clients/messenger_client.py
View File

@ -1,77 +0,0 @@
from abc import ABC, abstractmethod
from enum import Enum
from discord import Message, Attachment
class IClientMessage(ABC):
@abstractmethod
def __init__(self):
pass
class ClientMessage(IClientMessage):
def __init__(self):
self.author_name: str = ''
self.content: str = ''
self.image_urls: list = []
class IMessageBuilder(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod
def set_content(self, value: str):
pass
@abstractmethod
def set_author_name(self, value: str):
pass
@abstractmethod
def set_image_urls(self, value: list[str]):
pass
class DiscordMessageBuilder(IMessageBuilder):
def __init__(self, message: IClientMessage):
self.message = message
def set_author_name(self, value: str):
self.message.author_name = value
return self
def set_content(self, value: str):
self.message.content = value
return self
def set_image_urls(self, value: list[str]):
self.message.image_urls = value
return self
class IMessageAdapter(ABC):
@abstractmethod
def create_message(self, message: object) -> IClientMessage:
pass
@staticmethod
def validate_image_urls(urls_list: list[Attachment]) -> list[str]:
supported_image_formats = ('image/jpeg', 'image/png', 'image/webp', 'image/gif')
return [i_attachment.url for i_attachment in urls_list if i_attachment.content_type in supported_image_formats]
class DiscordMessageAdapter(IMessageAdapter):
def create_message(self, message: Message) -> IClientMessage:
client_message = ClientMessage()
message_builder = DiscordMessageBuilder(client_message)
urls_list = self.validate_image_urls(message.attachments)
message_builder.set_content(message.content)\
.set_author_name(message.author.name)\
.set_image_urls(urls_list)
return client_message
if __name__ == "__main__":
pass

39
reginaldCog/messenger_clients/services.py
View File

@ -1,39 +0,0 @@
import asyncio
from discord import Message
from reginaldCog.messenger_clients.messenger_client import ClientMessage, DiscordMessageAdapter as MessengerDiscordAdapter
from reginaldCog.llm_clients.llm_client import LMMClientType, MessengerClientMessageAdapter
class MessageService:
def __init__(self, message: Message, llm_client: LMMClientType = LMMClientType.OPENAI):
self.message = message
self.llm_client = llm_client
async def get_llm_response(self) -> str:
# Adapt discord.Message to ClientMessage domain object
client_message: ClientMessage = MessengerDiscordAdapter().create_message(self.message)
# Create prompt and prompt builder for this LLM client
prompt = self.llm_client.value.prompt_class()
prompt_builder = self.llm_client.value.prompt_builder_class(prompt)
# Adapt the messenger client message into LLM message and add it to prompt
llm_message = MessengerClientMessageAdapter(client_message, self.llm_client).to_message()
prompt_builder.add_message(llm_message)
# Call the LLM client; run in executor if method is blocking sync
llm_client_instance = self.llm_client.value()
loop = asyncio.get_running_loop()
# Assuming get_response is blocking - run in executor:
response = await loop.run_in_executor(None, llm_client_instance.get_response, prompt)
# Extract plain text from the response (assuming OpenAIResponseAdapter is present)
from reginaldCog.llm_clients.llm_client import OpenAIResponseAdapter
response_adapter = OpenAIResponseAdapter(response)
message_obj = response_adapter.to_message()
# Concatenate all textual outputs for sending to Discord
texts = [item.get("text", "") for item in message_obj.content.content_items if item.get("type") == "output_text"]
return "\n".join(texts) if texts else "Sorry, no response generated."

66
reginaldCog/reginald.py
View File

@ -16,6 +16,7 @@ from .weather import time_now, get_current_weather, get_weather_forecast
from .tools_description import TOOLS
from .debug_stuff import debug
CALLABLE_FUNCTIONS = {
# Dictionary with functions to call.
# You can use globals()[func_name](**args) instead, but that's too implicit.
@ -70,7 +71,7 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
async def on_message(self, message):
if message.author.bot or not message.guild:
return # Ignore bots and DMs
# ✅ Check if user is blacklisted
if await self.is_blacklisted(message.author):
return # Ignore message if user is explicitly blacklisted
@ -79,6 +80,7 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
if not (await self.is_admin(message) or await self.has_access(message.author)):
return # Ignore message if user has no permissions
guild = message.guild
channel_id = str(message.channel.id)
user_id = str(message.author.id)
@ -113,8 +115,8 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
await message.channel.send(random.choice(["Yes?", "How may I assist?", "You rang?"]))
return
explicit_invocation = True
# ✅ Passive Listening: Check if the message contains relevant keywords
# ✅ Passive Listening: Check if the message contains relevant keywords
elif self.should_reginald_interject(message_content):
prompt = message_content
explicit_invocation = False
@ -122,11 +124,11 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
else:
return # Ignore irrelevant messages
# ✅ Context Handling: Maintain conversation flow
# ✅ Context Handling: Maintain conversation flow
if memory and memory[-1]["user"] == user_name:
prompt = f"Continuation of the discussion:\n{prompt}"
# ✅ Prepare context messages
# ✅ Prepare context messages
formatted_messages = [{"role": "system", "content": self.get_reginald_persona()}]
if user_profile:
@ -143,10 +145,10 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
"content": f"[{summary['timestamp']}] Topics: {', '.join(summary['topics'])}\n{summary['summary']}"
})
formatted_messages += [{"role": "user", "content": f"{entry['user']}: {entry['content']}"} for entry in
memory]
formatted_messages += [{"role": "user", "content": f"{entry['user']}: {entry['content']}"} for entry in memory]
formatted_messages.append({"role": "user", "content": f"{user_name}: {prompt}"})
##################################################
# #
## Generate AI Response, put into response_text ##
@ -164,8 +166,7 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
memory.append({"user": "Reginald", "content": response_text})
if len(memory) > self.short_term_memory_limit:
summary = await self.summarize_memory(message, memory[:int(
self.short_term_memory_limit * self.summary_retention_ratio)])
summary = await self.summarize_memory(message, memory[:int(self.short_term_memory_limit * self.summary_retention_ratio)])
mid_memory.setdefault(channel_id, []).append({
"timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"),
"topics": self.extract_topics_from_summary(summary),
@ -173,20 +174,20 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
})
if len(mid_memory[channel_id]) > self.summary_retention_limit:
mid_memory[channel_id].pop(0)
memory = memory[-(self.short_term_memory_limit - int(
self.short_term_memory_limit * self.summary_retention_ratio)):]
memory = memory[-(self.short_term_memory_limit - int(self.short_term_memory_limit * self.summary_retention_ratio)):]
short_memory[channel_id] = memory
await self.send_split_message(message.channel, response_text)
def should_reginald_interject(self, message_content: str) -> bool:
"""Determines if Reginald should respond to a message based on keywords."""
direct_invocation = {
"reginald,"
}
"reginald,"
}
message_lower = message_content.lower()
return any(message_lower.startswith(invocation) for invocation in direct_invocation)
@debug
@ -226,8 +227,8 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
'content': func_result,
'tool_call_id': tool_call_id,
})
completion_args["messages"] = messages
completion_args["messages"] = messages
# 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)
@ -260,12 +261,11 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
await self.config.guild(ctx.guild).openai_api_key.set(api_key)
await ctx.send("OpenAI API key set successfully.")
@commands.command(name="reginald_set_listening_channel",
help="Set the channel where Reginald listens for messages.")
@commands.command(name="reginald_set_listening_channel", help="Set the channel where Reginald listens for messages.")
@commands.has_permissions(administrator=True)
async def set_listening_channel(self, ctx, channel: discord.TextChannel):
"""Sets the channel where Reginald will listen for passive responses."""
if not channel:
await ctx.send("❌ Invalid channel. Please mention a valid text channel.")
return
@ -273,13 +273,12 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
await self.config.guild(ctx.guild).listening_channel.set(channel.id)
await ctx.send(f"✅ Reginald will now listen only in {channel.mention}.")
@commands.command(name="reginald_get_listening_channel",
help="Check which channel Reginald is currently listening in.")
@commands.command(name="reginald_get_listening_channel", help="Check which channel Reginald is currently listening in.")
@commands.has_permissions(administrator=True)
async def get_listening_channel(self, ctx):
"""Displays the current listening channel."""
channel_id = await self.config.guild(ctx.guild).listening_channel()
if channel_id:
channel = ctx.guild.get_channel(channel_id)
if channel: # ✅ Prevents crash if channel was deleted
@ -289,17 +288,18 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
else:
await ctx.send("❌ No listening channel has been set.")
# async def send_long_message(self, ctx, message, prefix: str = ""):
# """Splits and sends a long message to avoid Discord's 2000-character limit."""
# chunk_size = 1900 # Leave some space for formatting
# if prefix:
# prefix_length = len(prefix)
# chunk_size -= prefix_length
#
# for i in range(0, len(message), chunk_size):
# chunk = message[i:i + chunk_size]
# await ctx.send(f"{prefix}{chunk}")
async def send_long_message(self, ctx, message, prefix: str = ""):
"""Splits and sends a long message to avoid Discord's 2000-character limit."""
chunk_size = 1900 # Leave some space for formatting
if prefix:
prefix_length = len(prefix)
chunk_size -= prefix_length
for i in range(0, len(message), chunk_size):
chunk = message[i:i + chunk_size]
await ctx.send(f"{prefix}{chunk}")
async def send_split_message(self, ctx, content: str, prefix: str = ""):
"""
Sends a long message to Discord while ensuring it does not exceed the 2000-character limit.
@ -351,4 +351,4 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
async def setup(bot):
"""✅ Correct async cog setup for Redbot"""
await bot.add_cog(ReginaldCog(bot))
await bot.add_cog(ReginaldCog(bot))