| """Conversation memory management""" | |
| from typing import List | |
| from langchain.messages import HumanMessage, AIMessage | |
| from config import logger_memory | |
| class ConversationMemory: | |
| """Manages conversation history with smart truncation""" | |
| def __init__(self, max_messages: int = 4): | |
| self.messages = [] | |
| self.max_messages = max_messages | |
| logger_memory.info(f"ConversationMemory initialized with max_messages={max_messages}") | |
| def add_user_message(self, content: str): | |
| """Add user message to history""" | |
| logger_memory.debug(f"Adding user message: {content[:50]}..." if len(content) > 50 else f"Adding user message: {content}") | |
| self.messages.append(HumanMessage(content=content)) | |
| current_count = len(self.messages) | |
| logger_memory.info(f"User message added. Current message count: {current_count}") | |
| self._truncate() | |
| def add_ai_message(self, content: str): | |
| """Add AI message to history""" | |
| content_preview = content[:50] + "..." if len(content) > 50 else content | |
| logger_memory.debug(f"Adding AI message: {content_preview}") | |
| self.messages.append(AIMessage(content=content)) | |
| current_count = len(self.messages) | |
| logger_memory.info(f"AI message added. Current message count: {current_count}") | |
| self._truncate() | |
| def _truncate(self): | |
| """Keep only recent messages to avoid context bloat""" | |
| if len(self.messages) > self.max_messages: | |
| removed_count = len(self.messages) - self.max_messages | |
| self.messages = self.messages[-self.max_messages:] | |
| logger_memory.info(f"Memory truncated: removed {removed_count} oldest messages, keeping {len(self.messages)} most recent") | |
| else: | |
| logger_memory.debug(f"No truncation needed. Current count: {len(self.messages)}, max: {self.max_messages}") | |
| def get_history(self) -> List: | |
| """Get formatted conversation history""" | |
| history = self.messages | |
| logger_memory.debug(f"Retrieving history: {len(history)} messages") | |
| return history | |
| def get_history_text(self) -> str: | |
| """Get history as formatted text""" | |
| if not self.messages: | |
| logger_memory.debug("No conversation history available") | |
| return "No previous conversation" | |
| history_text = [] | |
| for msg in self.messages: | |
| role = "User" if isinstance(msg, HumanMessage) else "Assistant" | |
| history_text.append(f"{role}: {msg.content}") | |
| formatted = "\n".join(history_text) | |
| logger_memory.debug(f"Formatted history: {len(formatted)} characters, {len(self.messages)} messages") | |
| return formatted | |