moha_mind/agent/connected_memory.py
Rama master · réplica de lectura del repositorio original.
"""Connected Memory Engine - links information across all memory categories.
When MohaMind learns something new, it automatically connects it to
related information across all memory files. This is what makes
MohaMind unique - it doesn't just store facts, it understands relationships.
"""
import re
from moha_mind.agent.memory import MemoryManager
from moha_mind.utils.logging_config import log
from moha_mind.utils.timezone import ksa_today_str
class ConnectedMemory:
def __init__(self, memory: MemoryManager):
self.memory = memory
self._link_rules = {
"birthday": self._handle_birthday,
"anniversary": self._handle_anniversary,
"expiry": self._handle_expiry,
"car": self._handle_car_info,
"wife": self._handle_family_info,
"kid": self._handle_family_info,
"child": self._handle_family_info,
"bill": self._handle_bill,
"subscription": self._handle_subscription,
"appointment": self._handle_appointment,
"doctor": self._handle_appointment,
"medication": self._handle_medication,
"task": self._handle_task,
"trip": self._handle_travel,
"travel": self._handle_travel,
"passport": self._handle_document,
"license": self._handle_document,
"visa": self._handle_document,
"pregnancy": self._handle_pregnancy,
"vaccination": self._handle_vaccination,
"gift": self._handle_gift,
"course": self._handle_learning,
"gym": self._handle_health_habit,
}
def process_new_info(self, text: str, source: str = "conversation") -> list[str]:
"""Analyze new information and create connections across memory files.
Returns a list of actions taken (for transparency).
"""
actions = []
text_lower = text.lower()
for keyword, handler in self._link_rules.items():
if keyword in text_lower:
try:
result = handler(text, source)
if result:
actions.extend(result)
except Exception as e:
log.warning(f"Connected memory rule '{keyword}' failed: {e}")
return actions
def _handle_birthday(self, text: str, source: str) -> list[str]:
"""Connect birthday info across occasions, relationships, shopping."""
actions = []
date_match = re.search(r"(\d{4}-\d{2}-\d{2}|\w+ \d{1,2}(?:,? \d{4})?)", text)
name_match = re.search(
r"(?:wife|husband|friend|mom|dad|mother|father|brother|sister|son|daughter|kid)\s+"
r"(?:named\s+)?(\w+)",
text,
re.IGNORECASE,
)
person_name = name_match.group(1) if name_match else ""
if date_match and person_name:
existing_occasions = self.memory.read("occasions")
entry = f"- {person_name}'s birthday: {date_match.group(1)}"
if entry.lower() not in existing_occasions.lower():
self.memory.append_to_section("occasions", "Birthdays", entry)
actions.append(f"Added {person_name}'s birthday to occasions")
actions.append(
f"Connected: I'll remind you about {person_name}'s birthday and suggest gift ideas when it's close"
)
return actions
def _handle_anniversary(self, text: str, source: str) -> list[str]:
actions = []
date_match = re.search(r"(\d{4}-\d{2}-\d{2})", text)
if date_match:
existing = self.memory.read("occasions")
entry = f"- Anniversary: {date_match.group(1)}"
if entry.lower() not in existing.lower():
self.memory.append_to_section("occasions", "Anniversaries", entry)
actions.append("Added anniversary to occasions")
actions.append("Connected: I'll suggest celebration ideas and gift reminders")
return actions
def _handle_expiry(self, text: str, source: str) -> list[str]:
actions = []
date_match = re.search(r"expir\w+(?:\s+(?:on|date)?)[:\s]+(\d{4}-\d{2}-\d{2})", text, re.IGNORECASE)
if date_match:
actions.append("Connected: Added to Expiry Guardian - will remind you 90, 30, 7 days before and on the day")
return actions
def _handle_car_info(self, text: str, source: str) -> list[str]:
actions = []
text_lower = text.lower()
service_keywords = ["oil change", "service", "maintenance", "tire", "brake", "battery"]
if any(kw in text_lower for kw in service_keywords):
today = ksa_today_str()
self.memory.append_to_section("vehicle", "Service History", f"- [{today}] {text.strip()}")
actions.append("Logged car service to vehicle history")
if "insurance" in text_lower:
actions.append("Connected: Will track car insurance expiry and remind you to renew")
if "registration" in text_lower:
actions.append("Connected: Will track registration renewal date")
return actions
def _handle_family_info(self, text: str, source: str) -> list[str]:
actions = []
text_lower = text.lower()
if "wife" in text_lower or "spouse" in text_lower:
for keyword in ["loves", "likes", "favorite", "favourite"]:
if keyword in text_lower:
existing = self.memory.read("family")
if "## Wife" in existing:
actions.append(
"Connected: Updated wife preferences - "
"will use for gift ideas, date suggestions, and shopping"
)
break
return actions
def _handle_bill(self, text: str, source: str) -> list[str]:
actions = []
amount_match = re.search(r"(\d+)\s*(?:sar|riyal|r)", text, re.IGNORECASE)
if amount_match:
actions.append("Connected: Added to finance tracker - will remind you before due date")
return actions
def _handle_subscription(self, text: str, source: str) -> list[str]:
actions = []
actions.append("Connected: Added to subscription tracker - will monitor renewals and price changes")
return actions
def _handle_appointment(self, text: str, source: str) -> list[str]:
actions = []
date_match = re.search(r"(\d{4}-\d{2}-\d{2})", text)
if date_match:
actions.append("Connected: Added appointment reminder - will notify you 1 day and 1 hour before")
if "family" in text.lower() or "wife" in text.lower() or "kid" in text.lower() or "child" in text.lower():
actions.append("Connected: Also added to family calendar")
return actions
def _handle_medication(self, text: str, source: str) -> list[str]:
actions = []
actions.append("Connected: Added to health tracker - will set up medication reminders")
return actions
def _handle_task(self, text: str, source: str) -> list[str]:
actions = []
date_match = re.search(r"(?:by|due|before|deadline)[:\s]+(\d{4}-\d{2}-\d{2})", text, re.IGNORECASE)
if date_match:
actions.append(
f"Connected: Task deadline set for {date_match.group(1)} - will remind you 3 days and 1 day before"
)
return actions
def _handle_travel(self, text: str, source: str) -> list[str]:
actions = []
date_match = re.search(r"(\d{4}-\d{2}-\d{2})", text)
if date_match:
actions.append("Connected: Added to travel tracker - will help with packing list and visa check")
return actions
def _handle_document(self, text: str, source: str) -> list[str]:
actions = []
date_match = re.search(r"expir\w+[:\s]+(\d{4}-\d{2}-\d{2})", text, re.IGNORECASE)
if date_match:
actions.append("Connected: Added to Expiry Guardian - will remind you months in advance")
return actions
def _handle_pregnancy(self, text: str, source: str) -> list[str]:
actions = []
week_match = re.search(r"week\s+(\d+)", text, re.IGNORECASE)
due_match = re.search(r"due[:\s]+(\d{4}-\d{2}-\d{2})", text, re.IGNORECASE)
if week_match or due_match:
actions.append(
"Connected: Updated pregnancy tracker - will provide weekly milestone updates and appointment reminders"
)
return actions
def _handle_vaccination(self, text: str, source: str) -> list[str]:
actions = []
date_match = re.search(r"(\d{4}-\d{2}-\d{2})", text)
if date_match:
actions.append("Connected: Added vaccination to family tracker - will remind you 3 days before")
return actions
def _handle_gift(self, text: str, source: str) -> list[str]:
actions = []
for person_keyword in ["wife", "mom", "dad", "friend", "brother", "sister"]:
if person_keyword in text.lower():
existing = self.memory.read("relationships")
if "## Gift Ideas" in existing:
self.memory.append_to_section("relationships", "Gift Ideas", f"- ({person_keyword}) {text.strip()}")
actions.append(f"Connected: Stored as gift idea for {person_keyword}")
break
return actions
def _handle_learning(self, text: str, source: str) -> list[str]:
actions = []
actions.append("Connected: Added to learning tracker - will help you schedule study time")
return actions
def _handle_health_habit(self, text: str, source: str) -> list[str]:
actions = []
actions.append("Connected: Updated health tracker - will include in weekly review patterns")
return actions
def get_connected_insights(self, query: str) -> list[str]:
"""Find connections across memory categories relevant to a query."""
insights = []
results = self.memory.search(query)
categories_found = set(r["category"] for r in results)
if len(categories_found) > 1:
cat_names = ", ".join(categories_found)
insights.append(f"This connects to your {cat_names} data")
if "birthday" in query.lower() or "anniversary" in query.lower():
shopping = self.memory.read("shopping")
relationships = self.memory.read("relationships")
if shopping or relationships:
insights.append("I have gift ideas and preferences saved - let me suggest something")
if "car" in query.lower() or "vehicle" in query.lower():
finances = self.memory.read("finances")
if finances and "insurance" in finances.lower():
insights.append("I can check your car insurance status too")
return insights