from collections import Counter from dataclasses import asdict, dataclass from datetime import datetime try: from langchain_core.chat_history import InMemoryChatMessageHistory except Exception: InMemoryChatMessageHistory = None @dataclass class MemoryEvent: iteration: int trainer: str query: str category: str priority: str timestamp: str class SimpleMemory: """Memoria simple para la simulación PokéSupport Town.""" def __init__(self): self.iteration = 0 self.conversation_history = [] self.event_history = [] self.detected_patterns = [] self.tickets = [] self.chat_history = ( InMemoryChatMessageHistory() if InMemoryChatMessageHistory is not None else None ) def reset(self): self.__init__() def add_conversation_turn(self, user_message, assistant_message): self.conversation_history.append({ "user": user_message, "assistant": assistant_message, }) if self.chat_history is not None: self.chat_history.add_user_message(user_message) self.chat_history.add_ai_message(assistant_message) def add_event(self, trainer, query, category, priority): self.iteration += 1 event = MemoryEvent( iteration=self.iteration, trainer=trainer, query=query, category=category, priority=priority, timestamp=datetime.now().isoformat(timespec="seconds"), ) self.event_history.append(asdict(event)) return asdict(event) def count_by_category(self): return Counter(event["category"] for event in self.event_history) def recent_events(self, limit=4): return self.event_history[-limit:] def recent_conversation(self, limit=6): if self.chat_history is None: return self.conversation_history[-limit:] return [ {"type": message.type, "content": message.content} for message in self.chat_history.messages[-limit:] ] def detect_pattern(self, category, threshold=3): category_count = self.count_by_category()[category] if category_count < threshold: return None existing_pattern = next( ( pattern for pattern in self.detected_patterns if pattern["category"] == category ), None, ) if existing_pattern: return existing_pattern pattern = { "category": category, "count": category_count, "message": ( f"Se han detectado {category_count} incidencias de tipo " f"'{category}'. Puede ser un problema general de la ciudad." ), } self.detected_patterns.append(pattern) return pattern def create_ticket(self, trainer, category, query, priority): ticket = { "id": f"T-{len(self.tickets) + 1:03d}", "trainer": trainer, "category": category, "query": query, "priority": priority, "status": "abierto", "iteration": self.iteration, } self.tickets.append(ticket) return ticket def summary(self): return { "iterations": self.iteration, "events": self.event_history, "category_counts": dict(self.count_by_category()), "patterns": self.detected_patterns, "tickets": self.tickets, }