Combine CrewAI's role-based agents with LangGraph's stateful workflows to build production-ready multi-agent systems.

How to Build a Multi-Agent System with CrewAI and LangGraph

Overview

Learn how to build a production-ready multi-agent system using CrewAI for role-based agent orchestration and LangGraph for stateful workflow management. This tutorial combines the best of both frameworks to create a robust AI automation pipeline.

Prerequisites

Python 3.10+
OpenAI API key (or other LLM provider)
Basic understanding of Python async/await

Installation

# Create virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install dependencies
pip install crewai langgraph langchain-openai pydantic python-dotenv

# Create .env file
echo "OPENAI_API_KEY=sk-..." > .env

Project Structure

multi-agent-system/
├── src/
│   ├── __init__.py
│   ├── agents.py           # CrewAI agent definitions
│   ├── tasks.py            # Task definitions
│   ├── workflow.py         # LangGraph workflow
│   └── tools.py            # Custom tools
├── config/
│   └── settings.yaml       # Configuration
├── outputs/
│   └── results/            # Output directory
├── main.py                 # Entry point
├── requirements.txt
└── README.md

Step 1: Define Agents with CrewAI

Create src/agents.py:

from crewai import Agent
from langchain_openai import ChatOpenAI
from src.tools import search_tool, file_tool

# Research agent
researcher = Agent(
    role='Senior Research Analyst',
    goal='Uncover cutting-edge developments in AI and technology',
    backstory='''You are a senior research analyst with deep expertise in 
    artificial intelligence and emerging technologies. You have a knack for 
    identifying emerging trends and understanding complex technical concepts. 
    Your work is used by industry leaders to make strategic decisions.''',
    verbose=True,
    allow_delegation=False,
    llm=ChatOpenAI(model="gpt-4o"),
    tools=[search_tool],
)

# Writer agent
writer = Agent(
    role='Technical Content Writer',
    goal='Create compelling, accurate content about technology',
    backstory='''You are a technical content writer specializing in AI and 
    technology. You excel at translating complex technical concepts into 
    clear, engaging content that resonates with both technical and 
    non-technical audiences.''',
    verbose=True,
    allow_delegation=True,
    llm=ChatOpenAI(model="gpt-4o"),
)

# Editor agent
editor = Agent(
    role='Content Editor and Quality Assurance',
    goal='Ensure content accuracy, clarity, and quality',
    backstory='''You are a meticulous content editor with a keen eye for detail. 
    You ensure all content meets the highest standards of accuracy, clarity, 
    and engagement before publication.''',
    verbose=True,
    allow_delegation=False,
    llm=ChatOpenAI(model="gpt-4o"),
)

# Manager agent (for hierarchical process)
manager = Agent(
    role='Project Manager',
    goal='Coordinate the team and ensure project success',
    backstory='''You are an experienced project manager who coordinates 
    cross-functional teams. You ensure deadlines are met, quality standards 
    are maintained, and the final deliverable exceeds expectations.''',
    verbose=True,
    allow_delegation=True,
    llm=ChatOpenAI(model="gpt-4o"),
)

Step 2: Define Custom Tools

Create src/tools.py:

from crewai_tools import tool
from langchain_community.tools import DuckDuckGoSearchRun
from crewai_tools import LangChainTool

@tool
def search_web(query: str, limit: int = 10) -> str:
    """Search the web for information on a given topic."""
    search = DuckDuckGoSearchRun(num_results=limit)
    return search.run(query)

@tool
def save_to_file(content: str, filename: str, directory: str = "outputs") -> str:
    """Save content to a file."""
    import os
    os.makedirs(directory, exist_ok=True)
    filepath = os.path.join(directory, filename)
    with open(filepath, 'w', encoding='utf-8') as f:
        f.write(content)
    return f"Saved to {filepath}"

@tool
def read_file(filename: str, directory: str = "outputs") -> str:
    """Read content from a file."""
    import os
    filepath = os.path.join(directory, filename)
    if not os.path.exists(filepath):
        return f"File not found: {filepath}"
    with open(filepath, 'r', encoding='utf-8') as f:
        return f.read()

# LangChain tool wrapper
search_tool = LangChainTool(
    tool=DuckDuckGoSearchRun(),
    name="Web Search",
    description="Search the web for information",
)

file_tool = save_to_file

Step 3: Define Tasks

Create src/tasks.py:

from crewai import Task
from src.agents import researcher, writer, editor

def create_research_task(topic: str, expected_output: str = None) -> Task:
    return Task(
        description=f'''
        Conduct comprehensive research on: {topic}
        
        Find the latest developments, key players, trends, and challenges.
        Gather information from at least 5 authoritative sources.
        ''',
        expected_output=expected_output or '''
        A detailed research report with:
        - Executive summary
        - Key findings (at least 10)
        - Source citations
        - Identified trends and patterns
        ''',
        agent=researcher,
        output_file='outputs/research_report.md',
    )

def create_writing_task(topic: str, audience: str, tone: str = "professional") -> Task:
    return Task(
        description=f'''
        Write a comprehensive article about: {topic}
        
        Target audience: {audience}
        Tone: {tone}
        Length: approximately 2000 words
        ''',
        expected_output='''
        A well-structured article with:
        - Engaging introduction
        - Clear sections with headings
        - Supporting evidence and examples
        - Actionable insights
        - Conclusion with next steps
        ''',
        agent=writer,
        context=[],  # Will be set when creating crew
        output_file='outputs/article_draft.md',
    )

def create_editing_task() -> Task:
    return Task(
        description='''
        Review and edit the article for accuracy, clarity, and quality.
        Check for factual errors, improve flow, and ensure consistency.
        ''',
        expected_output='''
        A polished, publication-ready article with:
        - All factual claims verified
        - Improved clarity and flow
        - Consistent tone and style
        - Proper formatting and structure
        ''',
        agent=editor,
        output_file='outputs/final_article.md',
    )

Step 4: Build LangGraph Workflow

Create src/workflow.py:

from typing import TypedDict, List, Annotated
from langgraph.graph import StateGraph, END, add_messages
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage
from crewai import Crew, Task, Process
from src.agents import researcher, writer, editor, manager
from src.tasks import create_research_task, create_writing_task, create_editing_task

class WorkflowState(TypedDict):
    """State for the multi-agent workflow."""
    messages: Annotated[List[BaseMessage], add_messages]
    topic: str
    audience: str
    tone: str
    research_output: str
    draft_output: str
    final_output: str
    status: str
    iteration: int

def research_node(state: WorkflowState) -> dict:
    """Execute research task."""
    research_task = create_research_task(state["topic"])
    
    crew = Crew(
        agents=[researcher],
        tasks=[research_task],
        process=Process.sequential,
        verbose=True,
    )
    
    result = crew.kickoff(inputs={"topic": state["topic"]})
    
    return {
        "research_output": result,
        "status": "research_complete",
        "iteration": state.get("iteration", 0) + 1,
    }

def write_node(state: WorkflowState) -> dict:
    """Execute writing task."""
    research_task = create_research_task(state["topic"])
    writing_task = create_writing_task(
        topic=state["topic"],
        audience=state["audience"],
        tone=state["tone"],
    )
    writing_task.context = [research_task]
    
    crew = Crew(
        agents=[researcher, writer],
        tasks=[research_task, writing_task],
        process=Process.sequential,
        verbose=True,
    )
    
    result = crew.kickoff(inputs={
        "topic": state["topic"],
        "audience": state["audience"],
        "tone": state["tone"],
    })
    
    return {
        "draft_output": result,
        "status": "draft_complete",
    }

def edit_node(state: WorkflowState) -> dict:
    """Execute editing task."""
    research_task = create_research_task(state["topic"])
    writing_task = create_writing_task(
        topic=state["topic"],
        audience=state["audience"],
        tone=state["tone"],
    )
    editing_task = create_editing_task()
    
    writing_task.context = [research_task]
    editing_task.context = [writing_task]
    
    crew = Crew(
        agents=[researcher, writer, editor],
        tasks=[research_task, writing_task, editing_task],
        process=Process.sequential,
        verbose=True,
    )
    
    result = crew.kickoff(inputs={
        "topic": state["topic"],
        "audience": state["audience"],
        "tone": state["tone"],
    })
    
    return {
        "final_output": result,
        "status": "complete",
    }

def should_continue(state: WorkflowState) -> str:
    """Determine next step."""
    if state.get("iteration", 0) >= 3:
        return "finalize"
    if state.get("status") == "draft_complete":
        return "edit"
    return "write"

def create_workflow():
    """Create the LangGraph workflow."""
    builder = StateGraph(WorkflowState)
    
    # Add nodes
    builder.add_node("research", research_node)
    builder.add_node("write", write_node)
    builder.add_node("edit", edit_node)
    
    # Define edges
    builder.set_entry_point("research")
    builder.add_edge("research", "write")
    builder.add_conditional_edges(
        "write",
        should_continue,
        {"write": "write", "edit": "edit", "finalize": END},
    )
    builder.add_edge("edit", END)
    
    return builder.compile()

# Create workflow instance
workflow = create_workflow()

Step 5: Main Entry Point

Create main.py:

from src.workflow import workflow
from langchain_core.messages import HumanMessage

def main():
    """Run the multi-agent workflow."""
    
    # Initial state
    initial_state = {
        "messages": [HumanMessage(content="Start the workflow")],
        "topic": "AI Agent Frameworks in 2026",
        "audience": "technical decision-makers",
        "tone": "professional yet accessible",
        "research_output": "",
        "draft_output": "",
        "final_output": "",
        "status": "pending",
        "iteration": 0,
    }
    
    # Run workflow
    print("Starting multi-agent workflow...")
    result = workflow.invoke(initial_state)
    
    print("\n" + "="*60)
    print("WORKFLOW COMPLETE")
    print("="*60)
    print(f"\nTopic: {result['topic']}")
    print(f"Status: {result['status']}")
    print(f"Iterations: {result['iteration']}")
    
    print("\n--- Final Output ---")
    print(result['final_output'][:1000] + "..." if len(result['final_output']) > 1000 else result['final_output'])
    
    return result

if __name__ == "__main__":
    main()

Step 6: Run the Workflow

# Run the workflow
python main.py

Advanced: Add Human-in-the-Loop

Modify src/workflow.py to add human review:

from langgraph.checkpoint.memory import MemorySaver

def create_workflow_with_human_review():
    """Create workflow with human review checkpoint."""
    builder = StateGraph(WorkflowState)
    
    builder.add_node("research", research_node)
    builder.add_node("write", write_node)
    builder.add_node("review", human_review_node)  # New node
    builder.add_node("edit", edit_node)
    
    builder.set_entry_point("research")
    builder.add_edge("research", "write")
    builder.add_edge("write", "review")  # Interrupt before review
    
    # Interrupt before review node
    builder.interrupt_before("review")
    
    builder.add_edge("review", "edit")
    builder.add_edge("edit", END)
    
    # Add checkpointer for persistence
    return builder.compile(checkpointer=MemorySaver())

def human_review_node(state: WorkflowState) -> dict:
    """Human review step."""
    print("\n" + "="*60)
    print("DRAFT READY FOR REVIEW")
    print("="*60)
    print(f"\n{state['draft_output'][:500]}...")
    print("\n--- PAUSED FOR HUMAN INPUT ---")
    
    # In production, this would integrate with a UI
    # For now, we'll auto-approve
    response = input("Approve draft? (yes/no): ").lower()
    
    if response == "yes":
        return {"status": "approved"}
    else:
        return {"status": "revision_needed"}

Step 7: Run with Human Review

from src.workflow import create_workflow_with_human_review
from langgraph.checkpoint.memory import MemorySaver

workflow = create_workflow_with_human_review()
checkpointer = MemorySaver()

config = {"configurable": {"thread_id": "session-1"}}
result = workflow.invoke(initial_state, config=config)

# Check if paused
snapshot = workflow.get_state(config)
if snapshot.next:
    print("Workflow paused for human review")
    
    # Resume after review
    result = workflow.invoke(None, config=config)

Output

The workflow will generate:

research_report.md - Detailed research findings
article_draft.md - First draft of the article
final_article.md - Polished, publication-ready article

Troubleshooting

Common Issues

API rate limits:

# Add retry logic
from crewai import Crew
import time

crew = Crew(agents=[researcher], tasks=[task])
for attempt in range(3):
    try:
        result = crew.kickoff()
        break
    except Exception as e:
        if attempt == 2:
            raise
        time.sleep(2 ** attempt)

Memory issues:

# Reduce context
writer = Agent(
    role='Writer',
    goal='Write content',
    backstory='...',
    llm=ChatOpenAI(model="gpt-4o-mini"),  # Use smaller model
)

Slow execution:

# Enable caching
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    cache=True,  # Enable caching
    memory=True,  # Enable memory
)

Best Practices

Start simple: Begin with sequential process, add complexity later
Use specific goals: Clear goals lead to better agent behavior
Enable verbose: Start verbose for debugging, reduce for production
Add error handling: Implement callbacks for error recovery
Use caching: Enable caching to save costs and improve speed
Test incrementally: Test each agent and task independently

Resources

CrewAI Documentation: https://docs.crewai.com/
LangGraph Documentation: https://langchain-ai.github.io/langgraph/
CrewAI Examples: https://github.com/crewAIInc/crewAI-examples
LangGraph Examples: https://github.com/langchain-ai/langgraph-examples

Last updated: May 2026