Introduction: Why Migrate from LangGraph to CrewAI?
LangGraph and CrewAI are two of the most popular frameworks for building multi-agent LLM applications, but they take fundamentally different approaches. LangGraph models agent workflows as explicit state graphs, giving developers fine-grained control over execution flow, state transitions, and conditional routing. CrewAI, on the other hand, embraces a role-based, task-driven paradigm that abstracts away much of the orchestration plumbing in favor of a more declarative, human-readable syntax.
This migration guide is for teams who have built production systems on LangGraph and want to transition to CrewAI — whether to reduce boilerplate, leverage CrewAI's built-in collaboration patterns, or simplify onboarding for non-engineering contributors. We'll walk through the conceptual mapping between the two frameworks, translate real code patterns, and cover the edge cases that trip up most teams during migration.
Understanding the Conceptual Differences
Before touching any code, it's critical to understand how the two frameworks differ in their mental models. A successful migration is not a line-by-line port — it's a re-architecture guided by CrewAI's idioms.
LangGraph: Graphs, Nodes, and State
LangGraph treats everything as a directed graph. You define State as a typed dictionary or Pydantic model, create node functions that accept and return state, and wire them together with edges. Conditional edges let you branch based on state contents. The framework gives you explicit control but requires you to specify every transition.
CrewAI: Roles, Tasks, and Crews
CrewAI organizes work around Agents (role-playing entities with a role, goal, and backstory), Tasks (units of work with expected outputs), and Crews (the orchestrator that binds agents and tasks together). Instead of manually wiring edges, you declare which agent owns which task, and CrewAI handles the execution order, context passing, and inter-agent communication.
Mapping Core Concepts
- LangGraph Node → CrewAI Agent + Task (a node typically does work; in CrewAI, an agent performs a task)
- LangGraph State → CrewAI Task outputs and crew-level context (passed automatically between tasks)
- LangGraph Edge → CrewAI task ordering (sequential by default, or process-based for parallel)
- Conditional Edge → CrewAI conditional task assignment or custom process logic
- LangGraph Tool → CrewAI Tool (both support function-based tools, but CrewAI integrates them at the agent level)
- LangGraph Checkpointer → CrewAI's built-in memory and flow state persistence
Setting Up Your Environment
Install CrewAI alongside your existing LangGraph setup so you can run both during the transition period. This lets you A/B test outputs and roll back if needed.
# Install CrewAI with all extras
pip install crewai crewai-tools
# Keep LangGraph installed during migration
pip install langgraph langchain-openai
# Verify installations
python -c "import crewai; print(crewai.__version__)"
python -c "import langgraph; print(langgraph.__version__)"
Set your API keys in your environment. CrewAI uses LiteLLM under the hood, so it supports the same providers you're likely already using.
import os
os.environ["OPENAI_API_KEY"] = "sk-your-key-here"
# CrewAI also supports Anthropic, Groq, Ollama, etc.
# os.environ["ANTHROPIC_API_KEY"] = "your-key"
Migrating a Simple Sequential Workflow
Let's start with the most common pattern: a sequential pipeline where one agent's output feeds into the next. This is where most teams begin their migration.
The Original LangGraph Implementation
Here's a typical LangGraph workflow with two nodes — a researcher and a writer — connected by a simple edge:
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langchain_openai import ChatOpenAI
class ResearchState(TypedDict):
topic: str
research_notes: str
final_article: str
llm = ChatOpenAI(model="gpt-4o")
def research_node(state: ResearchState) -> dict:
prompt = f"Research the following topic and provide detailed notes: {state['topic']}"
response = llm.invoke(prompt)
return {"research_notes": response.content}
def write_node(state: ResearchState) -> dict:
prompt = f"Write an article based on these notes:\n{state['research_notes']}"
response = llm.invoke(prompt)
return {"final_article": response.content}
graph = StateGraph(ResearchState)
graph.add_node("researcher", research_node)
graph.add_node("writer", write_node)
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "writer")
graph.add_edge("writer", END)
app = graph.compile()
result = app.invoke({"topic": "The future of quantum computing"})
print(result["final_article"])
The Equivalent CrewAI Implementation
Now here's the same workflow in CrewAI. Notice how the graph wiring disappears — you declare agents and tasks, and the crew handles execution order based on the task list:
from crewai import Agent, Task, Crew, Process
researcher = Agent(
role="Senior Research Analyst",
goal="Uncover detailed, accurate information about the given topic",
backstory="You are a meticulous researcher with 15 years of experience "
"synthesizing complex technical topics into clear notes.",
verbose=True,
llm="gpt-4o"
)
writer = Agent(
role="Technical Writer",
goal="Transform research notes into a compelling, well-structured article",
backstory="You are an award-winning technical writer known for making "
"complex subjects accessible to broad audiences.",
verbose=True,
llm="gpt-4o"
)
research_task = Task(
description="Research the topic: {topic}. Provide comprehensive notes "
"covering key concepts, current state, and future directions.",
expected_output="A detailed research brief with 5-7 key findings and "
"supporting context.",
agent=researcher
)
writing_task = Task(
description="Using the research brief provided, write a polished article "
"about the topic. The article should be engaging and informative.",
expected_output="A 800-1200 word article with a clear introduction, "
"body sections, and conclusion.",
agent=writer,
context=[research_task] # Explicitly pass prior task output
)
crew = Crew(
agents=[researcher, writer],
tasks=[research_task, writing_task],
process=Process.sequential,
verbose=True
)
result = crew.kickoff(inputs={"topic": "The future of quantum computing"})
print(result.raw)
The key differences to notice: CrewAI agents have rich personas (role, goal, backstory) that shape their behavior through system prompting. Tasks are self-contained descriptions with expected outputs. The context parameter on a task explicitly declares dependencies — this replaces LangGraph's implicit state-passing through edges.
Migrating Conditional Routing
Conditional edges are one of LangGraph's strengths. In CrewAI, you achieve similar behavior through conditional task assignment, custom tools, or the Flows feature for complex branching.
LangGraph Conditional Edge Example
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
query: str
category: str
response: str
def classify_node(state: State) -> dict:
# Simulated classification logic
if "code" in state["query"].lower():
return {"category": "technical"}
else:
return {"category": "general"}
def technical_node(state: State) -> dict:
return {"response": f"Technical answer for: {state['query']}"}
def general_node(state: State) -> dict:
return {"response": f"General answer for: {state['query']}"}
def route(state: State) -> str:
return "technical" if state["category"] == "technical" else "general"
graph = StateGraph(State)
graph.add_node("classify", classify_node)
graph.add_node("technical", technical_node)
graph.add_node("general", general_node)
graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route, {"technical": "technical", "general": "general"})
graph.add_edge("technical", END)
graph.add_edge("general", END)
app = graph.compile()
result = app.invoke({"query": "How do I write code in Python?"})
CrewAI Equivalent Using Flows
For conditional routing, CrewAI's Flows module is the closest analog. It gives you event-driven orchestration with explicit branching while still leveraging CrewAI's agent system:
from crewai import Agent, Task, Crew
from crewai.flow.flow import Flow, listen, start, router
class SupportFlow(Flow):
@start()
def classify_query(self):
query = self.state.get("query", "")
if "code" in query.lower():
self.state["category"] = "technical"
else:
self.state["category"] = "general"
return self.state
@router(classify_query)
def route_query(self):
if self.state["category"] == "technical":
return "technical"
return "general"
@listen("technical")
def handle_technical(self):
technical_agent = Agent(
role="Technical Support Engineer",
goal="Provide precise technical answers with code examples",
backstory="You are a senior engineer who excels at explaining "
"technical concepts with practical code.",
llm="gpt-4o"
)
task = Task(
description=f"Answer this technical query: {self.state['query']}",
expected_output="A technical answer with code examples if applicable.",
agent=technical_agent
)
crew = Crew(agents=[technical_agent], tasks=[task])
result = crew.kickoff()
self.state["response"] = result.raw
return self.state
@listen("general")
def handle_general(self):
general_agent = Agent(
role="Customer Support Specialist",
goal="Provide helpful, friendly answers to general questions",
backstory="You are a customer support expert known for clear, "
"empathetic communication.",
llm="gpt-4o"
)
task = Task(
description=f"Answer this general query: {self.state['query']}",
expected_output="A helpful, friendly response.",
agent=general_agent
)
crew = Crew(agents=[general_agent], tasks=[task])
result = crew.kickoff()
self.state["response"] = result.raw
return self.state
flow = SupportFlow()
result = flow.kickoff(inputs={"query": "How do I write code in Python?"})
print(result.state["response"])
Migrating Tools and Function Calling
Both frameworks support tool use, but the registration patterns differ. LangGraph typically uses LangChain tools attached to a model, while CrewAI tools are attached directly to agents.
LangGraph Tool Pattern
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
@tool
def search_web(query: str) -> str:
"""Search the web for information."""
# Simulated search
return f"Search results for: {query}"
@tool
def calculate(expression: str) -> str:
"""Evaluate a mathematical expression."""
try:
return str(eval(expression))
except Exception as e:
return f"Error: {e}"
llm = ChatOpenAI(model="gpt-4o")
agent = create_react_agent(llm, [search_web, calculate])
result = agent.invoke({"messages": [("user", "What is 15 * 23?")]})
CrewAI Tool Pattern
from crewai.tools import tool
from crewai import Agent, Task, Crew
@tool("Search Web")
def search_web(query: str) -> str:
"""Search the web for information on a given query."""
return f"Search results for: {query}"
@tool("Calculate")
def calculate(expression: str) -> str:
"""Evaluate a mathematical expression and return the result."""
try:
return str(eval(expression))
except Exception as e:
return f"Error: {e}"
# Tools are attached at the agent level
research_agent = Agent(
role="Research Assistant",
goal="Find accurate information and perform calculations as needed",
backstory="You are a precise assistant who uses tools to verify facts.",
tools=[search_web, calculate],
llm="gpt-4o"
)
task = Task(
description="Calculate 15 * 23 and explain the result.",
expected_output="The numerical result with a brief explanation.",
agent=research_agent
)
crew = Crew(agents=[research_agent], tasks=[task])
result = crew.kickoff()
print(result.raw)
Note the @tool decorator comes from crewai.tools and accepts a name argument. CrewAI also supports BaseTool subclasses for more complex tools and integrates with LangChain tools via crewai.tools.LangchainToolWrapper — useful during migration when you want to reuse existing tool implementations.
from crewai.tools import BaseTool
class CustomSearchTool(BaseTool):
name: str = "Custom Search"
description: str = "Searches a custom knowledge base for relevant documents."
def _run(self, query: str) -> str:
# Your custom logic here
return f"Custom search results for: {query}"
# Use directly in an agent
agent = Agent(
role="Researcher",
goal="Find information in the knowledge base",
backstory="You are an expert at finding relevant information.",
tools=[CustomSearchTool()],
llm="gpt-4o"
)
Migrating State Management and Memory
LangGraph's State object is the backbone of every workflow. In CrewAI, state is distributed across task outputs, crew-level memory, and (for Flows) the flow state object. Understanding this shift is essential for complex migrations.
LangGraph State with Reducers
from typing import TypedDict, Annotated, List
from operator import add
class ConversationState(TypedDict):
messages: Annotated[List[str], add]
summary: str
turn_count: int
def chat_node(state: ConversationState) -> dict:
new_message = f"Turn {state['turn_count']}: responding to {state['messages'][-1]}"
return {
"messages": [new_message],
"turn_count": state["turn_count"] + 1
}
CrewAI Memory and Context
CrewAI provides built-in short-term, long-term, and entity memory. For structured state that mirrors LangGraph's approach, use Flows with a typed state model:
from crewai import Agent, Task, Crew
from crewai.flow.flow import Flow, start, listen
from pydantic import BaseModel, Field
from typing import List
class ConversationState(BaseModel):
messages: List[str] = Field(default_factory=list)
summary: str = ""
turn_count: int = 0
class ConversationFlow(Flow[ConversationState]):
@start()
def initialize(self):
self.state.messages.append("Conversation started")
self.state.turn_count = 1
return self.state
@listen(initialize)
def process_turn(self):
agent = Agent(
role="Conversation Partner",
goal="Engage in meaningful dialogue",
backstory="You are a thoughtful conversationalist.",
llm="gpt-4o"
)
task = Task(
description=f"Respond to the conversation: {self.state.messages}",
expected_output="A thoughtful response message.",
agent=agent
)
crew = Crew(agents=[agent], tasks=[task], memory=True)
result = crew.kickoff()
self.state.messages.append(result.raw)
self.state.turn_count += 1
return self.state
flow = ConversationFlow()
result = flow.kickoff()
print(f"Turn count: {result.state.turn_count}")
print(f"Messages: {result.state.messages}")
For persistent memory across crew executions, enable the memory flag on your crew. CrewAI stores conversation history and entity knowledge, which can replace LangGraph's checkpointers for many use cases:
crew = Crew(
agents=[agent1, agent2],
tasks=[task1, task2],
memory=True,
verbose=True
)
Migrating Parallel Execution
LangGraph supports parallel execution through fan-out patterns (multiple edges from a single node). CrewAI handles this through the hierarchical process or by running tasks concurrently when they have no dependencies.
LangGraph Fan-Out Pattern
from langgraph.graph import StateGraph, START, END
from typing import TypedDict
class State(TypedDict):
topic: str
analysis_a: str
analysis_b: str
synthesis: str
def analyze_a(state: State) -> dict:
return {"analysis_a": f"Perspective A on {state['topic']}"}
def analyze_b(state: State) -> dict:
return {"analysis_b": f"Perspective B on {state['topic']}"}
def synthesize(state: State) -> dict:
return {"synthesis": f"Combining: {state['analysis_a']} + {state['analysis_b']}"}
graph = StateGraph(State)
graph.add_node("analyze_a", analyze_a)
graph.add_node("analyze_b", analyze_b)
graph.add_node("synthesize", synthesize)
graph.add_edge(START, "analyze_a")
graph.add_edge(START, "analyze_b") # Fan-out: parallel execution
graph.add_edge("analyze_a", "synthesize")
graph.add_edge("analyze_b", "synthesize") # Fan-in: waits for both
graph.add_edge("synthesize", END)
app = graph.compile()
result = app.invoke({"topic": "AI regulation"})
CrewAI Hierarchical Process
For parallel work, CrewAI's hierarchical process uses a manager agent that delegates tasks to worker agents concurrently:
from crewai import Agent, Task, Crew, Process
manager = Agent(
role="Project Manager",
goal="Coordinate analysis from multiple perspectives and synthesize results",
backstory="You are an expert project manager who delegates effectively "
"and synthesizes diverse viewpoints.",
llm="gpt-4o",
allow_delegation=True
)
analyst_a = Agent(
role="Economic Analyst",
goal="Analyze topics from an economic perspective",
backstory="You are an economist specializing in technology policy.",
llm="gpt-4o"
)
analyst_b = Agent(
role="Ethics Analyst",
goal="Analyze topics from an ethical and societal perspective",
backstory="You are a philosopher specializing in technology ethics.",
llm="gpt-4o"
)
task_a = Task(
description="Analyze {topic} from an economic perspective. "
"Consider market impacts, costs, and benefits.",
expected_output="A 300-word economic analysis.",
agent=analyst_a
)
task_b = Task(
description="Analyze {topic} from an ethical perspective. "
"Consider societal impacts, fairness, and rights.",
expected_output="A 300-word ethical analysis.",
agent=analyst_b
)
synthesis_task = Task(
description="Synthesize the economic and ethical analyses into a "
"unified report on {topic}.",
expected_output="A 500-word synthesis report combining both perspectives.",
agent=manager,
context=[task_a, task_b]
)
crew = Crew(
agents=[manager, analyst_a, analyst_b],
tasks=[task_a, task_b, synthesis_task],
process=Process.hierarchical,
manager_agent=manager,
verbose=True
)
result = crew.kickoff(inputs={"topic": "AI regulation"})
print(result.raw)
Migrating Human-in-the-Loop Patterns
LangGraph supports human-in-the-loop through interrupt_before and interrupt_after parameters. CrewAI achieves similar behavior through its human_input flag on tasks and custom validation callbacks.
LangGraph Interrupt Pattern
graph = StateGraph(State)
graph.add_node("draft", draft_node)
graph.add_node("review", review_node)
graph.add_edge(START, "draft")
graph.add_edge("draft", "review")
graph.add_edge("review", END)
app = graph.compile(
interrupt_before=["review"] # Pause before review node
)
# First invocation pauses
state = app.invoke({"content": "initial draft"})
# Human reviews and modifies state
state["content"] = human_edit(state["content"])
# Resume execution
final = app.invoke(state)
CrewAI Human Input Pattern
from crewai import Agent, Task, Crew
from crewai.tasks.conditional_task import ConditionalTask
writer = Agent(
role="Content Writer",
goal="Write high-quality content that meets user requirements",
backstory="You are a skilled writer who values feedback.",
llm="gpt-4o"
)
draft_task = Task(
description="Write a draft article about {topic}.",
expected_output="A draft article of 500-800 words.",
agent=writer,
human_input=True # Prompts for human feedback after execution
)
crew = Crew(agents=[writer], tasks=[draft_task], verbose=True)
result = crew.kickoff(inputs={"topic": "Renewable energy"})
For more granular control, use a custom output JSON parser or a validation function that can request re-execution:
def validate_output(output):
"""Custom validation that can trigger re-execution."""
if len(output.raw) < 100:
return (False, "Output too short, please expand.")
return (True, output.raw)
review_task = Task(
description="Review and refine the draft.",
expected_output="A polished final article.",
agent=writer,
output_pydantic=None,
output_json=None,
output_file=None,
callback=validate_output
)
Best Practices for a Smooth Migration
1. Migrate Incrementally, Not All at Once
Don't attempt a big-bang rewrite. Start with a single, self-contained workflow — ideally one with simple sequential logic — and migrate it end to end. This validates your CrewAI setup, tooling, and deployment pipeline before you tackle complex graphs.
2. Preserve Your Tool Implementations
Your custom tools (search, database queries, API integrations) represent significant engineering investment. CrewAI's LangchainToolWrapper lets you wrap existing LangChain tools directly, avoiding rewrites:
from crewai.tools import LangchainToolWrapper
from langchain_core.tools import Tool
# Existing LangChain tool
existing_search = Tool(
name="Search",
description="Search the knowledge base",
func=lambda q: f"Results for {q}"
)
# Wrap for CrewAI
crewai_search = LangchainToolWrapper(existing_search)
agent = Agent(
role="Researcher",
goal="Find information",
backstory="Expert researcher",
tools=[crewai_search],
llm="gpt-4o"
)
3. Map Your State Schema Explicitly
Before writing any CrewAI code, create a mapping document that lists every field in your LangGraph state and where it will live in CrewAI — task output, flow state, or crew memory. This prevents data loss during migration and surfaces hidden dependencies.
4. Test Output Parity Early
Run both implementations side by side with the same inputs and compare outputs. Use structured evaluation (not just eyeballing) to measure quality differences. Accept that outputs won't be identical — CrewAI's agent personas produce different stylistic results — but verify that factual accuracy and task completion are equivalent.
import json
def compare_outputs(langgraph_result, crewai_result, test_case):
"""Simple parity check for migration validation."""
checks = {
"test_case": test_case,
"langgraph_length": len(langgraph_result),
"crewai_length": len(crewai_result),
"length_ratio": len(crewai_result) / max(len(langgraph_result), 1),
"key_terms_present": all(
term.lower() in crewai_result.lower()
for term in ["introduction", "conclusion"] # Customize per task
)
}
return json.dumps(checks, indent=2)
# Example usage during migration testing
print(compare_outputs(lg_output, crew_output, "quantum_computing_article"))
5. Leverage CrewAI's Delegation for Complex Graphs
If your LangGraph workflow has more than 5-6 nodes with complex conditional routing, consider whether the explicit graph structure is actually necessary. Many LangGraph workflows that feel complex are really just sequential pipelines with occasional branching — CrewAI's hierarchical process with a manager agent can often express the same logic more naturally.
6. Handle Error and Retry Logic Explicitly
LangGraph's retry_policy and error edges don't have direct CrewAI equivalents. Wrap your crew execution in try/except blocks and implement retry logic at the application level:
import time
from crewai import Crew
def run_crew_with_retry(crew, inputs, max_retries=3, delay=2):
for attempt in range(max_retries):
try:
result = crew.kickoff(inputs=inputs)
return result
except Exception as e:
print(f"Attempt {attempt + 1} failed: {e}")
if attempt < max_retries - 1:
time.sleep(delay * (attempt + 1))
else:
raise
result = run_crew_with_retry(crew, {"topic": "AI regulation"})
7. Use CrewAI Flows for State-Heavy Workflows
If your LangGraph application relies heavily on complex state transitions, reducers, and conditional routing, CrewAI Flows are your best migration target. They provide the explicit control you're used to while still integrating with CrewAI's agent system. Reserve plain Crews for simpler sequential or hierarchical patterns.
8. Document Agent Personas Carefully
LangGraph nodes are anonymous functions — their behavior is defined entirely by their code. CrewAI agents have personas that significantly influence output quality and style. Invest time in crafting role, goal, and backstory fields. Vague personas produce vague results. Test multiple persona formulations and measure their impact on output quality.
Common Pitfalls and How to Avoid Them
Pitfall: Treating Migration as a Syntax Translation
The biggest mistake teams make is trying to mechanically translate LangGraph code into CrewAI line by line. This produces awkward, non-idiomatic CrewAI code that doesn't leverage the framework's strengths. Instead, understand what each part of your LangGraph workflow achieves, then express that goal using CrewAI's primitives.
Pitfall: Overusing Hierarchical Process
The hierarchical process is powerful but adds latency and cost (the manager agent makes additional LLM calls). Use it only when you genuinely need dynamic delegation. For fixed task orderings with known dependencies, sequential process with explicit context parameters is simpler and cheaper.
Pitfall: Ignoring Token Costs
CrewAI's agent personas add tokens to every call through system prompts. With many agents, this adds up. Monitor token usage during migration and optimize personas for conciseness. A 200-word backstory that doesn't improve output quality is pure cost.
# Before: verbose backstory (wasteful)
agent = Agent(
role="Writer",
goal="Write good content",
backstory="You graduated from Harvard with a degree in English literature "
"in 2015. After graduation, you worked at three different "
"publications..." # 200+ words of irrelevant detail
)
# After: concise, impactful backstory
agent = Agent(
role="Writer",
goal="Write clear, engaging content that serves the reader's needs",
backstory="Award-winning writer with 10 years of experience in "
"technical and editorial content."
)
Pitfall: Losing Observability
LangGraph integrates with LangSmith for tracing. CrewAI has its own telemetry but you may need to add custom logging to maintain your observability standards. Use CrewAI's verbose mode and integrate with your existing monitoring:
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("crewai_migration")
class LoggingCrew(Crew):
def kickoff(self, *args, **kwargs):
logger.info(f"Starting crew with {len(self.agents)} agents")
result = super().kickoff(*args, **kwargs)
logger.info(f"Crew completed. Output length: {len(result.raw)}")
return result
Conclusion
Migrating from LangGraph to CrewAI is not a mechanical code translation — it's a paradigm shift from graph-based orchestration to role-based collaboration. The frameworks solve the same fundamental problem (coordinating multiple LLM-powered steps) but with different philosophies: LangGraph gives you maximum control over execution flow, while CrewAI prioritizes developer ergonomics and natural-language expressiveness. Start your migration with a simple sequential workflow, preserve your existing tool investments using CrewAI's LangChain tool wrappers, and use Flows for state-heavy applications that need explicit routing control. Test output parity early and often, invest in crafting effective agent personas, and resist the urge to over-engineer with hierarchical processes when sequential execution suffices. The result of a thoughtful migration is typically a codebase that is shorter, more readable, and easier for non-engineers to understand and modify — which is, after all, the core value proposition that draws teams to CrewAI in the first place.