How to Build Memory-Powered Agentic AI That Learns Continuously Through Episodic Experiences and Semantic Patterns for Long-Term Autonomy
In this tutorial, we discover how to construct agentic techniques that suppose past a single interplay by using reminiscence as a core functionality. We stroll by how we design episodic reminiscence to retailer experiences and semantic reminiscence to seize long-term patterns, permitting the agent to evolve its behaviour over a number of classes. As we implement planning, appearing, revising, and reflecting, we see how the agent steadily adapts to person preferences and turns into extra autonomous. By the tip, we perceive how memory-driven reasoning helps us create brokers that really feel extra contextual, constant, and clever with each interplay. Check out the FULL CODES here.
import numpy as np
from collections import defaultdict
import json
from datetime import datetime
import pickle
class EpisodicMemory:
def __init__(self, capability=100):
self.capability = capability
self.episodes = []
def retailer(self, state, motion, end result, timestamp=None):
if timestamp is None:
timestamp = datetime.now().isoformat()
episode = {
'state': state,
'motion': motion,
'end result': end result,
'timestamp': timestamp,
'embedding': self._embed(state, motion, end result)
}
self.episodes.append(episode)
if len(self.episodes) > self.capability:
self.episodes.pop(0)
def _embed(self, state, motion, end result):
textual content = f"{state} {motion} {end result}".decrease()
return hash(textual content) % 10000
def retrieve_similar(self, query_state, okay=3):
if not self.episodes:
return []
query_emb = self._embed(query_state, "", "")
scores = [(abs(ep['embedding'] - query_emb), ep) for ep in self.episodes]
scores.kind(key=lambda x: x[0])
return [ep for _, ep in scores[:k]]
def get_recent(self, n=5):
return self.episodes[-n:]
class SemanticMemory:
def __init__(self):
self.preferences = defaultdict(float)
self.patterns = defaultdict(listing)
self.success_rates = defaultdict(lambda: {'success': 0, 'complete': 0})
def update_preference(self, key, worth, weight=1.0):
self.preferences[key] = 0.9 * self.preferences[key] + 0.1 * weight * worth
def record_pattern(self, context, motion, success):
pattern_key = f"{context}_{motion}"
self.patterns[context].append((motion, success))
self.success_rates[pattern_key]['total'] += 1
if success:
self.success_rates[pattern_key]['success'] += 1
def get_best_action(self, context):
if context not in self.patterns:
return None
action_scores = defaultdict(lambda: {'success': 0, 'complete': 0})
for motion, success in self.patterns[context]:
action_scores[action]['total'] += 1
if success:
action_scores[action]['success'] += 1
best_action = max(action_scores.gadgets(), key=lambda x: x[1]['success'] / max(x[1]['total'], 1))
return best_action[0] if best_action[1]['total'] > 0 else None
def get_preference(self, key):
return self.preferences.get(key, 0.0)
We outline the core reminiscence buildings that our agent depends on. We construct episodic reminiscence to seize particular experiences and semantic reminiscence to generalize patterns over time. As we set up these foundations, we put together the agent to be taught from interactions in the identical means people do. Check out the FULL CODES here.
class MemoryAgent:
def __init__(self):
self.episodic_memory = EpisodicMemory(capability=50)
self.semantic_memory = SemanticMemory()
self.current_plan = []
self.session_count = 0
def understand(self, user_input):
user_input = user_input.decrease()
if any(phrase in user_input for phrase in ['recommend', 'suggest', 'what should']):
intent = 'advice'
elif any(phrase in user_input for phrase in ['remember', 'prefer', 'like', 'favorite']):
intent = 'preference_update'
elif any(phrase in user_input for phrase in ['do', 'complete', 'finish', 'task']):
intent = 'task_execution'
else:
intent = 'dialog'
return {'intent': intent, 'uncooked': user_input}
def plan(self, state):
intent = state['intent']
user_input = state['raw']
similar_episodes = self.episodic_memory.retrieve_similar(user_input, okay=3)
plan = []
if intent == 'advice':
genre_prefs = {okay: v for okay, v in self.semantic_memory.preferences.gadgets() if 'genre_' in okay}
if genre_prefs:
best_genre = max(genre_prefs.gadgets(), key=lambda x: x[1])[0]
plan.append(('advocate', best_genre.change('genre_', '')))
else:
plan.append(('advocate', 'normal'))
elif intent == 'preference_update':
genres = ['sci-fi', 'fantasy', 'mystery', 'romance', 'thriller']
detected_genre = subsequent((g for g in genres if g in user_input), None)
if detected_genre:
plan.append(('update_preference', detected_genre))
elif intent == 'task_execution':
best_action = self.semantic_memory.get_best_action('process')
if best_action:
plan.append(('execute', best_action))
else:
plan.append(('execute', 'default'))
self.current_plan = plan
return plan
We assemble the agent’s notion and planning techniques. We course of the person’s enter, detect intent, and create plans by leveraging the recollections shaped earlier. We start shaping how the agent causes and decides its subsequent actions. Check out the FULL CODES here.
def act(self, motion):
action_type, param = motion
if action_type == 'advocate':
if param == 'normal':
return f"Let me be taught your preferences first! What genres do you get pleasure from?"
return f"Based in your preferences, I like to recommend exploring {param}!"
elif action_type == 'update_preference':
self.semantic_memory.update_preference(f'genre_{param}', 1.0, weight=1.0)
return f"Got it! I'll keep in mind you get pleasure from {param}."
elif action_type == 'execute':
return f"Executing process with technique: {param}"
return "Action accomplished"
def revise_plan(self, suggestions):
if 'no' in suggestions.decrease() or 'improper' in suggestions.decrease():
if self.current_plan:
action_type, param = self.current_plan[0]
if action_type == 'advocate':
genre_prefs = sorted(
[(k, v) for k, v in self.semantic_memory.preferences.items() if 'genre_' in k],
key=lambda x: x[1],
reverse=True
)
if len(genre_prefs) > 1:
new_genre = genre_prefs[1][0].change('genre_', '')
self.current_plan = [('recommend', new_genre)]
return True
return False
def replicate(self, state, motion, end result, success):
self.episodic_memory.retailer(state['raw'], str(motion), end result)
self.semantic_memory.record_pattern(state['intent'], str(motion), success)
We outline how the agent executes actions, revises its choices when suggestions contradicts expectations, and displays by storing experiences. We constantly enhance the agent’s behaviour by letting it be taught from each flip. Through this loop, we make the system adaptive and self-correcting. Check out the FULL CODES here.
def run_session(self, user_inputs):
self.session_count += 1
print(f"n{'='*60}")
print(f"SESSION {self.session_count}")
print(f"{'='*60}n")
outcomes = []
for i, user_input in enumerate(user_inputs, 1):
print(f"Turn {i}")
print(f"User: {user_input}")
state = self.understand(user_input)
plan = self.plan(state)
if not plan:
print("Agent: I'm undecided what to do with that.n")
proceed
response = self.act(plan[0])
print(f"Agent: {response}n")
success = 'advocate' in plan[0][0] or 'replace' in plan[0][0]
self.replicate(state, plan[0], response, success)
outcomes.append({
'flip': i,
'enter': user_input,
'intent': state['intent'],
'motion': plan[0],
'response': response
})
return outcomes
We simulate actual interactions through which the agent processes a number of person inputs inside a single session. We watch the understand → plan → act → replicate cycle unfold repeatedly. As we run classes, we see how the agent steadily turns into extra personalised and clever. Check out the FULL CODES here.
def evaluate_memory_usage(agent):
print("n" + "="*60)
print("MEMORY ANALYSIS")
print("="*60 + "n")
print(f"Episodic Memory:")
print(f" Total episodes saved: {len(agent.episodic_memory.episodes)}")
if agent.episodic_memory.episodes:
print(f" Oldest episode: {agent.episodic_memory.episodes[0]['timestamp']}")
print(f" Latest episode: {agent.episodic_memory.episodes[-1]['timestamp']}")
print(f"nSemantic Memory:")
print(f" Learned preferences: {len(agent.semantic_memory.preferences)}")
for pref, worth in sorted(agent.semantic_memory.preferences.gadgets(), key=lambda x: x[1], reverse=True)[:5]:
print(f" {pref}: {worth:.3f}")
print(f"n Action patterns discovered: {len(agent.semantic_memory.patterns)}")
print(f"n Success charges by context-action:")
for key, stats in listing(agent.semantic_memory.success_rates.gadgets())[:5]:
if stats['total'] > 0:
fee = stats['success'] / stats['total']
print(f" {key}: {fee:.2%} ({stats['success']}/{stats['total']})")
def compare_sessions(results_history):
print("n" + "="*60)
print("CROSS-SESSION ANALYSIS")
print("="*60 + "n")
for i, ends in enumerate(results_history, 1):
recommendation_quality = sum(1 for r in outcomes if 'preferences' in r['response'].decrease())
print(f"Session {i}:")
print(f" Turns: {len(outcomes)}")
print(f" Personalized responses: {recommendation_quality}")
We analyse how successfully the agent is utilizing its recollections. We verify saved episodes, discovered preferences, and success patterns to consider how the agent evolves. Check out the FULL CODES here.
def run_demo():
agent = MemoryAgent()
print("n
SCENARIO: Agent learns person preferences over a number of classes")
session1_inputs = [
"Hi, I'm looking for something to read",
"I really like sci-fi books",
"Can you recommend something?",
]
results1 = agent.run_session(session1_inputs)
session2_inputs = [
"I'm bored, what should I read?",
"Actually, I also enjoy fantasy novels",
"Give me a recommendation",
]
results2 = agent.run_session(session2_inputs)
session3_inputs = [
"What do you suggest for tonight?",
"I'm in the mood for mystery too",
"Recommend something based on what you know about me",
]
results3 = agent.run_session(session3_inputs)
evaluate_memory_usage(agent)
compare_sessions([results1, results2, results3])
print("n" + "="*60)
print("EPISODIC MEMORY RETRIEVAL TEST")
print("="*60 + "n")
question = "advocate sci-fi"
related = agent.episodic_memory.retrieve_similar(question, okay=3)
print(f"Query: '{question}'")
print(f"Retrieved {len(related)} related episodes:n")
for ep in related:
print(f" State: {ep['state']}")
print(f" Action: {ep['action']}")
print(f" Outcome: {ep['outcome'][:50]}...")
print()
if __name__ == "__main__":
print("="*60)
print("MEMORY & LONG-TERM AUTONOMY IN AGENTIC SYSTEMS")
print("="*60)
run_demo()
print("n
Tutorial full! Key takeaways:")
print(" • Episodic reminiscence shops particular experiences")
print(" • Semantic reminiscence generalizes patterns")
print(" • Agents enhance suggestions over classes")
print(" • Memory retrieval guides future choices")
We deliver all the things collectively by working a number of classes and testing reminiscence retrieval. We observe the agent enhance throughout interactions and refine suggestions primarily based on gathered data. This complete demo illustrates how long-term autonomy naturally arises from the reminiscence techniques now we have constructed.
In conclusion, we acknowledge how the mixture of episodic and semantic reminiscence permits us to construct brokers that be taught constantly and make more and more higher choices over time. We observe the agent refining suggestions, adapting plans, and retrieving previous experiences to enhance its responses session after session. Through these mechanisms, we see how long-term autonomy emerges from easy but efficient reminiscence buildings.
Check out the FULL CODES here. Feel free to try our GitHub Page for Tutorials, Codes and Notebooks. Also, be happy to observe us on Twitter and don’t overlook to be a part of our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
The submit How to Build Memory-Powered Agentic AI That Learns Continuously Through Episodic Experiences and Semantic Patterns for Long-Term Autonomy appeared first on MarkTechPost.
