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 β
π§ 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 β
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 β
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 β
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 β
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 β
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 β
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 β
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 β
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 β
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 β
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 β
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 β
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 β
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 β
- Experiment with Memory Types: Try different memory implementations for your use case
- Optimize Memory Management: Implement cleanup and optimization strategies
- Build Hybrid Systems: Combine multiple memory types for comprehensive context
- Add Persistence: Store memories in databases for long-term retention
- 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.