En este tutorial, nos sumergimos en la vanguardia de la IA agente mediante la construcción de un sistema de memoria “Zettelkasten”, una arquitectura “viva” que organiza la información de manera muy similar al cerebro humano. Vamos más allá de los métodos de recuperación estándar para construir un gráfico de conocimiento dinámico en el que un agente descompone de forma autónoma las entradas en hechos atómicos, los vincula semánticamente e incluso “duerme” para consolidar los recuerdos en conocimientos de orden superior. Al utilizar Gemini de Google, implementamos una solución sólida que aborda las limitaciones de API del mundo real, garantizando que nuestro agente almacene datos y también comprenda activamente el contexto en evolución de nuestros proyectos. Consulta los CÓDIGOS COMPLETOS aquí.
Comenzamos importando bibliotecas esenciales para la gestión de gráficos y la interacción del modelo de IA, al mismo tiempo que protegemos la entrada de nuestra clave API. Fundamentalmente, definimos una función retry_with_backoff sólida que maneja automáticamente los errores de límite de velocidad, lo que garantiza que nuestro agente se detenga y se recupere correctamente cuando se excede la cuota de API durante un procesamiento intensivo. Consulta los CÓDIGOS COMPLETOS aquí.
Definimos la estructura fundamental de MemoryNode para contener nuestro contenido, tipos e incrustaciones de vectores en una clase de datos organizada. Luego inicializamos la clase principal de RobustZettelkasten, establecimos el gráfico de red y configuramos el modelo de incrustación de Gemini que sirve como columna vertebral de nuestras capacidades de búsqueda semántica. Consulta los CÓDIGOS COMPLETOS aquí.
excepto: volver [text]
def _find_similar_nodes(self, incrustación, top_k=3, umbral=0.45): si no self.graph.nodes: regresar []
nodos = lista (self.graph.nodes (datos = Verdadero)) incrustaciones = [n[1][‘data’].embedding para n en nodos]valid_embeddings = [e for e in embeddings if len(e) > 0]
si no valid_embeddings: regresar []
sims = coseno_similitud([embedding]incrustaciones)[0]
índices_ordenados = np.argsort(sims)[::-1]
resultados = []
para idx en índices_ordenados[:top_k]: si sims[idx] > umbral: resultados.append((nodos[idx][0]sims[idx])) devuelve resultados def add_memory(self, user_input): self.step_counter += 1 print(f”\n🧠 [Step {self.step_counter}] Procesamiento: \”{user_input}\””) hechos = self._atomize_input(user_input) para hecho en hechos: print(f” -> Atom: {fact}”) emb = self._get_embedding(fact) candidatos = self._find_similar_nodes(emb) node_id = str(uuid.uuid4())[:6]
nodo = MemoryNode(id=nodo_id, contenido=hecho, tipo=”hecho”, incrustación=emb, marca de tiempo=self.step_counter) self.graph.add_node(nodo_id, datos=nodo, título=hecho, etiqueta=hecho[:15]+”…”) si candidatos: context_str = “\n”.join([f”ID {c[0]}: {self.graph.nodos[c[0]][‘data’].content}” para c en candidatos]) solicitud = f””” Estoy agregando: “{fact}” Memoria existente: {context_str} ¿Alguno de estos está directamente relacionado? En caso afirmativo, proporcione la etiqueta de relación. JSON: {{ “enlaces”: [{{ “target_id”: “ID”, “rel”: “label” }}] }} “”” respuesta = retry_with_backoff( self.model.generate_content, solicitud, generacion_config={“response_mime_type”: “application/json”} ) si respuesta: intente: links = json.loads(response.text).get(“links”, []) para enlace en enlaces: if self.graph.has_node(enlace[‘target_id’]): self.graph.add_edge(nodo_id, enlace[‘target_id’]etiqueta = enlace[‘rel’]) print(f” 🔗 Vinculado a {enlace[‘target_id’]} ({enlace[‘rel’]})”) excepto: pasar tiempo.dormir(1)
Construimos una canalización de ingesta que descompone las entradas complejas del usuario en hechos atómicos para evitar la pérdida de información. Inmediatamente incorporamos estos hechos y utilizamos nuestro agente para identificar y crear enlaces semánticos a nodos existentes, construyendo efectivamente un gráfico de conocimiento en tiempo real que imita la memoria asociativa. Consulta los CÓDIGOS COMPLETOS aquí.
clusters_procesados = set() para nodo_principal en nodos_de_alto grado: vecinos = lista(self.graph.neighbors(nodo_principal)) cluster_ids = tupla(ordenado([main_node] + vecinos)) si cluster_ids en clusters_procesados: continuar con clusters_procesados.add(cluster_ids) cluster_content = [self.graph.nodes[n][‘data’].content for n in cluster_ids]Prompt = f””” Genere un único resumen de información de alto nivel a partir de estos hechos. Hechos: {json.dumps(cluster_content)} JSON: {{ “insight”: “Su información aquí” }} “”” respuesta = retry_with_backoff( self.model.generate_content, Prompt, generate_config={“response_mime_type”: “application/json”} ) si respuesta: Pruebe: insight_text = json.loads(response.text).get(“insight”) si 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”) excepto: continuar time.sleep(1) def respuesta_query(self, query): print(f”\n🔍 Consultando: \”{query}\””) emb = self._get_embedding(query) candidatos = self._find_similar_nodes(emb, top_k=2) si no son candidatos: print(“No se encontró memoria relevante.”) devuelve contexto_relevante = set() para node_id, puntuación en candidatos: node_content = self.graph.nodes[node_id][‘data’].content relevante_context.add(f”- {node_content} (Coincidencia directa)”) para n1 en self.graph.neighbors(node_id): rel = self.graph[node_id][n1].get(‘label’, ‘relacionado’) contenido = self.graph.nodes[n1][‘data’].content relevante_context.add(f” – vinculado a través de ‘{rel}’ a: {content}”) context_text = “\n”.join(relevant_context) solicitud = f””” Respuesta basada SÓLO en el contexto. Pregunta: {query} Contexto: {context_text} “”” respuesta = retry_with_backoff(self.model.generate_content, solicitud) si respuesta: print(f”🤖 Agente Respuesta:\n{respuesta.texto}”)
Implementamos las funciones cognitivas de nuestro agente, permitiéndole “dormir” y consolidar densos grupos de memoria en conocimientos de orden superior. También definimos la lógica de consulta que atraviesa estas rutas conectadas, lo que permite al agente razonar a través de múltiples saltos en el gráfico para responder preguntas complejas. Consulta los CÓDIGOS COMPLETOS aquí.
“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(“— FASE 1: INGESTIÓN —“) para evento en eventos: Brain.add_memory(event) time.sleep(2) print(“— FASE 2: CONSOLIDACIÓN —“) Brain.consolidate_memory() print(“— FASE 3: RECUPERACIÓN —“) Brain.answer_query(“¿Cuál es la tecnología frontend actual de Apollo y por qué?”) print(“— FASE 4: VISUALIZACIÓN —“) Brain.show_graph()
Concluimos agregando un método de visualización que genera un gráfico HTML interactivo de la memoria de nuestro agente, lo que nos permite inspeccionar los nodos y bordes. Finalmente, ejecutamos un escenario de prueba que involucra una línea de tiempo del proyecto para verificar que nuestro sistema vincula correctamente conceptos, genera conocimientos y recupera el contexto correcto.
En conclusión, ahora tenemos un prototipo de “memoria viva” completamente funcional que trasciende el simple almacenamiento de una base de datos. Al permitir que nuestro agente vincule activamente conceptos relacionados y reflexione sobre sus experiencias durante una fase de “consolidación”, solucionamos el problema crítico del contexto fragmentado en interacciones de IA de larga duración. Este sistema demuestra que la verdadera inteligencia requiere potencia de procesamiento y una memoria estructurada y en evolución, lo que nos marca el camino para construir agentes autónomos más capaces y personalizados.
Consulta los CÓDIGOS COMPLETOS aquí. Además, no dude en seguirnos en Twitter y no olvide unirse a nuestro SubReddit de más de 100.000 ML y suscribirse a nuestro boletín. ¡Esperar! estas en telegrama? Ahora también puedes unirte a nosotros en Telegram.
Asif Razzaq es el director ejecutivo de Marktechpost Media Inc.. Como empresario e ingeniero visionario, Asif está comprometido a aprovechar el potencial de la inteligencia artificial para el bien social. Su esfuerzo más reciente es el lanzamiento de una plataforma de medios de inteligencia artificial, Marktechpost, que se destaca por su cobertura en profundidad del aprendizaje automático y las noticias sobre aprendizaje profundo que es técnicamente sólida y fácilmente comprensible para una amplia audiencia. La plataforma cuenta con más de 2 millones de visitas mensuales, lo que ilustra su popularidad entre el público.