5 mar 2026

Máquinas de Estado y Agentes: Construyendo Flujos de Trabajo Confiables con LangGraph

La mayoría de los tutoriales de agentes muestran un bucle simple: preguntar a Claude, analizar la respuesta, llamar a una herramienta, repetir. Esto funciona para demostraciones. En producción, sin embargo, necesitas determinismo, recuperación de errores, puertas de aprobación humana y auditabilidad.

LangGraph lleva las máquinas de estado a los flujos de trabajo de agentes. En lugar de un bucle ad-hoc mantenido por sentencias if, obtienes un grafo explícito: nodos con nombre (unidades de lógica), aristas tipadas (transiciones) y un esquema de estado compartido que fluye a través de toda la ejecución.

Por Qué Usar Máquinas de Estado para Agentes

El Problema del Bucle Ad-Hoc

Un bucle de agente típico se ve así:

messages = []
while True:
    response = claude.messages.create(model=..., messages=messages)
    if response.stop_reason == "end_turn":
        break
    for tool_call in get_tool_calls(response):
        result = execute_tool(tool_call)
        messages.append(tool_result(tool_call.id, result))

Esto es legible para dos o tres herramientas. Agrega cinco herramientas, rutas condicionales, un paso de aprobación humana y lógica de reintento — y tendrás cientos de líneas de flujo de control enredado.

El problema más profundo es el estado implícito. ¿En qué etapa está el agente? ¿Qué datos ha recopilado? ¿Qué decisiones ha tomado? Todo vive en messages — un blob sin tipo que cada nodo lee y al que agrega, sin esquema impuesto.

Las Máquinas de Estado como Solución

Una máquina de estado hace lo implícito explícito. Defines:

Nodos — unidades de lógica discretas. Cada nodo recibe el estado actual, hace una cosa y devuelve actualizaciones de estado.
Aristas — transiciones entre nodos, ya sea incondicionales (A → B siempre) o condicionales (si hay problemas: ir a revisión, si no: ir a resumen).
Estado — un diccionario tipado que fluye a través del grafo completo. El esquema se valida en cada paso.

Cuándo NO Usar LangGraph

LangGraph agrega sobrecarga. Para tareas simples de un solo paso (clasificar texto, extraer campos), una llamada directa a la API es más rápida y clara. Usa LangGraph cuando tu flujo de trabajo tenga:

Múltiples etapas distintas que deben ejecutarse en secuencia
Ramificación condicional basada en resultados intermedios
Pasos con participación humana
Lógica de recuperación de errores o reintento
Requisitos de auditabilidad

Fundamentos de LangGraph

Definiendo el Estado

El estado es la estructura de datos central en LangGraph:

from typing import TypedDict

class ResearchState(TypedDict):
    query: str
    research_notes: str
    draft: str
    review_feedback: str
    is_approved: bool

Los nodos devuelven diccionarios parciales. LangGraph fusiona estas actualizaciones en el estado en ejecución después de cada nodo.

Nodos

Un nodo es una función Python simple:

import anthropic

client = anthropic.Anthropic()

def research_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"Investiga este tema exhaustivamente: {state['query']}"
        }]
    )
    return {"research_notes": response.content[0].text}

def draft_node(state: ResearchState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": f"Escribe un borrador basado en estas notas:\n{state['research_notes']}"
        }]
    )
    return {"draft": response.content[0].text}

Aristas y Enrutamiento Condicional

Las aristas incondicionales son transiciones fijas. Las aristas condicionales usan una función de enrutamiento:

from typing import Literal

def route_after_check(state: ResearchState) -> Literal["human_review", "draft"]:
    if state.get("review_feedback"):
        return "human_review"
    return "draft"

workflow.add_conditional_edges(
    "research",
    route_after_check,
    {"human_review": "review_node", "draft": "draft_node"}
)

Construyendo y Ejecutando un Grafo

from langgraph.graph import StateGraph, END

workflow = StateGraph(ResearchState)
workflow.add_node("research", research_node)
workflow.add_node("draft", draft_node)
workflow.set_entry_point("research")
workflow.add_edge("research", "draft")
workflow.add_edge("draft", END)

graph = workflow.compile()
result = graph.invoke({"query": "¿Qué es el protocolo MCP?"})
print(result["draft"])

Construyendo un Flujo de Revisión de Documentos

Implementemos un ejemplo completo: un agente de revisión de documentos que extrae términos clave, verifica problemas de cumplimiento y genera un resumen final.

Diseñando el Estado

class DocumentState(TypedDict):
    document: str
    key_terms: list[str]
    compliance_issues: list[str]
    summary: str
    human_feedback: str
    is_approved: bool

Implementando los Nodos

import json

def extract_terms_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": (
                "Extrae exactamente 5 términos clave de este documento. "
                "Devuélvelos como un array JSON de strings.\n\n"
                f"Documento: {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        terms = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        terms = [t.strip() for t in response.content[0].text.split("\n") if t.strip()][:5]
    return {"key_terms": terms}


def check_compliance_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": (
                "Verifica este documento en busca de problemas de cumplimiento. "
                "Busca: PII (nombres, números de seguro social, correos, teléfonos), "
                "marcaciones confidenciales, afirmaciones no verificadas.\n\n"
                "Devuelve un array JSON de descripciones de problemas. "
                "Devuelve un array vacío [] si no se encuentran problemas.\n\n"
                f"Documento: {state['document']}"
            )
        }]
    )
    try:
        raw = response.content[0].text.strip()
        if raw.startswith("```"):
            raw = raw.split("```")[1]
            if raw.startswith("json"):
                raw = raw[4:]
        issues = json.loads(raw.strip())
    except (json.JSONDecodeError, IndexError):
        issues = []
    return {"compliance_issues": issues}


def human_review_node(state: DocumentState) -> dict:
    print("\n=== SE REQUIERE REVISIÓN HUMANA ===")
    for issue in state["compliance_issues"]:
        print(f"  - {issue}")
    return {
        "human_feedback": "Revisado y aprobado con redacción de PII",
        "is_approved": True
    }


def summarize_node(state: DocumentState) -> dict:
    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=256,
        messages=[{
            "role": "user",
            "content": f"Resume este documento en exactamente dos oraciones:\n\n{state['document']}"
        }]
    )
    return {"summary": response.content[0].text, "is_approved": True}

Añadiendo Enrutamiento Condicional y Ensamblando el Grafo

from langgraph.graph import StateGraph, END

def route_after_compliance(state: DocumentState) -> Literal["human_review", "summarize"]:
    if state.get("compliance_issues"):
        return "human_review"
    return "summarize"

workflow = StateGraph(DocumentState)
workflow.add_node("extract", extract_terms_node)
workflow.add_node("check", check_compliance_node)
workflow.add_node("review", human_review_node)
workflow.add_node("summarize", summarize_node)

workflow.set_entry_point("extract")
workflow.add_edge("extract", "check")
workflow.add_conditional_edges(
    "check",
    route_after_compliance,
    {"human_review": "review", "summarize": "summarize"}
)
workflow.add_edge("review", "summarize")
workflow.add_edge("summarize", END)

graph = workflow.compile()

result = graph.invoke({
    "document": "El paciente Juan García (DNI: 12345678A) tiene hipertensión.",
    "key_terms": [], "compliance_issues": [],
    "summary": "", "human_feedback": "", "is_approved": False,
})
print(result["summary"])

Puntos de Control Humanos e Interrupciones

Para pausar realmente la ejecución y esperar una decisión humana:

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["review"]
)

thread_config = {"configurable": {"thread_id": "doc-review-001"}}
result = graph.invoke(initial_state, config=thread_config)

# Inspeccionar el estado antes de reanudar
current_state = graph.get_state(thread_config)
print("Problemas:", current_state.values["compliance_issues"])

# El revisor toma una decisión
graph.update_state(
    thread_config,
    {"human_feedback": "Aprobado después de redacción manual de PII", "is_approved": True}
)

# Reanudar
final_result = graph.invoke(None, config=thread_config)

Persistencia y Despliegue en Producción

Verificación de Estado

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    graph = workflow.compile(checkpointer=checkpointer)
    result = graph.invoke(initial_state, config={"configurable": {"thread_id": "doc-001"}})

Recuperación de Errores

def check_compliance_node(state: DocumentState) -> dict:
    try:
        response = client.messages.create(...)
        issues = parse_issues(response)
        return {"compliance_issues": issues}
    except anthropic.APIError as e:
        print(f"Verificación de cumplimiento fallida: {e}")
        return {"compliance_issues": [], "human_feedback": f"Verificación automática fallida: {e}"}

Ejecución Asíncrona

import asyncio

async def process_document(doc: str) -> dict:
    return await graph.ainvoke({
        "document": doc, "key_terms": [], "compliance_issues": [],
        "summary": "", "human_feedback": "", "is_approved": False,
    })

async def batch_process(documents: list[str]) -> list[dict]:
    return await asyncio.gather(*[process_document(doc) for doc in documents])

Patrones Comunes y Errores Frecuentes

Fan-Out / Fan-In: Ejecuta múltiples nodos en paralelo y fusiona sus resultados usando Send.
Ramificación condicional excesiva: Si una función de enrutamiento tiene más de unas pocas ramas, divide el flujo en subgrafos.
Sobrecarga de estado: Almacena referencias (IDs de base de datos, claves S3) en lugar de grandes artefactos directamente.
Pruebas de nodos en aislamiento: Como los nodos son funciones simples, puedes probarlos unitariamente directamente.

def test_extract_terms_node():
    state = {
        "document": "El zorro marrón rápido salta sobre el perro perezoso.",
        "key_terms": [], "compliance_issues": [],
        "summary": "", "human_feedback": "", "is_approved": False,
    }
    result = extract_terms_node(state)
    assert "key_terms" in result
    assert isinstance(result["key_terms"], list)

LangGraph reemplaza los bucles ad-hoc con flujos de trabajo explícitos, depurables y reanudables. Empieza con un grafo lineal (A → B → C → FIN). Agrega una arista condicional cuando necesites ramificar. Añade un punto de control humano cuando necesites aprobación. La mayoría de los flujos de trabajo de agentes de producción necesitan exactamente estos tres patrones.