Una implementación de codificación para la creación de gráficos de conocimiento autoorganizados de Zettelkasten y mecanismos de consolidación del sueño

!pip install -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, field
from typing import List
from sklearn.metrics.pairwise import cosine_similarity
from IPython.display import display, HTML
from pyvis.network import Network
from google.api_core import exceptions


def retry_with_backoff(func, *args, **kwargs):
   max_retries = 5
   base_delay = 5
  
   for attempt in range(max_retries):
       try:
           return func(*args, **kwargs)
       except exceptions.ResourceExhausted:
           wait_time = base_delay * (2 ** attempt) + random.uniform(0, 1)
           print(f"   ⏳ Quota limit hit. Cooling down for {wait_time:.1f}s...")
           time.sleep(wait_time)
       except Exception as e:
           if "429" in str(e):
               wait_time = base_delay * (2 ** attempt) + random.uniform(0, 1)
               print(f"   ⏳ Quota limit hit (HTTP 429). Cooling down for {wait_time:.1f}s...")
               time.sleep(wait_time)
           else:
               print(f"   ⚠️ Unexpected Error: {e}")
               return None
   print("   ❌ Max retries reached.")
   return None


print("Enter your Google AI Studio API Key (Input will be hidden):")
API_KEY = getpass.getpass()


genai.configure(api_key=API_KEY)
MODEL_NAME = "gemini-2.5-flash" 
EMBEDDING_MODEL = "models/text-embedding-004"


print(f"✅ API Key configured. Using model: {MODEL_NAME}")

@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

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)

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)


   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}")

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}")


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