This guide delves into constructing intelligent agent systems that extend their reasoning beyond isolated interactions by embedding memory as a fundamental feature. We demonstrate how to architect episodic memory to archive distinct experiences and semantic memory to identify enduring patterns, enabling the agent to refine its behavior across multiple engagements. By integrating processes such as planning, execution, adjustment, and introspection, the agent progressively aligns with user preferences and gains autonomy. Ultimately, we reveal how memory-centric cognition fosters agents that deliver more coherent, context-aware, and insightful interactions over time.
Designing Memory Architectures for Adaptive Agents
At the heart of our agent lie two complementary memory modules. Episodic memory captures detailed records of individual interactions, preserving the context, actions taken, and outcomes. This allows the agent to recall specific past events. In contrast, semantic memory abstracts these experiences into generalized knowledge, such as user preferences and behavioral trends, which inform future decisions. This dual-memory framework mirrors human cognitive processes, equipping the agent to learn and evolve continuously.
import numpy as np
from collections import defaultdict
from datetime import datetime
class EpisodicMemory:
def __init__(self, max_capacity=100):
self.max_capacity = max_capacity
self.records = []
def add_episode(self, state, action, result, timestamp=None):
if timestamp is None:
timestamp = datetime.utcnow().isoformat()
episode = {
'state': state,
'action': action,
'result': result,
'timestamp': timestamp,
'embedding': self._create_embedding(state, action, result)
}
self.records.append(episode)
if len(self.records) > self.max_capacity:
self.records.pop(0)
def _create_embedding(self, state, action, result):
combined_text = f"{state} {action} {result}".lower()
return hash(combined_text) % 10000
def find_similar(self, query_state, top_k=3):
if not self.records:
return []
query_emb = self._create_embedding(query_state, "", "")
scored = [(abs(ep['embedding'] - query_emb), ep) for ep in self.records]
scored.sort(key=lambda x: x[0])
return [ep for _, ep in scored[:top_k]]
def recent_episodes(self, count=5):
return self.records[-count:]
class SemanticMemory:
def __init__(self):
self.user_preferences = defaultdict(float)
self.behavior_patterns = defaultdict(list)
self.performance_metrics = defaultdict(lambda: {'success': 0, 'attempts': 0})
def update_preference(self, key, value, influence=1.0):
self.user_preferences[key] = 0.9 * self.user_preferences[key] + 0.1 * influence * value
def log_pattern(self, context, action, was_successful):
pattern_key = f"{context}_{action}"
self.behavior_patterns[context].append((action, was_successful))
self.performance_metrics[pattern_key]['attempts'] += 1
if was_successful:
self.performance_metrics[pattern_key]['success'] += 1
def select_optimal_action(self, context):
if context not in self.behavior_patterns:
return None
action_stats = defaultdict(lambda: {'success': 0, 'attempts': 0})
for action, success in self.behavior_patterns[context]:
action_stats[action]['attempts'] += 1
if success:
action_stats[action]['success'] += 1
best_action = max(action_stats.items(), key=lambda x: x[1]['success'] / max(x[1]['attempts'], 1))
return best_action[0] if best_action[1]['attempts'] > 0 else None
def get_preference(self, key):
return self.user_preferences.get(key, 0.0)
Perception and Strategic Planning in the Agent
The agent’s ability to interpret user input and formulate a plan is crucial for meaningful interaction. We implement a perception module that classifies user intents such as seeking recommendations, updating preferences, or requesting task execution. Leveraging the memories, the planning component devises appropriate responses or actions, tailoring its strategy based on accumulated knowledge.
class MemoryDrivenAgent:
def __init__(self):
self.episodic_memory = EpisodicMemory(max_capacity=50)
self.semantic_memory = SemanticMemory()
self.active_plan = []
self.session_counter = 0
def interpret_input(self, user_text):
text = user_text.lower()
if any(keyword in text for keyword in ['recommend', 'suggest', 'what should']):
intent = 'recommendation'
elif any(keyword in text for keyword in ['remember', 'prefer', 'like', 'favorite']):
intent = 'update_preference'
elif any(keyword in text for keyword in ['do', 'complete', 'finish', 'task']):
intent = 'execute_task'
else:
intent = 'chat'
return {'intent': intent, 'content': text}
def formulate_plan(self, state):
intent = state['intent']
content = state['content']
similar_past = self.episodic_memory.find_similar(content, top_k=3)
plan = []
if intent == 'recommendation':
genre_preferences = {k: v for k, v in self.semantic_memory.user_preferences.items() if k.startswith('genre_')}
if genre_preferences:
top_genre = max(genre_preferences.items(), key=lambda x: x[1])[0]
plan.append(('recommend', top_genre.replace('genre_', '')))
else:
plan.append(('recommend', 'general'))
elif intent == 'update_preference':
genres = ['science fiction', 'fantasy', 'mystery', 'romance', 'thriller']
detected = next((g for g in genres if g in content), None)
if detected:
plan.append(('update_preference', detected))
elif intent == 'execute_task':
best_action = self.semantic_memory.select_optimal_action('task')
plan.append(('execute', best_action if best_action else 'default'))
else:
plan = []
self.active_plan = plan
return plan
Action Execution, Feedback Integration, and Learning
Once a plan is in place, the agent carries out the corresponding action. It also incorporates user feedback to adjust its strategy dynamically. This feedback loop, combined with continuous reflection and memory updates, enables the agent to self-correct and enhance its performance over time.
def execute_action(self, action):
action_type, parameter = action
if action_type == 'recommend':
if parameter == 'general':
return "I'd love to learn your tastes first! What genres do you prefer?"
return f"Based on what I know, I suggest exploring {parameter} titles!"
elif action_type == 'update_preference':
self.semantic_memory.update_preference(f'genre_{parameter}', 1.0)
return f"Thanks! I've noted that you enjoy {parameter}."
elif action_type == 'execute':
return f"Performing task using strategy: {parameter}"
return "Action completed."
def adjust_plan(self, user_feedback):
feedback_lower = user_feedback.lower()
if 'no' in feedback_lower or 'incorrect' in feedback_lower:
if self.active_plan:
action_type, _ = self.active_plan[0]
if action_type == 'recommend':
sorted_prefs = sorted(
[(k, v) for k, v in self.semantic_memory.user_preferences.items() if k.startswith('genre_')],
key=lambda x: x[1],
reverse=True
)
if len(sorted_prefs) > 1:
alternative_genre = sorted_prefs[1][0].replace('genre_', '')
self.active_plan = [('recommend', alternative_genre)]
return True
return False
def introspect(self, state, action, result, was_successful):
self.episodic_memory.add_episode(state['content'], str(action), result)
self.semantic_memory.log_pattern(state['intent'], str(action), was_successful)
Simulating Multi-Turn Sessions for Progressive Learning
To observe the agent’s evolving intelligence, we simulate sessions where it processes a series of user inputs. Each turn follows the cycle of perception, planning, action, and reflection. Over multiple sessions, the agent becomes increasingly adept at personalizing its responses and recommendations.
def conduct_session(self, user_messages):
self.session_counter += 1
print(f"n{'='*60}")
print(f"SESSION {self.session_counter}")
print(f"{'='*60}n")
session_log = []
for turn_index, message in enumerate(user_messages, 1):
print(f"Turn {turn_index}")
print(f"User: {message}")
current_state = self.interpret_input(message)
plan = self.formulate_plan(current_state)
if not plan:
print("Agent: I'm not sure how to respond to that.n")
continue
response = self.execute_action(plan[0])
print(f"Agent: {response}n")
success_flag = 'recommend' in plan[0][0] or 'update' in plan[0][0]
self.introspect(current_state, plan[0], response, success_flag)
session_log.append({
'turn': turn_index,
'input': message,
'intent': current_state['intent'],
'action': plan[0],
'response': response
})
return session_log
Assessing Memory Utilization and Agent Progress
Evaluating how the agent leverages its memories provides insight into its learning trajectory. We examine the volume and recency of episodic records, the breadth of learned preferences, and the success rates of action patterns. Additionally, comparing multiple sessions highlights improvements in personalization and response quality.
def analyze_memory(agent):
print("n" + "="*60)
print("MEMORY USAGE REPORT")
print("="*60 + "n")
print("Episodic Memory:")
print(f" Episodes stored: {len(agent.episodic_memory.records)}")
if agent.episodic_memory.records:
print(f" First episode timestamp: {agent.episodic_memory.records[0]['timestamp']}")
print(f" Most recent episode timestamp: {agent.episodic_memory.records[-1]['timestamp']}")
print("nSemantic Memory:")
print(f" Number of preferences learned: {len(agent.semantic_memory.user_preferences)}")
top_prefs = sorted(agent.semantic_memory.user_preferences.items(), key=lambda x: x[1], reverse=True)[:5]
for pref, val in top_prefs:
print(f" {pref}: {val:.3f}")
print(f"n Behavior patterns recorded: {len(agent.semantic_memory.behavior_patterns)}")
print("n Success rates per context-action:")
for key, stats in list(agent.semantic_memory.performance_metrics.items())[:5]:
if stats['attempts'] > 0:
rate = stats['success'] / stats['attempts']
print(f" {key}: {rate:.2%} ({stats['success']}/{stats['attempts']})")
def session_comparison(session_histories):
print("n" + "="*60)
print("SESSION-TO-SESSION PERFORMANCE")
print("="*60 + "n")
for idx, session in enumerate(session_histories, 1):
personalized_count = sum(1 for entry in session if 'preferences' in entry['response'].lower())
print(f"Session {idx}:")
print(f" Total turns: {len(session)}")
print(f" Personalized recommendations: {personalized_count}")
Comprehensive Demonstration: Learning Across Sessions
Bringing all components together, we run a demonstration where the agent interacts with a user over several sessions. The agent progressively refines its understanding of the user’s tastes, improving the relevance of its suggestions. We also test the retrieval of similar past experiences to showcase the power of episodic memory in guiding decisions.
def demo_agent():
agent = MemoryDrivenAgent()
print("n📚 DEMO: Agent progressively learns user preferences through multiple sessionsn")
session_one = [
"Hello, I'm searching for a good book to read.",
"I really enjoy science fiction novels.",
"Can you suggest something?"
]
results_one = agent.conduct_session(session_one)
session_two = [
"I'm feeling bored, what should I read?",
"I also like fantasy stories.",
"Please recommend a book."
]
results_two = agent.conduct_session(session_two)
session_three = [
"What do you suggest for tonight?",
"I'm also interested in mystery genres.",
"Recommend something based on what you know about me."
]
results_three = agent.conduct_session(session_three)
analyze_memory(agent)
session_comparison([results_one, results_two, results_three])
print("n" + "="*60)
print("EPISODIC MEMORY QUERY TEST")
print("="*60 + "n")
search_query = "recommend science fiction"
matches = agent.episodic_memory.find_similar(search_query, top_k=3)
print(f"Query: '{search_query}'")
print(f"Found {len(matches)} similar episodes:n")
for episode in matches:
print(f" State: {episode['state']}")
print(f" Action: {episode['action']}")
print(f" Result: {episode['result'][:50]}...n")
if __name__ == "__main__":
print("="*60)
print("MEMORY-ENABLED AGENTS WITH LONG-TERM AUTONOMY")
print("="*60)
demo_agent()
print("n✅ Tutorial complete! Key insights:")
print(" • Episodic memory archives detailed experiences")
print(" • Semantic memory abstracts and generalizes knowledge")
print(" • Agents enhance recommendations through repeated sessions")
print(" • Memory retrieval informs smarter future actions")
In summary, integrating episodic and semantic memory systems empowers agents to continuously learn and improve their decision-making capabilities. This approach enables the creation of autonomous systems that adapt to user preferences, refine their strategies, and recall relevant past experiences, resulting in more personalized and intelligent interactions over time.
