Implementa herramientas de gestión de tareas y calendario

En esta lección avanzada, integraremos las capacidades fundamentales de un agente de IA para la gestión de proyectos: la manipulación de tareas y la interacción con un calendario. Trascenderemos los ejemplos básicos para construir herramientas robustas, seguras y conectadas que un agente de LangGraph pueda utilizar para planificar, organizar y hacer seguimiento de un proyecto de manera autónoma. Abordaremos la persistencia de datos, la validación de entradas, la gestión de estados complejos y la coordinación entre diferentes herramientas, culminando en un sistema que puede razonar sobre plazos, dependencias y recursos.

Fundamentos de las herramientas de tareas y calendario en un agente

En el núcleo de un agente de gestión de proyectos se encuentran dos sistemas de información entrelazados: el sistema de tareas y el sistema de calendario. El primero gestiona el "qué" y el "quién" (descripción, estado, asignación, prioridad), mientras que el segundo gestiona el "cuándo" (fechas, duraciones, plazos). La potencia de un agente de IA reside en su capacidad para cruzar estos datos, razonando que, por ejemplo, si una tarea crítica se retrasa, las tareas dependientes y las reuniones asociadas en el calendario deben ser reajustadas. En LangGraph, cada una de estas capacidades se modela como una herramienta (tool) invocable por el LLM, la cual interactúa con un estado persistente.

La implementación efectiva requiere ir más allá de simples listas en memoria. Debemos considerar un almacenamiento persistente (como una base de datos SQLite, PostgreSQL o incluso un archivo JSON estructurado), un esquema de datos bien definido que modele relaciones (por ejemplo, tareas con subtareas o dependencias), y una lógica de negocio que impida estados inconsistentes (como asignar una tarea a un recurso no existente o programar una reunión en una fecha pasada). La herramienta no es solo una función, es una API interna con reglas claras.

Concepto clave: El estado compartido y el grafo de flujo

Imagina un tablero de proyecto físico (como un Kanban) junto con un calendario de pared. Cada vez que mueves una tarjeta de "En Progreso" a "Hecho", podrías tomar un rotulador y tachar la hora bloqueada en el calendario, liberando ese espacio. El Estado Compartido en LangGraph es la representación digital unificada de ese tablero y ese calendario. Es el único lugar donde se almacena la verdad sobre el proyecto. Todas las herramientas (mover tarea, agregar evento) leen y modifican este estado central. El Grafo de Flujo es el conjunto de reglas que dictan cómo el agente navega entre la decisión (ej: "necesito posponer esta reunión") y la ejecución de la herramienta correcta ("actualizar_evento_calendario").

Una analogía poderosa es la de un director de orquesta (el Grafo de Flujo) y los músicos (las Herramientas). El director no toca los instrumentos, pero conoce la partitura (el estado deseado) y da instrucciones específicas a cada músico para lograrlo. El estado compartido es la partitura misma, que todos pueden ver y que se actualiza con cada nota tocada. La clave para un sistema robusto es que el director (el grafo) pueda manejar instrucciones ambiguas del compositor (el usuario/LLM) y traducirlas en acciones precisas para los músicos (las herramientas), manteniendo la partitura coherente en todo momento.

Cómo funciona en la práctica: De la solicitud del usuario a la acción

Veamos el proceso paso a paso cuando un usuario solicita: "Programa una revisión de diseño para el prototipo A la próxima semana y asigna la tarea 'Finalizar assets' a María". Primero, el LLM del agente razona sobre esta solicitud y la descompone en intenciones accionables: 1) Crear un evento en el calendario, 2) Actualizar la asignación de una tarea existente. Luego, consulta el estado actual para obtener datos necesarios (¿existe la tarea 'Finalizar assets'?, ¿está libre María la próxima semana?).

El grafo dirige entonces la ejecución secuencial o paralela de las herramientas. Primero podría invocar crear_evento_calendario con parámetros extraídos: título="Revisión de diseño Prototipo A", fecha="próxima semana", duración=60, participantes=["María", "Usuario"]. La herramienta valida la fecha, la asigna y devuelve un ID de evento. Luego, el grafo invoca actualizar_tarea con: id_tarea=[ID], campo="asignado_a", nuevo_valor="María". Cada herramienta actualiza el estado compartido. Finalmente, el agente sintetiza una respuesta para el usuario confirmando ambas acciones y proporcionando los IDs de referencia, todo ello manteniendo un registro en el estado de la conversación.

Diseño e implementación de la herramienta de gestión de tareas

Una herramienta de tareas robusta debe soportar operaciones CRUD completas (Crear, Leer, Actualizar, Eliminar) y posiblemente operaciones complejas como la gestión de dependencias. Comenzaremos definiendo un esquema Pydantic para una tarea. Este esquema no solo estructura los datos, sino que también proporciona validación automática cuando el agente o las herramientas intentan crear o modificar una tarea, previniendo datos malformados desde el origen.

La herramienta en sí será una clase o conjunto de funciones que actúan sobre una capa de persistencia. Para este ejemplo, usaremos un almacenamiento en memoria con un diccionario, pero la estructura estará preparada para ser intercambiada por un cliente de base de datos. La clave es que cada función (herramienta) tenga una firma clara y un docstring detallado, ya que esto es lo que el LLM utilizará para entender cómo llamarla. Implementaremos lógica para evitar duplicados, manejar errores elegantes (por ejemplo, "tarea no encontrada") y actualizar campos específicos sin sobrescribir otros.

from typing import List, Optional, Dict, Any
from datetime import datetime
from enum import Enum
from pydantic import BaseModel, Field
import json

class EstadoTarea(str, Enum):
    PENDIENTE = "pendiente"
    EN_PROGRESO = "en_progreso"
    BLOQUEADA = "bloqueada"
    COMPLETADA = "completada"

class Tarea(BaseModel):
    id: str = Field(default_factory=lambda: f"tarea_{datetime.now().timestamp()}")
    titulo: str
    descripcion: Optional[str] = None
    estado: EstadoTarea = EstadoTarea.PENDIENTE
    asignado_a: Optional[str] = None
    prioridad: int = 3  # 1: Alta, 2: Media, 3: Baja
    fecha_creacion: datetime = Field(default_factory=datetime.now)
    fecha_limite: Optional[datetime] = None
    dependencias: List[str] = Field(default_factory=list)  # IDs de tareas de las que depende
    etiquetas: List[str] = Field(default_factory=list)

class GestorTareas:
    """Gestor del estado de las tareas del proyecto."""
    
    def __init__(self, archivo_almacenamiento: str = "tareas.json"):
        self.archivo = archivo_almacenamiento
        self.tareas: Dict[str, Tarea] = self._cargar_tareas()
    
    def _cargar_tareas(self) -> Dict[str, Tarea]:
        try:
            with open(self.archivo, 'r') as f:
                datos = json.load(f)
            return {tid: Tarea(**tarea) for tid, tarea in datos.items()}
        except FileNotFoundError:
            return {}
    
    def _guardar_tareas(self):
        with open(self.archivo, 'w') as f:
            json.dump({tid: tarea.dict() for tid, tarea in self.tareas.items()}, f, default=str)
    
    def crear_tarea(self, titulo: str, descripcion: Optional[str] = None, prioridad: int = 3, fecha_limite: Optional[datetime] = None) -> str:
        """Crea una nueva tarea y la almacena. Devuelve el ID de la tarea creada."""
        nueva_tarea = Tarea(titulo=titulo, descripcion=descripcion, prioridad=prioridad, fecha_limite=fecha_limite)
        self.tareas[nueva_tarea.id] = nueva_tarea
        self._guardar_tareas()
        return nueva_tarea.id
    
    def actualizar_tarea(self, id_tarea: str, **campos) -> bool:
        """Actualiza campos específicos de una tarea existente. Campos válidos: estado, asignado_a, prioridad, etc."""
        if id_tarea not in self.tareas:
            return False
        tarea = self.tareas[id_tarea]
        # Actualización segura, solo campos permitidos
        campos_permitidos = {'estado', 'asignado_a', 'prioridad', 'descripcion', 'fecha_limite', 'etiquetas'}
        for campo, valor in campos.items():
            if campo in campos_permitidos:
                setattr(tarea, campo, valor)
        self._guardar_tareas()
        return True
    
    def listar_tareas(self, estado: Optional[EstadoTarea] = None, asignado_a: Optional[str] = None) -> List[Dict[str, Any]]:
        """Lista tareas, opcionalmente filtradas por estado o asignación."""
        lista = []
        for tarea in self.tareas.values():
            if estado and tarea.estado != estado:
                continue
            if asignado_a and tarea.asignado_a != asignado_a:
                continue
            lista.append(tarea.dict())
        return lista
    
    def obtener_tarea(self, id_tarea: str) -> Optional[Dict[str, Any]]:
        """Obtiene los detalles de una tarea específica por su ID."""
        tarea = self.tareas.get(id_tarea)
        return tarea.dict() if tarea else None

# Ejemplo de uso como herramienta para LangGraph
from langchain.tools import tool

gestor = GestorTareas()

@tool
def crear_tarea(titulo: str, descripcion: str = "") -> str:
    """Crea una nueva tarea en el sistema de gestión del proyecto.
    
    Args:
        titulo: El título breve y descriptivo de la tarea.
        descripcion: (Opcional) Una descripción más detallada de lo que implica la tarea.
    
    Returns:
        str: El ID de la tarea recién creada o un mensaje de error.
    """
    try:
        tarea_id = gestor.crear_tarea(titulo=titulo, descripcion=descripcion)
        return f"Tarea '{titulo}' creada exitosamente con ID: {tarea_id}"
    except Exception as e:
        return f"Error al crear la tarea: {str(e)}"

@tool
def cambiar_estado_tarea(id_tarea: str, nuevo_estado: str) -> str:
    """Cambia el estado de una tarea existente (pendiente, en_progreso, completada, etc.)."""
    # Validar que el nuevo_estado sea uno permitido
    estados_validos = [e.value for e in EstadoTarea]
    if nuevo_estado not in estados_validos:
        return f"Estado '{nuevo_estado}' no válido. Estados permitidos: {', '.join(estados_validos)}"
    
    exito = gestor.actualizar_tarea(id_tarea, estado=nuevo_estado)
    if exito:
        return f"Estado de la tarea {id_tarea} actualizado a '{nuevo_estado}'."
    else:
        return f"No se encontró una tarea con ID: {id_tarea}"

Diseño e implementación de la herramienta de calendario

La herramienta de calendario debe manejar la complejidad temporal: solapamientos, duraciones, zonas horarias (para proyectos distribuidos) y recurrencia. Al igual que con las tareas, comenzamos con un modelo de datos robusto. Un evento de calendario no es solo un título y una fecha; debe incluir participantes, una ubicación (física o virtual), un estado de confirmación y un enlace a tareas relacionadas. La herramienta debe validar que un nuevo evento no choque con eventos existentes para los mismos participantes, a menos que se especifique lo contrario.

La implementación práctica implica un sistema de búsqueda de disponibilidad y una lógica de resolución de conflictos. Para mantener la simplicidad en este ejemplo, implementaremos una comprobación básica de solapamiento. En un sistema de producción, se integraría con una API como Google Calendar o Microsoft Graph. La herramienta debe devolver mensajes informativos: no solo "éxito" o "fracaso", sino "éxito, pero se solapa con la reunión X, se ha creado de todos modos" o "fracaso, el participante Y tiene un evento conflictivo".

from datetime import datetime, timedelta, date
from pydantic import BaseModel, validator
from typing import List, Optional
import json

class EventoCalendario(BaseModel):
    id: str = Field(default_factory=lambda: f"evento_{datetime.now().timestamp()}")
    titulo: str
    descripcion: Optional[str] = None
    inicio: datetime
    fin: datetime
    participantes: List[str] = Field(default_factory=list)
    ubicacion: Optional[str] = None
    tareas_relacionadas: List[str] = Field(default_factory=list)  # IDs de tareas
    recurrente: bool = False
    
    @validator('fin')
    def fin_despues_de_inicio(cls, v, values):
        if 'inicio' in values and v <= values['inicio']:
            raise ValueError('La fecha/hora de fin debe ser posterior a la de inicio.')
        return v

class GestorCalendario:
    """Gestor del calendario del proyecto."""
    
    def __init__(self, archivo_almacenamiento: str = "calendario.json"):
        self.archivo = archivo_almacenamiento
        self.eventos: Dict[str, EventoCalendario] = self._cargar_eventos()
    
    def _cargar_eventos(self) -> Dict[str, EventoCalendario]:
        try:
            with open(self.archivo, 'r') as f:
                datos = json.load(f)
            # Necesitamos parsear las cadenas de fecha/hora de vuelta a objetos datetime
            eventos = {}
            for eid, ev in datos.items():
                ev['inicio'] = datetime.fromisoformat(ev['inicio'])
                ev['fin'] = datetime.fromisoformat(ev['fin'])
                eventos[eid] = EventoCalendario(**ev)
            return eventos
        except FileNotFoundError:
            return {}
    
    def _guardar_eventos(self):
        with open(self.archivo, 'w') as f:
            json.dump({eid: evento.dict() for eid, evento in self.eventos.items()}, f, default=str)
    
    def crear_evento(self, titulo: str, inicio: datetime, fin: datetime, participantes: List[str] = [], descripcion: str = "") -> Dict[str, Any]:
        """Crea un nuevo evento en el calendario, con comprobación básica de solapamiento."""
        # Comprobación de solapamiento simple (para todos los participantes)
        conflictos = []
        for evento in self.eventos.values():
            if any(part in evento.participantes for part in participantes):
                # Si los intervalos de tiempo se solapan
                if not (fin <= evento.inicio or inicio >= evento.fin):
                    conflictos.append(evento.titulo)
        
        nuevo_evento = EventoCalendario(
            titulo=titulo,
            descripcion=descripcion,
            inicio=inicio,
            fin=fin,
            participantes=participantes
        )
        
        self.eventos[nuevo_evento.id] = nuevo_evento
        self._guardar_eventos()
        
        respuesta = {
            "id": nuevo_evento.id,
            "mensaje": f"Evento '{titulo}' creado exitosamente para {inicio.date()}.",
            "conflictos": conflictos
        }
        return respuesta
    
    def buscar_eventos(self, fecha: Optional[date] = None, participante: Optional[str] = None) -> List[Dict[str, Any]]:
        """Busca eventos por fecha o participante."""
        resultados = []
        for evento in self.eventos.values():
            if fecha and evento.inicio.date() != fecha:
                continue
            if participante and participante not in evento.participantes:
                continue
            resultados.append(evento.dict())
        # Ordenar por hora de inicio
        resultados.sort(key=lambda x: x['inicio'])
        return resultados

# Herramientas LangChain para el calendario
gestor_cal = GestorCalendario()

@tool
def programar_reunion(titulo: str, fecha_hora_inicio: str, duracion_minutos: int, participantes: List[str], descripcion: str = "") -> str:
    """Programa una nueva reunión en el calendario del proyecto.
    
    Args:
        titulo: Asunto de la reunión.
        fecha_hora_inicio: Fecha y hora de inicio en formato ISO (YYYY-MM-DDTHH:MM:SS).
        duracion_minutos: Duración de la reunión en minutos.
        participantes: Lista de nombres de los participantes.
        descripcion: (Opcional) Agenda o detalles de la reunión.
    
    Returns:
        str: Confirmación de la creación y advertencias de conflictos si las hay.
    """
    try:
        inicio = datetime.fromisoformat(fecha_hora_inicio)
        fin = inicio + timedelta(minutes=duracion_minutos)
        resultado = gestor_cal.crear_evento(titulo, inicio, fin, participantes, descripcion)
        
        mensaje = resultado["mensaje"]
        if resultado["conflictos"]:
            mensaje += f" ADVERTENCIA: Posible solapamiento con: {', '.join(resultado['conflictos'])}"
        return mensaje
    except ValueError as e:
        return f"Error en el formato de fecha o datos: {str(e)}"
    except Exception as e:
        return f"Error inesperado al programar la reunión: {str(e)}"

@tool
def consultar_calendario(fecha: Optional[str] = None, participante: Optional[str] = None) -> str:
    """Consulta los eventos del calendario. Puede filtrar por fecha (YYYY-MM-DD) o por participante."""
    fecha_obj = None
    if fecha:
        try:
            fecha_obj = datetime.strptime(fecha, "%Y-%m-%d").date()
        except ValueError:
            return f"Formato de fecha inválido. Use YYYY-MM-DD."
    
    eventos = gestor_cal.buscar_eventos(fecha=fecha_obj, participante=participante)
    
    if not eventos:
        filtro = []
        if fecha: filtro.append(f"fecha {fecha}")
        if participante: filtro.append(f"participante {participante}")
        return f"No hay eventos encontrados para {' y '.join(filtro) if filtro else 'el calendario'}."
    
    respuesta = ["Eventos encontrados:"]
    for ev in eventos:
        inicio_str = ev['inicio'].strftime("%Y-%m-%d %H:%M")
        fin_str = ev['fin'].strftime("%H:%M")
        respuesta.append(f"- [{inicio_str} - {fin_str}] {ev['titulo']} (Participantes: {', '.join(ev['participantes'])})")
    
    return "\n".join(respuesta)

Integración en un agente LangGraph con estado y memoria

Con las herramientas definidas, el siguiente paso es integrarlas en un grafo de flujo de LangGraph que orqueste su uso. Crearemos un estado que contenga no solo el historial de mensajes, sino también referencias a nuestros gestores (o los datos serializados de las tareas y eventos). Esto permite que el estado del grafo refleje el estado real del proyecto. El nodo central del grafo será un LLM (como GPT-4 o Claude) equipado con estas herramientas. La magia de LangGraph está en su capacidad de manejar ciclos y flujos condicionales, permitiendo que el agente realice múltiples pasos de razonamiento y uso de herramientas en respuesta a una sola consulta del usuario.

Configuraremos un flujo donde, después de que el LLM decida qué herramienta(s) usar, el grafo las ejecute, actualice el estado y luego decida si necesita realizar otra acción (un ciclo) o finalizar. También añadiremos un nodo de "supervisión" o "validación" que pueda revisar los cambios propuestos por el agente antes de comprometerlos, añadiendo una capa de seguridad en aplicaciones críticas. La memoria de conversación de largo plazo se mantendrá en el estado, permitiendo al agente hacer referencia a tareas y eventos mencionados anteriormente.

from typing import TypedDict, List, Annotated, Sequence
import operator
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langgraph.prebuilt import ToolExecutor, ToolInvocation
import json

# 1. Definir el Estado del Grafo
class EstadoAgente(TypedDict):
    """El estado del agente de gestión de proyectos."""
    mensajes: Annotated[List[BaseMessage], operator.add]  # Historial de la conversación
    # Podríamos añadir aquí referencias directas a los gestores si no usáramos herramientas con estado interno.
    # Para este ejemplo, las herramientas ya manejan su propia persistencia.

# 2. Inicializar el modelo y las herramientas
modelo = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0)
herramientas = [crear_tarea, cambiar_estado_tarea, programar_reunion, consultar_calendario]
modelo_con_herramientas = modelo.bind_tools(herramientas)
tool_executor = ToolExecutor(herramientas)

# 3. Definir la función del nodo del agente (LLM)
def nodo_agente(estado: EstadoAgente):
    """Nodo que llama al LLM para obtener la siguiente acción."""
    mensajes = estado['mensajes']
    respuesta = modelo_con_herramientas.invoke(mensajes)
    # Asegurarnos de que la respuesta del LLM se añade al historial
    return {"mensajes": [respuesta]}

# 4. Definir la función del nodo de ejecución de herramientas
def nodo_ejecutar_herramientas(estado: EstadoAgente):
    """Nodo que ejecuta las herramientas que el LLM decidió usar."""
    mensajes = estado['mensajes']
    ultimo_mensaje = mensajes[-1]
    
    # Asumimos que el último mensaje es del AI y contiene llamadas a herramientas
    tool_calls = ultimo_mensaje.tool_calls
    if not tool_calls:
        raise ValueError("No hay llamadas a herramientas para ejecutar.")
    
    # Ejecutar cada herramienta secuencialmente
    resultados = []
    for tool_call in tool_calls:
        # Construir la invocación
        invocation = ToolInvocation(
            tool=tool_call['name'],
            tool_input=tool_call['args'],
            id=tool_call['id'],
        )
        # Ejecutar
        output = tool_executor.invoke(invocation)
        resultados.append(output)
    
    # Crear mensajes de herramienta para el historial
    tool_messages = [
        ToolMessage(content=str(resultado), tool_call_id=tool_call['id'])
        for resultado, tool_call in zip(resultados, tool_calls)
    ]
    
    return {"mensajes": tool_messages}

# 5. Construir el grafo
grafo = StateGraph(EstadoAgente)

# Añadir nodos
grafo.add_node("agente", nodo_agente)
grafo.add_node("herramientas", nodo_ejecutar_herramientas)

# Establecer el punto de entrada
grafo.set_entry_point("agente")

# Definir las transiciones condicionales
def decidir_siguiente_paso(estado: EstadoAgente) -> str:
    """Decide si hay que ejecutar herramientas o finalizar."""
    ultimo_mensaje = estado['mensajes'][-1]
    # Si el último mensaje del AI tiene llamadas a herramientas, vamos al nodo de herramientas
    if hasattr(ultimo_mensaje, 'tool_calls') and ultimo_mens