Multi-Agent Systems & Production Orchestration: LangGraph vs AutoGen 0.4 vs CrewAI

from typing import TypedDict, Annotated, Sequence
import operator
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.postgres import PostgresSaver

# 1. Define Typed State Channels with Reducers
class AgentState(TypedDict):
    task: str
    code: str
    test_results: str
    iteration: int
    errors: Sequence[str]
    is_approved: bool

# 2. Define Node Worker Logic
def code_generator_node(state: AgentState):
    """Writes or refactors code based on task and prior test feedback."""
    iteration = state.get("iteration", 0) + 1
    # LLM generation call with context
    return {"iteration": iteration, "code": "# Generated safe code"}

def test_runner_node(state: AgentState):
    """Executes sandboxed unit tests via MCP or Docker container."""
    # Execute code in isolated sandbox
    tests_passed = True # Simulated run
    return {"test_results": "All 14 unit tests passed", "errors": []}

# 3. Define Conditional Routing Function
def should_continue(state: AgentState) -> str:
    """Evaluates whether to terminate, iterate, or request human review."""
    if state["iteration"] >= 5:
        return "human_approval" # Prevent runaway recursion
    if not state.get("errors"):
        return END # Success
    return "code_generator" # Self-correction loop

# 4. Compile Graph with Checkpointing
workflow = StateGraph(AgentState)
workflow.add_node("code_generator", code_generator_node)
workflow.add_node("test_runner", test_runner_node)

workflow.set_entry_point("code_generator")
workflow.add_edge("code_generator", "test_runner")
workflow.add_conditional_edges("test_runner", should_continue)

# Compile with persistent PostgreSQL state checkpointer
app = workflow.compile(checkpointer=PostgresSaver.from_conn_string("postgresql://..."))