Eine Codierungsimplementierung zum Aufbau selbstorganisierender Zettelkasten-Wissensgraphen und Schlafkonsolidierungsmechanismen

!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 Checklist
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"   ⚠️ Surprising Error: {e}")
               return None
   print("   ❌ Max retries reached.")
   return None


print("Enter your Google AI Studio API Key (Enter can 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}")

@dataclass
class MemoryNode:
   id: str
   content material: str
   kind: str
   embedding: Checklist(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):
       consequence = retry_with_backoff(
           genai.embed_content,
           mannequin=EMBEDDING_MODEL,
           content material=textual content
       )
       return consequence('embedding') if consequence else (0.0) * 768

def _atomize_input(self, textual content):
       immediate = f"""
       Break the next textual content into unbiased 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 (textual content)
       besides:
           return (textual content)


   def _find_similar_nodes(self, embedding, top_k=3, threshold=0.45):
       if not self.graph.nodes: return ()
      
       nodes = listing(self.graph.nodes(information=True))
       embeddings = (n(1)('information').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, information=node, title=truth, label=truth(:15)+"...")
          
           if candidates:
               context_str = "n".be part of((f"ID {c(0)}: {self.graph.nodes(c(0))('information').content material}" for c in candidates))
               immediate = f"""
               I'm including: "{truth}"
               Present 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:
                       move
          
           time.sleep(1)

def consolidate_memory(self):
       print(f"n💤 (Consolidation Part) Reflecting...")
       high_degree_nodes = (n for n, d in self.graph.diploma() 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)('information').content material for n in cluster_ids)
          
           immediate = f"""
           Generate a single high-level perception abstract from these information.
           Info: {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, information=insight_node, title=f"INSIGHT: {insight_text}", label="INSIGHT", coloration="#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)('information').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)('information').content material
               relevant_context.add(f"  - linked by way of '{rel}' to: {content material}")
              
       context_text = "n".be part of(relevant_context)
       immediate = f"""
       Reply based mostly 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}")

def show_graph(self):
       attempt:
           internet = Community(pocket book=True, cdn_resources="distant", top="500px", width="100%", bgcolor="#222222", font_color="white")
           for n, information in self.graph.nodes(information=True):
               coloration = "#97c2fc" if information('information').kind == 'truth' else "#ff7f7f"
               internet.add_node(n, label=information.get('label', ''), title=information('information').content material, coloration=coloration)
           for u, v, information in self.graph.edges(information=True):
               internet.add_edge(u, v, label=information.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 mission 'Apollo' goals to construct a dashboard for monitoring photo voltaic panel effectivity.",
   "We selected React for the frontend as a result of the crew is aware of it effectively.",
   "The backend have to be Python to assist the info science libraries.",
   "Shopper referred to as. They're sad with React efficiency on low-end units.",
   "We're switching the frontend to Svelte for higher efficiency."
)


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()