hello-agents/Co-creation-projects/dongyu23-MADF/app/agent/memory.py

from collections import deque

class SharedMemory:
    def __init__(self, n_participants):
        self.window_size = n_participants
        # Context Window: Always holds the last N messages for context (Sliding)
        self.context_window = deque(maxlen=n_participants)
        # Summary Buffer: Accumulates messages to be summarized (Batch)
        self.summary_buffer = [] 
        
        self.summary_history = [] # Stores the summaries generated by the moderator
        self.all_history = []     # Stores all messages for record keeping

    def add_message(self, speaker_name, content):
        message = {"speaker": speaker_name, "content": content}
        self.context_window.append(message)
        self.summary_buffer.append(message)
        self.all_history.append(message)

    def is_ready_for_summary(self):
        """Check if we have enough new messages to trigger a summary."""
        return len(self.summary_buffer) >= self.window_size

    def get_messages_for_summary(self):
        """Return the batch of messages to be summarized."""
        return self.summary_buffer

    def clear_summary_buffer(self):
        """Clear the summary buffer after summarization."""
        self.summary_buffer = []

    def add_summary(self, summary):
        self.summary_history.append(summary)

    def get_summaries(self):
        return self.summary_history
    
    def get_context_str(self):
        """Returns a string representation of summaries + current sliding window for context."""
        context = "【过往总结】\n"
        if not self.summary_history:
            context += "(暂无)\n"
        for s in self.summary_history:
            context += f"- {s}\n"
        
        context += "\n【近期讨论】\n"
        if not self.context_window:
            context += "(暂无)\n"
        for m in self.context_window:
            context += f"{m['speaker']}: {m['content']}\n"
        
        return context


class PrivateMemory:
    def __init__(self, n_participants):
        self.window_size = n_participants
        self.thoughts = []
        self.speeches = []

    def add_speech(self, content):
        self.speeches.append(content)

    def get_speech_history_str(self):
        if not self.speeches:
            return "暂无过往发言。"
        
        history = "【我之前的发言】\n"
        for i, speech in enumerate(self.speeches[-3:], 1): # Last 3 speeches
            history += f"发言{i}: {speech}\n"
        return history

    def add_thought(self, thought_json):
        self.thoughts.append(thought_json)
        if len(self.thoughts) > self.window_size:
            self.thoughts.pop(0)

    def get_thoughts(self):
        return self.thoughts

    def get_recent_thought_str(self):
        if not self.thoughts:
            return "暂无过往思考。"
        
        last_thought = self.thoughts[-1]
        return f"上次思考: {last_thought.get('focus', 'N/A')} | 态度: {last_thought.get('attitude', 'N/A')}"