Sistema de memoria para tus agentes de IA

mi-agente/
├── agente.py              ← el código del agente
├── memoria/
│   ├── largo_plazo.json   ← hechos permanentes
│   └── sesion_actual.json ← log de sesión (se crea al iniciar)

{
  "usuario": {
    "nombre": "María",
    "empresa": "AgenciaX",
    "industria": "Marketing digital",
    "preferencias": {
      "idioma_codigo": "español",
      "formato_reportes": "markdown",
      "modelo_favorito": "claude-sonnet-4-6"
    }
  },
  "negocio": {
    "clientes_activos": 12,
    "herramientas_usadas": ["Notion", "HubSpot", "Zapier"],
    "objetivos_2026": "Escalar a 25 clientes sin contratar más staff"
  },
  "aprendido": [
    "Prefiere reportes cortos, máximo 1 página",
    "No usar jerga técnica en entregables para clientes",
    "Los lunes hace review semanal"
  ]
}

import json
import os
from datetime import datetime

import json
import os
from datetime import datetime

# Ruta base para todos los archivos de memoria
MEMORIA_DIR = "memoria"

def cargar_memoria(tipo: str) -> dict:
    """Carga un archivo de memoria por tipo: 'activa', 'sesion', 'largo_plazo'"""
    ruta = os.path.join(MEMORIA_DIR, f"{tipo}.json")

    # Si el archivo no existe, retornar estructura vacía
    if not os.path.exists(ruta):
        return {"hechos": [], "preferencias": {}, "ultima_actualizacion": None}

    with open(ruta, "r", encoding="utf-8") as f:
        return json.load(f)  # Leer y parsear el JSON

def guardar_memoria(tipo: str, datos: dict) -> None:
    """Guarda el estado actual de la memoria en disco"""
    os.makedirs(MEMORIA_DIR, exist_ok=True)  # Crear carpeta si no existe
    ruta = os.path.join(MEMORIA_DIR, f"{tipo}.json")

    # Actualizar timestamp antes de guardar
    datos["ultima_actualizacion"] = datetime.now().isoformat()

    with open(ruta, "w", encoding="utf-8") as f:
        json.dump(datos, f, ensure_ascii=False, indent=2)  # Guardar con formato legible

def agregar_hecho(hecho: str, tipo: str = "largo_plazo") -> None:
    """Agrega un hecho nuevo a la memoria del tipo especificado"""
    memoria = cargar_memoria(tipo)

    # Evitar duplicados exactos
    if hecho not in memoria["hechos"]:
        memoria["hechos"].append(hecho)
        guardar_memoria(tipo, memoria)
        print(f"✓ Hecho guardado en memoria {tipo}: {hecho}")

def cargar_memoria_largo_plazo(ruta="memoria/largo_plazo.json"):
    """Carga la memoria permanente al inicio de cada sesión."""
    if os.path.exists(ruta):
        with open(ruta, "r", encoding="utf-8") as f:
            return json.load(f)  # Retornar el dict completo con usuario, negocio y aprendido
    return {"usuario": {}, "negocio": {}, "aprendido": []}  # Estructura vacía si no existe

def actualizar_memoria(memoria, nueva_info, ruta="memoria/largo_plazo.json"):
    """Guarda nueva información permanente aprendida durante la sesión."""
    if "aprendido" not in memoria:
        memoria["aprendido"] = []  # Inicializar lista si no existe
    memoria["aprendido"].append({
        "fecha": datetime.now().strftime("%Y-%m-%d"),  # Fecha en formato ISO
        "hecho": nueva_info
    })
    os.makedirs("memoria", exist_ok=True)  # Crear carpeta si no existe
    with open(ruta, "w", encoding="utf-8") as f:
        json.dump(memoria, f, ensure_ascii=False, indent=2)  # Guardar con indentación legible

def iniciar_sesion(objetivo):
    """Crea un log de sesión nuevo al arrancar el agente."""
    sesion = {
        "inicio": datetime.now().isoformat(),  # Timestamp de inicio
        "objetivo": objetivo,
        "pasos": []  # Lista vacía que se irá llenando con cada acción
    }
    os.makedirs("memoria", exist_ok=True)
    with open("memoria/sesion_actual.json", "w", encoding="utf-8") as f:
        json.dump(sesion, f, ensure_ascii=False, indent=2)
    return sesion

def registrar_paso(accion, resultado):
    """Registra cada paso del agente en el log de sesión para poder reanudar si falla."""
    ruta = "memoria/sesion_actual.json"
    if os.path.exists(ruta):
        with open(ruta, "r", encoding="utf-8") as f:
            sesion = json.load(f)
        sesion["pasos"].append({
            "timestamp": datetime.now().isoformat(),
            "accion": accion,
            "resultado": resultado[:200]  # Solo primeros 200 chars para no inflar el archivo
        })
        with open(ruta, "w", encoding="utf-8") as f:
            json.dump(sesion, f, ensure_ascii=False, indent=2)

import anthropic
import json
import re
from memory_utils import cargar_memoria_largo_plazo, iniciar_sesion, registrar_paso, actualizar_memoria

def construir_system_prompt(memoria):
    """Construye el prompt de sistema con la memoria cargada."""
    usuario = memoria.get("usuario", {})
    negocio = memoria.get("negocio", {})
    aprendido = memoria.get("aprendido", [])

    hechos = "\n".join([f"- {h}" if isinstance(h, str) else f"- {h.get('hecho', h)}"
                        for h in aprendido[-10:]])  # últimos 10 hechos

    return f"""Eres un asistente especializado para {usuario.get('nombre', 'el usuario')}.

CONTEXTO DEL NEGOCIO:
- Empresa: {negocio.get('empresa', 'no especificada')}
- Industria: {negocio.get('industria', 'no especificada')}
- Herramientas: {', '.join(negocio.get('herramientas_usadas', []))}

LO QUE YA SABES DE SESIONES ANTERIORES:
{hechos if hechos else 'Primera sesión — sin historial previo'}

PREFERENCIAS DEL USUARIO:
{json.dumps(usuario.get('preferencias', {}), ensure_ascii=False, indent=2)}

Si durante esta sesión aprendes algo nuevo y permanente sobre el usuario o su negocio,
indícalo con el prefijo: [MEMORIZAR: hecho_aprendido]
"""

def correr_agente_con_memoria(objetivo):
    memoria = cargar_memoria_largo_plazo()
    sesion = iniciar_sesion(objetivo)
    system_prompt = construir_system_prompt(memoria)

    client = anthropic.Anthropic()
    mensajes = [{"role": "user", "content": objetivo}]

    respuesta = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        system=system_prompt,  # ← memoria inyectada aquí
        tools=tools,
        messages=mensajes
    )

    # Detectar nuevos hechos a memorizar
    for bloque in respuesta.content:
        if hasattr(bloque, 'text') and '[MEMORIZAR:' in bloque.text:
            hechos = re.findall(r'\[MEMORIZAR: (.+?)\]', bloque.text)
            for hecho in hechos:
                actualizar_memoria(memoria, hecho)
                print(f"  💾 Memorizado: {hecho}")

    return respuesta