On this tutorial, we dive into the reducing fringe of Agentic AI by constructing a “Zettelkasten” reminiscence system, a “dwelling” structure that organizes info very like the human mind. We transfer past customary retrieval strategies to assemble a dynamic data graph the place an agent autonomously decomposes inputs into atomic information, hyperlinks them semantically, and even “sleeps” to consolidate recollections into higher-order insights. Utilizing Google’s Gemini, we implement a sturdy answer that addresses real-world API constraints, making certain our agent shops knowledge and likewise actively understands the evolving context of our tasks. Try the FULL CODES right here.
!pip set up -q -U google-generativeai networkx pyvis scikit-learn numpy
import os
import json
import uuid
import time
import getpass
import random
import networkx as nx
import numpy as np
import google.generativeai as genai
from dataclasses import dataclass, subject
from typing import Record
from sklearn.metrics.pairwise import cosine_similarity
from IPython.show import show, HTML
from pyvis.community import Community
from google.api_core import exceptions
def retry_with_backoff(func, *args, **kwargs):
max_retries = 5
base_delay = 5
for try in vary(max_retries):
attempt:
return func(*args, **kwargs)
besides exceptions.ResourceExhausted:
wait_time = base_delay * (2 ** try) + random.uniform(0, 1)
print(f" ⏳ Quota restrict hit. Cooling down for {wait_time:.1f}s...")
time.sleep(wait_time)
besides Exception as e:
if "429" in str(e):
wait_time = base_delay * (2 ** try) + random.uniform(0, 1)
print(f" ⏳ Quota restrict hit (HTTP 429). Cooling down for {wait_time:.1f}s...")
time.sleep(wait_time)
else:
print(f" ⚠️ Sudden Error: {e}")
return None
print(" ❌ Max retries reached.")
return None
print("Enter your Google AI Studio API Key (Enter will likely be hidden):")
API_KEY = getpass.getpass()
genai.configure(api_key=API_KEY)
MODEL_NAME = "gemini-2.5-flash"
EMBEDDING_MODEL = "fashions/text-embedding-004"
print(f"✅ API Key configured. Utilizing mannequin: {MODEL_NAME}")We start by importing important libraries for graph administration and AI mannequin interplay, whereas additionally securing our API key enter. Crucially, we outline a sturdy retry_with_backoff perform that routinely handles charge restrict errors, making certain our agent gracefully pauses and recovers when the API quota is exceeded throughout heavy processing. Try the FULL CODES right here.
@dataclass
class MemoryNode:
id: str
content material: str
sort: str
embedding: Record[float] = subject(default_factory=record)
timestamp: int = 0
class RobustZettelkasten:
def __init__(self):
self.graph = nx.Graph()
self.mannequin = genai.GenerativeModel(MODEL_NAME)
self.step_counter = 0
def _get_embedding(self, textual content):
end result = retry_with_backoff(
genai.embed_content,
mannequin=EMBEDDING_MODEL,
content material=textual content
)
return end result['embedding'] if end result else [0.0] * 768We outline the elemental MemoryNode construction to carry our content material, sorts, and vector embeddings in an organized knowledge class. We then initialize the principle RobustZettelkasten class, establishing the community graph and configuring the Gemini embedding mannequin that serves because the spine of our semantic search capabilities. Try the FULL CODES right here.
def _atomize_input(self, textual content):
immediate = f"""
Break the next textual content into impartial atomic information.
Output JSON: {{ "information": ["fact1", "fact2"] }}
Textual content: "{textual content}"
"""
response = retry_with_backoff(
self.mannequin.generate_content,
immediate,
generation_config={"response_mime_type": "utility/json"}
)
attempt:
return json.masses(response.textual content).get("information", []) if response else [text]
besides:
return [text]
def _find_similar_nodes(self, embedding, top_k=3, threshold=0.45):
if not self.graph.nodes: return []
nodes = record(self.graph.nodes(knowledge=True))
embeddings = [n[1]['data'].embedding for n in nodes]
valid_embeddings = [e for e in embeddings if len(e) > 0]
if not valid_embeddings: return []
sims = cosine_similarity([embedding], embeddings)[0]
sorted_indices = np.argsort(sims)[::-1]
outcomes = []
for idx in sorted_indices[:top_k]:
if sims[idx] > threshold:
outcomes.append((nodes[idx][0], sims[idx]))
return outcomes
def add_memory(self, user_input):
self.step_counter += 1
print(f"n🧠 [Step {self.step_counter}] Processing: "{user_input}"")
information = self._atomize_input(user_input)
for truth in information:
print(f" -> Atom: {truth}")
emb = self._get_embedding(truth)
candidates = self._find_similar_nodes(emb)
node_id = str(uuid.uuid4())[:6]
node = MemoryNode(id=node_id, content material=truth, sort="truth", embedding=emb, timestamp=self.step_counter)
self.graph.add_node(node_id, knowledge=node, title=truth, label=truth[:15]+"...")
if candidates:
context_str = "n".be a part of([f"ID {c[0]}: {self.graph.nodes[c[0]]['data'].content material}" for c in candidates])
immediate = f"""
I'm including: "{truth}"
Current Reminiscence:
{context_str}
Are any of those straight associated? If sure, present the connection label.
JSON: {{ "hyperlinks": [{{ "target_id": "ID", "rel": "label" }}] }}
"""
response = retry_with_backoff(
self.mannequin.generate_content,
immediate,
generation_config={"response_mime_type": "utility/json"}
)
if response:
attempt:
hyperlinks = json.masses(response.textual content).get("hyperlinks", [])
for hyperlink in hyperlinks:
if self.graph.has_node(hyperlink['target_id']):
self.graph.add_edge(node_id, hyperlink['target_id'], label=hyperlink['rel'])
print(f" 🔗 Linked to {hyperlink['target_id']} ({hyperlink['rel']})")
besides:
go
time.sleep(1)We assemble an ingestion pipeline that decomposes advanced consumer inputs into atomic information to forestall info loss. We instantly embed these information and use our agent to determine and create semantic hyperlinks to present nodes, successfully constructing a data graph in actual time that mimics associative reminiscence. Try the FULL CODES right here.
def consolidate_memory(self):
print(f"n💤 [Consolidation Phase] Reflecting...")
high_degree_nodes = [n for n, d in self.graph.degree() if d >= 2]
processed_clusters = set()
for main_node in high_degree_nodes:
neighbors = record(self.graph.neighbors(main_node))
cluster_ids = tuple(sorted([main_node] + neighbors))
if cluster_ids in processed_clusters: proceed
processed_clusters.add(cluster_ids)
cluster_content = [self.graph.nodes[n]['data'].content material for n in cluster_ids]
immediate = f"""
Generate a single high-level perception abstract from these information.
Details: {json.dumps(cluster_content)}
JSON: {{ "perception": "Your perception right here" }}
"""
response = retry_with_backoff(
self.mannequin.generate_content,
immediate,
generation_config={"response_mime_type": "utility/json"}
)
if response:
attempt:
insight_text = json.masses(response.textual content).get("perception")
if insight_text:
insight_id = f"INSIGHT-{uuid.uuid4().hex[:4]}"
print(f" ✨ Perception: {insight_text}")
emb = self._get_embedding(insight_text)
insight_node = MemoryNode(id=insight_id, content material=insight_text, sort="perception", embedding=emb)
self.graph.add_node(insight_id, knowledge=insight_node, title=f"INSIGHT: {insight_text}", label="INSIGHT", colour="#ff7f7f")
self.graph.add_edge(insight_id, main_node, label="abstracted_from")
besides:
proceed
time.sleep(1)
def answer_query(self, question):
print(f"n🔍 Querying: "{question}"")
emb = self._get_embedding(question)
candidates = self._find_similar_nodes(emb, top_k=2)
if not candidates:
print("No related reminiscence discovered.")
return
relevant_context = set()
for node_id, rating in candidates:
node_content = self.graph.nodes[node_id]['data'].content material
relevant_context.add(f"- {node_content} (Direct Match)")
for n1 in self.graph.neighbors(node_id):
rel = self.graph[node_id][n1].get('label', 'associated')
content material = self.graph.nodes[n1]['data'].content material
relevant_context.add(f" - linked by way of '{rel}' to: {content material}")
context_text = "n".be a part of(relevant_context)
immediate = f"""
Reply primarily based ONLY on context.
Query: {question}
Context:
{context_text}
"""
response = retry_with_backoff(self.mannequin.generate_content, immediate)
if response:
print(f"🤖 Agent Reply:n{response.textual content}")We implement the cognitive features of our agent, enabling it to “sleep” and consolidate dense reminiscence clusters into higher-order insights. We additionally outline the question logic that traverses these linked paths, permitting the agent to motive throughout a number of hops within the graph to reply advanced questions. Try the FULL CODES right here.
def show_graph(self):
attempt:
internet = Community(pocket book=True, cdn_resources="distant", top="500px", width="100%", bgcolor="#222222", font_color="white")
for n, knowledge in self.graph.nodes(knowledge=True):
colour = "#97c2fc" if knowledge['data'].sort == 'truth' else "#ff7f7f"
internet.add_node(n, label=knowledge.get('label', ''), title=knowledge['data'].content material, colour=colour)
for u, v, knowledge in self.graph.edges(knowledge=True):
internet.add_edge(u, v, label=knowledge.get('label', ''))
internet.present("memory_graph.html")
show(HTML("memory_graph.html"))
besides Exception as e:
print(f"Graph visualization error: {e}")
mind = RobustZettelkasten()
occasions = [
"The project 'Apollo' aims to build a dashboard for tracking solar panel efficiency.",
"We chose React for the frontend because the team knows it well.",
"The backend must be Python to support the data science libraries.",
"Client called. They are unhappy with React performance on low-end devices.",
"We are switching the frontend to Svelte for better performance."
]
print("--- PHASE 1: INGESTION ---")
for occasion in occasions:
mind.add_memory(occasion)
time.sleep(2)
print("--- PHASE 2: CONSOLIDATION ---")
mind.consolidate_memory()
print("--- PHASE 3: RETRIEVAL ---")
mind.answer_query("What's the present frontend know-how for Apollo and why?")
print("--- PHASE 4: VISUALIZATION ---")
mind.show_graph()We wrap up by including a visualization methodology that generates an interactive HTML graph of our agent’s reminiscence, permitting us to examine the nodes and edges. Lastly, we execute a take a look at state of affairs involving a challenge timeline to confirm that our system accurately hyperlinks ideas, generates insights, and retrieves the appropriate context.
In conclusion, we now have a completely purposeful “Residing Reminiscence” prototype that transcends easy database storage. By enabling our agent to actively hyperlink associated ideas and replicate on its experiences throughout a “consolidation” part, we resolve the essential drawback of fragmented context in long-running AI interactions. This technique demonstrates that true intelligence requires processing energy and a structured, evolving reminiscence, marking the way in which for us to construct extra succesful, personalised autonomous brokers.
Try the FULL CODES right here. Additionally, be at liberty to observe us on Twitter and don’t overlook to hitch our 100k+ ML SubReddit and Subscribe to our E-newsletter. Wait! are you on telegram? now you may be a part of us on telegram as nicely.
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 reputation amongst audiences.
Elevate your perspective with NextTech Information, the place innovation meets perception.
Uncover the most recent breakthroughs, get unique updates, and join with a world community of future-focused thinkers.
Unlock tomorrow’s tendencies at the moment: learn extra, subscribe to our e-newsletter, and grow to be a part of the NextTech neighborhood at NextTech-news.com
