| from typing import List, Dict, Optional | |
| from collections import deque | |
| class ConversationMemory: | |
| """Manages conversation history with limited context window.""" | |
| def __init__(self, max_messages: int = 10): | |
| """ | |
| Initialize conversation memory. | |
| Args: | |
| max_messages: Maximum number of messages to keep in memory | |
| """ | |
| self.max_messages = max_messages | |
| self.messages: deque = deque(maxlen=max_messages) | |
| self.user_name: Optional[str] = None | |
| self.intent: Optional[str] = None | |
| def add_message(self, role: str, content: str): | |
| """ | |
| Add a message to memory. | |
| Args: | |
| role: "user" or "assistant" | |
| content: Message content | |
| """ | |
| self.messages.append({ | |
| "role": role, | |
| "content": content | |
| }) | |
| def get_context(self) -> List[Dict]: | |
| """Get all messages in memory.""" | |
| return list(self.messages) | |
| def get_formatted_context(self) -> str: | |
| """Get formatted context for LLM.""" | |
| formatted = [] | |
| for msg in self.messages: | |
| formatted.append(f"{msg['role'].upper()}: {msg['content']}") | |
| return "\n".join(formatted) | |
| def set_user_name(self, name: str): | |
| """Set user name extracted from conversation.""" | |
| self.user_name = name | |
| def set_intent(self, intent: str): | |
| """Set detected intent.""" | |
| self.intent = intent | |
| def clear(self): | |
| """Clear all messages.""" | |
| self.messages.clear() | |
| self.user_name = None | |
| self.intent = None | |
| def get_summary(self) -> Dict: | |
| """Get memory summary.""" | |
| return { | |
| "message_count": len(self.messages), | |
| "user_name": self.user_name, | |
| "intent": self.intent, | |
| "messages": self.get_context() | |
| } | |