AI Agent Orchestration Frameworks 2026: LangGraph vs CrewAI vs AutoGen Complete Guide

from langgraph.graph import StateGraph
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    task: str
    result: str

graph_builder = StateGraph(AgentState)

# Define nodes
def process_task(state: AgentState) -> AgentState:
    # Task processing logic
    return {"result": "Task completed"}

# Define conditional edges
def should_continue(state: AgentState) -> str:
    if state["result"]:
        return "end"
    return "retry"

graph_builder.add_node("process", process_task)
graph_builder.add_conditional_edges("process", should_continue)

graph = graph_builder.compile()
result = graph.invoke({"messages": [], "task": "analyze"})

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

class DocumentState(TypedDict):
    document: str
    analysis: str
    insights: list

graph = StateGraph(DocumentState)
llm = ChatOpenAI(model="gpt-4")

def analyze_document(state: DocumentState) -> DocumentState:
    response = llm.invoke(f"Analyze: {state['document']}")
    return {"analysis": response.content}

def extract_insights(state: DocumentState) -> DocumentState:
    response = llm.invoke(f"Extract key insights from: {state['analysis']}")
    return {"insights": [item.strip() for item in response.content.split('\n')]}

graph.add_node("analyze", analyze_document)
graph.add_node("extract", extract_insights)
graph.add_edge("analyze", "extract")
graph.add_edge("extract", END)

workflow = graph.compile()

from crewai import Agent, Task, Crew
from crewai_tools import tool

class ResearchTools:
    @tool
    def search_web(query: str) -> str:
        """Perform web search"""
        return f"Search results for: {query}"

# Define agents
researcher = Agent(
    role="Research Analyst",
    goal="Find accurate information about topics",
    backstory="Expert researcher with 10 years of experience",
    tools=[ResearchTools.search_web()],
    verbose=True
)

writer = Agent(
    role="Content Writer",
    goal="Write engaging and informative content",
    backstory="Professional writer for tech publications"
)

# Define tasks
research_task = Task(
    description="Research AI agents in 2026",
    agent=researcher,
    expected_output="Detailed research report"
)

writing_task = Task(
    description="Write a blog post based on research",
    agent=writer,
    context=[research_task]
)

# Organize crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    verbose=2
)

result = crew.kickoff()

from autogen import AssistantAgent, UserProxyAgent

# Create assistant agent
assistant = AssistantAgent(
    name="Scientist",
    system_message="You are a helpful AI scientist assistant"
)

# Create user proxy agent
user_proxy = UserProxyAgent(
    name="User",
    human_input_mode="TERMINATE",
    code_execution_config={"use_docker": False}
)

# Initiate conversation
user_proxy.initiate_chat(
    assistant,
    message="Write Python code to analyze a CSV file and generate statistics"
)

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

# Multiple specialist agents
data_scientist = AssistantAgent(
    name="DataScientist",
    system_message="You are an expert data scientist"
)

engineer = AssistantAgent(
    name="Engineer",
    system_message="You are an expert software engineer"
)

# Group chat setup
groupchat = GroupChat(
    agents=[data_scientist, engineer],
    messages=[],
    max_round=10
)

manager = GroupChatManager(groupchat=groupchat)

user_proxy.initiate_chat(
    manager,
    message="Build a machine learning pipeline for sales prediction"
)

version: 1.0
name: 'Customer Support Workflow'

nodes:
  - id: input
    type: input
    config:
      title: 'Customer Query'

  - id: classify
    type: llm
    config:
      model: gpt-4
      prompt: 'Classify this query: {{input.query}}'

  - id: route
    type: switch
    config:
      cases:
        - value: technical
          next: technical_agent
        - value: billing
          next: billing_agent

  - id: technical_agent
    type: agent
    config:
      role: 'Technical Support'

  - id: output
    type: output
    config:
      title: 'Response'

from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI

class ArticleState(TypedDict):
    article: str
    summary: str
    sentiment: str

graph = StateGraph(ArticleState)
llm = ChatOpenAI()

def summarize(state):
    result = llm.invoke(f"Summarize: {state['article']}")
    return {"summary": result.content}

def analyze_sentiment(state):
    result = llm.invoke(f"Analyze sentiment: {state['summary']}")
    return {"sentiment": result.content}

graph.add_node("summarize", summarize)
graph.add_node("sentiment", analyze_sentiment)
graph.add_edge("summarize", "sentiment")
graph.add_edge("sentiment", END)

workflow = graph.compile()
result = workflow.invoke({"article": "...", "summary": "", "sentiment": ""})

from crewai import Agent, Task, Crew

summarizer = Agent(
    role="News Summarizer",
    goal="Create accurate summaries of news articles"
)

analyst = Agent(
    role="Sentiment Analyst",
    goal="Analyze emotional tone of content"
)

summary_task = Task(
    description="Summarize the article",
    agent=summarizer
)

sentiment_task = Task(
    description="Analyze sentiment",
    agent=analyst,
    context=[summary_task]
)

crew = Crew(
    agents=[summarizer, analyst],
    tasks=[summary_task, sentiment_task]
)

result = crew.kickoff()

from autogen import AssistantAgent, UserProxyAgent, GroupChat, GroupChatManager

summarizer = AssistantAgent(name="Summarizer")
analyst = AssistantAgent(name="Sentiment Analyst")

groupchat = GroupChat(agents=[summarizer, analyst], messages=[])
manager = GroupChatManager(groupchat=groupchat)

user_proxy = UserProxyAgent(name="Admin", human_input_mode="TERMINATE")
user_proxy.initiate_chat(
    manager,
    message="Summarize this article and analyze its sentiment: ..."
)

# Good: Single responsibility
analyzer_agent = Agent(
    role="Data Analyzer",
    goal="Extract insights from structured data"
)

# Avoid: Too broad responsibility
universal_agent = Agent(
    role="Universal Assistant",
    goal="Do everything"
)

from crewai_tools import tool

@tool
def fetch_market_data(symbol: str) -> str:
    """Fetch market data for a specific symbol"""
    # Implementation
    pass

@tool
def analyze_trends(data: str) -> str:
    """Analyze market trends from data"""
    # Implementation
    pass

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_external_api(endpoint: str):
    # API call implementation
    pass

# Compatibility wrapper example
class LegacyAgentWrapper:
    def __init__(self, new_agent):
        self.agent = new_agent

    def execute(self, task):
        # Legacy interface
        return self.agent.run(task)