Implementa bucles para tareas iterativas

Introducción: El Poder de la Iteración en Agentes de IA

En la construcción de agentes de IA avanzados con LangGraph, la capacidad de realizar tareas repetitivas y adaptativas es fundamental. Hasta ahora, hemos explorado flujos lineales y bifurcaciones condicionales. Sin embargo, los problemas del mundo real rara vez se resuelven en un solo paso. Tareas como la búsqueda iterativa de información, el refinamiento progresivo de una respuesta, o la ejecución de un proceso hasta cumplir un criterio, requieren de bucles. En esta lección, profundizaremos en la implementación de bucles dentro de LangGraph, pasando de la teoría a la práctica con ejemplos complejos y funcionales.

Un bucle, en el contexto de LangGraph, no es un simple while de programación tradicional. Es un ciclo controlado y observable dentro del grafo de estado del agente. Esto permite que el agente "recuerde" sus iteraciones pasadas, evalúe su progreso y decida cuándo continuar o finalizar. Dominar este patrón es lo que separa a un agente que ejecuta una instrucción de uno que puede persistir en la resolución de un problema complejo, emulando un razonamiento más profundo y metódico.

Concepto Clave: El Grafo como Máquina de Estados Iterativa

Para entender los bucles en LangGraph, debemos visualizar el grafo no como una secuencia fija, sino como una máquina de estados finitos con retroalimentación. El núcleo de esta funcionalidad es la capacidad de un nodo (o un conjunto de nodos) de apuntar a un nodo anterior, creando un ciclo en el grafo dirigido. El estado (State) del agente es el vehículo que viaja por este ciclo, mutando en cada iteración. La clave para evitar bucles infinitos es la función de condición de parada (conditional edge), que evalúa el estado y decide si el flujo debe continuar en el ciclo o salir hacia una ruta de finalización.

Una analogía del mundo real sería un investigador que está redactando un informe. Su proceso no es lineal: 1) Escribe un borrador (estado inicial). 2) Lo revisa (nodo de procesamiento). 3) Evalúa: ¿Está lo suficientemente bueno? (condición de parada). Si la respuesta es NO, regresa al paso 2, añadiendo correcciones y anotaciones al borrador (modifica el estado). Si la respuesta es SÍ, procede a 4) Enviar el informe (nodo final). El "estado" aquí es el borrador del informe, que se enriquece iterativamente. LangGraph formaliza y automatiza este tipo de flujo de trabajo cíclico para un agente de IA.

Cómo Funciona en la Práctica: Anatomía de un Bucle LangGraph

Implementar un bollo requiere definir claramente tres componentes: el cuerpo del bucle, la condición de continuación y la condición de salida. En LangGraph, esto se traduce típicamente en una estructura donde un nodo principal (como un agente de LLM con herramientas) está conectado a sí mismo a través de una arista condicional. Esta arista es gestionada por una función router que, basándose en el estado, decide si se debe "REITERAR" o "FINALIZAR". El estado debe incluir un contador o un indicador de progreso (como una lista de pasos completados) para que la lógica de decisión tenga base.

Veamos un ejemplo paso a paso para un agente que resuelve un problema matemático complejo descomponiéndolo: 1) El estado inicial contiene el problema. 2) El nodo "Resolver Paso" analiza el problema, identifica el siguiente sub-problema más simple, lo resuelve y añade la solución a una lista en el estado. 3) La función should_continue examina el estado: ¿Quedan sub-problemas por resolver? Si sí, dirige el flujo de vuelta a "Resolver Paso". Si no, dirige el flujo al nodo "Sintetizar Respuesta". 4) El nodo "Sintetizar Respuesta" toma todas las soluciones parciales de la lista en el estado y genera una respuesta final coherente. Este patrón asegura que el agente trabaje de forma metódica y autónoma.

Tip Crítico: El diseño del estado es la parte más importante al implementar bucles. Debes incluir explícitamente los campos que actuarán como "memoria de trabajo" para la iteración (ej., `steps_taken`, `remaining_tasks`, `partial_result`) y los campos que almacenan el resultado final. Planifica cómo cada ciclo modificará estos campos.

Código en Acción: Agente Iterativo de Investigación y Síntesis

A continuación, presentamos un ejemplo completo y funcional de un agente que realiza una investigación iterativa sobre un tema, consultando una herramienta de búsqueda en cada ciclo y sintetizando la información progresivamente hasta alcanzar un criterio de profundidad o exhaustividad.

from typing import TypedDict, List, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
import operator
import sys

# 1. Definir el Estado del Grafo
class AgentState(TypedDict):
    """Estado del agente para el bucle de investigación."""
    # Historial de mensajes para el LLM
    messages: Annotated[List, add_messages]
    # Tema de investigación original
    original_query: str
    # Lista de ángulos o sub-preguntas a investigar
    research_angles: List[str]
    # Ángulos ya investigados
    completed_angles: List[str]
    # Resultados acumulados de cada búsqueda
    findings: List[str]
    # Síntesis acumulativa
    comprehensive_summary: str
    # Contador de iteraciones para límite de seguridad
    iteration_count: int

# 2. Inicializar Modelo y Herramientas
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)
search_tool = TavilySearchResults(max_results=2)
tools = [search_tool]
llm_with_tools = llm.bind_tools(tools)

# 3. Definir la Función Nodo: Un Paso de Investigación
def research_one_angle(state: AgentState):
    """Ejecuta un ciclo de investigación para un ángulo pendiente."""
    print(f"\n[DEBUG] Iniciando ciclo de investigación. Iteración: {state['iteration_count']}", file=sys.stderr)

    # 1. Seleccionar el próximo ángulo a investigar
    current_angle = state['research_angles'][0]
    updated_angles = state['research_angles'][1:]
    updated_completed = state['completed_angles'] + [current_angle]

    # 2. Formular pregunta para la herramienta de búsqueda
    research_prompt = f"{state['original_query']}, enfocándome específicamente en: {current_angle}. Proporciona información factual y detallada."
    print(f"[DEBUG] Investigando ángulo: {current_angle}", file=sys.stderr)

    # 3. Invocar al LLM para decidir usar la herramienta (en un caso real, aquí iría la lógica de decición)
    # Para simplificar, asumimos que siempre busca.
    search_result = search_tool.invoke(research_prompt)
    result_str = f"Ángulo: {current_angle}\nResultados: {search_result}\n---"

    # 4. Actualizar el estado
    new_findings = state['findings'] + [result_str]
    new_iteration = state['iteration_count'] + 1

    # 5. Crear mensaje para el historial
    ai_message = {"role": "assistant", "content": f"He investigado '{current_angle}'. Encontré {len(search_result)} resultados."}

    return {
        "research_angles": updated_angles,
        "completed_angles": updated_completed,
        "findings": new_findings,
        "iteration_count": new_iteration,
        "messages": state['messages'] + [ai_message]
    }

# 4. Definir la Función de Enrutamiento (Condición)
def decide_to_continue(state: AgentState):
    """Decide si el bucle debe continuar o finalizar."""
    # Condición 1: ¿Quedan ángulos por investigar?
    angles_remaining = len(state['research_angles']) > 0
    # Condición 2: Límite de seguridad para evitar bucles infinitos
    under_iteration_limit = state['iteration_count'] < 5

    if angles_remaining and under_iteration_limit:
        print(f"[DEBUG] Continuando. Ángulos restantes: {state['research_angles']}", file=sys.stderr)
        return "research_step"  # Nombre del nodo al que volver
    else:
        print(f"[DEBUG] Finalizando bucle. Motivo: Ángulos restantes={angles_remaining}, Iteración < 5={under_iteration_limit}", file=sys.stderr)
        return "synthesize"

# 5. Definir el Nodo de Síntesis Final
def synthesize_findings(state: AgentState):
    """Sintetiza todos los hallazgos en un resumen coherente."""
    print(f"\n[DEBUG] Sintetizando {len(state['findings'])} hallazgos.", file=sys.stderr)
    all_findings_text = "\n\n".join(state['findings'])

    synthesis_prompt = f"""
    Basándote en las siguientes investigaciones fragmentadas sobre '{state['original_query']}',
    genera un informe final comprehensivo, estructurado y bien redactado.

    Investigaciones:
    {all_findings_text}
    """
    synthesis_response = llm.invoke(synthesis_prompt)

    final_message = {"role": "assistant", "content": f"# Informe Final\n\n{synthesis_response.content}"}
    return {
        "comprehensive_summary": synthesis_response.content,
        "messages": state['messages'] + [final_message]
    }

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

# Agregar nodos
workflow.add_node("research_step", research_one_angle)
workflow.add_node("synthesize", synthesize_findings)

# Establecer punto de entrada
workflow.set_entry_point("research_step")

# Agregar arista condicional DESPUÉS de `research_step`
workflow.add_conditional_edges(
    "research_step",
    decide_to_continue,
    {
        "research_step": "research_step",  # Bucle a sí mismo
        "synthesize": "synthesize"        # Salida del bucle
    }
)
# Conectar el nodo de síntesis al final
workflow.add_edge("synthesize", END)

# Compilar el grafo
app = workflow.compile()

# 7. Ejecutar el Agente
initial_state = AgentState(
    messages=[{"role": "user", "content": "Investiga las implicaciones éticas de la IA generativa."}],
    original_query="Implicaciones éticas de la IA generativa",
    research_angles=["Sesgo algorítmico y discriminación", "Propiedad intelectual y derechos de autor", "Impacto en el empleo y la economía", "Desinformación y deepfakes"],
    completed_angles=[],
    findings=[],
    comprehensive_summary="",
    iteration_count=0
)

print("=== INICIANDO EJECUCIÓN DEL AGENTE ITERATIVO ===")
final_state = app.invoke(initial_state, config={"recursion_limit": 50})
print("\n=== SÍNTESIS FINAL ===")
print(final_state["comprehensive_summary"][:500] + "...")  # Muestra primeras 500 chars

Patrones Avanzados: Bucles Anidados y Paralelismo Controlado

Para problemas de mayor complejidad, LangGraph permite la construcción de bucles anidados. Imagina un agente que, para resolver una tarea principal (bucle externo), necesita completar varias subtareas, cada una de las cuales puede requerir su propio proceso iterativo (bucle interno). La implementación requiere grafo jerárquicos o un diseño de estado más sofisticado que lleve la cuenta del contexto de cada nivel. Por ejemplo, un agente de desarrollo de software podría tener un bucle externo para "Implementar función" y un bucle interno para "Depurar error", donde la salida del bucle interno alimenta la siguiente iteración del bucle externo.

Otro patrón poderoso es el paralelismo controlado dentro de un ciclo. Aunque LangGraph ejecuta nodos secuencialmente por defecto, puedes diseñar un bucle donde en cada iteración se lance un conjunto de tareas independientes (por ejemplo, consultar múltiples fuentes de datos) cuyos resultados se agregan antes de la siguiente evaluación. Esto se logra teniendo un nodo que, basado en el estado, bifurca el flujo a varios nodos hermanos que se ejecutan en paralelo (usando concurrent o map en versiones avanzadas), y un nodo posterior que converge sus resultados de vuelta al estado, listo para la evaluación de la condición de continuación.

# Esquema conceptual para un bucle con paso de paralelización
from langgraph.graph import StateGraph, END
from typing import List
import asyncio

class ParallelState(TypedDict):
    task_list: List[str]
    pending_subtasks: List[List[str]]
    results: List[str]
    phase: str  # 'expand', 'parallel_work', 'aggregate'

def expand_into_subtasks(state: ParallelState):
    """Convierte una tarea principal en subtareas independientes."""
    # Lógica para descomponer state['task_list'][0] en una lista de subtareas
    subtasks = [f"Subtarea_{i}_para_{state['task_list'][0]}" for i in range(3)]
    return {
        "pending_subtasks": state['pending_subtasks'] + [subtasks],
        "task_list": state['task_list'][1:]  # Elimina la tarea procesada
    }

def process_subtask_batch(state: ParallelState):
    """Procesa un lote de subtareas en paralelo (simulado)."""
    current_batch = state['pending_subtasks'][0]
    # En un caso real, aquí se lanzarían varias herramientas o LLM calls concurrentes.
    batch_results = [f"Resultado_de_{subtask}" for subtask in current_batch]
    return {
        "results": state['results'] + batch_results,
        "pending_subtasks": state['pending_subtasks'][1:],  # Elimina el lote procesado
        "phase": "aggregate"
    }

# El grafo controlaría el flujo: expand -> process_subtask_batch -> (condición) -> expand o END.

Errores Comunes y Cómo Evitarlos

Al implementar bucles en LangGraph, varios errores sutiles pueden causar comportamientos inesperados o fallos catastróficos. Aquí detallamos los más frecuentes:

1. Bucle Infinito por Condición de Parada Mal Definida: La función router siempre devuelve la acción que reactiva el ciclo. Solución: Incluye siempre un criterio de salida absoluto, como un contador de iteraciones máximo (`iteration_count`), y asegúrate de que el estado se modifique de tal forma que la condición eventualmente se cumpla (ej., vaciar una lista de `pending_tasks`).

2. Estado que no Progresa (Loop Estático): Cada iteración debe modificar el estado de manera significativa. Un error común es olvidar actualizar el campo que la función `decide_to_continue` evalúa. Solución: En cada nodo del ciclo, actualiza explícitamente al menos un campo que sea leído por la condición de continuación (ej., mover un ítem de `pending` a `completed`).

3. Pérdida de Contexto o Historial: En ciclos largos, el historial de mensajes puede crecer demasiado, excediendo el contexto del LLM o ralentizando el sistema. Solución: Implementa una estrategia de resumen o "ventana deslizante". Mantén en `messages` solo los intercambios esenciales, y almacena el detalle extenso en otro campo del estado (como `findings`), que puede ser resumido periódicamente.

4. Mala Gestión de Herramientas en Ciclos: Invocar la misma herramienta con los mismos parámetros en cada ciclo, llevando a resultados idénticos. Solución: Diseña la lógica del nodo para que cada iteración formule una consulta diferente, basada en el progreso acumulado. Usa campos del estado como `completed_angles` o `last_result` para variar la solicitud a la herramienta.

5. Ignorar la Recursión del Sistema: LangGraph tiene un límite de recursión por defecto para proteger la estabilidad. Un bucle muy largo puede alcanzarlo. Solución: Al invocar el grafo, configura el parámetro `config={"recursion_limit": N}` con un valor adecuado (ej., 50 o 100). Mejor aún, diseña tu condición de parada para que el bucle termine mucho antes de acercarse a ese límite.

Tip de Depuración: Siempre incluye sentencias `print` o logging detallado en las funciones de nodo y enrutamiento, mostrando los valores clave del estado y la decisión tomada. Usa `file=sys.stderr` para no contaminar la salida del agente. Visualizar el grafo con `app.get_graph().draw_mermaid_png()` es invaluable para entender el flujo cíclico.

Checklist de Dominio

Antes de considerar que dominas la implementación de bucles en LangGraph, verifica que puedes realizar y comprendes cada uno de los siguientes puntos:

• Puedo diseñar un esquema de estado (State) que incluya campos específicos para controlar un bucle (listas de pendientes/completados, contador, resultado parcial).
• Sé construir un grafo donde un nodo apunta condicionalmente a sí mismo, creando explícitamente un ciclo.
• Puedo escribir una función de enrutamiento (conditional edge function) que evalúe el estado y devuelva diferentes destinos, incluyendo la opción de continuar el bucle.
• Implemento de forma robusta un límite máximo de iteraciones (contador en el estado) como salvaguarda contra bucles infinitos.
• Sé cómo modificar el estado dentro del cuerpo del bucle para garantizar que cada iteración sea diferente y progrese hacia la condición de salida.
• Puedo explicar la diferencia entre un bucle simple y un patrón de bucle anidado o con procesamiento paralelo interno.
• He depurado y solucionado al menos un problema de bucle infinito o estado estancado en mi propio código.
• Puedo documentar el flujo iterativo de mi agente, incluyendo las condiciones de entrada, proceso y salida del ciclo.