File size: 2,186 Bytes
9d2e12a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
63
64
65
import os
from typing import List

def chunk_text(text: str, chunk_size: int = 800, overlap: int = 150) -> List[str]:
    """Simple chunker by characters. Returns chunks with overlap."""
    if not text:
        return []
    text = text.replace("\r\n", "\n").replace("\r", "\n")
    chunks = []
    start = 0
    length = len(text)
    while start < length:
        end = start + chunk_size
        chunk = text[start:end]
        chunks.append(chunk.strip())
        start = end - overlap
        if start < 0:
            start = 0
    return [c for c in chunks if c]


class ConversationMemory:
    """Window buffer memory to store last k conversation turns."""
    
    def __init__(self, window_size: int = 10):
        """Initialize with a specific window size (number of past interactions to remember).
        
        Args:
            window_size: Number of past question-answer pairs to keep in memory
        """
        self.window_size = window_size
        self.history = []
    
    def add_interaction(self, question: str, answer: str):
        """Add a question-answer pair to memory."""
        self.history.append({"question": question, "answer": answer})
        # Keep only the last window_size interactions
        if len(self.history) > self.window_size:
            self.history = self.history[-self.window_size:]
    
    def get_history(self) -> List[dict]:
        """Get the current conversation history."""
        return self.history
    
    def get_formatted_history(self) -> str:
        """Get formatted conversation history for context."""
        if not self.history:
            return ""
        
        formatted = ["Previous conversation:"]
        for i, interaction in enumerate(self.history, 1):
            formatted.append(f"\nTurn {i}:")
            formatted.append(f"Student: {interaction['question']}")
            formatted.append(f"Assistant: {interaction['answer']}")
        
        return "\n".join(formatted)
    
    def clear(self):
        """Clear all conversation history."""
        self.history = []
    
    def __len__(self):
        """Return the number of interactions in memory."""
        return len(self.history)