first attempt at separating all memory into its own module separate from core

This commit is contained in:
AllfatherHatt 2025-03-15 17:50:43 +01:00
parent 2f78408c77
commit 4bd380a8a7
2 changed files with 329 additions and 308 deletions

325
reginaldCog/memory.py Normal file
View File

@ -0,0 +1,325 @@
import re
import random
import datetime
import discord
import openai
from collections import Counter
from redbot.core import commands, Config
from openai import OpenAIError
class MemoryMixin:
"""Handles all memory-related functions for Reginald."""
def __init__(self, config: Config):
self.config = config
self.short_term_memory_limit = 100
self.summary_retention_limit = 25
self.summary_retention_ratio = 0.8
@commands.command(name="reginald_clear_short", help="Clears short-term memory for this channel.")
@commands.has_permissions(administrator=True)
async def clear_short_memory(self, ctx):
"""Clears short-term memory for this channel."""
async with self.config.guild(ctx.guild).short_term_memory() as short_memory:
short_memory[ctx.channel.id] = []
await ctx.send("Short-term memory for this channel has been cleared.")
@commands.command(name="reginald_set_limit", help="Set the short-term memory message limit.")
@commands.has_permissions(administrator=True)
async def set_short_term_memory_limit(self, ctx, limit: int):
"""Allows an admin to change the short-term memory limit dynamically."""
if limit < 5:
await ctx.send("⚠️ The short-term memory limit must be at least 5.")
return
self.short_term_memory_limit = limit
await ctx.send(f"✅ Short-term memory limit set to {limit} messages.")
@commands.command(name="reginald_memory_limit", help="Displays the current short-term memory message limit.")
async def get_short_term_memory_limit(self, ctx):
"""Displays the current short-term memory limit."""
await ctx.send(f"📏 **Current Short-Term Memory Limit:** {self.short_term_memory_limit} messages.")
@commands.command(name="reginald_clear_mid", help="Clears mid-term memory (summarized logs).")
@commands.has_permissions(administrator=True)
async def clear_mid_memory(self, ctx):
async with self.config.guild(ctx.guild).mid_term_memory() as mid_memory:
mid_memory[ctx.channel.id] = ""
await ctx.send("Mid-term memory for this channel has been cleared.")
@commands.command(name="reginald_summary", help="Displays a selected mid-term summary for this channel.")
async def get_mid_term_summary(self, ctx, index: int):
"""Fetch and display a specific mid-term memory summary by index."""
async with self.config.guild(ctx.guild).mid_term_memory() as mid_memory:
summaries = mid_memory.get(str(ctx.channel.id), [])
# Check if there are summaries
if not summaries:
await ctx.send("⚠️ No summaries available for this channel.")
return
# Validate index (1-based for user-friendliness)
if index < 1 or index > len(summaries):
await ctx.send(f"⚠️ Invalid index. Please provide a number between **1** and **{len(summaries)}**.")
return
# Fetch the selected summary
selected_summary = summaries[index - 1] # Convert to 0-based index
# Format output correctly
formatted_summary = (
f"📜 **Summary {index} of {len(summaries)}**\n"
f"📅 **Date:** {selected_summary['timestamp']}\n"
f"🔍 **Topics:** {', '.join(selected_summary['topics']) or 'None'}\n"
f"📝 **Summary:**\n\n"
f"{selected_summary['summary']}"
)
await self.send_long_message(ctx, formatted_summary)
@commands.command(name="reginald_summaries", help="Lists available summaries for this channel.")
async def list_mid_term_summaries(self, ctx):
"""Displays a brief list of all available mid-term memory summaries."""
async with self.config.guild(ctx.guild).mid_term_memory() as mid_memory:
summaries = mid_memory.get(str(ctx.channel.id), [])
if not summaries:
await ctx.send("⚠️ No summaries available for this channel.")
return
summary_list = "\n".join(
f"**{i+1}.** 📅 {entry['timestamp']} | 🔍 Topics: {', '.join(entry['topics']) or 'None'}"
for i, entry in enumerate(summaries)
)
await ctx.send(f"📚 **Available Summaries:**\n{summary_list[:2000]}")
@commands.command(name="reginald_clear_long", help="Clears all long-term stored knowledge.")
@commands.has_permissions(administrator=True)
async def clear_long_memory(self, ctx):
async with self.config.guild(ctx.guild).long_term_profiles() as long_memory:
long_memory.clear()
await ctx.send("All long-term memory has been erased.")
@commands.command(name="reginald_reset_all", help="Completely resets all memory.")
@commands.has_permissions(administrator=True)
async def reset_all_memory(self, ctx):
async with self.config.guild(ctx.guild).short_term_memory() as short_memory:
short_memory.clear()
async with self.config.guild(ctx.guild).mid_term_memory() as mid_memory:
mid_memory.clear()
async with self.config.guild(ctx.guild).long_term_profiles() as long_memory:
long_memory.clear()
await ctx.send("All memory has been completely reset.")
@commands.command(name="reginald_recall", help="Recalls what Reginald knows about a user.")
async def recall_user(self, ctx, user: discord.User):
async with self.config.guild(ctx.guild).long_term_profiles() as long_memory:
profile = long_memory.get(str(user.id), {}).get("summary", "No stored information on this user.")
await ctx.send(f"📜 **Memory Recall for {user.display_name}:** {profile}")
@commands.command(name="reginald_forget", help="Forgets a specific user's long-term profile.")
@commands.has_permissions(administrator=True)
async def forget_user(self, ctx, user: discord.User):
async with self.config.guild(ctx.guild).long_term_profiles() as long_memory:
if str(user.id) in long_memory:
del long_memory[str(user.id)]
await ctx.send(f"Reginald has forgotten all stored information about {user.display_name}.")
else:
await ctx.send(f"No stored knowledge about {user.display_name} to delete.")
async def summarize_memory(self, ctx, messages):
"""✅ Generates a structured, compact summary of past conversations for mid-term storage."""
summary_prompt = (
"Summarize the following conversation into a structured, concise format that retains key details while maximizing brevity. "
"The summary should be **organized** into clear sections: "
"\n\n📌 **Key Takeaways:** Important facts or conclusions reached."
"\n🔹 **Disputed Points:** Areas where opinions or facts conflicted."
"\n🗣️ **Notable User Contributions:** Key statements from users that shaped the discussion."
"\n📜 **Additional Context:** Any other relevant information."
"\n\nEnsure the summary is **dense but not overly verbose**. Avoid unnecessary repetition while keeping essential meaning intact."
)
summary_text = "\n".join(f"{msg['user']}: {msg['content']}" for msg in messages)
try:
api_key = await self.config.guild(ctx.guild).openai_api_key()
if not api_key:
print("🛠️ DEBUG: No API key found for summarization.")
return (
"It appears that I have not been furnished with the necessary credentials to carry out this task. "
"Might I suggest consulting an administrator to rectify this unfortunate oversight?"
)
client = openai.AsyncClient(api_key=api_key)
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": summary_prompt},
{"role": "user", "content": summary_text}
],
max_tokens=2048
)
summary_content = response.choices[0].message.content.strip()
if not summary_content:
print("🛠️ DEBUG: Empty summary received from OpenAI.")
return (
"Ah, an unusual predicament indeed! It seems that my attempt at summarization has resulted in "
"a void of information. I shall endeavor to be more verbose next time."
)
return summary_content
except OpenAIError as e:
error_message = f"OpenAI Error: {e}"
print(f"🛠️ DEBUG: {error_message}") # Log error to console
reginald_responses = [
f"Regrettably, I must inform you that I have encountered a bureaucratic obstruction whilst attempting to summarize:\n\n{error_message}",
f"It would seem that a most unfortunate technical hiccup has befallen my faculties in the matter of summarization:\n\n{error_message}",
f"Ah, it appears I have received an urgent memorandum stating that my summarization efforts have been thwarted:\n\n{error_message}",
f"I regret to inform you that my usual eloquence is presently obstructed by an unforeseen complication while summarizing:\n\n{error_message}"
]
return random.choice(reginald_responses)
def extract_topics_from_summary(self, summary):
"""Dynamically extracts the most important topics from a summary."""
# 🔹 Extract all words from summary
keywords = re.findall(r"\b\w+\b", summary.lower())
# 🔹 Count word occurrences
word_counts = Counter(keywords)
# 🔹 Remove unimportant words (common filler words)
stop_words = {"the", "and", "of", "in", "to", "is", "on", "for", "with", "at", "by", "it", "this", "that", "his", "her"}
filtered_words = {word: count for word, count in word_counts.items() if word not in stop_words and len(word) > 2}
# 🔹 Take the 5 most frequently used words as "topics"
topics = sorted(filtered_words, key=filtered_words.get, reverse=True)[:5]
return topics
def select_relevant_summaries(self, summaries, prompt):
"""Selects the most relevant summaries based on topic matching, frequency, and recency weighting."""
max_summaries = 5 if len(prompt) > 50 else 3 # Use more summaries if the prompt is long
current_time = datetime.datetime.now()
def calculate_weight(summary):
"""Calculate a weighted score for a summary based on relevance, recency, and frequency."""
topic_match = sum(1 for topic in summary["topics"] if topic in prompt.lower()) # Context match score
frequency_score = len(summary["topics"]) # More topics = likely more important
timestamp = datetime.datetime.strptime(summary["timestamp"], "%Y-%m-%d %H:%M")
recency_factor = max(0.1, 1 - ((current_time - timestamp).days / 365)) # Older = lower weight
return (topic_match * 2) + (frequency_score * 1.5) + (recency_factor * 3)
# Apply the weighting function and sort by highest weight
weighted_summaries = sorted(summaries, key=calculate_weight, reverse=True)
return weighted_summaries[:max_summaries] # Return the top-scoring summaries
def extract_fact_from_response(self, response_text):
"""
Extracts potential long-term knowledge from Reginald's response.
This filters out generic responses and focuses on statements about user preferences, traits, and history.
"""
# Define patterns that suggest factual knowledge (adjust as needed)
fact_patterns = [
r"I recall that you (.*?)\.", # "I recall that you like chess."
r"You once mentioned that you (.*?)\.", # "You once mentioned that you enjoy strategy games."
r"Ah, you previously stated that (.*?)\.", # "Ah, you previously stated that you prefer tea over coffee."
r"As I remember, you (.*?)\.", # "As I remember, you studied engineering."
r"I believe you (.*?)\.", # "I believe you enjoy historical fiction."
r"I seem to recall that you (.*?)\.", # "I seem to recall that you work in software development."
r"You have indicated in the past that you (.*?)\.", # "You have indicated in the past that you prefer single-malt whisky."
r"From what I remember, you (.*?)\.", # "From what I remember, you dislike overly sweet desserts."
r"You previously mentioned that (.*?)\.", # "You previously mentioned that you train in martial arts."
r"It is my understanding that you (.*?)\.", # "It is my understanding that you have a preference for Linux systems."
r"If I am not mistaken, you (.*?)\.", # "If I am not mistaken, you studied philosophy."
]
for pattern in fact_patterns:
match = re.search(pattern, response_text, re.IGNORECASE)
if match:
return match.group(1) # Extract the meaningful fact
return None # No strong fact found
@commands.command(name="reginald_memory_status", help="Displays a memory usage summary.")
async def memory_status(self, ctx):
async with self.config.guild(ctx.guild).short_term_memory() as short_memory, \
self.config.guild(ctx.guild).mid_term_memory() as mid_memory, \
self.config.guild(ctx.guild).long_term_profiles() as long_memory:
short_count = sum(len(v) for v in short_memory.values())
mid_count = sum(len(v) for v in mid_memory.values())
long_count = len(long_memory)
status_message = (
f"📊 **Memory Status:**\n"
f"- **Short-Term Messages Stored:** {short_count}\n"
f"- **Mid-Term Summaries Stored:** {mid_count}\n"
f"- **Long-Term Profiles Stored:** {long_count}\n"
)
await ctx.send(status_message)
def normalize_fact(self, fact: str) -> str: # ✅ Now it's a proper method
"""Cleans up facts for better duplicate detection."""
return re.sub(r"\s+", " ", fact.strip().lower()) # Removes excess spaces)
async def update_long_term_memory(self, ctx, user_id: str, fact: str, source_message: str, timestamp: str):
"""Ensures long-term memory updates are structured, preventing overwrites and tracking historical changes."""
fact = self.normalize_fact(fact) # ✅ Normalize before comparison
async with self.config.guild(ctx.guild).long_term_profiles() as long_memory:
if user_id not in long_memory:
long_memory[user_id] = {"facts": []}
user_facts = long_memory[user_id]["facts"]
for entry in user_facts:
if self.normalize_fact(entry["fact"]) == fact:
entry["last_updated"] = timestamp
return
# Check for conflicting facts (same topic but different details)
conflicting_entry = None
for entry in user_facts:
existing_keywords = set(entry["fact"].lower().split())
new_keywords = set(fact.lower().split())
# If there's significant overlap in keywords, assume it's a conflicting update
if len(existing_keywords & new_keywords) >= 2:
conflicting_entry = entry
break
if "previous_versions" not in conflicting_entry:
# ✅ If contradiction found, archive the previous version
conflicting_entry["previous_versions"].append({
"fact": conflicting_entry["fact"],
"source": conflicting_entry["source"],
"timestamp": conflicting_entry["timestamp"]
})
conflicting_entry["fact"] = fact # Store the latest fact
conflicting_entry["source"] = source_message
conflicting_entry["timestamp"] = timestamp
conflicting_entry["last_updated"] = timestamp
else:
# ✅ Otherwise, add it as a new fact
user_facts.append({
"fact": fact,
"source": source_message,
"timestamp": timestamp,
"last_updated": timestamp,
"previous_versions": []
})

View File

@ -10,16 +10,14 @@ from redbot.core import Config, commands
from openai import OpenAIError from openai import OpenAIError
from .permissions import PermissionsMixin from .permissions import PermissionsMixin
from .blacklist import BlacklistMixin from .blacklist import BlacklistMixin
from .memory import MemoryMixin
class ReginaldCog(commands.Cog, PermissionsMixin, BlacklistMixin): class ReginaldCog(commands.Cog, PermissionsMixin, BlacklistMixin, MemoryMixin):
def __init__(self, bot): def __init__(self, bot):
self.bot = bot self.bot = bot
self.config = Config.get_conf(self, identifier=71717171171717) self.config = Config.get_conf(self, identifier=71717171171717)
self.default_listening_channel = 1085649787388428370 # self.default_listening_channel = 1085649787388428370 #
self.memory_locks = {} # ✅ Prevents race conditions per channel self.memory_locks = {} # ✅ Prevents race conditions per channel
self.short_term_memory_limit = 100 # ✅ Now retains 100 messages
self.summary_retention_limit = 25 # ✅ Now retains 25 summaries
self.summary_retention_ratio = 0.8 # ✅ 80% summarization, 20% retention
# ✅ Properly Registered Configuration Keys # ✅ Properly Registered Configuration Keys
default_global = {"openai_model": "gpt-4o-mini"} default_global = {"openai_model": "gpt-4o-mini"}
@ -35,6 +33,8 @@ class ReginaldCog(commands.Cog, PermissionsMixin, BlacklistMixin):
} }
self.config.register_global(**default_global) self.config.register_global(**default_global)
self.config.register_guild(**default_guild) self.config.register_guild(**default_guild)
MemoryMixin.__init__(self, self.config)
async def is_admin(self, ctx): async def is_admin(self, ctx):
admin_role_id = await self.config.guild(ctx.guild).admin_role() admin_role_id = await self.config.guild(ctx.guild).admin_role()
@ -165,129 +165,6 @@ class ReginaldCog(commands.Cog, PermissionsMixin, BlacklistMixin):
return any(message_lower.startswith(invocation) for invocation in direct_invocation) return any(message_lower.startswith(invocation) for invocation in direct_invocation)
async def summarize_memory(self, ctx, messages):
"""✅ Generates a structured, compact summary of past conversations for mid-term storage."""
summary_prompt = (
"Summarize the following conversation into a structured, concise format that retains key details while maximizing brevity. "
"The summary should be **organized** into clear sections: "
"\n\n📌 **Key Takeaways:** Important facts or conclusions reached."
"\n🔹 **Disputed Points:** Areas where opinions or facts conflicted."
"\n🗣️ **Notable User Contributions:** Key statements from users that shaped the discussion."
"\n📜 **Additional Context:** Any other relevant information."
"\n\nEnsure the summary is **dense but not overly verbose**. Avoid unnecessary repetition while keeping essential meaning intact."
)
summary_text = "\n".join(f"{msg['user']}: {msg['content']}" for msg in messages)
try:
api_key = await self.config.guild(ctx.guild).openai_api_key()
if not api_key:
print("🛠️ DEBUG: No API key found for summarization.")
return (
"It appears that I have not been furnished with the necessary credentials to carry out this task. "
"Might I suggest consulting an administrator to rectify this unfortunate oversight?"
)
client = openai.AsyncClient(api_key=api_key)
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": summary_prompt},
{"role": "user", "content": summary_text}
],
max_tokens=2048
)
summary_content = response.choices[0].message.content.strip()
if not summary_content:
print("🛠️ DEBUG: Empty summary received from OpenAI.")
return (
"Ah, an unusual predicament indeed! It seems that my attempt at summarization has resulted in "
"a void of information. I shall endeavor to be more verbose next time."
)
return summary_content
except OpenAIError as e:
error_message = f"OpenAI Error: {e}"
print(f"🛠️ DEBUG: {error_message}") # Log error to console
reginald_responses = [
f"Regrettably, I must inform you that I have encountered a bureaucratic obstruction whilst attempting to summarize:\n\n{error_message}",
f"It would seem that a most unfortunate technical hiccup has befallen my faculties in the matter of summarization:\n\n{error_message}",
f"Ah, it appears I have received an urgent memorandum stating that my summarization efforts have been thwarted:\n\n{error_message}",
f"I regret to inform you that my usual eloquence is presently obstructed by an unforeseen complication while summarizing:\n\n{error_message}"
]
return random.choice(reginald_responses)
def extract_topics_from_summary(self, summary):
"""Dynamically extracts the most important topics from a summary."""
# 🔹 Extract all words from summary
keywords = re.findall(r"\b\w+\b", summary.lower())
# 🔹 Count word occurrences
word_counts = Counter(keywords)
# 🔹 Remove unimportant words (common filler words)
stop_words = {"the", "and", "of", "in", "to", "is", "on", "for", "with", "at", "by", "it", "this", "that", "his", "her"}
filtered_words = {word: count for word, count in word_counts.items() if word not in stop_words and len(word) > 2}
# 🔹 Take the 5 most frequently used words as "topics"
topics = sorted(filtered_words, key=filtered_words.get, reverse=True)[:5]
return topics
def select_relevant_summaries(self, summaries, prompt):
"""Selects the most relevant summaries based on topic matching, frequency, and recency weighting."""
max_summaries = 5 if len(prompt) > 50 else 3 # Use more summaries if the prompt is long
current_time = datetime.datetime.now()
def calculate_weight(summary):
"""Calculate a weighted score for a summary based on relevance, recency, and frequency."""
topic_match = sum(1 for topic in summary["topics"] if topic in prompt.lower()) # Context match score
frequency_score = len(summary["topics"]) # More topics = likely more important
timestamp = datetime.datetime.strptime(summary["timestamp"], "%Y-%m-%d %H:%M")
recency_factor = max(0.1, 1 - ((current_time - timestamp).days / 365)) # Older = lower weight
return (topic_match * 2) + (frequency_score * 1.5) + (recency_factor * 3)
# Apply the weighting function and sort by highest weight
weighted_summaries = sorted(summaries, key=calculate_weight, reverse=True)
return weighted_summaries[:max_summaries] # Return the top-scoring summaries
def extract_fact_from_response(self, response_text):
"""
Extracts potential long-term knowledge from Reginald's response.
This filters out generic responses and focuses on statements about user preferences, traits, and history.
"""
# Define patterns that suggest factual knowledge (adjust as needed)
fact_patterns = [
r"I recall that you (.*?)\.", # "I recall that you like chess."
r"You once mentioned that you (.*?)\.", # "You once mentioned that you enjoy strategy games."
r"Ah, you previously stated that (.*?)\.", # "Ah, you previously stated that you prefer tea over coffee."
r"As I remember, you (.*?)\.", # "As I remember, you studied engineering."
r"I believe you (.*?)\.", # "I believe you enjoy historical fiction."
r"I seem to recall that you (.*?)\.", # "I seem to recall that you work in software development."
r"You have indicated in the past that you (.*?)\.", # "You have indicated in the past that you prefer single-malt whisky."
r"From what I remember, you (.*?)\.", # "From what I remember, you dislike overly sweet desserts."
r"You previously mentioned that (.*?)\.", # "You previously mentioned that you train in martial arts."
r"It is my understanding that you (.*?)\.", # "It is my understanding that you have a preference for Linux systems."
r"If I am not mistaken, you (.*?)\.", # "If I am not mistaken, you studied philosophy."
]
for pattern in fact_patterns:
match = re.search(pattern, response_text, re.IGNORECASE)
if match:
return match.group(1) # Extract the meaningful fact
return None # No strong fact found
async def generate_response(self, api_key, messages): async def generate_response(self, api_key, messages):
model = await self.config.openai_model() model = await self.config.openai_model()
try: try:
@ -315,123 +192,6 @@ class ReginaldCog(commands.Cog, PermissionsMixin, BlacklistMixin):
] ]
return random.choice(reginald_responses) return random.choice(reginald_responses)
@commands.command(name="reginald_clear_short", help="Clears short-term memory for this channel.")
@commands.has_permissions(administrator=True)
async def clear_short_memory(self, ctx):
async with self.config.guild(ctx.guild).short_term_memory() as short_memory:
short_memory[ctx.channel.id] = []
await ctx.send("Short-term memory for this channel has been cleared.")
@commands.command(name="reginald_clear_mid", help="Clears mid-term memory (summarized logs).")
@commands.has_permissions(administrator=True)
async def clear_mid_memory(self, ctx):
async with self.config.guild(ctx.guild).mid_term_memory() as mid_memory:
mid_memory[ctx.channel.id] = ""
await ctx.send("Mid-term memory for this channel has been cleared.")
@commands.command(name="reginald_clear_long", help="Clears all long-term stored knowledge.")
@commands.has_permissions(administrator=True)
async def clear_long_memory(self, ctx):
async with self.config.guild(ctx.guild).long_term_profiles() as long_memory:
long_memory.clear()
await ctx.send("All long-term memory has been erased.")
@commands.command(name="reginald_reset_all", help="Completely resets all memory.")
@commands.has_permissions(administrator=True)
async def reset_all_memory(self, ctx):
async with self.config.guild(ctx.guild).short_term_memory() as short_memory:
short_memory.clear()
async with self.config.guild(ctx.guild).mid_term_memory() as mid_memory:
mid_memory.clear()
async with self.config.guild(ctx.guild).long_term_profiles() as long_memory:
long_memory.clear()
await ctx.send("All memory has been completely reset.")
@commands.command(name="reginald_memory_status", help="Displays a memory usage summary.")
async def memory_status(self, ctx):
async with self.config.guild(ctx.guild).short_term_memory() as short_memory, \
self.config.guild(ctx.guild).mid_term_memory() as mid_memory, \
self.config.guild(ctx.guild).long_term_profiles() as long_memory:
short_count = sum(len(v) for v in short_memory.values())
mid_count = sum(len(v) for v in mid_memory.values())
long_count = len(long_memory)
status_message = (
f"📊 **Memory Status:**\n"
f"- **Short-Term Messages Stored:** {short_count}\n"
f"- **Mid-Term Summaries Stored:** {mid_count}\n"
f"- **Long-Term Profiles Stored:** {long_count}\n"
)
await ctx.send(status_message)
def normalize_fact(self, fact: str) -> str: # ✅ Now it's a proper method
"""Cleans up facts for better duplicate detection."""
return re.sub(r"\s+", " ", fact.strip().lower()) # Removes excess spaces)
async def update_long_term_memory(self, ctx, user_id: str, fact: str, source_message: str, timestamp: str):
"""Ensures long-term memory updates are structured, preventing overwrites and tracking historical changes."""
fact = self.normalize_fact(fact) # ✅ Normalize before comparison
async with self.config.guild(ctx.guild).long_term_profiles() as long_memory:
if user_id not in long_memory:
long_memory[user_id] = {"facts": []}
user_facts = long_memory[user_id]["facts"]
for entry in user_facts:
if self.normalize_fact(entry["fact"]) == fact:
entry["last_updated"] = timestamp
return
# Check for conflicting facts (same topic but different details)
conflicting_entry = None
for entry in user_facts:
existing_keywords = set(entry["fact"].lower().split())
new_keywords = set(fact.lower().split())
# If there's significant overlap in keywords, assume it's a conflicting update
if len(existing_keywords & new_keywords) >= 2:
conflicting_entry = entry
break
if "previous_versions" not in conflicting_entry:
# ✅ If contradiction found, archive the previous version
conflicting_entry["previous_versions"].append({
"fact": conflicting_entry["fact"],
"source": conflicting_entry["source"],
"timestamp": conflicting_entry["timestamp"]
})
conflicting_entry["fact"] = fact # Store the latest fact
conflicting_entry["source"] = source_message
conflicting_entry["timestamp"] = timestamp
conflicting_entry["last_updated"] = timestamp
else:
# ✅ Otherwise, add it as a new fact
user_facts.append({
"fact": fact,
"source": source_message,
"timestamp": timestamp,
"last_updated": timestamp,
"previous_versions": []
})
@commands.command(name="reginald_recall", help="Recalls what Reginald knows about a user.")
async def recall_user(self, ctx, user: discord.User):
async with self.config.guild(ctx.guild).long_term_profiles() as long_memory:
profile = long_memory.get(str(user.id), {}).get("summary", "No stored information on this user.")
await ctx.send(f"📜 **Memory Recall for {user.display_name}:** {profile}")
@commands.command(name="reginald_forget", help="Forgets a specific user's long-term profile.")
@commands.has_permissions(administrator=True)
async def forget_user(self, ctx, user: discord.User):
async with self.config.guild(ctx.guild).long_term_profiles() as long_memory:
if str(user.id) in long_memory:
del long_memory[str(user.id)]
await ctx.send(f"Reginald has forgotten all stored information about {user.display_name}.")
else:
await ctx.send(f"No stored knowledge about {user.display_name} to delete.")
@commands.guild_only() @commands.guild_only()
@commands.has_permissions(manage_guild=True) @commands.has_permissions(manage_guild=True)
@commands.command(help="Set the OpenAI API key") @commands.command(help="Set the OpenAI API key")
@ -439,70 +199,6 @@ class ReginaldCog(commands.Cog, PermissionsMixin, BlacklistMixin):
"""Allows an admin to set the OpenAI API key for Reginald.""" """Allows an admin to set the OpenAI API key for Reginald."""
await self.config.guild(ctx.guild).openai_api_key.set(api_key) await self.config.guild(ctx.guild).openai_api_key.set(api_key)
await ctx.send("OpenAI API key set successfully.") await ctx.send("OpenAI API key set successfully.")
@commands.command(name="reginald_set_limit", help="Set the short-term memory message limit.")
@commands.has_permissions(administrator=True)
async def set_short_term_memory_limit(self, ctx, limit: int):
"""Allows an admin to change the short-term memory limit dynamically."""
if limit < 5:
await ctx.send("⚠️ The short-term memory limit must be at least 5.")
return
self.short_term_memory_limit = limit
await ctx.send(f"✅ Short-term memory limit set to {limit} messages.")
@commands.command(name="reginald_memory_limit", help="Displays the current short-term memory message limit.")
async def get_short_term_memory_limit(self, ctx):
"""Displays the current short-term memory limit."""
await ctx.send(f"📏 **Current Short-Term Memory Limit:** {self.short_term_memory_limit} messages.")
@commands.command(name="reginald_summary", help="Displays a selected mid-term summary for this channel.")
async def get_mid_term_summary(self, ctx, index: int):
"""Fetch and display a specific mid-term memory summary by index."""
async with self.config.guild(ctx.guild).mid_term_memory() as mid_memory:
summaries = mid_memory.get(str(ctx.channel.id), [])
# Check if there are summaries
if not summaries:
await ctx.send("⚠️ No summaries available for this channel.")
return
# Validate index (1-based for user-friendliness)
if index < 1 or index > len(summaries):
await ctx.send(f"⚠️ Invalid index. Please provide a number between **1** and **{len(summaries)}**.")
return
# Fetch the selected summary
selected_summary = summaries[index - 1] # Convert to 0-based index
# Format output correctly
formatted_summary = (
f"📜 **Summary {index} of {len(summaries)}**\n"
f"📅 **Date:** {selected_summary['timestamp']}\n"
f"🔍 **Topics:** {', '.join(selected_summary['topics']) or 'None'}\n"
f"📝 **Summary:**\n\n"
f"{selected_summary['summary']}"
)
await self.send_long_message(ctx, formatted_summary)
@commands.command(name="reginald_summaries", help="Lists available summaries for this channel.")
async def list_mid_term_summaries(self, ctx):
"""Displays a brief list of all available mid-term memory summaries."""
async with self.config.guild(ctx.guild).mid_term_memory() as mid_memory:
summaries = mid_memory.get(str(ctx.channel.id), [])
if not summaries:
await ctx.send("⚠️ No summaries available for this channel.")
return
summary_list = "\n".join(
f"**{i+1}.** 📅 {entry['timestamp']} | 🔍 Topics: {', '.join(entry['topics']) or 'None'}"
for i, entry in enumerate(summaries)
)
await ctx.send(f"📚 **Available Summaries:**\n{summary_list[:2000]}")
@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) @commands.has_permissions(administrator=True)