Integra memoria y lógica condicional para priorización

Tipo de lección: Video

Introducción: La Priorización Inteligente en Agentes de IA

En el desarrollo de agentes de IA avanzados, especialmente para dominios complejos como la gestión de proyectos, la capacidad de tomar decisiones contextuales y priorizar tareas de forma dinámica es lo que separa un simple chatbot de un asistente verdaderamente autónomo. Hasta ahora, has aprendido a construir agentes con herramientas y memoria. En esta lección integradora, daremos el salto crucial: combinar la memoria de estado de LangGraph con lógica condicional para crear un agente que no solo recuerde el historial de un proyecto, sino que también evalúe, clasifique y decida qué acción tomar a continuación en función de reglas de negocio complejas. Este es el núcleo de un sistema de IA que puede gestionar, no solo responder.

Imagina un gestor de proyectos humano. No procesa tareas en el orden en que llegan. Evalúa la urgencia, el impacto, los recursos disponibles, las dependencias y los plazos. Nuestro agente emulará este proceso. Utilizaremos la estructura de grafo de LangGraph para definir caminos de decisión explícitos, donde el nodo siguiente no está predeterminado por una secuencia lineal, sino por el resultado de una función condicional (conditional edge) que inspecciona el estado actual de la memoria. El proyecto final será un agente que gestiona un backlog, prioriza tareas automáticamente y sugiere acciones, todo manteniendo un contexto rico y persistente.

Concepto Clave: El Estado como Fuente de Verdad para la Decisión

El concepto fundamental aquí es tratar el estado compartido (o memoria) del grafo como la única fuente de verdad sobre la que se basan todas las decisiones. En LangGraph, este estado es un diccionario de Python que persiste y se modifica a lo largo de la ejecución. La lógica condicional actúa como un guardián en las transiciones (edges) entre nodos. Este guardián examina el estado actual y, en función de su contenido, dirige el flujo de ejecución por una rama u otra del grafo. Es análogo a un semáforo inteligente en una red de carreteras: no deja pasar a todos los coches (llamadas de función) por igual; en su lugar, lee datos en tiempo real (tráfico, emergencias) y envía cada vehículo por la ruta óptima.

En nuestro contexto de gestión de proyectos, el estado contendrá el backlog de tareas, cada una con atributos como prioridad, dificultad estimada, plazo y estado. Un nodo de "Evaluación" calculará una puntuación de prioridad. Luego, una arista condicional preguntará: "¿La tarea con mayor puntuación tiene una prioridad 'CRÍTICA'?" Dependiendo de la respuesta (True o False), el grafo se dirigirá a un nodo de "Acción Inmediata" o a un nodo de "Planificación Programada". Esta separación del cálculo (en los nodos) de la decisión (en las aristas) es lo que hace que los grafos de LangGraph sean tan poderosos y fáciles de depurar para lógicas de negocio complejas.

Tip Clave: Diseña tu estado pensando en qué datos necesitarán tus funciones condicionales para tomar decisiones. A menudo, es útil incluir un campo como `"decision"` o `"next_step"` en el estado, que sea llenado por un nodo y leído por la arista condicional para enrutar el flujo. Esto desacopla la lógica y hace el grafo más modular.

Cómo Funciona en la Práctica: Paso a Paso en el Grafo

Vamos a desglosar el flujo de nuestro agente de gestión de proyectos. Primero, el usuario interactúa con un nodo de entrada (orchestrator) que recibe una consulta como "Tengo estas tareas nuevas" o "¿Qué debería hacer ahora?". Este nodo actualiza el estado con la información nueva o recupera el contexto existente. Luego, el flujo pasa a un nodo de analizador de prioridades. Aquí es donde reside la inteligencia: este nodo ejecuta una función que, basándose en reglas predefinidas (ej: `puntuación = (urgencia * 3) + (impacto * 2) - días_de_retraso`), calcula una puntuación para cada tarea pendiente y actualiza el estado con una lista ordenada.

Ahora viene el corazón de la lógica condicional. El grafo sale del analizador y llega a una bifurcación. Esta bifurcación está definida por una función router (enrutador). Esta función, que se asigna a una conditional edge, examina el estado. Podría tener una lógica como: "Si existe al menos una tarea con prioridad 'CRÍTICA' y está sin asignar, dirígete al nodo 'asignar_recurso_urgente'. De lo contrario, si la tarea de mayor puntucción tiene un plazo dentro de las próximas 24 horas, dirígete al nodo 'recordar_plazo_cercano'. Si ninguna de las anteriores es cierta, dirígete al nodo 'planificar_sprint_semanal'." Cada uno de estos nodos destino es una herramienta especializada o un proceso que realiza acciones concretas (escribir en una base de datos, enviar un mensaje, etc.) y actualiza el estado para reflejar lo realizado.

Finalmente, después de ejecutar la acción correspondiente, el grafo puede volver al nodo de orquestación para esperar la siguiente interacción del usuario, creando así un ciclo de conversación persistente y con estado. Este diseño permite que el agente "recuerde" las decisiones anteriores, el estado del proyecto, y adapte su comportamiento futuro en consecuencia, demostrando una forma primitiva pero efectiva de razonamiento iterativo y gestión de contexto a largo plazo.

Código en Acción: Implementación del Agente de Priorización

A continuación, un ejemplo completo y funcional de un grafo LangGraph que implementa la lógica descrita. Definiremos el esquema del estado, las herramientas/nodos y las transiciones condicionales.

from typing import TypedDict, List, Annotated, Literal
from langgraph.graph import StateGraph, END
from datetime import datetime, timedelta
import operator

# 1. DEFINICIÓN DEL ESTADO (Memoria)
class ProjectState(TypedDict):
    """El estado compartido y persistente de nuestro agente."""
    messages: Annotated[List[str], operator.add]  # Historial de la conversación
    backlog: List[dict]  # Lista de tareas del proyecto
    prioritized_tasks: List[dict]  # Lista de tareas después del análisis
    next_action: str  # Decisión tomada por la lógica condicional
    last_updated: datetime

# 2. DEFINICIÓN DE HERRAMIENTAS/NODOS
def add_tasks_to_backlog(state: ProjectState):
    """Nodo: Añade nuevas tareas al backlog desde los mensajes."""
    new_message = state["messages"][-1]
    # (Simulación: parsear mensaje para extraer tareas)
    # En un caso real, usarías un LLM o un parser estructurado aquí.
    simulated_tasks = [
        {"id": 1, "desc": "Arreglar bug crítico en login", "urgency": 5, "impact": 5, "deadline": datetime.now() + timedelta(hours=2), "status": "unassigned"},
        {"id": 2, "desc": "Actualizar documentación API", "urgency": 2, "impact": 3, "deadline": datetime.now() + timedelta(days=5), "status": "unassigned"},
        {"id": 3, "desc": "Revisar PR #123", "urgency": 3, "impact": 2, "deadline": datetime.now() + timedelta(days=1), "status": "unassigned"}
    ]
    state["backlog"].extend(simulated_tasks)
    state["last_updated"] = datetime.now()
    return {"backlog": state["backlog"], "last_updated": state["last_updated"]}

def analyze_and_prioritize(state: ProjectState):
    """Nodo: Analiza el backlog y calcula una puntuación de prioridad."""
    for task in state["backlog"]:
        # Fórmula de priorización simple
        days_until_deadline = (task["deadline"] - datetime.now()).days
        priority_score = (task["urgency"] * 3) + (task["impact"] * 2) - max(0, -days_until_deadline) * 5
        task["priority_score"] = priority_score
        # Clasificación categórica basada en puntuación
        if priority_score > 15:
            task["priority_category"] = "CRITICAL"
        elif priority_score > 10:
            task["priority_category"] = "HIGH"
        else:
            task["priority_category"] = "NORMAL"

    # Ordenar backlog por puntuación de prioridad (descendente)
    sorted_backlog = sorted(state["backlog"], key=lambda x: x["priority_score"], reverse=True)
    state["prioritized_tasks"] = sorted_backlog
    return {"prioritized_tasks": state["prioritized_tasks"]}

def handle_critical_task(state: ProjectState):
    """Nodo: Acción para tareas críticas."""
    critical_task = state["prioritized_tasks"][0]
    action = f"🚨 ASIGNACIÓN INMEDIATA: Tarea '{critical_task['desc']}' asignada al ingeniero senior. Notificación enviada."
    state["next_action"] = action
    state["backlog"][0]["status"] = "assigned_urgently"  # Actualizar estado de la tarea
    return {"next_action": state["next_action"], "backlog": state["backlog"]}

def handle_high_priority_task(state: ProjectState):
    """Nodo: Acción para tareas de alta prioridad."""
    high_task = state["prioritized_tasks"][0]
    action = f"⚠️ PLANIFICAR PARA HOY: Tarea '{high_task['desc']}' agregada a la agenda del día. Revisar recursos."
    state["next_action"] = action
    return {"next_action": state["next_action"]}

def handle_normal_task(state: ProjectState):
    """Nodo: Acción para tareas normales."""
    normal_task = state["prioritized_tasks"][0]
    action = f"📅 AGENDAR PARA SPRINT: Tarea '{normal_task['desc']}' añadida al backlog de planificación del próximo sprint."
    state["next_action"] = action
    return {"next_action": state["next_action"]}

# 3. FUNCIÓN DE ENRUTAMIENTO CONDICIONAL (Conditional Edge)
def route_after_analysis(state: ProjectState) -> Literal["critical", "high", "normal", "empty"]:
    """Determina la ruta a seguir basándose en el estado priorizado."""
    if not state["prioritized_tasks"]:
        return "empty"
    top_task = state["prioritized_tasks"][0]
    if top_task["priority_category"] == "CRITICAL":
        return "critical"
    elif top_task["priority_category"] == "HIGH":
        return "high"
    else:
        return "normal"

# 4. CONSTRUCCIÓN DEL GRAFO
workflow = StateGraph(ProjectState)

# Añadir nodos
workflow.add_node("add_tasks", add_tasks_to_backlog)
workflow.add_node("prioritize", analyze_and_prioritize)
workflow.add_node("critical_action", handle_critical_task)
workflow.add_node("high_action", handle_high_priority_task)
workflow.add_node("normal_action", handle_normal_task)

# Definir flujo principal y ramas condicionales
workflow.set_entry_point("add_tasks")
workflow.add_edge("add_tasks", "prioritize")
# La arista condicional sale de 'prioritize' y se enruta por la función 'route_after_analysis'
workflow.add_conditional_edges(
    "prioritize",
    route_after_analysis,
    {
        "critical": "critical_action",
        "high": "high_action",
        "normal": "normal_action",
        "empty": END
    }
)
# Conectar los nodos de acción al final del grafo
workflow.add_edge("critical_action", END)
workflow.add_edge("high_action", END)
workflow.add_edge("normal_action", END)

# Compilar el grafo
app = workflow.compile()

# 5. EJECUCIÓN DE PRUEBA
if __name__ == "__main__":
    # Estado inicial
    initial_state: ProjectState = {
        "messages": ["Usuario: Añade estas tareas nuevas al sistema."],
        "backlog": [],
        "prioritized_tasks": [],
        "next_action": "",
        "last_updated": datetime.now()
    }
    # Ejecutar el grafo
    final_state = app.invoke(initial_state)
    print("Estado Final:")
    print(f"  Tareas Prioritizadas: {[t['desc'] for t in final_state['prioritized_tasks']]}")
    print(f"  Categoría de la Top: {final_state['prioritized_tasks'][0]['priority_category']}")
    print(f"  Próxima Acción Decidida: {final_state['next_action']}")

Explicación del Código de Ejemplo

El código anterior construye un grafo operativo. El estado ProjectState es nuestro contenedor de memoria. El grafo comienza en add_tasks, que simula la adición de tareas. Luego fluye incondicionalmente a prioritize, donde se calculan las puntuaciones y categorías. Aquí es donde la memoria se enriquece. La función route_after_analysis actúa como el guardián condicional: lee la categoría de la tarea principal del estado y devuelve un string literal que LangGraph usa para elegir el siguiente nodo (critical_action, high_action, etc.). Cada nodo de acción actualiza el estado con la decisión tomada, cerrando el ciclo de "análisis-decisión-acción".

Errores Comunes y Cómo Evitarlos

Al integrar memoria y lógica condicional, varios errores sutiles pueden aparecer. Aquí están los más comunes y sus soluciones.

1. Estado Mutado Inesperadamente (Efectos Secundarios Peligrosos): Las funciones de nodo y enrutamiento reciben el estado. Modificarlo directamente sin usar el mecanismo de retorno de LangGraph puede causar inconsistencias. Solución: Siempre devuelve un diccionario con las claves del estado que quieres actualizar. LangGraph se encarga de la fusión de forma predecible. Evita las mutaciones in-place como state['key'].append(value) a menos que uses campos anotados con operator.add.

2. Lógica de Enrutamiento Demasiado Acoplada: Poner toda la lógica de decisión dentro de la función route_after_analysis puede hacerla enorme y frágil. Solución: Usa los nodos para preparar el estado con campos de decisión claros (ej: state['recommended_flow'] = 'critical'). La función de enrutamiento puede entonces ser simple, solo leyendo este campo. Esto separa las preocupaciones y facilita las pruebas.

3. No Gestionar Todos los Caminos Condicionales: Si tu función de enrutamiento puede devolver un valor no mapeado en el diccionario de destinos, el grafo lanzará una excepción. Solución: Asegúrate de que el diccionario de mapeo en add_conditional_edges cubra todos los valores de retorno posibles de tu función de enrutamiento. Incluye siempre un camino por defecto o maneja explícitamente casos como listas vacías o estados de error.

4. Olvidar la Persistencia del Estado entre Ejecuciones: En una aplicación real, el estado debe ser persistido (en base de datos, archivo, memoria distribuida) entre invocaciones del grafo. Si no lo haces, el agente "olvidará" todo al reiniciar. Solución: Integra un checkpointer de LangGraph. Configúralo para que serialice el estado después de cada ejecución y lo cargue al inicio de una nueva sesión, usando un ID de hilo (thread_id) para identificar cada conversación o proyecto único.

5. Condiciones Basadas en Datos No Actualizados: Un error clásico es que una arista condicional evalúe una parte del estado que un nodo anterior no ha actualizado aún en el ciclo actual. Solución: Dibuja tu grafo en papel. Sigue el flujo de datos y asegúrate de que cualquier nodo que necesite un dato esté conectado después del nodo que lo produce. Usa el depurador de LangGraph para inspeccionar el estado en cada paso.

Checklist de Dominio

Antes de considerar esta lección integrada, verifica que puedes realizar o explicar cada uno de los siguientes puntos:

• Diseñar un esquema de estado TypedDict que contenga todos los datos necesarios para la memoria a largo plazo y la toma de decisiones de un agente específico.
• Implementar al menos dos nodos que modifiquen el estado de manera aditiva (usando Annotated con operator.add) y no aditiva.
• Escribir una función de enrutamiento condicional que tome el estado como argumento y devuelva un literal de string, dictando la rama del grafo a seguir.
• Construir un grafo con StateGraph que utilice add_conditional_edges para crear una bifurcación con al menos tres posibles destinos.
• Explicar la diferencia entre una arista fija (add_edge) y una arista condicional, y cuándo usar cada una.
• Describir cómo un checkpointer permite la persistencia del estado a través de múltiples ejecuciones de un agente conversacional.
• Identificar y solucionar un problema de mutación de estado inadecuada dentro de un nodo del grafo.
• Extender la lógica de priorización para que considere dependencias entre tareas almacenadas en el estado, no solo atributos individuales.