Skip to content

Chatbot Memory - Persistent Conversation Context ​

Build intelligent chatbots that remember conversations and maintain context across interactions

🧠 What is Chatbot Memory? ​

Chatbot Memory in LangChain refers to the system's ability to store, retrieve, and utilize conversation history to maintain context across multiple interactions. It's what transforms a stateless question-answering system into an intelligent, context-aware conversational agent.

Simple Analogy: Think of chatbot memory as the "working memory" of a human conversation - just like how you remember what was discussed earlier in a conversation to maintain coherent dialogue, chatbot memory allows AI to reference past interactions and maintain conversational flow.

🎯 Memory Architecture Overview ​

text
                    🧠 LANGCHAIN MEMORY ARCHITECTURE 🧠
                    (Persistent Conversation Intelligence)

    β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
    β”‚                     πŸ’Ύ MEMORY STORAGE                          β”‚
    β”‚                   "Conversation Persistence"                    β”‚
    β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
                         β”‚
    β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β–Όβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
    β”‚         🎯 MEMORY TYPES                 β”‚
    β””β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜
       β”‚        β”‚        β”‚        β”‚
       β–Ό        β–Ό        β–Ό        β–Ό
  β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
  β”‚πŸ“ BUFFERβ”‚πŸ“‹ SUMM  β”‚πŸ‘€ ENTITYβ”‚πŸ” VECTORβ”‚
  β”‚ MEMORY  β”‚ MEMORY  β”‚ MEMORY  β”‚ MEMORY  β”‚
  β”‚         β”‚         β”‚         β”‚         β”‚
  β”‚Recent   β”‚AI-Gen   β”‚People & β”‚Semantic β”‚
  β”‚History  β”‚Summary  β”‚Objects  β”‚Search   β”‚
  β””β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”΄β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”˜
       β”‚         β”‚         β”‚         β”‚
       β–Ό         β–Ό         β–Ό         β–Ό
  β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”
  β”‚                  πŸ”„ MEMORY OPERATIONS                          β”‚
  β”‚                                                               β”‚
  β”‚ πŸ’­ Store      πŸ“– Retrieve    πŸ” Search      🧹 Clean          β”‚
  β”‚ New Info     Past Context   Relevant       Old Data          β”‚
  β”‚ Auto-save    Smart Recall   History        Optimize          β”‚
  β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜

πŸ’Ύ Core Memory Types ​

πŸ“ Conversation Buffer Memory ​

Buffer Memory stores recent conversation history in its raw form, providing immediate access to recent context.

Basic Buffer Memory Implementation ​

python
from langchain.memory import ConversationBufferMemory
from langchain.schema import BaseMessage

# Initialize buffer memory
memory = ConversationBufferMemory(return_messages=True)

# Add conversations
memory.chat_memory.add_user_message("Hi, I'm looking for help with Python programming.")
memory.chat_memory.add_ai_message("I'd be happy to help you with Python! What specific topic would you like to learn about?")
memory.chat_memory.add_user_message("I want to learn about data structures.")
memory.chat_memory.add_ai_message("Great choice! Python has several built-in data structures like lists, dictionaries, sets, and tuples. Which one interests you most?")

# Retrieve conversation history
print("Chat History:")
print(memory.buffer)

Custom Buffer Memory with Limits ​

python
class LimitedBufferMemory(ConversationBufferMemory):
    def __init__(self, max_messages: int = 10, **kwargs):
        super().__init__(**kwargs)
        self.max_messages = max_messages
    
    def save_context(self, inputs: dict, outputs: dict):
        super().save_context(inputs, outputs)
        
        # Trim messages if exceeding limit
        messages = self.chat_memory.messages
        if len(messages) > self.max_messages:
            # Keep most recent messages
            self.chat_memory.messages = messages[-self.max_messages:]
    
    def get_context_summary(self) -> str:
        """Get a summary of current context"""
        messages = self.chat_memory.messages
        if not messages:
            return "No conversation history"
        
        user_messages = [msg.content for msg in messages if msg.type == "human"]
        ai_messages = [msg.content for msg in messages if msg.type == "ai"]
        
        return f"Conversation: {len(user_messages)} user messages, {len(ai_messages)} AI responses"

# Usage
limited_memory = LimitedBufferMemory(max_messages=6)

πŸ“‹ Conversation Summary Memory ​

Summary Memory uses AI to create condensed summaries of conversations, allowing for longer-term context while maintaining efficiency.

Basic Summary Memory ​

python
from langchain.memory import ConversationSummaryMemory
from langchain.llms import OpenAI

# Initialize with LLM for summarization
llm = OpenAI(temperature=0)
summary_memory = ConversationSummaryMemory(llm=llm)

# Add conversation and automatically summarize
summary_memory.save_context(
    {"input": "Tell me about machine learning"},
    {"output": "Machine learning is a subset of artificial intelligence that enables computers to learn and make decisions from data without being explicitly programmed for every task."}
)

summary_memory.save_context(
    {"input": "What are the main types?"},
    {"output": "The main types are supervised learning (with labeled data), unsupervised learning (finding patterns in unlabeled data), and reinforcement learning (learning through trial and error)."}
)

# Get current summary
print("Conversation Summary:")
print(summary_memory.buffer)

Advanced Summary Memory with Custom Prompts ​

python
class CustomSummaryMemory(ConversationSummaryMemory):
    def __init__(self, llm, summary_style="detailed", **kwargs):
        super().__init__(llm=llm, **kwargs)
        self.summary_style = summary_style
        self.summary_prompts = {
            "brief": "Summarize this conversation in 1-2 sentences focusing on key topics:",
            "detailed": "Create a comprehensive summary including main topics, decisions made, and action items:",
            "bullet": "Summarize this conversation as bullet points covering main topics:"
        }
    
    def predict_new_summary(self, messages, existing_summary):
        prompt = self.summary_prompts.get(self.summary_style, self.summary_prompts["detailed"])
        
        # Custom summarization logic
        conversation_text = "\n".join([f"{msg.type}: {msg.content}" for msg in messages])
        
        summary_prompt = f"""
        {prompt}
        
        Existing summary: {existing_summary}
        New conversation:
        {conversation_text}
        
        Updated summary:
        """
        
        return self.llm.predict(summary_prompt)

# Usage
custom_summary = CustomSummaryMemory(llm=llm, summary_style="bullet")

πŸ‘€ Entity Memory ​

Entity Memory tracks specific entities (people, places, concepts) mentioned in conversations, maintaining a knowledge graph of relationships.

Basic Entity Memory ​

python
from langchain.memory import ConversationEntityMemory

# Initialize entity memory
entity_memory = ConversationEntityMemory(llm=llm)

# Add conversation with entities
entity_memory.save_context(
    {"input": "Hi, I'm John. I work at Google as a software engineer."},
    {"output": "Nice to meet you, John! It's great to connect with someone from Google. What kind of software engineering do you focus on?"}
)

entity_memory.save_context(
    {"input": "I work on machine learning infrastructure, specifically TensorFlow optimization."},
    {"output": "That's fascinating! TensorFlow is such an important framework. Are you working on performance improvements or new features?"}
)

# Check stored entities
print("Stored Entities:")
print(entity_memory.entity_store.store)

Custom Entity Extraction ​

python
import re
from typing import Dict, List

class AdvancedEntityMemory:
    def __init__(self, llm):
        self.llm = llm
        self.entities = {}
        self.entity_patterns = {
            'person': r'\b[A-Z][a-z]+ [A-Z][a-z]+\b',
            'company': r'\b(?:Google|Microsoft|Apple|Amazon|Meta|Tesla|Netflix)\b',
            'technology': r'\b(?:Python|JavaScript|React|TensorFlow|PyTorch|Docker|Kubernetes)\b',
            'location': r'\b[A-Z][a-z]+(?:, [A-Z][a-z]+)?\b'
        }
    
    def extract_entities(self, text: str) -> Dict[str, List[str]]:
        """Extract entities from text using patterns and NLP"""
        extracted = {}
        
        for entity_type, pattern in self.entity_patterns.items():
            matches = re.findall(pattern, text)
            if matches:
                extracted[entity_type] = list(set(matches))
        
        return extracted
    
    def update_entity_knowledge(self, entities: Dict[str, List[str]], context: str):
        """Update entity knowledge with new context"""
        for entity_type, entity_list in entities.items():
            for entity in entity_list:
                if entity not in self.entities:
                    self.entities[entity] = {
                        'type': entity_type,
                        'contexts': [],
                        'relationships': []
                    }
                
                self.entities[entity]['contexts'].append(context)
    
    def get_entity_context(self, entity_name: str) -> str:
        """Get relevant context for a specific entity"""
        if entity_name in self.entities:
            entity_info = self.entities[entity_name]
            return f"Known information about {entity_name} ({entity_info['type']}): " + \
                   " | ".join(entity_info['contexts'][-3:])  # Last 3 contexts
        return f"No information available about {entity_name}"
    
    def save_context(self, inputs: dict, outputs: dict):
        """Save conversation context and extract entities"""
        conversation_text = f"User: {inputs.get('input', '')} AI: {outputs.get('output', '')}"
        
        entities = self.extract_entities(conversation_text)
        self.update_entity_knowledge(entities, conversation_text)

# Usage
advanced_entity_memory = AdvancedEntityMemory(llm)

πŸ” Vector Store Memory ​

Vector Memory uses embeddings to store and retrieve conversationally relevant information based on semantic similarity.

Basic Vector Memory ​

python
from langchain.memory import VectorStoreRetrieverMemory
from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.docstore import InMemoryDocstore

# Initialize vector store
embeddings = OpenAIEmbeddings()
embedding_size = 1536
index = faiss.IndexFlatL2(embedding_size)
vectorstore = FAISS(embeddings.embed_query, index, InMemoryDocstore({}), {})

# Create vector memory
vector_memory = VectorStoreRetrieverMemory(
    vectorstore=vectorstore,
    k=3  # Retrieve top 3 most relevant memories
)

# Add memories
vector_memory.save_context(
    {"input": "I love working with neural networks"},
    {"output": "Neural networks are fascinating! They're inspired by how our brains work."}
)

vector_memory.save_context(
    {"input": "What's the best way to prevent overfitting?"},
    {"output": "Common techniques include regularization, dropout, early stopping, and cross-validation."}
)

# Retrieve relevant memories
relevant_memories = vector_memory.load_memory_variables(
    {"input": "How can I improve my deep learning model?"}
)
print("Relevant memories:", relevant_memories)

Advanced Vector Memory with Custom Embeddings ​

python
class SemanticMemory:
    def __init__(self, embeddings_model, similarity_threshold=0.8):
        self.embeddings = embeddings_model
        self.memories = []
        self.memory_embeddings = []
        self.similarity_threshold = similarity_threshold
    
    def add_memory(self, input_text: str, output_text: str, metadata: dict = None):
        """Add a new memory with semantic embedding"""
        memory_text = f"Input: {input_text} Output: {output_text}"
        embedding = self.embeddings.embed_query(memory_text)
        
        memory_entry = {
            'input': input_text,
            'output': output_text,
            'metadata': metadata or {},
            'timestamp': datetime.now(),
            'embedding': embedding
        }
        
        self.memories.append(memory_entry)
        self.memory_embeddings.append(embedding)
    
    def retrieve_relevant_memories(self, query: str, k: int = 3) -> List[dict]:
        """Retrieve memories most similar to the query"""
        if not self.memories:
            return []
        
        query_embedding = self.embeddings.embed_query(query)
        
        # Calculate similarities
        similarities = []
        for i, memory_embedding in enumerate(self.memory_embeddings):
            similarity = self._cosine_similarity(query_embedding, memory_embedding)
            similarities.append((i, similarity))
        
        # Sort by similarity and filter by threshold
        similarities.sort(key=lambda x: x[1], reverse=True)
        relevant_memories = []
        
        for idx, similarity in similarities[:k]:
            if similarity >= self.similarity_threshold:
                memory = self.memories[idx].copy()
                memory['similarity_score'] = similarity
                relevant_memories.append(memory)
        
        return relevant_memories
    
    def _cosine_similarity(self, vec1, vec2):
        """Calculate cosine similarity between two vectors"""
        import numpy as np
        return np.dot(vec1, vec2) / (np.linalg.norm(vec1) * np.linalg.norm(vec2))
    
    def get_memory_summary(self) -> dict:
        """Get summary statistics about stored memories"""
        return {
            'total_memories': len(self.memories),
            'memory_types': list(set(m.get('metadata', {}).get('type', 'general') for m in self.memories)),
            'oldest_memory': min(m['timestamp'] for m in self.memories) if self.memories else None,
            'newest_memory': max(m['timestamp'] for m in self.memories) if self.memories else None
        }

# Usage
semantic_memory = SemanticMemory(embeddings)

πŸ”„ Memory Management Strategies ​

🧹 Memory Optimization ​

Memory Cleanup and Maintenance ​

python
from datetime import datetime, timedelta
from typing import Dict, Any

class MemoryManager:
    def __init__(self, max_memories: int = 1000, cleanup_interval_hours: int = 24):
        self.max_memories = max_memories
        self.cleanup_interval = timedelta(hours=cleanup_interval_hours)
        self.last_cleanup = datetime.now()
        self.importance_scores = {}
    
    def should_cleanup(self) -> bool:
        """Check if cleanup is needed"""
        return (datetime.now() - self.last_cleanup) > self.cleanup_interval
    
    def calculate_importance_score(self, memory: dict) -> float:
        """Calculate importance score for a memory"""
        score = 0.0
        
        # Recency score (more recent = higher score)
        days_old = (datetime.now() - memory['timestamp']).days
        recency_score = max(0, 1 - (days_old / 30))  # Decay over 30 days
        score += recency_score * 0.4
        
        # Frequency score (how often memory is accessed)
        access_count = memory.get('access_count', 0)
        frequency_score = min(1.0, access_count / 10)  # Normalize to 0-1
        score += frequency_score * 0.3
        
        # Content length score (longer conversations might be more important)
        content_length = len(memory.get('input', '') + memory.get('output', ''))
        length_score = min(1.0, content_length / 500)  # Normalize
        score += length_score * 0.2
        
        # User rating score (if available)
        user_rating = memory.get('metadata', {}).get('user_rating', 0.5)
        score += user_rating * 0.1
        
        return score
    
    def cleanup_memories(self, memories: List[dict]) -> List[dict]:
        """Remove least important memories to stay under limit"""
        if len(memories) <= self.max_memories:
            return memories
        
        # Calculate importance scores
        scored_memories = []
        for memory in memories:
            score = self.calculate_importance_score(memory)
            scored_memories.append((memory, score))
        
        # Sort by importance and keep top memories
        scored_memories.sort(key=lambda x: x[1], reverse=True)
        cleaned_memories = [memory for memory, score in scored_memories[:self.max_memories]]
        
        self.last_cleanup = datetime.now()
        return cleaned_memories
    
    def merge_similar_memories(self, memories: List[dict], similarity_threshold: float = 0.9) -> List[dict]:
        """Merge very similar memories to reduce redundancy"""
        if not memories:
            return memories
        
        merged_memories = []
        processed_indices = set()
        
        for i, memory1 in enumerate(memories):
            if i in processed_indices:
                continue
            
            similar_memories = [memory1]
            
            for j, memory2 in enumerate(memories[i+1:], i+1):
                if j in processed_indices:
                    continue
                
                similarity = self._calculate_text_similarity(
                    memory1['input'] + memory1['output'],
                    memory2['input'] + memory2['output']
                )
                
                if similarity >= similarity_threshold:
                    similar_memories.append(memory2)
                    processed_indices.add(j)
            
            # Merge similar memories
            merged_memory = self._merge_memory_group(similar_memories)
            merged_memories.append(merged_memory)
            processed_indices.add(i)
        
        return merged_memories
    
    def _calculate_text_similarity(self, text1: str, text2: str) -> float:
        """Simple text similarity calculation"""
        words1 = set(text1.lower().split())
        words2 = set(text2.lower().split())
        
        intersection = words1 & words2
        union = words1 | words2
        
        return len(intersection) / len(union) if union else 0.0
    
    def _merge_memory_group(self, memories: List[dict]) -> dict:
        """Merge a group of similar memories"""
        if len(memories) == 1:
            return memories[0]
        
        # Combine inputs and outputs
        combined_input = " | ".join([m['input'] for m in memories])
        combined_output = " | ".join([m['output'] for m in memories])
        
        # Use most recent timestamp
        latest_timestamp = max(m['timestamp'] for m in memories)
        
        # Combine metadata
        combined_metadata = {}
        for memory in memories:
            combined_metadata.update(memory.get('metadata', {}))
        combined_metadata['merged_count'] = len(memories)
        
        return {
            'input': combined_input,
            'output': combined_output,
            'timestamp': latest_timestamp,
            'metadata': combined_metadata,
            'access_count': sum(m.get('access_count', 0) for m in memories)
        }

# Usage
memory_manager = MemoryManager(max_memories=500)

πŸ”„ Memory Integration Patterns ​

Multi-Memory System ​

python
class HybridMemorySystem:
    def __init__(self, llm, embeddings):
        # Different memory types for different purposes
        self.buffer_memory = ConversationBufferMemory(k=5)  # Recent context
        self.summary_memory = ConversationSummaryMemory(llm=llm)  # Long-term overview
        self.entity_memory = AdvancedEntityMemory(llm)  # Entity tracking
        self.vector_memory = SemanticMemory(embeddings)  # Semantic retrieval
        
        self.memory_manager = MemoryManager()
    
    def save_context(self, inputs: dict, outputs: dict):
        """Save context across all memory systems"""
        # Save to all memory types
        self.buffer_memory.save_context(inputs, outputs)
        self.summary_memory.save_context(inputs, outputs)
        self.entity_memory.save_context(inputs, outputs)
        
        # Add to vector memory with metadata
        self.vector_memory.add_memory(
            inputs.get('input', ''),
            outputs.get('output', ''),
            metadata={'timestamp': datetime.now()}
        )
    
    def load_memory_variables(self, inputs: dict) -> dict:
        """Load relevant memories from all systems"""
        memory_context = {}
        
        # Recent conversation context
        buffer_context = self.buffer_memory.load_memory_variables(inputs)
        memory_context['recent_history'] = buffer_context.get('history', '')
        
        # Conversation summary
        summary_context = self.summary_memory.load_memory_variables(inputs)
        memory_context['conversation_summary'] = summary_context.get('history', '')
        
        # Relevant entities
        input_text = inputs.get('input', '')
        entities = self.entity_memory.extract_entities(input_text)
        entity_contexts = []
        for entity_list in entities.values():
            for entity in entity_list:
                context = self.entity_memory.get_entity_context(entity)
                entity_contexts.append(context)
        memory_context['relevant_entities'] = " | ".join(entity_contexts)
        
        # Semantically similar memories
        relevant_memories = self.vector_memory.retrieve_relevant_memories(input_text)
        memory_context['similar_conversations'] = [
            f"Previous: {m['input']} -> {m['output']}" 
            for m in relevant_memories[:2]
        ]
        
        return memory_context
    
    def get_comprehensive_context(self, query: str) -> str:
        """Get formatted context for LLM prompt"""
        memory_vars = self.load_memory_variables({'input': query})
        
        context_parts = []
        
        if memory_vars.get('recent_history'):
            context_parts.append(f"Recent conversation:\n{memory_vars['recent_history']}")
        
        if memory_vars.get('conversation_summary'):
            context_parts.append(f"Conversation summary:\n{memory_vars['conversation_summary']}")
        
        if memory_vars.get('relevant_entities'):
            context_parts.append(f"Relevant entities:\n{memory_vars['relevant_entities']}")
        
        if memory_vars.get('similar_conversations'):
            context_parts.append(f"Similar past conversations:\n" + 
                               "\n".join(memory_vars['similar_conversations']))
        
        return "\n\n".join(context_parts)

# Usage
hybrid_memory = HybridMemorySystem(llm, embeddings)

# Save a conversation
hybrid_memory.save_context(
    {'input': 'I need help with Python data analysis'},
    {'output': 'I can help you with Python data analysis! Are you working with pandas, numpy, or other libraries?'}
)

# Get comprehensive context for next interaction
context = hybrid_memory.get_comprehensive_context('What are the best practices for data visualization?')

🎯 Real-World Applications ​

πŸ’Ό Customer Support Memory ​

python
class CustomerSupportMemory:
    def __init__(self, llm, embeddings):
        self.hybrid_memory = HybridMemorySystem(llm, embeddings)
        self.customer_profiles = {}
        self.ticket_history = {}
    
    def initialize_customer_session(self, customer_id: str, customer_data: dict):
        """Initialize memory for a customer support session"""
        self.customer_profiles[customer_id] = {
            'name': customer_data.get('name', 'Customer'),
            'account_type': customer_data.get('account_type', 'Standard'),
            'previous_issues': customer_data.get('previous_issues', []),
            'preferences': customer_data.get('preferences', {}),
            'session_start': datetime.now()
        }
    
    def add_support_interaction(self, customer_id: str, agent_input: str, customer_response: str, issue_type: str):
        """Add a support interaction to memory"""
        # Save to hybrid memory system
        inputs = {'input': agent_input, 'customer_id': customer_id, 'issue_type': issue_type}
        outputs = {'output': customer_response}
        
        self.hybrid_memory.save_context(inputs, outputs)
        
        # Track ticket history
        if customer_id not in self.ticket_history:
            self.ticket_history[customer_id] = []
        
        self.ticket_history[customer_id].append({
            'timestamp': datetime.now(),
            'agent_input': agent_input,
            'customer_response': customer_response,
            'issue_type': issue_type
        })
    
    def get_customer_context(self, customer_id: str, current_query: str) -> str:
        """Get comprehensive customer context for agent"""
        context_parts = []
        
        # Customer profile
        if customer_id in self.customer_profiles:
            profile = self.customer_profiles[customer_id]
            context_parts.append(f"Customer: {profile['name']} ({profile['account_type']} account)")
            
            if profile['previous_issues']:
                context_parts.append(f"Previous issues: {', '.join(profile['previous_issues'])}")
        
        # Recent conversation context
        memory_context = self.hybrid_memory.get_comprehensive_context(current_query)
        if memory_context:
            context_parts.append(f"Conversation context:\n{memory_context}")
        
        # Ticket history summary
        if customer_id in self.ticket_history:
            recent_tickets = self.ticket_history[customer_id][-3:]  # Last 3 tickets
            ticket_summaries = []
            for ticket in recent_tickets:
                ticket_summaries.append(f"- {ticket['issue_type']}: {ticket['agent_input'][:100]}...")
            
            context_parts.append(f"Recent tickets:\n" + "\n".join(ticket_summaries))
        
        return "\n\n".join(context_parts)

# Usage
support_memory = CustomerSupportMemory(llm, embeddings)

# Initialize customer session
support_memory.initialize_customer_session('cust_123', {
    'name': 'John Smith',
    'account_type': 'Premium',
    'previous_issues': ['billing', 'technical']
})

# Add interactions
support_memory.add_support_interaction(
    'cust_123',
    'I understand you\'re having trouble with login. Can you tell me what error message you see?',
    'It says "Invalid credentials" but I\'m sure my password is correct.',
    'technical'
)

πŸŽ“ Educational Chatbot Memory ​

python
class EducationalMemory:
    def __init__(self, llm, embeddings):
        self.hybrid_memory = HybridMemorySystem(llm, embeddings)
        self.learning_profiles = {}
        self.knowledge_tracking = {}
    
    def create_learning_profile(self, student_id: str, profile_data: dict):
        """Create a learning profile for a student"""
        self.learning_profiles[student_id] = {
            'name': profile_data.get('name', 'Student'),
            'grade_level': profile_data.get('grade_level', 'Unknown'),
            'subjects': profile_data.get('subjects', []),
            'learning_style': profile_data.get('learning_style', 'visual'),
            'knowledge_level': profile_data.get('knowledge_level', {}),
            'goals': profile_data.get('goals', []),
            'session_start': datetime.now()
        }
        
        self.knowledge_tracking[student_id] = {
            'topics_covered': [],
            'difficulty_progression': [],
            'areas_of_strength': [],
            'areas_for_improvement': [],
            'quiz_scores': []
        }
    
    def track_learning_interaction(self, student_id: str, topic: str, question: str, response: str, understanding_level: int):
        """Track a learning interaction"""
        # Save to memory
        inputs = {
            'input': question,
            'student_id': student_id,
            'topic': topic,
            'understanding_level': understanding_level
        }
        outputs = {'output': response}
        
        self.hybrid_memory.save_context(inputs, outputs)
        
        # Update knowledge tracking
        if student_id in self.knowledge_tracking:
            tracking = self.knowledge_tracking[student_id]
            
            if topic not in tracking['topics_covered']:
                tracking['topics_covered'].append(topic)
            
            tracking['difficulty_progression'].append({
                'topic': topic,
                'understanding_level': understanding_level,
                'timestamp': datetime.now()
            })
            
            # Update strengths and weaknesses
            if understanding_level >= 8:
                if topic not in tracking['areas_of_strength']:
                    tracking['areas_of_strength'].append(topic)
            elif understanding_level <= 5:
                if topic not in tracking['areas_for_improvement']:
                    tracking['areas_for_improvement'].append(topic)
    
    def get_personalized_context(self, student_id: str, current_topic: str) -> str:
        """Get personalized learning context"""
        context_parts = []
        
        # Student profile
        if student_id in self.learning_profiles:
            profile = self.learning_profiles[student_id]
            context_parts.append(
                f"Student: {profile['name']} (Grade {profile['grade_level']}, {profile['learning_style']} learner)"
            )
            
            if profile['goals']:
                context_parts.append(f"Learning goals: {', '.join(profile['goals'])}")
        
        # Knowledge tracking
        if student_id in self.knowledge_tracking:
            tracking = self.knowledge_tracking[student_id]
            
            context_parts.append(f"Topics covered: {', '.join(tracking['topics_covered'][-5:])}")
            
            if tracking['areas_of_strength']:
                context_parts.append(f"Strengths: {', '.join(tracking['areas_of_strength'])}")
            
            if tracking['areas_for_improvement']:
                context_parts.append(f"Areas for improvement: {', '.join(tracking['areas_for_improvement'])}")
        
        # Conversation memory
        memory_context = self.hybrid_memory.get_comprehensive_context(f"Topic: {current_topic}")
        if memory_context:
            context_parts.append(f"Previous learning context:\n{memory_context}")
        
        return "\n\n".join(context_parts)
    
    def suggest_next_topic(self, student_id: str) -> str:
        """Suggest next topic based on learning progress"""
        if student_id not in self.knowledge_tracking:
            return "Let's start with fundamentals!"
        
        tracking = self.knowledge_tracking[student_id]
        profile = self.learning_profiles.get(student_id, {})
        
        # Analyze recent performance
        recent_progress = tracking['difficulty_progression'][-5:]
        avg_understanding = sum(p['understanding_level'] for p in recent_progress) / len(recent_progress) if recent_progress else 5
        
        if avg_understanding >= 8:
            return "You're doing great! Ready for a more advanced topic?"
        elif avg_understanding <= 5:
            return "Let's review some fundamentals to strengthen your foundation."
        else:
            return "You're making good progress! Let's continue building on what you've learned."

# Usage
edu_memory = EducationalMemory(llm, embeddings)

# Create student profile
edu_memory.create_learning_profile('student_456', {
    'name': 'Alice Johnson',
    'grade_level': '10',
    'subjects': ['mathematics', 'physics'],
    'learning_style': 'visual',
    'goals': ['understand calculus', 'prepare for physics exam']
})

πŸš€ Getting Started with Memory ​

πŸ“š Quick Implementation ​

python
from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain
from langchain.llms import OpenAI

# Initialize components
llm = OpenAI(temperature=0.7)
memory = ConversationBufferMemory()

# Create conversation chain with memory
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# Have a conversation
response1 = conversation.predict(input="Hi, I'm learning about Python.")
print(response1)

response2 = conversation.predict(input="What are the best practices for functions?")
print(response2)

response3 = conversation.predict(input="Can you give me an example based on what we discussed?")
print(response3)

🎯 Next Steps ​

  1. Experiment with Memory Types: Try different memory implementations for your use case
  2. Optimize Memory Management: Implement cleanup and optimization strategies
  3. Build Hybrid Systems: Combine multiple memory types for comprehensive context
  4. Add Persistence: Store memories in databases for long-term retention
  5. Monitor Performance: Track memory usage and retrieval effectiveness

πŸ“– Additional Resources ​

  • Memory Performance Tuning: Optimization strategies and benchmarks
  • Advanced Entity Extraction: Using spaCy and custom NLP models
  • Vector Database Integration: Pinecone, Weaviate, and other vector stores
  • Production Memory Systems: Scalable memory architectures

Master chatbot memory to build intelligent, context-aware conversational AI that remembers, learns, and provides personalized experiences across interactions.

Released under the MIT License.