Guía interactiva: Tu primer agente con LangChain + LangGraph

Nivel 1 · Fundamentos

Tu primer agente con create_agent

Objetivo: instalar LangChain y crear tu primer agente con la función create_agent.

create_agent (del paquete langchain) es hoy la forma recomendada de crear un agente ReAct: te da un modelo con capacidad de usar herramientas y razonar en varios pasos, con un sistema de middleware flexible para extenderlo. Por debajo corre sobre LangGraph.

1. Instala las dependencias

TERMINAL

pip install langchain langchain-openai
export OPENAI_API_KEY="sk-tu-api-key-aqui"

2. Tu primer agente

agente.py

from langchain.agents import create_agent

agent = create_agent(
    model="gpt-4o",
    tools=[],
)

resultado = agent.invoke({
    "messages": [{"role": "user", "content": "¿Qué es un agente de IA, en una frase?"}]
})

print(resultado["messages"][-1].content)

Todo agente creado con create_agent trabaja con un estado de mensajes (messages) — el mismo formato que usás en LangGraph, lo que hace que subir de nivel hacia grafos más complejos sea un paso natural, no una reescritura.

Reto

Cambiá model por otro proveedor (por ejemplo un modelo de Anthropic vía langchain-anthropic) sin cambiar nada más del código.

Nivel 2 · Personalidad

System prompt e instrucciones

Objetivo: moldear el comportamiento del agente con un prompt de sistema y parámetros del modelo.

agente.py

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o", temperature=0.3, max_tokens=400)

agent = create_agent(
    model=model,
    tools=[],
    system_prompt=(
        "Sos un asistente técnico de soporte. Respondé en español, "
        "breve y con pasos numerados. Si no sabés algo, decilo."
    ),
)

resultado = agent.invoke({"messages": [{"role": "user", "content": "La app se cierra al abrir la cámara"}]})
print(resultado["messages"][-1].content)

Pasar un objeto ChatOpenAI en vez de un string te da control fino sobre temperature, max_tokens y otros parámetros del proveedor.

Reto

Escribí un system prompt que rechace responder cualquier pregunta fuera de un dominio específico que elijas.

Nivel 3 · Herramientas

Herramientas con @tool

Objetivo: crear una herramienta custom con el decorador @tool de langchain_core.

agente.py

from langchain_core.tools import tool
from langchain.agents import create_agent

@tool
def clima_actual(ciudad: str) -> str:
    """Devuelve el clima actual reportado para una ciudad dada."""
    datos = {"Ciudad de Guatemala": "22°C, parcialmente nublado"}
    return datos.get(ciudad, "No tengo datos para esa ciudad")

agent = create_agent(model="gpt-4o", tools=[clima_actual])

resultado = agent.invoke({
    "messages": [{"role": "user", "content": "¿Cómo está el clima en Ciudad de Guatemala?"}]
})
print(resultado["messages"][-1].content)

El agente decide solo cuándo llamar clima_actual — el docstring de la función es la descripción que el modelo usa para decidirlo, igual que en la mayoría de frameworks de agentes modernos.

Reto

Agregá una segunda herramienta buscar_producto(nombre: str) y verificá que el agente combine ambas en una sola respuesta cuando se le pregunten las dos cosas a la vez.

Nivel 4 · Memoria

Memoria con checkpointer

Objetivo: persistir el historial de conversación entre llamadas usando un checkpointer y un thread_id.

Como create_agent corre sobre LangGraph, la memoria se maneja con un checkpointer: guarda el estado completo del grafo (incluyendo mensajes) asociado a un thread_id, para que puedas retomar la misma conversación después.

agente_memoria.py

from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent

agent = create_agent(
    model="gpt-4o",
    tools=[],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "cliente-42"}}

agent.invoke({"messages": [{"role": "user", "content": "Busco una laptop para diseño gráfico"}]}, config)
r = agent.invoke({"messages": [{"role": "user", "content": "¿Cuál es la más barata de las que mencionaste?"}]}, config)
print(r["messages"][-1].content)

InMemorySaver es ideal para prototipos — en producción se usa PostgresSaver o SqliteSaver para que el historial sobreviva a un reinicio del proceso.

Reto

Cambiá el thread_id a mitad de la conversación y confirmá que el agente "olvida" el contexto anterior — así comprobás que la memoria está atada al hilo, no al proceso.

Nivel 5 · Salida estructurada

response_format con Pydantic

Objetivo: forzar que el agente devuelva un objeto validado en vez de texto libre.

agente.py

from pydantic import BaseModel
from langchain.agents import create_agent

class TicketSoporte(BaseModel):
    categoria: str
    urgencia: str
    resumen: str

agent = create_agent(model="gpt-4o", tools=[], response_format=TicketSoporte)

resultado = agent.invoke({
    "messages": [{"role": "user", "content": "Se cierra la app al subir foto de perfil, es urgente"}]
})

ticket: TicketSoporte = resultado["structured_response"]
print(ticket.categoria, ticket.urgencia, ticket.resumen)

LangChain elige automáticamente la estrategia correcta: si el modelo soporta salida estructurada nativa (como los JSON schemas de OpenAI) la usa directamente; si no, envuelve tu schema como una tool interna que el modelo debe invocar.

Reto

Definí un modelo Pydantic para extraer rating, sentimiento y si menciona un problema específico de una reseña de producto.

Nivel 6 · Streaming

Streaming con .stream()

Objetivo: mostrar la respuesta del agente en tiempo real, token por token.

stream.py

for evento, metadata in agent.stream(
    {"messages": [{"role": "user", "content": "Explicame qué es RAG en 3 pasos"}]},
    stream_mode="messages",
):
    if evento.content:
        print(evento.content, end="", flush=True)

stream_mode="messages" te da los tokens del modelo a medida que se generan. Otros modos como "updates" te muestran cada paso del grafo (útil para depurar qué herramienta se ejecutó y cuándo). Para async, usá agent.astream() dentro de una función async def.

Reto

Envolvé el streaming en un endpoint de FastAPI con StreamingResponse, usando astream() en vez de stream().

Nivel 7 · Multi-agente

Handoffs entre agentes

Objetivo: transferir el control de una conversación de un agente especializado a otro usando el patrón de "handoff".

En LangGraph, un handoff es una tool especial que, al ejecutarse, no solo devuelve un resultado sino que le dice al grafo "de ahora en adelante, que siga otro agente". Se implementa devolviendo un objeto Command con goto (a quién transferir) y update (qué agregar al estado compartido).

handoff.py

from langchain_core.tools import tool
from langgraph.types import Command
from langgraph.prebuilt import InjectedState

@tool
def transferir_a_soporte(state: InjectedState) -> Command:
    """Transferir la conversación al especialista de soporte técnico."""
    return Command(
        goto="agente_soporte",
        update={"messages": state["messages"]},
        graph=Command.PARENT,
    )

# El agente principal recibe esta tool; cuando decide usarla,
# el control pasa directo al nodo "agente_soporte" del grafo

A diferencia de "agente como herramienta" (donde el agente A llama al agente B y espera su respuesta), un handoff transfiere el control completo — el agente B sigue la conversación directamente con el usuario.

Reto

Creá un segundo handoff transferir_a_ventas y probá una conversación donde el agente principal decida a cuál de los dos transferir.

Nivel 8 · Orquestador

Supervisor con StateGraph

Objetivo: construir un orquestador explícito que enruta cada mensaje al especialista correcto usando un grafo de estados.

Cuando necesitás ver y controlar cada decisión de enrutamiento explícitamente (útil para depurar y trazar en producción), construís el supervisor directamente con StateGraph en vez de dejar que un agente "decida solo".

orquestador.py

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END

class Estado(TypedDict):
    mensaje: str
    respuesta: str
    siguiente: str

def nodo_supervisor(estado: Estado) -> Estado:
    # El supervisor usa el modelo con salida estructurada para decidir el ruteo
    decision = supervisor_agent.invoke({"messages": [{"role": "user", "content": estado["mensaje"]}]})
    estado["siguiente"] = decision["structured_response"].agente  # "ventas" | "soporte" | "facturacion"
    return estado

def ruta(estado: Estado) -> Literal["ventas", "soporte", "facturacion"]:
    return estado["siguiente"]

grafo = StateGraph(Estado)
grafo.add_node("supervisor", nodo_supervisor)
grafo.add_node("ventas", nodo_ventas)
grafo.add_node("soporte", nodo_soporte)
grafo.add_node("facturacion", nodo_facturacion)

grafo.set_entry_point("supervisor")
grafo.add_conditional_edges("supervisor", ruta, {"ventas": "ventas", "soporte": "soporte", "facturacion": "facturacion"})
grafo.add_edge("ventas", END)
grafo.add_edge("soporte", END)
grafo.add_edge("facturacion", END)

app = grafo.compile()
print(app.invoke({"mensaje": "Me cobraron dos veces la suscripción"}))

Para el mismo problema, también existe el paquete langgraph-supervisor, que ya trae este patrón empaquetado si no necesitás personalizar la lógica de ruteo.

Reto

Agregá una arista de vuelta de cada especialista al supervisor (en vez de END) para soportar conversaciones multi-turno con varios especialistas en la misma sesión.

Nivel 9 · Producción

Trazas, errores y guardrails

Objetivo: instrumentar el grafo con LangSmith y manejar errores de forma robusta.

TERMINAL

export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY="tu-api-key"
export LANGSMITH_PROJECT="agente-soporte-prod"
python orquestador.py
# Cada nodo del grafo, cada tool call y cada tokens generado
# queda trazado y visible en smith.langchain.com

reintentos.py

from langchain_core.runnables import RunnableConfig

app_con_reintentos = app.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
)

resultado = app_con_reintentos.invoke(
    {"mensaje": "..."},
    config=RunnableConfig(recursion_limit=15),  # evita loops infinitos del grafo
)

recursion_limit — evita que el grafo entre en un ciclo infinito de nodos
with_retry — reintentos automáticos con backoff ante errores transitorios del proveedor
Checkpointer persistente (Postgres) — para poder auditar y retomar conversaciones tras un fallo

Reto

Configurá LangSmith en el orquestador del Nivel 8 y revisá la traza completa de una conversación con enrutamiento a 2 especialistas distintos.

Nivel 10 · Arquitectura final

Supervisor + load balancer en producción

Objetivo: desplegar el supervisor y cada especialista como servicios independientes, balanceados y con checkpointer compartido.

1. Cada especialista como su propio servicio

servicio_soporte.py

from fastapi import FastAPI
from pydantic import BaseModel
from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver

app = FastAPI()
checkpointer = PostgresSaver.from_conn_string("postgresql://user:pass@db:5432/agentes")
agente_soporte = create_agent(model="gpt-4o", tools=[], checkpointer=checkpointer)

class Consulta(BaseModel):
    mensaje: str
    thread_id: str

@app.post("/consultar")
async def consultar(c: Consulta):
    config = {"configurable": {"thread_id": c.thread_id}}
    r = await agente_soporte.ainvoke({"messages": [{"role": "user", "content": c.mensaje}]}, config)
    return {"respuesta": r["messages"][-1].content}

# uvicorn servicio_soporte:app --host 0.0.0.0 --port 8001

Usar un checkpointer compartido (Postgres) en vez de InMemorySaver es lo que permite correr varias réplicas del mismo servicio sin que cada una tenga su propia memoria aislada e inconsistente.

2. Load balancer entre réplicas

nginx.conf

upstream agentes_soporte {
    least_conn;
    server soporte-1:8001;
    server soporte-2:8001;
    server soporte-3:8001;
}
server {
    listen 80;
    location /soporte/ { proxy_pass http://agentes_soporte/; }
}

3. Arquitectura completa

arquitectura final

Cliente
  │
  ▼
Supervisor (StateGraph, servicio propio)
  │
  ▼
Cola de trabajos (Redis / SQS)
  │
  ├──▶ Load Balancer ──▶ [Worker Ventas x3]
  ├──▶ Load Balancer ──▶ [Worker Soporte x3]
  └──▶ Load Balancer ──▶ [Worker Facturación x3]
                              │
                              ▼
                   PostgresSaver compartido (memoria)
                              │
                              ▼
                    LangSmith (trazas de todo el flujo)

Con esto tenés el camino completo: de un create_agent() de una línea en el Nivel 1, a un supervisor multi-agente en producción, con memoria persistente, escalado horizontal y trazas de punta a punta.

Reto final

Separá los 3 especialistas del Nivel 8 en servicios FastAPI independientes con checkpointer Postgres compartido, y montá 2 réplicas de uno de ellos detrás de nginx.

Volver a Guías →