Building Multi-Agent Systems with LangGraph & CrewAI: Stateful Graphs and Autonomous Teams

Why Sequential Chains Fail in Production

• No Self-Correction: If a code generator agent produces code that fails a unit test, a linear chain cannot loop back to reflect, inspect the trace, and re-attempt generation.
• Lost Context Across Specialists: An agent that researches code, an agent that writes SQL queries, and an agent that audits cybersecurity have completely different system prompts, toolsets, and reasoning budgets. Cramming everything into one single prompt produces hallucination.
• Lack of State Checkpointing: If an agent pipeline crashes midway through a 15-minute workflow, you shouldn't have to restart from scratch and burn thousands of API tokens.

State Machines & Cyclical Graphs in LangGraph

• State: A shared, immutable data structure passed between nodes.
• Nodes: Functions or specialized agents that read the current state, perform work (call tools, generate thoughts), and return state updates.
• Edges: Conditional routers that determine which node executes next based on state values.
• Checkpointers: Persistence layers (Postgres, SQLite, Redis) that save snapshots of state after every turn, enabling time-travel debugging and human-in-the-loop approvals.

       ┌────────────────────────┐
       │   Supervisor Router    │◄─────────────────┐
       └───────────┬────────────┘                  │
                   │                               │
        ┌──────────┴──────────┐                    │
        ▼                     ▼                    │ (Loop back for review)
┌──────────────┐      ┌──────────────┐             │
│ Coder Agent  │      │ Tester Agent │─────────────┘
└──────────────┘      └──────────────┘

Hands-On: Building a Self-Healing Code Generator in LangGraph

python
# multi_agent_graph.py
import operator
from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

# 1. Define the Global Graph State
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    code: str
    test_results: str
    iterations: int

llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

# 2. Define Specialist Nodes
def coder_node(state: AgentState) -> dict:
    """Specialist node focused on writing or revising Python code."""
    iterations = state.get("iterations", 0) + 1
    code = state.get("code", "")
    test_results = state.get("test_results", "")

    if not code:
        prompt = f"Write a clean, typed Python function for: {state['messages'][-1].content}"
    else:
        prompt = f"Your previous code failed tests with:\n{test_results}\n\nExisting code:\n{code}\n\nFix all bugs and return only Python code."

    response = llm.invoke([HumanMessage(content=prompt)])
    return {
        "code": response.content,
        "iterations": iterations,
        "messages": [AIMessage(content=f"Coder generated iteration {iterations}")]
    }

def tester_node(state: AgentState) -> dict:
    """Specialist node that evaluates generated code against test requirements."""
    code = state["code"]
    prompt = f"Evaluate this code for bugs, edge cases, and typing errors:\n{code}\nIf 100% correct, reply exactly 'PASS'. Otherwise describe the error."

    response = llm.invoke([HumanMessage(content=prompt)])
    result = response.content.strip()

    return {
        "test_results": result,
        "messages": [AIMessage(content=f"Tester evaluation: {result}")]
    }

# 3. Define Conditional Routing Logic
def should_continue(state: AgentState) -> str:
    """Decide whether to loop back to Coder, or terminate."""
    if "PASS" in state["test_results"]:
        return "success"
    if state["iterations"] >= 3:
        return "max_retries_reached"
    return "revise_code"

# 4. Construct the Cyclical Graph
workflow = StateGraph(AgentState)

workflow.add_node("coder", coder_node)
workflow.add_node("tester", tester_node)

workflow.set_entry_point("coder")
workflow.add_edge("coder", "tester")

# Conditional cyclical edge
workflow.add_conditional_edges(
    "tester",
    should_continue,
    {
        "revise_code": "coder",   # Loop back!
        "success": END,           # Complete!
        "max_retries_reached": END
    }
)

# 5. Compile with In-Memory Checkpointing
checkpointer = MemorySaver()
app = workflow.compile(checkpointer=checkpointer)

Human-in-the-Loop Review & Time-Travel Debugging

python
# Interrupt execution right before the production deployment node
app = workflow.compile(
    interrupt_before=["deploy_to_production"],
    checkpointer=checkpointer
)

# Thread execution pauses automatically and saves state
thread_config = {"configurable": {"thread_id": "tx_session_101"}}
for event in app.stream(initial_input, thread_config):
    print(event)

# The human reviews the proposed diff in the dashboard:
# To resume, simply provide user approval or amended state:
app.invoke(None, thread_config)

LangGraph vs CrewAI: Which Should You Choose?

┌──────────────────┬─────────────────────────────┬─────────────────────────────┐
│ Feature          │ LangGraph                   │ CrewAI                      │
├──────────────────┼─────────────────────────────┼─────────────────────────────┤
│ Core Metaphor    │ State Machines & Graphs     │ Persona-based Roleplay      │
│ Learning Curve   │ Steeper (Engineering-first) │ Gentle (Configuration-first)│
│ Control & Loops  │ Microscopic, exact control  │ High-level autonomous delegation│
│ Persistence      │ Native pluggable checkpoint │ Built-in memory systems     │
│ Best For         │ Complex enterprise backends │ Fast prototypes & content   │
└──────────────────┴─────────────────────────────┴─────────────────────────────┘

• Choose CrewAI when you want rapid persona-based orchestration (e.g., "You are a Senior Technical Researcher, collaborate with the Content Editor"), where high-level delegation and autonomy matter more than rigid deterministic control.
• Choose LangGraph when building mission-critical enterprise workflows that require fine-grained state machines, deterministic validation loops, exact tool routing, and audited checkpointing.

Share this article

Comments