Cómo construir un sistema de memoria de agente autoorganizado para el razonamiento de IA a largo plazo

En este tutorial, construimos un sistema de memoria autoorganizado para un agente que va más allá de almacenar el historial de conversaciones sin procesar y, en cambio, estructura las interacciones en unidades de conocimiento persistentes y significativas. Diseñamos el sistema para que el razonamiento y la gestión de la memoria estén claramente separados, permitiendo que un componente dedicado extraiga, comprima y organice la información. Al mismo tiempo, el agente principal se centra en responder al usuario. Usamos almacenamiento estructurado con SQLite, agrupación basada en escenas y consolidación de resumen, y mostramos cómo un agente puede mantener un contexto útil a lo largo de horizontes prolongados sin depender de una recuperación opaca de solo vectores.

importar sqlite3 importar json importar re desde fecha y hora importar fecha y hora desde escribir importar lista, dictar desde getpass importar getpass desde openai importar OpenAI OPENAI_API_KEY = getpass(“Ingrese su clave API de OpenAI: “).strip() client = OpenAI(api_key=OPENAI_API_KEY) def llm(prompt, Temperature=0.1, max_tokens=500): return client.chat.completions.create( modelo=”gpt-4o-mini”, mensajes=[{“role”: “user”, “content”: prompt}]temperatura=temperatura, max_tokens=max_tokens ).opciones[0].mensaje.content.strip()

Configuramos el tiempo de ejecución principal importando todas las bibliotecas necesarias y recopilando de forma segura la clave API en el momento de la ejecución. Inicializamos el cliente del modelo de lenguaje y definimos una única función auxiliar que estandariza todas las llamadas del modelo. Nos aseguramos de que todos los componentes posteriores dependan de esta interfaz compartida para un comportamiento de generación consistente.

clase MemoryDB: def __init__(self): self.db = sqlite3.connect(“:memory:”) self.db.row_factory = sqlite3.Row self._init_schema() def _init_schema(self): self.db.execute(“”” CREAR TABLA mem_cells (id INTEGER PRIMARY KEY, escena TEXTO, cell_type TEXTO, prominencia REAL, contenido TEXTO, creado_en TEXTO ) “””) self.db.execute(“”” CREAR TABLA mem_scenes ( escena TEXTO CLAVE PRIMARIA, TEXTO resumen, actualizado_en TEXTO ) “””) self.db.execute(“”” CREAR TABLA VIRTUAL mem_cells_fts USANDO fts5(contenido, escena, tipo_celda) “””) def insert_cell(self, celda): self.db.execute( “INSERT INTO mem_cells VALUES(NULL,?,?,?,?,?)”, (celda[“scene”]celúla[“cell_type”]celúla[“salience”]json.dumps(celda[“content”]), datetime.utcnow().isoformat() ) ) self.db.execute( “INSERT INTO mem_cells_fts VALUES(?,?,?)”, ( json.dumps(cell[“content”]), celúla[“scene”]celúla[“cell_type”]
) ) self.db.commit()

Definimos una base de datos de memoria estructurada que conserva la información a través de interacciones. Creamos tablas para unidades de memoria atómica, escenas de nivel superior y un índice de búsqueda de texto completo para permitir la recuperación simbólica. También implementamos la lógica para insertar nuevas entradas de memoria en una forma normalizada y consultable.

def get_scene(self, escena): return self.db.execute( “SELECCIONAR * DE mem_scenes DONDE escena=?”, (escena,) ).fetchone() def upsert_scene(self, escena, resumen): self.db.execute(“”” INSERTAR EN mem_scenes VALORES(?,?,?) EN CONFLICTO(escena) ACTUALIZAR CONFIGURACIÓN resumen=excluido.summary, actualizado_at=excluido.actualizado_at “””, (escena, resumen, fecha y hora.utcnow().isoformat())) self.db.commit() def retrieve_scene_context(self, consulta, límite=6): tokens = re.findall(r”[a-zA-Z0-9]+”, consulta) si no son tokens: devolver []

fts_query = ” O “.join(tokens) filas = self.db.execute(“”” SELECCIONAR escena, contenido DE mem_cells_fts ¿DÓNDE mem_cells_fts COINCIDIR? LIMIT ? “””, (fts_query, límite)).fetchall() si no filas: filas = self.db.execute(“”” SELECCIONAR escena, contenido DE mem_cells ORDEN POR prominencia DESC LIMIT ? “””, (límite,)).fetchall() devolver filas def retrieve_scene_summary(self, escena): fila = self.get_scene(escena) devolver fila[“summary”] si la fila es otra “”

Nos centramos en la recuperación de memoria y la lógica de mantenimiento de escenas. Implementamos una búsqueda segura de texto completo al desinfectar las consultas de los usuarios y agregar una estrategia alternativa cuando no se encuentran coincidencias léxicas. También exponemos métodos de ayuda para obtener resúmenes de escenas consolidados para la construcción de contexto a largo plazo.

clase MemoryManager: def __init__(self, db: MemoryDB): self.db = db def extract_cells(self, usuario, asistente) -> Lista[Dict]: solicitud = f””” Convierte esta interacción en celdas de memoria estructuradas. Devuelve una matriz JSON con objetos que contienen: – escena – tipo de celda (hecho, plan, preferencia, decisión, tarea, riesgo) – prominencia (0-1) – contenido (comprimido, factual) Usuario: {usuario} Asistente: {asistente} “”” raw = llm(prompt) raw = re.sub(r”“`json|“`”, “”, raw) try: cell = json.loads (sin procesar) devuelve celdas si es una instancia (celdas, lista) en caso contrario []
excepto Excepción: regresar []

def consolidate_scene(self, escena): filas = self.db.db.execute( “SELECCIONAR contenido DE mem_cells DONDE escena=? ORDENAR POR prominencia DESC”, (escena,) ).fetchall() si no filas: devolver celdas = [json.loads(r[“content”]) para r en filas]solicitud = f””” Resume esta escena de memoria en menos de 100 palabras. Mantenla estable y reutilizable para razonamientos futuros. Celdas: {celdas} “”” resumen = llm(prompt, temperatura=0.05) self.db.upsert_scene(escena, resumen) def update(self, usuario, asistente): celdas = self.extract_cells(usuario, asistente) para celda en celdas: self.db.insert_cell(cell) para escena en set(c[“scene”] para c en celdas): self.consolidate_scene(escena)

Implementamos el componente de gestión de memoria dedicado responsable de estructurar la experiencia. Extraemos representaciones de memoria compacta de las interacciones, las almacenamos y las consolidamos periódicamente en resúmenes de escenas estables. Nos aseguramos de que la memoria evolucione incrementalmente sin interferir con el flujo de respuesta del agente.

clase WorkerAgent: def __init__(self, db: MemoryDB, mem_manager: MemoryManager): self.db = db self.mem_manager = mem_manager def respuesta(self, user_input): recordado = self.db.retrieve_scene_context(user_input) escenas = set(r[“scene”] para r en recuperado) resúmenes = “\n”.join( f”[{scene}]\n{self.db.retrieve_scene_summary(scene)}” para escena en escenas) solicitud = f””” Eres un agente inteligente con memoria a largo plazo. Memoria relevante: {resúmenes} Usuario: {user_input} “”” asistente_respuesta = llm(prompt) self.mem_manager.update(user_input, asistente_reply) return asistente_reply db = MemoryDB() memoria_manager = MemoryManager(db) agente = WorkerAgent(db, memoria_manager) print(agent.answer(“Estamos creando un agente que recuerda proyectos a largo plazo.”)) print(agent.answer(“Debería organizar las conversaciones en temas automáticamente.”)) print(agent.answer(“Este sistema de memoria debería soportar el razonamiento futuro.”)) para la fila en db.db.execute(“SELECT * FROM mem_scenes”): print(dict(row))

Definimos el agente trabajador que realiza el razonamiento sin dejar de ser consciente de la memoria. Recuperamos escenas relevantes, reunimos resúmenes contextuales y generamos respuestas basadas en conocimientos a largo plazo. Luego cerramos el ciclo devolviendo la interacción al administrador de memoria para que el sistema mejore continuamente con el tiempo.

En este tutorial, demostramos cómo un agente puede curar activamente su propia memoria y convertir interacciones pasadas en conocimiento estable y reutilizable en lugar de registros de chat efímeros. Permitimos que la memoria evolucionara a través de la consolidación y el recuerdo selectivo, lo que respalda un razonamiento más consistente y fundamentado en todas las sesiones. Este enfoque proporciona una base práctica para construir sistemas agentes de larga duración y puede ampliarse naturalmente con mecanismos para el olvido, una memoria relacional más rica u orquestación basada en gráficos a medida que el sistema crece en complejidad.

Consulte los códigos completos. 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.