On this tutorial, we discover tips on how to construct agentic techniques that assume 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 periods. As we implement planning, appearing, revising, and reflecting, we see how the agent step by step adapts to person preferences and turns into extra autonomous. By the top, we perceive how memory-driven reasoning helps us create brokers that really feel extra contextual, constant, and clever with each interplay. Try 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, consequence, timestamp=None):
if timestamp is None:
timestamp = datetime.now().isoformat()
episode = {
'state': state,
'motion': motion,
'consequence': consequence,
'timestamp': timestamp,
'embedding': self._embed(state, motion, consequence)
}
self.episodes.append(episode)
if len(self.episodes) > self.capability:
self.episodes.pop(0)
def _embed(self, state, motion, consequence):
textual content = f"{state} {motion} {consequence}".decrease()
return hash(textual content) % 10000
def retrieve_similar(self, query_state, ok=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.type(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.objects(), 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 method people do. Try 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, ok=3)
plan = []
if intent == 'advice':
genre_prefs = {ok: v for ok, v in self.semantic_memory.preferences.objects() if 'genre_' in ok}
if genre_prefs:
best_genre = max(genre_prefs.objects(), key=lambda x: x[1])[0]
plan.append(('suggest', best_genre.substitute('genre_', '')))
else:
plan.append(('suggest', 'common'))
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('activity')
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. Try the FULL CODES here.
def act(self, motion):
action_type, param = motion
if action_type == 'suggest':
if param == 'common':
return f"Let me be taught your preferences first! What genres do you get pleasure from?"
return f"Based mostly 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"Received it! I will keep in mind you get pleasure from {param}."
elif action_type == 'execute':
return f"Executing activity with technique: {param}"
return "Motion accomplished"
def revise_plan(self, suggestions):
if 'no' in suggestions.decrease() or 'fallacious' in suggestions.decrease():
if self.current_plan:
action_type, param = self.current_plan[0]
if action_type == 'suggest':
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].substitute('genre_', '')
self.current_plan = [('recommend', new_genre)]
return True
return False
def replicate(self, state, motion, consequence, success):
self.episodic_memory.retailer(state['raw'], str(motion), consequence)
self.semantic_memory.record_pattern(state['intent'], str(motion), success)
We outline how the agent executes actions, revises its selections when suggestions contradicts expectations, and displays by storing experiences. We constantly enhance the agent’s behaviour by letting it be taught from each flip. By this loop, we make the system adaptive and self-correcting. Try 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"Flip {i}")
print(f"Consumer: {user_input}")
state = self.understand(user_input)
plan = self.plan(state)
if not plan:
print("Agent: I am unsure what to do with that.n")
proceed
response = self.act(plan[0])
print(f"Agent: {response}n")
success="suggest" 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 periods, we see how the agent step by step turns into extra personalised and clever. Try the FULL CODES here.
def evaluate_memory_usage(agent):
print("n" + "="*60)
print("MEMORY ANALYSIS")
print("="*60 + "n")
print(f"Episodic Reminiscence:")
print(f" Complete episodes saved: {len(agent.episodic_memory.episodes)}")
if agent.episodic_memory.episodes:
print(f" Oldest episode: {agent.episodic_memory.episodes[0]['timestamp']}")
print(f" Newest episode: {agent.episodic_memory.episodes[-1]['timestamp']}")
print(f"nSemantic Reminiscence:")
print(f" Realized preferences: {len(agent.semantic_memory.preferences)}")
for pref, worth in sorted(agent.semantic_memory.preferences.objects(), key=lambda x: x[1], reverse=True)[:5]:
print(f" {pref}: {worth:.3f}")
print(f"n Motion 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.objects())[: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, leads to 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" Customized responses: {recommendation_quality}")
We analyse how successfully the agent is utilizing its recollections. We verify saved episodes, discovered preferences, and success patterns to guage how the agent evolves. Try the FULL CODES here.
def run_demo():
agent = MemoryAgent()
print("n📚 SCENARIO: Agent learns person preferences over a number of periods")
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 = "suggest sci-fi"
comparable = agent.episodic_memory.retrieve_similar(question, ok=3)
print(f"Question: '{question}'")
print(f"Retrieved {len(comparable)} comparable episodes:n")
for ep in comparable:
print(f" State: {ep['state']}")
print(f" Motion: {ep['action']}")
print(f" Consequence: {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(" • Brokers enhance suggestions over periods")
print(" • Reminiscence retrieval guides future selections")
We deliver the whole lot collectively by working a number of periods and testing reminiscence retrieval. We observe the agent enhance throughout interactions and refine suggestions based mostly on gathered information. This complete demo illustrates how long-term autonomy naturally arises from the reminiscence techniques we’ve 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 selections over time. We observe the agent refining suggestions, adapting plans, and retrieving previous experiences to enhance its responses session after session. By these mechanisms, we see how long-term autonomy emerges from easy but efficient reminiscence buildings.
Try the FULL CODES here. Be happy to take a look at our GitHub Page for Tutorials, Codes and Notebooks. Additionally, be happy to observe us on Twitter and don’t overlook to affix our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Asif Razzaq is the CEO of Marktechpost Media Inc.. As a visionary entrepreneur and engineer, Asif is dedicated to harnessing the potential of Synthetic Intelligence for social good. His most up-to-date endeavor is the launch of an Synthetic Intelligence Media Platform, Marktechpost, which stands out for its in-depth protection of machine studying and deep studying information that’s each technically sound and simply comprehensible by a large viewers. The platform boasts of over 2 million month-to-month views, illustrating its recognition amongst audiences.