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

Updating feedback

Browse Source
This commit is contained in:
AllfatherHatt 2026-03-16 13:20:03 +01:00
parent 83dcdd9277
commit 18137c054b
1 changed files with 45 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

81
reginaldCog/reginald.py
View File

@ -25,7 +25,6 @@ CALLABLE_FUNCTIONS = {
DEFAULT_MODEL = "gpt-5-mini-2025-08-07"
DEFAULT_MAX_COMPLETION_TOKENS = 2000
TYPING_HEARTBEAT_INTERVAL_SECONDS = 8
STATUS_UPDATE_MIN_INTERVAL_SECONDS = 1.5
@ -100,12 +99,6 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
}
return tool_statuses.get(tool_name, "Reginald is consulting an external source...")
async def typing_heartbeat(self, channel: discord.TextChannel):
while True:
with suppress(discord.HTTPException):
await channel.trigger_typing()
await asyncio.sleep(TYPING_HEARTBEAT_INTERVAL_SECONDS)
def make_status_updater(
self, status_message: discord.Message
) -> Callable[[str, bool], Awaitable[None]]:
@ -215,46 +208,62 @@ class ReginaldCog(PermissionsMixin, BlacklistMixin, MemoryMixin, commands.Cog):
status_message = await message.channel.send(self.get_thinking_status_message())
status_update = self.make_status_updater(status_message)
typing_task = asyncio.create_task(self.typing_heartbeat(message.channel))
try:
response_text = None
if hasattr(message.channel, "typing"):
try:
async with message.channel.typing():
response_text = await self.generate_response(
api_key,
formatted_messages,
status_update=status_update,
)
except (discord.HTTPException, AttributeError):
# Fall back to normal processing if typing indicator isn't available.
response_text = await self.generate_response(
api_key,
formatted_messages,
status_update=status_update,
)
else:
response_text = await self.generate_response(
api_key,
formatted_messages,
status_update=status_update,
)
try:
await self.send_split_message(message.channel, response_text)
finally:
typing_task.cancel()
with suppress(asyncio.CancelledError):
await typing_task
if status_message is not None:
with suppress(discord.HTTPException):
await status_message.delete()
memory.append({"user": user_name, "content": prompt})
memory.append({"user": "Reginald", "content": response_text})
try:
memory.append({"user": user_name, "content": prompt})
memory.append({"user": "Reginald", "content": response_text})
if len(memory) > self.short_term_memory_limit:
summary_batch_size = int(self.short_term_memory_limit * self.summary_retention_ratio)
summary = await self.summarize_memory(message, memory[:summary_batch_size])
if len(memory) > self.short_term_memory_limit:
summary_batch_size = int(self.short_term_memory_limit * self.summary_retention_ratio)
summary = await self.summarize_memory(message, memory[:summary_batch_size])
mid_term_summaries.append(
{
"timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"),
"topics": self.extract_topics_from_summary(summary),
"summary": summary,
}
)
mid_term_summaries.append(
{
"timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M"),
"topics": self.extract_topics_from_summary(summary),
"summary": summary,
}
)
retained_count = max(1, self.short_term_memory_limit - summary_batch_size)
memory = memory[-retained_count:]
retained_count = max(1, self.short_term_memory_limit - summary_batch_size)
memory = memory[-retained_count:]
async with self.config.guild(guild).short_term_memory() as short_memory, self.config.guild(
guild
).mid_term_memory() as mid_memory:
short_memory[channel_id] = memory
mid_memory[channel_id] = mid_term_summaries[-self.summary_retention_limit :]
await self.send_split_message(message.channel, response_text)
if status_message is not None:
with suppress(discord.HTTPException):
await status_message.delete()
async with self.config.guild(guild).short_term_memory() as short_memory, self.config.guild(
guild
).mid_term_memory() as mid_memory:
short_memory[channel_id] = memory
mid_memory[channel_id] = mid_term_summaries[-self.summary_retention_limit :]
except Exception as error:
print(f"DEBUG: Memory persistence failed after response delivery: {error}")
def should_reginald_interject(self, message_content: str) -> bool:
direct_invocation = {"reginald,"}