Agentes Autónomos de IA con LangGraph y CrewAI: De Cero a Producción

2. Paso a paso: conecta el StateGraph con enrutamiento condicional

La sección 1 definió el estado y los dos nodos. Un grafo se convierte en agente cuando los conectas con aristas, incluida la que vuelve atrás, que es justamente la razón para elegir LangGraph en lugar de una cadena. Aquí está el ensamblaje completo, reutilizando exactamente coder_node y reviewer_node de la sección 1:

from typing import TypedDict, List

from langgraph.graph import StateGraph, START, END


class AgentState(TypedDict):
    task: str
    code: str
    review_notes: List[str]
    approved: bool


def coder_node(state: AgentState):
    # Generates or modifies implementation code
    return {"code": "def solve(): pass", "review_notes": []}


def reviewer_node(state: AgentState):
    # Inspects code against the quality checklist
    is_valid = "def solve" in state["code"]
    return {
        "approved": is_valid,
        "review_notes": ["Looks good"] if is_valid else ["Missing function"],
    }


def route_after_review(state: AgentState) -> str:
    # The conditional edge: approve -> finish, reject -> loop back to coder
    return END if state["approved"] else "coder"


graph = StateGraph(AgentState)
graph.add_node("coder", coder_node)
graph.add_node("reviewer", reviewer_node)
graph.add_edge(START, "coder")
graph.add_edge("coder", "reviewer")
graph.add_conditional_edges(
    "reviewer",
    route_after_review,
    {"coder": "coder", END: END},
)

app = graph.compile()

result = app.invoke({
    "task": "Write solve()",
    "code": "",
    "review_notes": [],
    "approved": False,
})
print(result["approved"], result["review_notes"])

Lee las conexiones, no los nodos: los nodos devuelven actualizaciones parciales del estado, las aristas definen los caminos posibles y la función enrutadora es el único lugar donde vive el flujo de control. Cambia el revisor determinista por una llamada a un LLM y la topología no cambia; por eso la puerta con intervención humana de los objetivos de este tutorial encaja limpiamente como un nodo más entre el revisor y END, pausando el grafo antes de que se ejecute algo irreversible.

Añade la puerta humana que prometían los objetivos

Un nodo revisor que a su vez es un LLM todavía deja que el grafo corra de principio a fin sin supervisión. Para cualquier acción irreversible, inserta un nodo humano que pause el grafo con el interrupt de LangGraph: la ejecución se detiene, el estado queda guardado en un punto de control y una persona lo reanuda con una decisión. Necesita un checkpointer para que el estado en pausa sobreviva:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command, interrupt


def human_gate(state: AgentState):
    decision = interrupt({
        "question": "Approve this code for the next step?",
        "code": state["code"],
    })
    return {"approved": decision == "approve"}


# graph.add_node("human", human_gate) and route reviewer -> human -> END/coder
app = graph.compile(checkpointer=MemorySaver())

config = {"configurable": {"thread_id": "review-1"}}
app.invoke(
    {"task": "Write solve()", "code": "", "review_notes": [],
     "approved": False},
    config=config,
)  # runs until the gate, then pauses with state saved

# A human reviews the code and resumes the same thread:
app.invoke(Command(resume="approve"), config=config)

El thread_id es la clave de reanudación: el mismo grafo, el mismo estado, continuado horas después por otro proceso si hace falta. Esa propiedad de pausar y reanudar es la diferencia concreta entre un agente de demostración y uno que puedes dejar corriendo cerca de sistemas de producción.

3. Un ejemplo completo

El mismo flujo en CrewAI cambia el grafo por roles. Donde LangGraph te pedía declarar estado y aristas, CrewAI pide agentes con metas y tareas en secuencia: la pareja programador/revisor de la sección 1, expresada como una tripulación:

from crewai import Agent, Crew, Process, Task

coder = Agent(
    role="Python Developer",
    goal="Write correct, minimal Python for the assigned task",
    backstory="Senior Python engineer who favors small, testable functions.",
    verbose=True,
)

reviewer = Agent(
    role="Code Reviewer",
    goal="Reject any code that fails the quality checklist",
    backstory="Staff engineer known for strict, specific review notes.",
    verbose=True,
)

write_task = Task(
    description="Implement the function described here: {task}",
    expected_output="A single Python code block with the implementation.",
    agent=coder,
)

review_task = Task(
    description=(
        "Review the implementation produced by the previous task. "
        "Check: correct signature, edge cases handled, no placeholder code."
    ),
    expected_output="Either APPROVED, or a numbered list of required fixes.",
    agent=reviewer,
)

crew = Crew(
    agents=[coder, reviewer],
    tasks=[write_task, review_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(
    inputs={"task": "Write solve() returning the sum of even numbers in a list"}
)
print(result.raw)

Ejecuta ambas versiones con la misma tarea y el compromiso del que habla la sección de preguntas frecuentes se vuelve tangible. La versión de LangGraph te da el bucle gratis —un rechazo regresa al programador hasta la aprobación o el límite de recursión— además de un objeto de estado que puedes inspeccionar, persistir y reanudar. La versión de CrewAI es un tercio del código y se lee como un organigrama, pero la disciplina de reintentos vive en las instrucciones del revisor y no en un flujo de control exigible. Prototipa en CrewAI; endurece en LangGraph.

4. Solución de errores

• El grafo nunca termina. El revisor nunca aprueba, así que la arista condicional entra en bucle para siempre. Dos protecciones, ambas baratas: un contador de intentos máximos guardado en el estado y consultado por el enrutador, y un techo duro en la llamada de invoke: app.invoke(state, config={"recursion_limit": 10}). Un bucle sin contador es un error esperando una factura de tokens.
• Los campos del estado vuelven vacíos o se reinician entre nodos. Los nodos deben devolver actualizaciones parciales, solo las claves que cambiaron, y los campos de lista que deben acumularse necesitan un reductor, declarado como Annotated[List[str], operator.add] en el TypedDict. Devolver un diccionario de estado nuevo y completo desde un nodo descarta en silencio el trabajo de todos los demás nodos.
• El revisor de CrewAI lo aprueba todo. Los agentes que interpretan un rol son complacientes por defecto. Afina la meta («rechaza cualquier código que no cumpla la lista de calidad»), detalla la lista en la descripción de la tarea y fija expected_output para forzar un formato de veredicto —APPROVED o una lista numerada de correcciones— de modo que aprobar sea una decisión, no un cumplido.
• ImportError o caos de versiones tras instalar ambos frameworks. LangGraph y CrewAI arrastran dependencias de LangChain que se solapan. Instálalos juntos en un entorno virtual nuevo, deja que el resolvedor elija versiones compatibles en una sola pasada y congela el resultado (pip freeze) antes de que el proyecto crezca alrededor de un conjunto que funciona.