LangGraph no es otra librería de «IA generativa» para hacer demos. Es una herramienta para orquestar agentes que necesitan tomar decisiones en tiempo real, ejecutar herramientas externas y ajustar su plan sobre la marcha. Si alguna vez has intentado que un LLM haga algo más complejo que responder preguntas, sabes que los prompts lineales se quedan cortos rápido. LangGraph resuelve ese problema.
Qué es LangGraph y cuándo tiene sentido usarlo
LangGraph es una librería construida sobre LangChain que modela el comportamiento de un agente como un grafo de estado. Los nodos son pasos de procesamiento (llamar al modelo, ejecutar una herramienta, tomar una decisión) y las aristas son las transiciones entre ellos, que pueden ser condicionales.
La diferencia clave con LangChain puro es que LangGraph permite ciclos. Un pipeline de LangChain ejecuta pasos en orden y termina. Un grafo de LangGraph puede volver atrás, repetir nodos y bifurcarse según lo que devuelva el modelo. Eso es lo que necesitas cuando el agente tiene que decidir en tiempo real qué herramienta usar y si el resultado que ha obtenido es suficiente o no.
Úsalo cuando el flujo de trabajo no sea lineal: el agente necesita buscar información, evaluar si es suficiente, buscar más si no lo es y solo entonces redactar la respuesta. Eso no cabe en una cadena.
El patrón ReAct: razonar y actuar en bucle
ReAct (Reason + Act) es el patrón más habitual para construir agentes con LLMs. El bucle es sencillo: el modelo piensa qué necesita, elige una herramienta, la ejecuta, observa el resultado y vuelve a pensar. Así hasta que considera que tiene suficiente información para responder.
Funciona mejor que los prompts lineales para tareas complejas porque el modelo puede corregirse. Si la primera búsqueda no da lo que necesitaba, puede reformular y volver a intentarlo. Con un prompt estático eso no es posible.
Eso sí, tiene limitaciones. El agente puede quedarse atascado en un bucle si el modelo no converge. También puede «alucinar» herramientas que no existen en tu sistema. Por eso los límites de iteraciones no son opcionales en producción.
Construir un agente ReAct con LangGraph paso a paso
El bloque central es StateGraph. Defines un estado compartido entre todos los nodos, añades los nodos y configuras las aristas, incluyendo las condicionales.
El estado compartido
Usa un TypedDict para el estado. Lo mínimo que necesitas:
from typing import TypedDict, Annotated, List
from langchain_core.messages import BaseMessage
import operator
class AgentState(TypedDict):
messages: Annotated[List[BaseMessage], operator.add]
next_action: str
El campo messages acumula el historial completo. next_action lo usará la arista condicional para decidir si el agente continúa o termina.
El nodo llm_call
Este nodo llama al modelo con el historial actual y devuelve la respuesta, que puede ser una llamada a herramienta o la respuesta final:
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tools = llm.bind_tools(tools)
def llm_call(state: AgentState) -> AgentState:
response = llm_with_tools.invoke(state["messages"])
return {"messages": [response]}
El nodo tool_executor
Cuando el modelo pide ejecutar una herramienta, este nodo se encarga:
from langchain_core.messages import ToolMessage
from langchain_core.tools import tool
@tool
def buscar_web(query: str) -> str:
"""Busca información en internet."""
# Tu implementación aquí
return resultado
def tool_executor(state: AgentState) -> AgentState:
last_message = state["messages"][-1]
tool_results = []
for tool_call in last_message.tool_calls:
result = buscar_web.invoke(tool_call["args"])
tool_results.append(
ToolMessage(content=str(result), tool_call_id=tool_call["id"])
)
return {"messages": tool_results}
La arista condicional
La función should_continue decide si el agente sigue ejecutando herramientas o ya tiene la respuesta final:
def should_continue(state: AgentState) -> str:
last_message = state["messages"][-1]
if hasattr(last_message, "tool_calls") and last_message.tool_calls:
return "tools"
return "end"
Montar el grafo
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
workflow = StateGraph(AgentState)
workflow.add_node("llm", llm_call)
workflow.add_node("tools", tool_executor)
workflow.set_entry_point("llm")
workflow.add_conditional_edges(
"llm",
should_continue,
{"tools": "tools", "end": END}
)
workflow.add_edge("tools", "llm")
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)
Con esto tienes un agente ReAct funcional que puede usar una herramienta de búsqueda web y sabe cuándo parar.
Gateway multi-modelo con LiteLLM y FastAPI
En producción, depender de un solo modelo es un problema. Si sube el precio, si aumenta la latencia o si necesitas una capacidad específica de otro modelo, cambiar requiere reescribir código. LiteLLM lo resuelve: una interfaz única, muchos modelos por detrás.
Por qué cambiar de modelo en producción
Los motivos son prácticos. GPT-4o puede ser demasiado caro para tareas de clasificación simple que Llama 3 resuelve igual de bien. Claude es mejor en textos largos con mucho contexto. Gemini puede tener latencias más bajas en ciertas regiones. Tener un gateway te da la flexibilidad de enrutar según la tarea sin cambiar el código del agente.
Configurar LiteLLM
import litellm
# Cambiar de modelo es solo cambiar el string
response_openai = litellm.completion(
model="gpt-4o",
messages=[{"role": "user", "content": "Hola"}]
)
response_claude = litellm.completion(
model="claude-opus-4",
messages=[{"role": "user", "content": "Hola"}]
)
response_llama = litellm.completion(
model="ollama/llama3",
messages=[{"role": "user", "content": "Hola"}]
)
La respuesta tiene siempre el mismo formato. Tu agente no necesita saber qué modelo está usando.
FastAPI como servidor compatible con OpenAI
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List, Optional
api = FastAPI()
class Message(BaseModel):
role: str
content: str
class ChatRequest(BaseModel):
model: str = "gpt-4o"
messages: List[Message]
thread_id: Optional[str] = "default"
@api.post("/v1/chat/completions")
async def chat(request: ChatRequest):
config = {"configurable": {"thread_id": request.thread_id}}
state = {"messages": [
HumanMessage(content=request.messages[-1].content)
]}
result = app.invoke(state, config=config)
last_message = result["messages"][-1]
return {
"choices": [{"message": {"role": "assistant", "content": last_message.content}}]
}
Cualquier cliente compatible con la API de OpenAI puede apuntar a este servidor sin cambiar nada. Útil para integraciones con herramientas de terceros que ya usan el SDK de OpenAI. Para más ideas sobre cómo exponer interfaces para aplicaciones Python, NiceGUI y Streamlit también son opciones viables si necesitas algo visual.
Desplegar el agente en producción
Gestión de sesiones
MemorySaver guarda el estado en memoria y es perfecto para desarrollo. En producción no sirve: si el proceso se reinicia, pierdes el contexto de todas las conversaciones. Cambia a AsyncPostgresSaver o RedisSaver:
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
async with AsyncPostgresSaver.from_conn_string(DATABASE_URL) as checkpointer:
app = workflow.compile(checkpointer=checkpointer)
Límites de seguridad
Sin límites, un agente puede consumir miles de tokens en un bucle infinito. Añade siempre un máximo de iteraciones y un timeout por herramienta:
from langgraph.graph import StateGraph
import asyncio
MAX_ITERATIONS = 10
def tool_executor_safe(state: AgentState) -> AgentState:
if state.get("iteration_count", 0) >= MAX_ITERATIONS:
return {"messages": [ToolMessage(content="Límite de iteraciones alcanzado.", tool_call_id="limit")]}
try:
result = asyncio.wait_for(ejecutar_herramienta(state), timeout=30.0)
except asyncio.TimeoutError:
result = "La herramienta tardó demasiado y fue cancelada."
return {"messages": [result], "iteration_count": state.get("iteration_count", 0) + 1}
Logging de cada paso
Para auditoría, necesitas saber exactamente qué decidió el agente en cada paso y qué devolvió cada herramienta. LangGraph emite eventos que puedes capturar con stream:
async for event in app.astream(state, config=config):
for node_name, node_output in event.items():
logging.info(f"Nodo: {node_name} | Output: {node_output}")
Gestión de errores que importan
JSON inválido del LLM
A veces el modelo devuelve una llamada a herramienta con JSON mal formado. No pases eso directamente al tool_executor. Valida antes:
import json
def parse_tool_call_safe(tool_call_str: str) -> dict:
try:
return json.loads(tool_call_str)
except json.JSONDecodeError as e:
logging.error(f"JSON inválido en tool_call: {e}")
return {}
Reintentos con backoff exponencial
Las APIs de LLMs devuelven errores de rate limit con cierta frecuencia. tenacity lo gestiona sin complicaciones:
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def llamar_llm_con_reintento(messages):
return llm_with_tools.invoke(messages)
Fallback a modelo más barato
Si el modelo principal falla, LiteLLM puede intentarlo con otro automáticamente:
response = litellm.completion(
model="gpt-4o",
messages=messages,
fallbacks=["claude-haiku-4", "ollama/llama3"]
)
Interrupciones humanas
Hay acciones que no quieres que el agente ejecute sin confirmación: borrar datos, hacer pagos, enviar emails. LangGraph tiene interrupt_before para eso:
app = workflow.compile(
checkpointer=checkpointer,
interrupt_before=["tool_ejecutar_pago"]
)
# El agente se detiene antes del nodo indicado.
# Reanudas después de confirmar:
app.invoke(None, config=config) # Continúa desde el checkpoint
Costes y cómo no arruinarte
Cada vuelta del bucle ReAct consume tokens: el historial completo más la respuesta del modelo. Si el agente da diez vueltas con un historial de 2.000 tokens cada vez, estás pagando 20.000 tokens por consulta. A eso súmale que esto puede pasar cientos de veces por hora.
El prompt caching reduce el coste cuando una parte del contexto se repite (las instrucciones del sistema, la descripción de las herramientas). Anthropic y OpenAI lo ofrecen; con LiteLLM lo activas con un parámetro. Para más detalle sobre cómo Python para computación científica puede integrarse con estas herramientas de IA, el stack es compatible sin grandes adaptaciones.
La otra palanca es el enrutamiento por tarea. No mandes todo a GPT-4o. Si la tarea es clasificar un texto en tres categorías, un modelo pequeño lo hace igual de bien por una fracción del coste. Define criterios claros (longitud del contexto, complejidad del razonamiento requerido) y enruta en consecuencia desde el gateway.
def elegir_modelo(tarea: str) -> str:
if tarea in ["clasificacion", "extraccion_simple"]:
return "gpt-4o-mini"
elif tarea in ["redaccion_larga", "analisis_legal"]:
return "claude-opus-4"
return "gpt-4o"
Con LangGraph tienes el grafo, los límites y los puntos de control. Con LiteLLM tienes la flexibilidad de modelo. Con FastAPI tienes la API. Juntos forman una base sólida para agentes que funcionan de verdad en producción, no solo en un notebook.
Imagen: Pexels / Google DeepMind