A Coding Implementation on Building Self-Organizing Zettelkasten Knowledge Graphs and Sleep-Consolidation Mechanisms
Why It Matters
By giving autonomous agents a dynamic, graph‑based memory that can consolidate and retrieve insights, the approach mitigates fragmented context and enables more coherent, long‑term interactions, a critical step toward truly intelligent assistants.
A Coding Implementation on Building Self-Organizing Zettelkasten Knowledge Graphs and Sleep-Consolidation Mechanisms
By Asif Razzaq
In this tutorial, we dive into the cutting edge of Agentic AI by building a “Zettelkasten” memory system, a “living” architecture that organizes information much like the human brain. We move beyond standard retrieval methods to construct a dynamic knowledge graph where an agent autonomously decomposes inputs into atomic facts, links them semantically, and even “sleeps” to consolidate memories into higher‑order insights. Using Google’s Gemini, we implement a robust solution that addresses real‑world API constraints, ensuring our agent stores data and also actively understands the evolving context of our projects.
Full code: https://github.com/Marktechpost/AI-Tutorial-Codes-Included/blob/main/Agentic%20AI%20Memory/Agentic_Zettelkasten_Memory_Martechpost.ipynb
1. Setup and Core Classes
@dataclass
class MemoryNode:
id: str
content: str
type: str
embedding: List[float] = field(default_factory=list)
timestamp: int = 0
class RobustZettelkasten:
def __init__(self):
self.graph = nx.Graph()
self.model = genai.GenerativeModel(MODEL_NAME)
self.step_counter = 0
def _get_embedding(self, text):
result = retry_with_backoff(
genai.embed_content,
model=EMBEDDING_MODEL,
content=text
)
return result['embedding'] if result else [0.0] * 768
We define the fundamental MemoryNode structure to hold our content, types, and vector embeddings in an organized data class. The RobustZettelkasten class creates the network graph and configures the Gemini embedding model that powers our semantic search.
2. Ingestion Pipeline
def _atomize_input(self, text):
prompt = f"""
Break the following text into independent atomic facts.
Output JSON: {{ "facts": ["fact1", "fact2"] }}
Text: "{text}"
"""
response = retry_with_backoff(
self.model.generate_content,
prompt,
generation_config={"response_mime_type": "application/json"}
)
try:
return json.loads(response.text).get("facts", []) if response else [text]
except:
return [text]
def _find_similar_nodes(self, embedding, top_k=3, threshold=0.45):
if not self.graph.nodes: return []
nodes = list(self.graph.nodes(data=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]
results = []
for idx in sorted_indices[:top_k]:
if sims[idx] > threshold:
results.append((nodes[idx][0], sims[idx]))
return results
def add_memory(self, user_input):
self.step_counter += 1
print(f"\n [Step {self.step_counter}] Processing: \"{user_input}\"")
facts = self._atomize_input(user_input)
for fact in facts:
print(f" -> Atom: {fact}")
emb = self._get_embedding(fact)
candidates = self._find_similar_nodes(emb)
node_id = str(uuid.uuid4())[:6]
node = MemoryNode(id=node_id, content=fact, type='fact', embedding=emb, timestamp=self.step_counter)
self.graph.add_node(node_id, data=node, title=fact, label=fact[:15]+"...")
if candidates:
context_str = "\n".join([f"ID {c[0]}: {self.graph.nodes[c[0]]['data'].content}" for c in candidates])
prompt = f"""
I am adding: "{fact}"
Existing Memory:
{context_str}
Are any of these directly related? If yes, provide the relationship label.
JSON: {{ "links": [{{ "target_id": "ID", "rel": "label" }}] }}
"""
response = retry_with_backoff(
self.model.generate_content,
prompt,
generation_config={"response_mime_type": "application/json"}
)
if response:
try:
links = json.loads(response.text).get("links", [])
for link in links:
if self.graph.has_node(link['target_id']):
self.graph.add_edge(node_id, link['target_id'], label=link['rel'])
print(f" Linked to {link['target_id']} ({link['rel']})")
except:
pass
time.sleep(1)
The ingestion pipeline breaks complex inputs into atomic facts, embeds them, and creates semantic links to existing nodes, building a knowledge graph in real time.
3. Consolidation (Sleep) Phase
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 = list(self.graph.neighbors(main_node))
cluster_ids = tuple(sorted([main_node] + neighbors))
if cluster_ids in processed_clusters: continue
processed_clusters.add(cluster_ids)
cluster_content = [self.graph.nodes[n]['data'].content for n in cluster_ids]
prompt = f"""
Generate a single high-level insight summary from these facts.
Facts: {json.dumps(cluster_content)}
JSON: {{ "insight": "Your insight here" }}
"""
response = retry_with_backoff(
self.model.generate_content,
prompt,
generation_config={"response_mime_type": "application/json"}
)
if response:
try:
insight_text = json.loads(response.text).get("insight")
if insight_text:
insight_id = f"INSIGHT-{uuid.uuid4().hex[:4]}"
print(f" Insight: {insight_text}")
emb = self._get_embedding(insight_text)
insight_node = MemoryNode(id=insight_id, content=insight_text, type='insight', embedding=emb)
self.graph.add_node(insight_id, data=insight_node, title=f"INSIGHT: {insight_text}", label="INSIGHT", color="#ff7f7f")
self.graph.add_edge(insight_id, main_node, label="abstracted_from")
except:
continue
time.sleep(1)
During consolidation, the agent abstracts high‑level insights from densely connected clusters, adding “insight” nodes that capture emergent knowledge.
4. Query Answering
def answer_query(self, query):
print(f"\n Querying: \"{query}\"")
emb = self._get_embedding(query)
candidates = self._find_similar_nodes(emb, top_k=2)
if not candidates:
print("No relevant memory found.")
return
relevant_context = set()
for node_id, score in candidates:
node_content = self.graph.nodes[node_id]['data'].content
relevant_context.add(f"- {node_content} (Direct Match)")
for n1 in self.graph.neighbors(node_id):
rel = self.graph[node_id][n1].get('label', 'related')
content = self.graph.nodes[n1]['data'].content
relevant_context.add(f" - linked via '{rel}' to: {content}")
context_text = "\n".join(relevant_context)
prompt = f"""
Answer based ONLY on context.
Question: {query}
Context:
{context_text}
"""
response = retry_with_backoff(self.model.generate_content, prompt)
if response:
print(f" Agent Answer:\n{response.text}")
The query function retrieves the most relevant facts and their linked nodes, then asks the LLM to answer using only that context.
5. Visualization
def show_graph(self):
try:
net = Network(notebook=True, cdn_resources='remote', height="500px", width="100%", bgcolor='#222222', font_color='white')
for n, data in self.graph.nodes(data=True):
color = "#97c2fc" if data['data'].type == 'fact' else "#ff7f7f"
net.add_node(n, label=data.get('label', ''), title=data['data'].content, color=color)
for u, v, data in self.graph.edges(data=True):
net.add_edge(u, v, label=data.get('label', ''))
net.show("memory_graph.html")
display(HTML("memory_graph.html"))
except Exception as e:
print(f"Graph visualization error: {e}")
An interactive HTML graph lets us inspect nodes, edges, and their relationships.
6. End‑to‑End Demo
brain = RobustZettelkasten()
events = [
"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 event in events:
brain.add_memory(event)
time.sleep(2)
print("--- PHASE 2: CONSOLIDATION ---")
brain.consolidate_memory()
print("--- PHASE 3: RETRIEVAL ---")
brain.answer_query("What is the current frontend technology for Apollo and why?")
print("--- PHASE 4: VISUALIZATION ---")
brain.show_graph()
The demo ingests a series of project updates, consolidates the graph into insights, answers a specific query, and visualizes the resulting memory structure.
Conclusion
We now have a fully functional “Living Memory” prototype that goes beyond simple database storage. By enabling the agent to actively link related concepts and reflect during a consolidation phase, we solve the problem of fragmented context in long‑running AI interactions. This system demonstrates that true intelligence requires both processing power and a structured, evolving memory, paving the way for more capable, personalized autonomous agents.
About the Author
Asif Razzaq – CEO of Marktechpost Media Inc. A visionary entrepreneur and engineer committed to harnessing AI for social good. He leads the Artificial Intelligence Media Platform, Marktechpost, known for in‑depth, technically sound coverage of machine learning and deep learning news. The platform attracts over 2 million monthly views.
Comments
Want to join the conversation?
Loading comments...