On this tutorial, we dive into the slicing fringe of Agentic AI by constructing a “Zettelkasten” reminiscence system, a “dwelling” structure that organizes info very like the human mind. We transfer past normal 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 resolution that addresses real-world API constraints, guaranteeing our agent shops knowledge and in addition actively understands the evolving context of our initiatives. Take a look at the FULL CODES 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, discipline
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 probably 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 operate that mechanically handles price restrict errors, guaranteeing our agent gracefully pauses and recovers when the API quota is exceeded throughout heavy processing. Take a look at the FULL CODES here.
@dataclass
class MemoryNode:
id: str
content material: str
kind: str
embedding: Record[float] = discipline(default_factory=listing)
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 basic MemoryNode construction to carry our content material, sorts, and vector embeddings in an organized knowledge class. We then initialize the primary RobustZettelkasten class, establishing the community graph and configuring the Gemini embedding mannequin that serves because the spine of our semantic search capabilities. Take a look at the FULL CODES 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": "software/json"}
)
attempt:
return json.hundreds(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 = listing(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, kind="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 instantly 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": "software/json"}
)
if response:
attempt:
hyperlinks = json.hundreds(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 complicated person inputs into atomic information to forestall info loss. We instantly embed these information and use our agent to establish and create semantic hyperlinks to present nodes, successfully constructing a data graph in actual time that mimics associative reminiscence. Take a look at the FULL CODES 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 = listing(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.
Information: {json.dumps(cluster_content)}
JSON: {{ "perception": "Your perception right here" }}
"""
response = retry_with_backoff(
self.mannequin.generate_content,
immediate,
generation_config={"response_mime_type": "software/json"}
)
if response:
attempt:
insight_text = json.hundreds(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, kind="perception", embedding=emb)
self.graph.add_node(insight_id, knowledge=insight_node, title=f"INSIGHT: {insight_text}", label="INSIGHT", shade="#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 through '{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 capabilities 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 related paths, permitting the agent to motive throughout a number of hops within the graph to reply complicated questions. Take a look at the FULL CODES 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):
shade = "#97c2fc" if knowledge['data'].kind == 'truth' else "#ff7f7f"
internet.add_node(n, label=knowledge.get('label', ''), title=knowledge['data'].content material, shade=shade)
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 technique that generates an interactive HTML graph of our agent’s reminiscence, permitting us to examine the nodes and edges. Lastly, we execute a check state of affairs involving a mission timeline to confirm that our system accurately hyperlinks ideas, generates insights, and retrieves the correct context.
In conclusion, we now have a completely purposeful “Dwelling Reminiscence” prototype that transcends easy database storage. By enabling our agent to actively hyperlink associated ideas and replicate on its experiences throughout a “consolidation” section, we remedy the vital 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.
Take a look at the FULL CODES here. Additionally, be at liberty 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 reputation amongst audiences.
