File size: 2,755 Bytes
9d5041f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 | """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
|