Memory Buffer

class ConversationHistory:
    def __init__(self, max_history: int = 5):
        """
        Initialize the conversation history with a maximum limit.
        :param max_history: Maximum number of queries to retain in history.
        """
        self.max_history = max_history
        self.history: List[Dict[str, str]] = []

{
    "query": "How do I solve the two sum problem?",
    "response": "Use a hash map to store complements..."
}

self.history = [
    {"query": "What is binary search?", "response": "..."},
    {"query": "How do I implement it?", "response": "..."},
    {"query": "What's the time complexity?", "response": "..."}
]

def add_query(self, query: str, response: str):
    """
    Add a new query and response to the history.
    :param query: The user's query.
    :param response: The system's response.
    """
    self.history.append({"query": query, "response": response})
    if len(self.history) > self.max_history:
        # Remove the oldest query if history exceeds the limit
        self.history.pop(0)

def get_context(self) -> str:
    """
    Generate a context string from the conversation history.
    :return: A formatted context string.
    """
    context = ""
    for entry in self.history:
        context += f"User: {entry['query']}\nSystem: {entry['response']}\n"
    return context.strip()

User: What is the two sum problem?
System: The two sum problem asks you to find two numbers in an array that add up to a target value...
User: How do I optimize it?
System: Use a hash map to achieve O(n) time complexity...
User: What about space complexity?
System: The hash map approach uses O(n) space in the worst case...

def clear(self):
    """Clear the conversation history."""
    self.history = []

class RAGEngine:
    def __init__(
        self,
        retriever: LeetCodeRetriever,
        max_history: int = 3,
        # ... other parameters
    ):
        self.conversation_history = ConversationHistory(max_history)

def generate_enhanced_prompt(self, query: str, context: List[Solution]) -> str:
    """Generate a structured prompt incorporating context and conversation history."""
    # Retrieve the conversation history
    history_context = self.conversation_history.get_context()
    
    # Generate the base prompt based on the mode
    if self.mode == "reasoning":
        base_prompt = PromptTemplates.reasoning_prompt(query, context)
    else:
        base_prompt = PromptTemplates.general_prompt(query, context)
    
    # Enhance the prompt with conversation history
    enhanced_prompt = (
        f"Conversation History:\n{history_context}\n\n"
        f"Query: {query}\n\n"
        f"Context: {context}\n\n"
        f"Instruction: {base_prompt}"
    )
    
    return enhanced_prompt

def answer_question(self, query: str, k: int = 5, min_confidence: float = 0.6) -> str:
    # ... retrieval and generation logic ...
    
    # Get response from Ollama
    response = self.call_ollama(prompt)
    
    # Add the response to the conversation history
    self.conversation_history.add_query(query, response)
    
    # Filter response if in reasoning mode
    if self.mode == "reasoning":
        response = self.filter_reasoning_response(response)
    
    return f"Generated Solution:\n{response}"

Query: ~20-50 tokens
Response: ~100-300 tokens
Total per entry: ~120-350 tokens

from rag_engine import RAGEngine
from src.DSAAssistant.components.retriever2 import LeetCodeRetriever

# Initialize with max_history=3
retriever = LeetCodeRetriever()
rag_engine = RAGEngine(retriever, max_history=3)
rag_engine.set_mode("general")

# Query 1
answer1 = rag_engine.answer_question("What is the two sum problem?")
print(answer1)

# Query 2 (references previous context)
answer2 = rag_engine.answer_question("How do I optimize it?")
print(answer2)

# Query 3 (builds on previous two)
answer3 = rag_engine.answer_question("What about space complexity?")
print(answer3)

# View conversation history
print("\nConversation History:")
print(rag_engine.conversation_history.get_context())

# Get current history length
history_len = len(rag_engine.conversation_history.history)
print(f"Current history length: {history_len}")

# Max history setting
max_history = rag_engine.conversation_history.max_history
print(f"Max history: {max_history}")

# Check if history is at capacity
if history_len >= max_history:
    print("History is full - oldest entries will be evicted")

# Have a conversation
rag_engine.answer_question("What is binary search?")
rag_engine.answer_question("How do I implement it?")

# Clear history to start fresh
rag_engine.conversation_history.clear()

# Next query has no context
rag_engine.answer_question("What is merge sort?")
# This will be treated as a new conversation

# Long conversation mode (10 entries)
rag_engine_long = RAGEngine(retriever, max_history=10)

# Minimal context mode (1 entry)
rag_engine_minimal = RAGEngine(retriever, max_history=1)

# No history (stateless)
rag_engine_stateless = RAGEngine(retriever, max_history=0)

# Typical memory per history entry
entry_size = len(query) + len(response)  # Characters
avg_entry = 400  # ~400 characters average

# For max_history=3
total_memory = 3 × 400 = 1200 characters (~1.2 KB)

# On first query, history is empty
context = rag_engine.conversation_history.get_context()
print(context)  # Returns empty string ""

# Reduce max_history for verbose responses
rag_engine = RAGEngine(retriever, max_history=2)  # Instead of 3

# Set max_history=0 for stateless operation
rag_engine = RAGEngine(retriever, max_history=0)

# Every query is independent
rag_engine.answer_question("Query 1")
rag_engine.answer_question("Query 2")  # No context from Query 1