CrewAI Thinks in People. LangGraph Thinks in Process.

Everyone is debating CrewAI vs LangGraph. That debate is almost always framed wrong.

The two frameworks are not competing on capability. They are competing on mental model, and picking the wrong one means you will fight the framework the entire time you are building.

CrewAI thinks in people. LangGraph thinks in process.

When you build in CrewAI, the first question you ask is WHO. You define agents by role, goal, and backstory. The framework figures out coordination. When you build in LangGraph, the first question you ask is WHAT. What happens next, and under what condition. You are drawing a flowchart in code. Every branch is explicit. Every transition is something you wrote.

Are you trying to replicate a team or replicate a process? If you can draw it as an org chart, reach for CrewAI. If you can draw it as a flowchart, reach for LangGraph.

Everything else in this post follows from that distinction.

CrewAI: thinking in people

CrewAI (currently at 1.14.7) gives you three execution modes. Each one reveals a different assumption about how agents should coordinate.

Sequential: the assembly line

Sequential mode is the default and the most predictable. Tasks run in order. Each task's output feeds the next. You must assign a task.agent explicitly. The framework does not decide who does what — you do. The wiring is static and visible.

This is the mode to reach for when your pipeline is linear and you want to minimize surprises.

# pip install crewai==1.14.7
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool

search_tool = SerperDevTool()

researcher = Agent(
    role="Research Analyst",
    goal="Find accurate, up-to-date competitive intelligence about {company}",
    backstory="A meticulous analyst with a talent for surfacing signal from noise.",
    tools=[search_tool],
    allow_delegation=False,
)
analyst = Agent(
    role="Strategy Analyst",
    goal="Interpret research findings and identify strategic implications for {company}",
    backstory="Former management consultant who turns raw data into structured insight.",
    allow_delegation=False,
)
writer = Agent(
    role="Report Writer",
    goal="Write a clear, concise competitive intelligence report",
    backstory="Technical writer skilled at translating dense analysis into executive prose.",
    allow_delegation=False,
)
critic = Agent(
    role="Quality Reviewer",
    goal="Identify gaps, factual errors, and weak reasoning in the draft report",
    backstory="Devil's advocate and ruthless editor who raises the bar on every deliverable.",
    allow_delegation=False,
)

research_task = Task(
    description="Research {company}: recent news, product launches, leadership changes, financials.",
    expected_output="Bullet-point research notes with source URLs, max 500 words.",
    agent=researcher,
)
analysis_task = Task(
    description="Analyse the research notes. Identify SWOT themes and strategic risks.",
    expected_output="Structured analysis: strengths, weaknesses, opportunities, threats.",
    agent=analyst,
    context=[research_task],
)
writing_task = Task(
    description="Write a 3-section competitive intelligence report: Overview, SWOT, Recommendations.",
    expected_output="Polished report in markdown, ~600 words.",
    agent=writer,
    context=[analysis_task],
)
review_task = Task(
    description="Review the draft report. List specific improvements required.",
    expected_output="Numbered critique list. Mark 'APPROVED' if no major issues remain.",
    agent=critic,
    context=[writing_task],
)

crew = Crew(
    agents=[researcher, analyst, writer, critic],
    tasks=[research_task, analysis_task, writing_task, review_task],
    process=Process.sequential,
    output_log_file="./intel_log.json",
    verbose=True,
)

result = crew.kickoff(inputs={"company": "Notion"})
print(result.raw)

Notice allow_delegation=False on every agent. Without that flag, any agent can spontaneously hand off work to another — useful in theory, a debugging nightmare in practice. Set it to False unless you have a specific reason not to, and keep Process.sequential for anything where the order actually matters.

One gotcha worth knowing before it bites you: context window overflow in sequential mode fails silently. The LLM hallucinates rather than erroring. Build in output length discipline via expected_output and watch your token counts.

Hierarchical: the manager who delegates

Hierarchical mode introduces a manager agent. The manager runs a ReAct loop with exactly two injected tools: "Delegate work to coworker" and "Ask question to coworker". The manager sees every worker agent's role and goal. The task list you define is a hint — not a wiring diagram. The manager decides who does what at runtime.

This buys you flexibility. It costs you tokens — roughly 2-3x versus sequential for an equivalent workflow. The manager's ReAct loop adds multiple LLM calls on top of the worker calls.

# pip install crewai==1.14.7
from crewai import Agent, Task, Crew, Process

researcher = Agent(
    role="Research Analyst",
    goal="Find competitive intelligence about {company}",
    backstory="A meticulous analyst who surfaces signal from noise.",
    allow_delegation=False,
)
analyst = Agent(
    role="Strategy Analyst",
    goal="Interpret research and identify strategic implications",
    backstory="Former consultant who structures raw data into insight.",
    allow_delegation=False,
)
writer = Agent(
    role="Report Writer",
    goal="Write a competitive intelligence report",
    backstory="Technical writer who translates dense analysis into prose.",
    allow_delegation=False,
)
critic = Agent(
    role="Quality Reviewer",
    goal="Find gaps and errors in the draft report",
    backstory="Ruthless editor who raises the bar on every deliverable.",
    allow_delegation=False,
)

# Tasks — no explicit agent assignment; the manager decides who does what
research_task = Task(
    description="Research {company}: news, products, leadership, financials.",
    expected_output="Bullet-point notes with source URLs, max 500 words.",
)
analysis_task = Task(
    description="Analyse research notes. Identify SWOT themes and risks.",
    expected_output="Structured SWOT analysis.",
)
writing_task = Task(
    description="Write a 3-section competitive intelligence report in markdown.",
    expected_output="Polished ~600-word report.",
)
review_task = Task(
    description="Review the draft. List improvements. Mark APPROVED if clean.",
    expected_output="Numbered critique list.",
)

crew = Crew(
    agents=[researcher, analyst, writer, critic],
    tasks=[research_task, analysis_task, writing_task, review_task],
    process=Process.hierarchical,
    manager_llm="gpt-4o",
    output_log_file="./intel_log.json",
    verbose=True,
)

result = crew.kickoff(inputs={"company": "Notion"})
print(result.raw)

The tasks no longer have an agent= field. The manager is now the coordinator. This is a fundamentally different contract: you are trusting the LLM to wire the right people to the right work, at runtime, every time. That non-determinism is fine for exploration. It is expensive and unreliable for production workflows where you need auditability. Two things to watch out for: QA agents tend to approve drafts — the "devil's advocate" role is aspirational, not mechanical. And allow_delegation=True on worker agents causes cascading delegation chains that are extremely hard to debug.

Flows: conditional orchestration between crews

Flows are CrewAI's answer to multi-crew conditional pipelines. You get event-driven orchestration via Python decorators: @start(), @router(), @listen(). State is a typed Pydantic model. You can call crew.kickoff() inside a @listen method. Mixing Flows and Crews is the production pattern when you need real branching.

# pip install crewai==1.14.7 pydantic
from crewai import Agent, Task, Crew, Process
from crewai.flow.flow import Flow, start, router, listen
from pydantic import BaseModel


class IntelState(BaseModel):
    company: str = ""
    topic_type: str = ""
    result: str = ""


def build_technical_crew(company: str) -> Crew:
    researcher = Agent(role="Tech Researcher", goal=f"Deep-dive {company} tech stack",
                       backstory="Ex-engineer who reads source code for fun.", allow_delegation=False)
    writer = Agent(role="Tech Writer", goal="Explain technical findings clearly",
                   backstory="Bridges engineering depth and executive readability.", allow_delegation=False)
    return Crew(
        agents=[researcher, writer],
        tasks=[
            Task(description=f"Research {company} APIs, infra, and engineering blog.", expected_output="Tech notes.", agent=researcher),
            Task(description="Summarise technical findings in a 2-page brief.", expected_output="Tech brief.", agent=writer),
        ],
        process=Process.sequential,
    )


def build_strategic_crew(company: str) -> Crew:
    analyst = Agent(role="Strategy Analyst", goal=f"Assess {company} market position",
                    backstory="McKinsey-trained with a nose for strategic leverage.", allow_delegation=False)
    writer = Agent(role="Strategy Writer", goal="Write actionable strategic recommendations",
                   backstory="Turns analysis into board-ready language.", allow_delegation=False)
    return Crew(
        agents=[analyst, writer],
        tasks=[
            Task(description=f"Analyse {company} market share, pricing, and partnerships.", expected_output="SWOT notes.", agent=analyst),
            Task(description="Write a strategic brief with 3 recommendations.", expected_output="Strategy brief.", agent=writer),
        ],
        process=Process.sequential,
    )


class IntelFlow(Flow[IntelState]):

    @start()
    def classify_topic(self):
        keywords = ["api", "infra", "stack", "architecture", "sdk"]
        self.state.topic_type = (
            "technical" if any(k in self.state.company.lower() for k in keywords)
            else "strategic"
        )
        return self.state.topic_type

    @router(classify_topic)
    def route_by_type(self):
        return self.state.topic_type

    @listen("technical")
    def run_technical_crew(self):
        crew = build_technical_crew(self.state.company)
        result = crew.kickoff(inputs={"company": self.state.company})
        self.state.result = result.raw

    @listen("strategic")
    def run_strategic_crew(self):
        crew = build_strategic_crew(self.state.company)
        result = crew.kickoff(inputs={"company": self.state.company})
        self.state.result = result.raw


flow = IntelFlow()
flow.kickoff(inputs={"company": "Stripe"})
print(flow.state.result)

Flows give you the conditional routing that neither sequential nor hierarchical mode offers cleanly. But notice: the routing logic in classify_topic is still keyword matching on a string. You are doing the branching in Python. The crew handles the people coordination inside each branch.

One more thing before we move on: CrewAI's memory system uses LanceDB with LLM-driven scope assignment. The LLM decides what goes into short-term versus long-term memory. That is non-deterministic by design, and it causes "memory bleed" in production — agents recalling context from unrelated prior runs. If your pipeline requires reliable state isolation between invocations, this is a problem you need to design around, not ignore.

LangGraph: thinking in process

LangGraph (1.0 went GA in October 2025) is a graph execution engine. There are no agents in the CrewAI sense. There are nodes. Nodes are plain Python functions that receive state and return state. The graph controls what runs when.

StateGraph and AgentState

The foundation is StateGraph and a TypedDict for state. Every field in the TypedDict is a slot the graph can read and write. By default, fields use last-write-wins semantics. If you annotate a field with Annotated[list, add_messages], the reducer deduplicates by message ID instead of overwriting.

# pip install langgraph langchain-openai
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")


class AgentState(TypedDict):
    company: str
    research_notes: str
    analysis: str
    draft: str
    review_feedback: str
    iteration: int


def research_node(state: AgentState) -> AgentState:
    notes = llm.invoke(f"Research competitive intelligence on {state['company']}. Return bullet-point notes.").content
    return {"research_notes": notes, "iteration": state.get("iteration", 0)}


def analysis_node(state: AgentState) -> AgentState:
    analysis = llm.invoke(f"Analyse these research notes and produce a SWOT:\n{state['research_notes']}").content
    return {"analysis": analysis}


def writer_node(state: AgentState) -> AgentState:
    context = f"Analysis:\n{state['analysis']}"
    if state.get("review_feedback"):
        context += f"\n\nPrevious review feedback to address:\n{state['review_feedback']}"
    draft = llm.invoke(f"Write a competitive intelligence report (markdown, ~600 words).\n{context}").content
    return {"draft": draft}


def critic_node(state: AgentState) -> AgentState:
    feedback = llm.invoke(
        f"Review this competitive intelligence report. List specific issues.\n"
        f"End with APPROVED or REVISE.\n\n{state['draft']}"
    ).content
    return {"review_feedback": feedback, "iteration": state["iteration"] + 1}


def should_revise(state: AgentState) -> Literal["writer", "__end__"]:
    if "APPROVED" in state["review_feedback"].upper() or state["iteration"] >= 3:
        return "__end__"
    return "writer"


graph = (
    StateGraph(AgentState)
    .add_node("researcher", research_node)
    .add_node("analyst", analysis_node)
    .add_node("writer", writer_node)
    .add_node("critic", critic_node)
    .add_edge(START, "researcher")
    .add_edge("researcher", "analyst")
    .add_edge("analyst", "writer")
    .add_edge("writer", "critic")
    .add_conditional_edges("critic", should_revise, {"writer": "writer", "__end__": END})
    .compile()
)

result = graph.invoke({"company": "Notion", "iteration": 0})
print(result["draft"])

The same four-agent competitive intelligence pipeline, now as a state machine. The researcher, analyst, writer, and critic are not agents with backstories. They are named execution nodes. The should_revise function is the critic edge's routing logic. It reads review_feedback from state and returns a string key that maps to the next node. If the critic says APPROVED or we hit three iterations, we exit. Otherwise we loop back to the writer.

Every path through this graph exists as a line of code you wrote. There is no ambiguity about what happens next.

Conditional edges

The routing function pattern is the core of LangGraph's power. The function receives state and returns a string. That string is looked up in a dict you provide to add_conditional_edges. The dict maps strings to node names or END. This is how you build every branch in the graph — explicit, testable, debuggable.

The downside is the boilerplate. A simple 2-step workflow is 10-20 lines before you write any business logic. That verbosity is also the point: every decision is a named function, not an emergent behavior. When something goes wrong, the Pregel stack trace tells you exactly which node and which edge.

Schema rigidity is a real pain in production. If you need to add a new field to AgentState with a custom reducer, existing checkpoints become incompatible. You need a migration strategy before you change reducers on a live system. LangGraph does not handle this for you.

The same pipeline, two mental models

Both examples above solve the same problem: competitive intelligence research with a review loop. But the mental model is completely different.

In CrewAI, you hired people. You wrote job descriptions. The framework figured out who talks to whom. You did not write a single line of routing logic. When you added the critic, you added a person. The framework wired them in.

In LangGraph, you drew a flowchart. You wrote should_revise. You defined every edge. When you added the critic, you added a node and two edges. The framework executed what you drew.

Neither approach is strictly better. They answer different questions. CrewAI answers "who is responsible?" LangGraph answers "what happens next?" The right choice depends entirely on which question is more natural for your problem.

Where each wins — production evidence

CrewAI in production

PwC deployed CrewAI agents for code generation tasks and moved accuracy from 10% to 70%. The gain came from specialist role separation — the framework's natural model matched the problem. IBM used CrewAI with WatsonX for federal benefits processing, where the role-based mental model aligned with how the actual team was organized. Gelato used hierarchical crews to map SKUs across product catalogues, collapsing what had been a 9-24 month manual process to a 90% automated reduction.

These are all problems where the org chart framing is correct. The team IS the solution.

LangGraph in production

Klarna ran LangGraph at 2.3 million conversations per month. Resolution time dropped from 11 minutes to 2 minutes. They reported $40M in profit improvement from reduced headcount. Worth noting: in 2025, they partially reversed that decision when they over-cut human agents and service quality suffered. The framework did not cause that — the organizational decision did. The technical results were real.

Uber used LangGraph to build internal developer tooling that saved 21,000 developer hours. AppFolio moved text-to-data accuracy from 40% to 80% using LangGraph for structured extraction workflows. A European RegTech firm used LangGraph with PostgreSQL checkpointing for compliance workflows and reported zero silent data corruption — a meaningful claim in a regulated industry where CrewAI's non-deterministic memory would be a hard blocker.

LangGraph's PyPI download numbers tell a similar story: 34-38 million downloads per month versus CrewAI's 5-12 million. The ratio is roughly 6-7x. LangGraph is winning the production adoption race, though much of that is driven by LangChain ecosystem lock-in.

Performance numbers

For equivalent 3-agent workflows, CrewAI uses approximately 56% more tokens per request. The cost comes from prepending role, goal, and backstory on every agent call. For a content drafting use case where quality is the constraint, that overhead is acceptable. For a high-volume inference pipeline, it is not.

Latency on GPT-4o with 3 agents and ~500 tokens of output: LangGraph parallel at 4.2 seconds, CrewAI hierarchical at 7.8 seconds, CrewAI sequential at 13.1 seconds. Sequential is the slowest because it is strictly serial. Hierarchical adds manager overhead. LangGraph's parallel execution wins when nodes have no data dependency.

Going to production in LangGraph

Two things separate a LangGraph prototype from a production system: checkpointing and human-in-the-loop. Both require design decisions you have to make before you build, not after.

Checkpointing

LangGraph's checkpointing ladder is straightforward. MemorySaver for development — in-process, no persistence. SqliteSaver for single-instance production. PostgresSaver for multi-instance, horizontally scaled deployments. The thread ID in configurable is your session key. Every node execution is a checkpoint. If the process restarts mid-graph, the next invocation resumes from the last checkpoint.

Human-in-the-loop with interrupt()

interrupt() inside a node suspends the graph and surfaces a payload to the caller. The graph is paused at that node. You resume it with Command(resume=value). The critical footgun: the entire node re-executes on resume. Every line before interrupt() runs again. That means any side effects before the interrupt — LLM calls, writes, API requests — happen twice. Keep everything before interrupt() idempotent.

# pip install langgraph langgraph-checkpoint-sqlite langchain-openai
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.sqlite import SqliteSaver
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")


class AgentState(TypedDict):
    company: str
    research_notes: str
    analysis: str
    draft: str
    review_feedback: str
    iteration: int


def research_node(state: AgentState) -> AgentState:
    notes = llm.invoke(f"Research {state['company']} competitively. Bullet points.").content
    return {"research_notes": notes, "iteration": state.get("iteration", 0)}


def analysis_node(state: AgentState) -> AgentState:
    analysis = llm.invoke(f"SWOT analysis:\n{state['research_notes']}").content
    return {"analysis": analysis}


def writer_node(state: AgentState) -> AgentState:
    context = state["analysis"]
    if state.get("review_feedback"):
        context += f"\n\nAddress this feedback:\n{state['review_feedback']}"
    draft = llm.invoke(f"Write 600-word competitive intel report:\n{context}").content
    return {"draft": draft}


def review_node(state: AgentState) -> AgentState:
    # AI pre-review (idempotent — safe to re-run on resume)
    ai_feedback = llm.invoke(
        f"Review this report. List issues, then end with APPROVED or REVISE.\n\n{state['draft']}"
    ).content

    # Human-in-the-loop gate: entire node re-executes on resume,
    # so everything above must be idempotent.
    human_decision = interrupt({
        "question": "Approve this report for delivery?",
        "ai_review": ai_feedback,
        "draft_preview": state["draft"][:500],
    })

    if human_decision == "approve":
        return {"review_feedback": "APPROVED by human.", "iteration": state["iteration"] + 1}
    return {"review_feedback": f"Human requested changes: {human_decision}",
            "iteration": state["iteration"] + 1}


def should_revise(state: AgentState) -> Literal["writer", "__end__"]:
    if "APPROVED" in state["review_feedback"].upper() or state["iteration"] >= 3:
        return "__end__"
    return "writer"


checkpointer = SqliteSaver.from_conn_string("./intel_checkpoints.db")

graph = (
    StateGraph(AgentState)
    .add_node("researcher", research_node)
    .add_node("analyst", analysis_node)
    .add_node("writer", writer_node)
    .add_node("reviewer", review_node)
    .add_edge(START, "researcher")
    .add_edge("researcher", "analyst")
    .add_edge("analyst", "writer")
    .add_edge("writer", "reviewer")
    .add_conditional_edges("reviewer", should_revise, {"writer": "writer", "__end__": END})
    .compile(checkpointer=checkpointer)
)

thread = {"configurable": {"thread_id": "intel-notion-001"}}

# First run — pauses at interrupt() inside review_node
for event in graph.stream({"company": "Notion", "iteration": 0}, config=thread):
    print(event)

# Human inspects the draft, then resumes
for event in graph.stream(Command(resume="approve"), config=thread):
    print(event)

SSE streaming with FastAPI

astream_events(version="v2") gives you per-token streaming. The v2 API is the current stable interface — langgraph.prebuilt is not deprecated and still the right import path for prebuilt components. Here is the full FastAPI SSE pattern:

# pip install fastapi uvicorn langgraph langgraph-checkpoint-sqlite langchain-openai
import json
from typing import TypedDict

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.sqlite import SqliteSaver

app = FastAPI()
llm = ChatOpenAI(model="gpt-4o")


class IntelState(MessagesState):
    company: str
    research_notes: str
    draft: str


def research_node(state: IntelState) -> IntelState:
    notes = llm.invoke(f"Research {state.get('company', 'the company')} competitively.").content
    return {"research_notes": notes}


def writer_node(state: IntelState) -> IntelState:
    draft = llm.invoke(
        f"Write a competitive intel report based on:\n{state['research_notes']}"
    ).content
    return {"draft": draft}


checkpointer = SqliteSaver.from_conn_string("./intel_stream.db")

graph = (
    StateGraph(IntelState)
    .add_node("researcher", research_node)
    .add_node("writer", writer_node)
    .add_edge(START, "researcher")
    .add_edge("researcher", "writer")
    .add_edge("writer", END)
    .compile(checkpointer=checkpointer)
)


class RequestBody(BaseModel):
    query: str
    session_id: str


@app.post("/intel/stream")
async def stream_intelligence(body: RequestBody):
    async def event_gen():
        config = {"configurable": {"thread_id": body.session_id}}
        initial_state = {
            "messages": [HumanMessage(content=body.query)],
            "company": body.query,
        }
        async for event in graph.astream(
            initial_state,
            config=config,
            stream_mode="updates",
        ):
            yield f"data: {json.dumps(event)}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        event_gen(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

The X-Accel-Buffering: no header matters when you are running behind nginx. Without it, nginx buffers your SSE stream and the client sees nothing until the buffer flushes.

The 3 migration triggers

Teams that start on CrewAI and migrate to LangGraph almost always hit the same three triggers. If any of these are true for your system, the migration is not optional — it is just a matter of when.

1. More than 3 agents sharing the same variable.

CrewAI's task context system passes outputs forward through context=[task] references. This is linear. When three or more agents need to read and write the same piece of state, you end up with awkward workarounds: serializing shared state into task descriptions, parsing it back out, losing type safety. LangGraph's TypedDict state is a first-class shared store. Every node reads from it and writes back to it. There is no workaround needed.

2. Any variable needs to survive a service restart.

CrewAI has no native checkpointing. If your process crashes mid-run, the workflow starts over. For short workflows with cheap LLM calls, this is annoying. For multi-minute workflows with expensive API calls, it is a hard reliability problem. LangGraph's checkpoint system exists specifically for this. MemorySaver is not production — but SqliteSaver and PostgresSaver are.

3. Conditional routing based on agent output.

CrewAI Flows let you do conditional routing with @router() and @listen(). But once your routing logic depends on structured output from a crew — parsing an LLM response to decide which branch to take — you are fighting the abstraction. LangGraph's conditional edges take a Python function that reads typed state. The routing logic is just code. You can unit test it without running the LLM.

If you hit all three triggers, you are not choosing between frameworks anymore. You are choosing between migrating now or migrating after your production system has been unreliable for six months.

The question to ask

The framework decision reduces to one question: what is the natural representation of your problem?

If you can draw it as an org chart — roles, responsibilities, handoffs between specialists — CrewAI is the right tool. The framework was built around that mental model. You get quick setup, readable code that domain experts can understand, and a sensible default for coordination.

If you can draw it as a flowchart — states, transitions, conditions, branches — LangGraph is the right tool. The boilerplate is real, but every line of it is a decision you made explicitly. When something goes wrong in production, you know exactly where to look.

Most debates about these frameworks treat capability as the axis of comparison. It is not. Both frameworks can build most things. The question is which one stops fighting you when you build. Match the mental model to the problem, and the framework disappears into the background. Pick the wrong one, and every feature you add will feel like swimming upstream.

CrewAI got you thinking about your team. LangGraph got you thinking about your process. Pick the one that matches how the problem already lives in your head.