ruta ai/agentes / módulo 3

A3 — Agentes (LangGraph)

Outcome del módulo: construir un agente real con LangGraph — estado tipado, nodos, edges condicionales, tool-calling, memoria y límite de pasos — que decide, en vez de ejecutar un pipeline fijo de antemano.

Concepto 1 — Qué es un agente (loop pensar → actuar → observar)

Idea núcleo: un agente es un LLM corriendo en loop que decide una acción, la ejecuta, observa el resultado, y con eso decide la siguiente acción — a diferencia de un pipeline de pasos fijos definidos de antemano.

Entender (capa 1)

Texto: un pipeline tradicional (chain) ejecuta pasos A→B→C siempre en ese orden. Un agente, en el patrón ReAct (Reason + Act), en cada turno: (1) razona sobre el estado actual y decide qué hacer, (2) actúa (llama una tool, o responde), (3) observa el resultado de la acción, y vuelve a (1) hasta decidir que terminó. El “camino” no está fijo de antemano — lo decide el modelo en cada paso según lo que va observando.

Visual:

flowchart LR
    P[Pensar: decidir próxima acción] --> A[Actuar: ejecutar tool o responder]
    A --> O[Observar: leer el resultado]
    O --> P

Video: AI Agentic Design Patterns: ReAct Explained | Reasoning + Acting — CodeCraft Academy

Ejemplo:

def run_agent_loop(pregunta: str, max_steps: int = 3) -> str:
    historial = [{"role": "user", "content": pregunta}]
    for _ in range(max_steps):
        decision = llm_decide(historial)          # pensar
        if decision["action"] == "respond":
            return decision["content"]
        resultado = ejecutar_tool_mock(decision["action"], decision["args"])  # actuar
        historial.append({"role": "tool", "content": resultado})              # observar
    return "no se resolvió en el límite de pasos"

Fijar (capa 2)

Nota atómica:

Feynman: “Un pipeline es una receta de cocina: siempre los mismos pasos en el mismo orden. Un agente es un cocinero: prueba, decide qué le falta, corrige, prueba de nuevo, hasta que el plato está listo.”

Aplicar (capa 3)

Escribí a mano (sin framework) un loop de máximo 3 iteraciones que llame a un LLM, parseé su decisión (acción + argumentos), y ejecute una tool mock según esa decisión.

Límites

Concepto 2 — LangGraph: state, nodes, edges, conditional edges

Idea núcleo: LangGraph modela el agente como un grafo con un estado tipado explícito; los nodos son funciones que transforman ese estado, los edges (incluidos los condicionales) definen el flujo, y END — el sentinel importado de langgraph.graph, no el string "END" — marca la salida del grafo.

Entender (capa 1)

Texto: StateGraph se construye sobre un esquema de estado (típicamente un TypedDict). Cada nodo es una función state -> dict que devuelve una actualización parcial del estado. Los edges conectan nodos; los conditional edges usan una función de routing que lee el estado y devuelve el nombre del próximo nodo — o el sentinel END para terminar. Punto crítico: END es un objeto especial que LangGraph reconoce internamente al compilar el grafo. Comparar o devolver el string "END" a mano no termina el grafo salvo que vos mismo hayas definido un nodo llamado literalmente "END" — hay que importar y usar el sentinel real.

Visual:

flowchart TD
    Start([entry point]) --> Agent["Nodo: agent llama al LLM"]
    Agent -->|conditional edge| Decision{should_continue}
    Decision -->|hay tool_calls| Tools["Nodo: tools"]
    Decision -->|no hay tool_calls| End(["END sentinel"])
    Tools --> Agent

Video: LangGraph Tutorial - Implementing Conditional Edges for Dynamic Navigation — Mohamed Naji Aboo

Ejemplo:

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    steps: int

def call_model(state: AgentState) -> dict:
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response], "steps": state["steps"] + 1}

def should_continue(state: AgentState) -> str:
    last = state["messages"][-1]
    if getattr(last, "tool_calls", None):
        return "tools"
    return END                    # sentinel real, NUNCA el string "END"

graph = StateGraph(AgentState)
graph.add_node("agent", call_model)
graph.add_node("tools", ToolNode(tools))
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
graph.add_edge("tools", "agent")

app = graph.compile()

Fijar (capa 2)

Nota atómica:

Feynman: “END no es una etiqueta de texto, es una llave especial que abre la puerta de salida. Si escribís la palabra ‘END’ en un papel, no abre nada; necesitás la llave de verdad, que se importa del módulo.”

Aplicar (capa 3)

Armá un grafo mínimo de 2 nodos (agent, tools) con un conditional edge que, cuando no hay tool_calls, termine en el END real importado. Compilalo e invocalo con un mensaje simple.

Límites

Concepto 3 — Tool-calling desde el agente

Idea núcleo: el LLM no ejecuta tools directamente — devuelve una decisión estructurada (tool_calls) que tu código ejecuta; el resultado se reinyecta al estado como un ToolMessage para que el modelo lo lea en el siguiente paso.

Entender (capa 1)

Texto: con bind_tools, el modelo recibe el schema de cada tool disponible y, cuando decide usarlas, responde con tool_calls (nombre + argumentos), no ejecuta nada por sí mismo. El nodo ToolNode (prebuilt de LangGraph) lee esos tool_calls del último mensaje, ejecuta la función Python real correspondiente, y agrega el resultado al estado como ToolMessage. El siguiente paso del agent ve ese resultado en el historial y decide de nuevo.

Visual:

flowchart LR
    LLM["Modelo decide: tool_calls"] --> Node["ToolNode ejecuta la función real"]
    Node --> Msg["ToolMessage con el resultado"]
    Msg --> LLM

Video: LangGraph Tutorial: Introduction to Tool Use with ToolNode — Mohamed Naji Aboo

Ejemplo:

from langchain_core.tools import tool

@tool
def buscar_clima(ciudad: str) -> str:
    """Devuelve el clima actual de una ciudad dado su nombre."""
    return clima_api(ciudad)

tools = [buscar_clima]
llm_with_tools = llm.bind_tools(tools)

# el nodo agent llama a llm_with_tools.invoke(mensajes)
# si el modelo decide usar la tool, el mensaje de respuesta trae .tool_calls poblado
# ToolNode(tools) lo detecta y ejecuta buscar_clima(ciudad=...) automáticamente

Fijar (capa 2)

Nota atómica:

Feynman: “El modelo es un cliente que llena una orden de pedido con lo que quiere. El mesero (tu código) es quien va a la cocina, prepara el plato de verdad, y se lo trae de vuelta a la mesa.”

Aplicar (capa 3)

Definí una tool con @tool y bindeala al LLM en tu grafo del Concepto 2. Verificá con una pregunta que la amerite que el modelo la invoca, y con una que no la amerite que el modelo responde directo sin llamarla.

Límites

Concepto 4 — Memoria / estado compartido

Idea núcleo: el estado del grafo (incluido el historial de mensajes) es la memoria de corto plazo del agente dentro de una ejecución; persistir memoria entre ejecuciones distintas requiere un checkpointer.

Entender (capa 1)

Texto: dentro de una sola invocación (app.invoke(...)), el estado (AgentState) fluye de nodo a nodo y acumula el historial de mensajes — esa es la memoria de corto plazo del agente, todo en RAM. Para que el agente recuerde algo entre invocaciones distintas (una conversación multi-turno real, con el usuario volviendo horas después), LangGraph usa un checkpointer (ej. MemorySaver en memoria, o un checkpointer de Postgres para producción) que persiste el estado asociado a un thread_id. Al invocar de nuevo con el mismo thread_id, el grafo carga el estado guardado y continúa desde ahí.

Visual:

flowchart LR
    R1["Run 1: state + mensajes"] --> CP["Checkpoint guardado por thread_id"]
    CP --> R2["Run 2: mismo thread_id, carga el checkpoint"]
    R2 --> R3["El agente recuerda el turno anterior"]

Video: LangGraph Tutorial: Conversation State with Checkpoints & Memory — Mohamed Naji Aboo

Ejemplo:

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "usuario-123"}}

app.invoke({"messages": [("user", "mi nombre es Josse")], "steps": 0}, config=config)
# ... más tarde, misma conversación ...
app.invoke({"messages": [("user", "¿cómo me llamo?")], "steps": 0}, config=config)
# el modelo ve el historial completo del thread_id "usuario-123" y responde "Josse"

Fijar (capa 2)

Nota atómica:

Feynman: “El estado dentro de una corrida es como la memoria de corto plazo mientras hablás. El checkpointer es la libreta donde anotás la conversación para poder seguirla mañana, identificada por el número de conversación (thread_id).”

Aplicar (capa 3)

Corré el mismo grafo dos veces con el mismo thread_id y confirmá que el segundo invoke recuerda información dada en el primero.

Límites

Concepto 5 — Multi-step y límites de pasos (evitar loops infinitos)

Idea núcleo: un agente sin límite de pasos puede loopear indefinidamente (llamando tools sin converger a una respuesta) — hay que capar pasos explícitamente, tanto en el propio estado como con el recursion_limit del compilador.

Entender (capa 1)

Texto: nada garantiza que el modelo decida terminar. Una tool que falla repetidamente, un bug en el prompt, o un caso ambiguo pueden hacer que el agente reintente para siempre. Dos capas de protección: (1) un contador steps en el propio AgentState, incrementado en cada paso del nodo agent, chequeado en la función de routing para forzar END si se pasa de un máximo; (2) el recursion_limit que LangGraph acepta en la config de invoke, como red de seguridad adicional a nivel del framework.

Visual:

flowchart TD
    S["steps += 1 en cada paso del agente"] --> Chk{"steps >= MAX_STEPS?"}
    Chk -->|sí| End(["END forzado"])
    Chk -->|no| Cont["sigue evaluando tool_calls normalmente"]

Video: LangGraph Tutorial: Mastering Recursion Limits — Avoiding Infinite Loops — Mohamed Naji Aboo

Ejemplo:

MAX_STEPS = 6

def should_continue(state: AgentState) -> str:
    if state["steps"] >= MAX_STEPS:
        return END                          # corta antes de evaluar tool_calls
    last = state["messages"][-1]
    if getattr(last, "tool_calls", None):
        return "tools"
    return END

app.invoke(estado_inicial, config={"configurable": {"thread_id": "t1"}, "recursion_limit": 25})

Fijar (capa 2)

Nota atómica:

Feynman: “El contador propio es el freno de mano que vos calibrás para tu caso. El recursion_limit es el airbag que salta si todo lo demás falló — no querés depender solo del airbag.”

Aplicar (capa 3)

Forzá un caso donde el agente querría loopear (una tool que siempre devuelve un error que invita a reintentar) y verificá que el límite de pasos corta la ejecución en el máximo definido, devolviendo algo razonable.

Límites

Concepto 6 — Human-in-the-loop (compacto)

Idea núcleo: para acciones sensibles o irreversibles, el grafo se pausa antes de ejecutar el nodo crítico y espera aprobación humana explícita antes de continuar.

Entender (capa 1)

Texto: LangGraph soporta interrupt_before (o interrupt_after) al compilar el grafo, indicando en qué nodo pausar. Al llegar ahí, la ejecución se detiene y el estado queda guardado (requiere checkpointer); un humano revisa ese estado por fuera, y la ejecución se reanuda invocando de nuevo con el mismo thread_id (típicamente pasando None como input para continuar donde quedó).

Visual:

flowchart LR
    Node["Nodo tools: acción sensible"] -->|interrupt_before| Pause["Grafo pausado"]
    Pause --> Human["Humano revisa el estado"]
    Human -->|aprueba| Resume["Grafo continúa"]

Video: LangGraph interrupt: building human-in-the-loop agents — LangChain

Ejemplo:

app = graph.compile(checkpointer=checkpointer, interrupt_before=["tools"])

config = {"configurable": {"thread_id": "t1"}}
app.invoke({"messages": [("user", "cancelá mi suscripción")], "steps": 0}, config=config)
# se pausa antes de ejecutar el nodo "tools"

# ... humano revisa state y aprueba ...
app.invoke(None, config=config)   # continúa desde donde quedó

Nota atómica: Front: ¿qué resuelve interrupt_before que no resuelve un chequeo dentro de la tool? Back: pausa la ejecución completa del grafo ANTES de que la tool corra, dejando el estado persistido para revisión externa — un chequeo interno de la tool ya la estaría ejecutando.

Feynman: “Es un semáforo en rojo justo antes de la acción que no se puede deshacer. El grafo se detiene solo, alguien mira, y recién ahí se le da luz verde para seguir.”

Aplicar (capa 3)

Agregá interrupt_before=["tools"] sobre la tool más sensible de tu agente y simulá la aprobación manual reinvocando con None.

Límites

Build step

Convertí el retrieval de A1/A2 en una tool (buscar_en_corpus). Armá un agente LangGraph (state, nodo agent, nodo tools, conditional edge con END real) que decida cuándo usar esa tool y cuándo responder directo, con un límite de pasos explícito (Concepto 5).

Checklist de dominio

Marcás cuando podés explicar (Feynman) + aplicar cada uno:

Salida verificable del módulo: un agente LangGraph que recibe una pregunta, decide si necesita el RAG (tool) o no, ejecuta hasta N pasos como máximo sin loopear, y puede explicar en voz alta por qué usó END como sentinel y no como string.