Cómo construir un agente con LangGraph Python: tutorial para producción

LangGraph es el framework de orquestación de agentes de código abierto desarrollado por LangChain AI, publicado bajo licencia MIT. A diferencia de cadenas simples, un grafo LangGraph puede bifurcarse, retroceder a nodos anteriores y mantener estado persistente entre ejecuciones. Esto lo hace adecuado para flujos de trabajo largos, agentes de múltiples pasos y sistemas donde la intervención humana es parte del proceso. La versión 1.2.7, lanzada el 30 de junio de 2026, es la base de este tutorial.

El agente que construirás sigue el patrón de StateGraph: un grafo dirigido donde cada nodo es una función Python que recibe el estado actual y devuelve una actualización parcial. Las aristas condicionales enrutan el flujo según el valor del estado tras cada nodo. Los tres nodos del agente son: analizar_documento (invoca al LLM con el contenido), revision_humana (pausa y espera aprobación) y finalizar (cierra el flujo). La función enrutar actúa como función de ruteo en add_conditional_edges, pero no se registra como nodo del grafo. El checkpointer de PostgreSQL guarda el estado completo tras cada nodo, lo que permite reanudar el agente desde cualquier punto sin perder información.

• Python 3.10 o superior (LangGraph requiere Python >=3.10)
• PostgreSQL local o en la nube (Supabase, Neon o Railway funcionan)
• Una clave de API de Anthropic (obtenla en console.anthropic.com)
• pip o uv instalado en tu entorno

1. Define el estado del agente

El estado es el único dato que viaja entre nodos y que el checkpointer persiste en PostgreSQL tras cada paso. Usa TypedDict para que LangGraph pueda serializarlo. El campo messages usa el reductor add_messages de LangGraph: en vez de reemplazar la lista completa, agrega los mensajes nuevos al historial existente, lo que implementa la memoria de conversación de forma automática. Mantén el estado mínimo: cada campo se escribe en la base de datos en cada checkpoint.

# state.py
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    documento: str
    confianza: float
    revision_requerida: bool
    estado: Literal["procesando", "esperando_revision", "listo", "error"]

2. Conecta PostgreSQL como checkpointer

Reemplaza InMemorySaver por PostgresSaver para tener persistencia real. La primera vez que ejecutes setup(), el checkpointer crea las tablas necesarias en tu base de datos. Los parámetros autocommit=True y row_factory=dict_row son obligatorios: sin ellos, el checkpointer lanza errores de tipo en tiempo de ejecución.

# checkpointer.py
import os
from dotenv import load_dotenv
from psycopg import Connection
from psycopg.rows import dict_row
from langgraph.checkpoint.postgres import PostgresSaver

load_dotenv()

conn = Connection.connect(
    os.environ["DATABASE_URL"],
    autocommit=True,
    row_factory=dict_row
)
checkpointer = PostgresSaver(conn)
checkpointer.setup()  # Crea las tablas en la primera ejecucion

3. Define los nodos del agente

Cada nodo es una función Python que recibe el estado completo y devuelve un diccionario con los campos que actualiza. El nodo analizar_documento llama al LLM con el documento y extrae un nivel de confianza del texto de respuesta. Si la confianza está bajo el umbral definido, el nodo de enrutamiento deriva al flujo de revisión humana. En producción, usa salida estructurada del LLM para extraer la confianza con mayor fiabilidad que el parseo de texto.

# nodes.py
import os
from dotenv import load_dotenv
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, SystemMessage
from state import AgentState

load_dotenv()

# Actualiza el nombre del modelo con la version mas reciente disponible
MODELO = "claude-3-5-sonnet-20241022"
UMBRAL_CONFIANZA = 0.7

llm = ChatAnthropic(
    model=MODELO,
    api_key=os.environ["ANTHROPIC_API_KEY"]
)

def analizar_documento(state: AgentState) -> dict:
    sistema = SystemMessage(
        content="Analiza el documento. Al final escribe CONFIANZA: seguido de un numero del 0 al 1."
    )
    respuesta = llm.invoke([sistema, HumanMessage(content=state["documento"])])
    texto = respuesta.content
    # En produccion, usa salida estructurada del LLM
    confianza = float(texto.split("CONFIANZA:")[-1].strip().split()[0])
    return {
        "messages": [respuesta],
        "confianza": confianza,
        "revision_requerida": confianza < UMBRAL_CONFIANZA,
        "estado": "procesando"
    }

def enrutar(state: AgentState) -> str:
    return "revision_humana" if state["revision_requerida"] else "finalizar"

def finalizar(state: AgentState) -> dict:
    return {"estado": "listo"}

Antes de llevar el agente a producción, verifica que el flujo completo funciona: que el agente analiza el documento, detecta baja confianza, se interrumpe correctamente y reanuda cuando el revisor aprueba. La siguiente prueba usa InMemorySaver para no depender de una base de datos externa.

# test_agente.py
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from langgraph.graph import StateGraph, START, END
from nodes import analizar_documento, enrutar, finalizar
from revision import revision_humana
from state import AgentState

def construir_grafo(checkpointer):
    builder = StateGraph(AgentState)
    builder.add_node("analizar", analizar_documento)
    builder.add_node("revision_humana", revision_humana)
    builder.add_node("finalizar", finalizar)
    builder.add_edge(START, "analizar")
    builder.add_conditional_edges("analizar", enrutar)
    builder.add_edge("revision_humana", "finalizar")
    builder.add_edge("finalizar", END)
    return builder.compile(checkpointer=checkpointer)

def test_flujo_revision():
    grafo = construir_grafo(InMemorySaver())
    config = {"configurable": {"thread_id": "test-001"}}
    input_prueba = {
        "documento": "Texto ambiguo sin datos concretos.",
        "messages": [],
        "confianza": 0.0,
        "revision_requerida": False,
        "estado": "procesando"
    }
    for evento in grafo.stream(input_prueba, config=config):
        print("Evento:", evento)
    # El revisor aprueba: Command(resume=True) reanuda el flujo
    for evento in grafo.stream(Command(resume=True), config=config):
        print("Reanudado:", evento)

Checklist de producción LangGraph (Cognitiva)

1. PostgresSaver activo: nunca uses InMemorySaver en producción. El estado debe persistir entre reinicios del proceso.
2. recursion_limit configurado: incluye recursion_limit con valor 25 en la config de invocación para evitar bucles infinitos.
3. Variables de entorno sin valores escritos en el código: DATABASE_URL y ANTHROPIC_API_KEY deben venir de .env o del sistema, nunca escritas en el código fuente.
4. Al menos un nodo interrupt() para cada acción con efectos irreversibles: envío de correos, aprobación de pagos, modificación de registros externos.
5. Prueba de reanudación validada: confirma que Command(resume=valor) recupera el estado correcto tras un reinicio del proceso antes de pasar a producción.

Una vez que el checklist está verde, empaqueta el agente en un contenedor Docker con tu API FastAPI, apunta DATABASE_URL a tu PostgreSQL en la nube y despliega en la plataforma de tu elección. El agente escala horizontalmente porque el estado vive en la base de datos, no en la memoria del proceso. Cada instancia adicional puede atender un thread_id distinto sin interferir con las otras sesiones activas.