Desarrolla un agente con memoria para asistente personal

Introducción: Más Allá del Estado Efímero

En el desarrollo de agentes de IA avanzados, la capacidad de recordar más allá de una sola sesión de conversación o ejecución es lo que separa a un prototipo interesante de una herramienta verdaderamente útil. Hasta ahora, has trabajado con el estado del grafo, que es volátil y se reinicia con cada ejecución. En esta lección, daremos el salto crucial hacia la persistencia, construyendo un asistente personal que recuerda tus preferencias, tus conversaciones pasadas y el contexto de tareas anteriores, incluso después de que el programa se haya detenido y reiniciado.

Integraremos memoria persistente en un agente construido con LangGraph, utilizando bases de datos para almacenar el estado del grafo y, de manera más crítica, el historial de mensajes. Esto permite que tu asistente mantenga una "relación" continua contigo, aprendiendo de cada interacción. Abordaremos no solo el "cómo" técnico, utilizando checkpointers y memoria de conversación, sino también el "qué" estratégico: decidir qué información merece ser recordada y cómo estructurarla para un acceso eficiente y relevante en el futuro.

El objetivo final es transformar nuestro agente de un solucionador de tasks aislado en un compañero digital con contexto continuo. Este es un patrón fundamental para aplicaciones como asistentes de productividad, coaches de aprendizaje, o agentes de soporte al cliente que necesitan un historial de interacciones con el usuario.

Concepto Clave: Checkpointers y Memoria de Larga Duración

Imagina que estás leyendo una novela compleja. Tu marcador de páginas (bookmark) te permite cerrar el libro y, al reabrirlo, continuar exactamente donde lo dejaste, sin perder el hilo de la trama. En LangGraph, un checkpointer es ese marcador de páginas digital. Su función principal es serializar y guardar el estado del grafo (los valores de todos los nodos y el flujo de ejecución) en un almacenamiento persistente (como una base de datos SQL o en memoria). Esto permite pausar y reanudar ejecuciones largas o complejas.

Sin embargo, para un asistente personal, necesitamos más que solo el estado de la máquina. Necesitamos el contenido de la conversación. Aquí es donde combinamos el checkpointer con una memoria de conversación persistente. Piensa en esto como tu diario personal o el historial de chat de una aplicación de mensajería. Cada intercambio entre tú (el usuario) y el asistente se almacena de forma estructurada. Cuando inicias una nueva sesión, el agente no comienza con una pizarra en blanco; carga el historial relevante, proporcionando contexto inmediato. Esta memoria no es un simple volcado de texto; en un sistema avanzado, se puede indexar, buscar y resumir para extraer solo la información más relevante para la consulta actual.

Tip: Distingue entre el estado del grafo (el "dónde está" y "qué datos temporales tiene" el proceso) y el historial de la conversación (el "qué se dijo"). Para un asistente, el historial es a menudo más valioso. El checkpointer persiste el estado del proceso; un `ChatMessageHistory` persistente guarda la conversación. Los usaremos juntos.

Cómo Funciona en la Práctica: Arquitectura de un Asistente con Memoria

Vamos a desglosar el flujo paso a paso de nuestro asistente personal con memoria persistente. Primero, al inicializar el agente, configuramos dos componentes clave: un checkpointer (por ejemplo, `SqliteSaver`) que se encargará de guardar el estado de ejecución, y un backend de memoria (como `RedisChatMessageHistory`) que almacenará la secuencia de mensajes. Cada usuario tendrá un `thread_id` único (como un ID de usuario o un nombre de sesión) que actuará como llave para recuperar tanto su estado de grafo guardado como su historial de chat.

Cuando un usuario envía un mensaje por primera vez, el grafo se ejecuta desde el principio. Al finalizar la ejecución, el checkpointer guarda automáticamente el estado final. Simultáneamente, el mensaje del usuario y la respuesta del asistente se añaden al almacén de historial de mensajes persistente. En la siguiente interacción del mismo usuario, la magia ocurre: LangGraph puede reanudar desde el checkpoint guardado usando el `thread_id`, y el nodo de inicio del grafo recupera los últimos `N` mensajes del historial persistente, inyectándolos en el contexto de la nueva ejecución. Esto crea la ilusión de una conversación continua.

Un detalle crucial es la gestión del contextocompresión o resumen de memoria. Por ejemplo, podemos mantener una ventana deslizante de los últimos 20 intercambios en el contexto inmediato, y periódicamente generar un "resumen de la relación" que capture hechos clave, preferencias y temas recurrentes, almacenando este resumen también de forma persistente y cargándolo al inicio de cada sesión.

Código en Acción: Asistente Personal con SQLite y Resumen de Memoria

A continuación, un ejemplo completo y funcional de un asistente personal que recuerda conversaciones usando SQLite para checkpoints y memoria, e implementa un resumen básico para gestionar contextos largos.

Configuración Inicial y Definición del Estado

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_community.chat_message_histories import SQLChatMessageHistory
from langchain_openai import ChatOpenAI
from typing import TypedDict, Annotated, List
import operator
from datetime import datetime
import os

# 1. Definir el Estado del Grafo
class AgentState(TypedDict):
    messages: Annotated[List, operator.add]  # Historial de la conversación actual
    user_id: str  # Identificador único del usuario
    summary: str  # Resumen acumulativo de la conversación a largo plazo

# 2. Configurar Persistencia
# Checkpointer para el estado del grafo
checkpointer = SqliteSaver.from_conn_string(":memory:")  # Usa un archivo (ej. "assistant.db") para producción

# Función para obtener el historial de chat persistente por user_id
def get_persistent_chat_history(user_id: str):
    connection_string = "sqlite:///chat_history.db"
    return SQLChatMessageHistory(session_id=user_id, connection_string=connection_string)

# 3. Nodo: Cargar Memoria y Preparar Contexto
def load_memory_node(state: AgentState):
    user_id = state['user_id']
    # Cargar historial persistente
    chat_history = get_persistent_chat_history(user_id)
    all_messages = chat_history.messages

    # Estrategia de ventana deslizante: tomar los últimos 6 mensajes brutos
    recent_messages = all_messages[-6:] if len(all_messages) > 6 else all_messages

    # Combinar el resumen previo (si existe) con los mensajes recientes
    system_context = f"Resumen de conversación previa: {state.get('summary', 'No hay resumen previo.')}"
    full_message_list = [SystemMessage(content=system_context)] + recent_messages

    return {"messages": full_message_list}

# 4. Nodo: Invocar el LLM (Asistente)
def call_llm_node(state: AgentState):
    llm = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0.7)
    response = llm.invoke(state['messages'])
    return {"messages": [response]}

# 5. Nodo: Guardar Respuesta y Actualizar Resumen
def save_and_summarize_node(state: AgentState):
    user_id = state['user_id']
    chat_history = get_persistent_chat_history(user_id)

    # Los dos últimos elementos en 'messages' son: la última pregunta del usuario y la respuesta del asistente
    new_messages = state['messages'][-2:]  # Asume que el flujo añade en orden
    for msg in new_messages:
        chat_history.add_message(msg)  # Guarda de forma persistente

    # Lógica de resumen periódico (ejemplo simplificado)
    # Cada 4 intercambios, generar un nuevo resumen
    all_messages = chat_history.messages
    if len(all_messages) >= 8:  # 4 intercambios (4 user + 4 assistant)
        summary_llm = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0)
        summary_prompt = f"""
        Resume la siguiente conversación entre el usuario {user_id} y un asistente de IA.
        Extrae preferencias clave, hechos importantes sobre el usuario, tareas pendientes mencionadas y temas de interés.
        Mantén el resumen conciso y estructurado.

        Conversación:
        {" ".join([m.content for m in all_messages])}
        """
        new_summary = summary_llm.invoke(summary_prompt).content
        return {"summary": new_summary}
    return {"summary": state.get('summary', '')}

Construcción del Grafo y Ejecución

# 6. Construir el Grafo
workflow = StateGraph(AgentState)

# Añadir nodos
workflow.add_node("load_memory", load_memory_node)
workflow.add_node("generate_response", call_llm_node)
workflow.add_node("persist_data", save_and_summarize_node)

# Definir el flujo
workflow.set_entry_point("load_memory")
workflow.add_edge("load_memory", "generate_response")
workflow.add_edge("generate_response", "persist_data")
workflow.add_edge("persist_data", END)

# Compilar el grafo CON el checkpointer
app = workflow.compile(checkpointer=checkpointer)

# 7. Ejecutar el Agente con Memoria
def chat_with_assistant(user_id: str, user_input: str):
    # Configurar el estado inicial para este thread/hilo de conversación
    initial_state = AgentState(messages=[HumanMessage(content=user_input)], user_id=user_id, summary="")
    # Configuración para reanudar o iniciar un nuevo thread
    config = {"configurable": {"thread_id": user_id}}

    # Ejecutar el grafo. Si existe un checkpoint para este thread_id, se reanuda desde allí.
    final_state = app.invoke(initial_state, config)

    # La respuesta del asistente es el último mensaje AI en el estado
    ai_messages = [msg for msg in final_state["messages"] if isinstance(msg, AIMessage)]
    last_ai_response = ai_messages[-1].content if ai_messages else "No response generated."
    return last_ai_response, final_state.get('summary', '')

# Ejemplo de uso en una conversación multi-sesión
if __name__ == "__main__":
    # Sesión 1 del usuario "alice"
    print("--- Sesión 1 de Alice ---")
    response1, summary1 = chat_with_assistant("alice", "Hola, me llamo Alice. Mi color favorito es el azul y quiero aprender a cocinar pasta.")
    print(f"Asistente: {response1}")
    print(f"Resumen actual: {summary1}\n")

    # Simular que el programa se detiene y se reinicia aquí...

    # Sesión 2 del usuario "alice" (días después)
    print("--- Sesión 2 de Alice (días después) ---")
    # Nota: No mencionamos el color ni la pasta. El asistente lo RECUERDA del resumen/historial.
    response2, summary2 = chat_with_assistant("alice", "¿Qué me recomendabas que hiciera?")
    print(f"Asistente: {response2}")  # Debería referirse a aprender a cocinar pasta.
    print(f"Resumen actual: {summary2}")

Errores Comunes y Cómo Evitarlos

1. No Limpiar o Rotar la Memoria Persistente: Almacenar cada mensaje para siempre puede llevar a bases de datos gigantescas y ralentizar las consultas. Los historiales muy largos también complican la recuperación del contexto relevante. Solución: Implementa una política de retención. Puedes archivar y luego eliminar mensajes antiguos (ej. mayores a 90 días), o comprimir conversaciones muy antiguas en un único registro de resumen. Usa la estrategia de ventana deslizante + resumen como en nuestro ejemplo.

2. Confundir Thread_ID con User_ID: Usar un identificador demasiado genérico (como "default") hace que todos los usuarios compartan el mismo historial y estado de grafo, causando caos. Inversamente, crear un nuevo `thread_id` para cada mensaje del mismo usuario impide la memoria a largo plazo. Solución: Asigna un `thread_id` único y estable por entidad de conversación. Para un asistente personal, el `user_id` es una buena opción. Para un agente de soporte, podría ser `user_id + ticket_id`.

3. No Gestionar la Ventana de Contexto del LLM: Cargar automáticamente toda la historia de conversación (cientos de mensajes) en el prompt superará rápidamente el límite de tokens del modelo, causando errores de costo o truncamiento. Solución: Siempre implementa un mecanismo de selección o compresión. La ventana deslizante de los últimos N mensajes es un buen inicio. Para memoria a largo plazo, usa un sistema de recuperación (RAG) sobre tu historial: incrusta mensajes pasados y recupera solo los más relevantes a la consulta actual.

4. Persistir Datos Sensibles Sin Cifrado: Almacenar conversaciones completas en una base de datos SQLite o Redis sin protección puede ser un riesgo grave de privacidad si el sistema es comprometido. Solución: Evalúa qué necesita ser recordado. Considera almacenar solo metadatos o resúmenes anonimizados. Si debes guardar el contenido completo, utiliza cifrado en reposo (encryption at rest) para tu base de datos y cifra campos sensibles a nivel de aplicación antes de guardarlos.

5. Olvidar la Atomicidad en Operaciones de Guardado: En escenarios de alta concurrencia, si el guardado del checkpoint y el guardado del historial de mensajes no son atómicos, podrías tener estados inconsistentes (ej., el checkpoint hace referencia a mensajes que no se guardaron). Solución: Diseña tu grafo para que la persistencia ocurra en un nodo dedicado (como nuestro `persist_data`), al final del flujo. Considera el uso de transacciones de base de datos si usas el mismo backend para ambos. LangGraph's checkpointer ya maneja la atomicidad del estado del grafo.

Checklist de Dominio

• Identifico claramente la llave de persistencia (`thread_id`) y la asocio correctamente a una entidad de conversación (usuario, sesión, ticket).
• He separado conceptual y técnicamente la persistencia del estado del grafo (checkpointer) de la persistencia del historial de conversación (ChatMessageHistory).
• He implementado una estrategia de gestión de contexto para el LLM, como ventana deslizante, resumen o recuperación basada en embeddings, para no exceder los límites de tokens.
• Mi asistente puede mantener el contexto a través de múltiples ejecuciones del programa, demostrando memoria de preferencias y hechos establecidos en sesiones anteriores.
• He considerado la privacidad y retención de datos, estableciendo políticas sobre qué se guarda, por cuánto tiempo y si se necesita cifrado.
• He probado la recuperación tras un reinicio simulando la detención y nueva puesta en marcha del servicio, verificando que el agente recuerda la interacción previa.
• He manejado posibles errores de concurrencia para el mismo `thread_id`, asegurando que las escrituras no corrompan el estado.
• El resumen de memoria o el mecanismo de recuperación extrae información verdaderamente útil para futuras interacciones, y no solo ruido.