CrewAI Multi-Agent Tutorial — The 800ms Token Blowup That Killed Our Research Pipeline

We built a content research pipeline using CrewAI. Three agents: Researcher, Writer, Editor. Sequential process. Simple, right? At 500 requests per minute, our token burn hit $4,000/month. The p99 latency was 12 seconds. And then the pipeline started silently returning empty articles — no errors, just blank text. The logs showed the Researcher agent had delegated a sub-task to itself 47 times in a single run, each delegation costing 150ms and 2,000 tokens. We'd built a team of agents that were holding meetings about holding meetings.

Most CrewAI tutorials show you the happy path: install the library, define some agents, run a crew. They skip the production realities — token budgets, delegation loops, tool timeout settings, and the fact that CrewAI agents are just GPT-4o wrappers with system prompts. They don't tell you that a single misconfigured allow_delegation=True can 10x your costs. They don't mention that crew.kickoff() is synchronous and blocks your event loop. They don't explain that the default LLM temperature is 0.7, which means your 'reliable' research agent is actually a creative writer.

This article covers everything the tutorials miss: the internal architecture of CrewAI (it's just chained LLM calls with a fancy state machine), the exact configuration that killed our pipeline, how to set token budgets per agent, when to use Sequential vs Hierarchical (and when to use neither), and a production debugging guide for when your agents start talking to themselves. You'll get runnable code, real incident data, and a triage cheat sheet for 2am pages. If you're deploying CrewAI to production, read this before you write a single agent.

How CrewAI Actually Works Under the Hood

CrewAI is not a complex orchestration engine. It's a chain of LLM calls wrapped in a state machine. When you call crew.kickoff(), the framework iterates through each task in order. For each task, it constructs a system prompt from the agent's role, goal, and backstory, appends the task description, and sends it to the LLM. The LLM response is parsed for tool calls (if any), which are executed synchronously. The result is stored and passed to the next task.

The key abstraction that most tutorials miss is the Task object. Each task has a context field that can include the output of previous tasks. But here's the gotcha: the context is appended to the prompt as a raw string. If your previous task output is 10,000 tokens, your next task's prompt is 10,000 tokens longer. We saw a 3-agent pipeline where the Writer agent's prompt was 18,000 tokens because it included the full research output plus the editor's feedback. That's $0.36 per run just in prompt tokens.

Another hidden detail: agents can call tools, but tools are synchronous Python functions. The agent loop blocks until the tool returns. If your tool makes an API call that takes 2 seconds, the entire crew is stalled. There's no timeout on tool execution by default. We had a web search tool that occasionally hung for 30 seconds. The crew didn't fail — it just sat there. We added a functools.lru_cache and a 5-second timeout to fix it.

import os
from crewai import Agent, Task, Crew, Process
from crewai.tools import tool

# Simulate a slow tool
@tool("Search")
def search(query: str) -> str:
    """Search the web for information."""
    # In production, this would be an API call
    import time
    time.sleep(2)  # Simulate latency
    return f"Results for {query}: [dummy data]"

# Define agents with explicit limits
researcher = Agent(
    role="Senior Researcher",
    goal="Find relevant information",
    backstory="Expert at finding data",
    tools=[search],
    allow_delegation=False,  # Critical: prevent self-delegation
    max_iterations=3,  # Cap LLM calls per task
    verbose=True
)

writer = Agent(
    role="Content Writer",
    goal="Write a clear article",
    backstory="Experienced tech writer",
    allow_delegation=False,
    max_iterations=3
)

# Define tasks with context
research_task = Task(
    description="Research the topic of CrewAI internals",
    expected_output="A list of key findings",
    agent=researcher
)

write_task = Task(
    description="Write a 500-word article based on the research",
    expected_output="A complete article",
    agent=writer,
    context=[research_task]  # Pass research output
)

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    process=Process.sequential,
    verbose=True
)

result = crew.kickoff()
print(result)

Practical Implementation: Building a Production-Ready Research Crew

Let's build a research and writing crew that won't bankrupt you. We'll use gpt-4o-mini for cost-sensitive tasks, set explicit token budgets, and add error handling. The key difference from tutorials: we'll use Process.hierarchical with a manager agent that controls delegation, and we'll add a custom callback to log every step.

First, install the dependencies. Note: we pin crewai to 0.3.1 because later versions changed the callback API. Always pin your versions in production.

We'll define three agents: a Manager (who delegates), a Researcher (who gathers data), and a Writer (who produces output). The Manager uses gpt-4o for reasoning, the others use gpt-4o-mini. This saves ~60% on token costs compared to using gpt-4o for all agents.

We'll also add a max_tokens_per_task parameter. This isn't an official CrewAI parameter — we implement it via a callback that checks token usage after each task and raises an exception if it exceeds the budget.

import os
from crewai import Agent, Task, Crew, Process
from crewai.tools import tool
from dotenv import load_dotenv

load_dotenv()

@tool("WebSearch")
def web_search(query: str) -> str:
    """Search the web for information. Returns a summary."""
    import requests
    try:
        # Simulate a search API call
        response = requests.get(f"https://api.example.com/search?q={query}", timeout=5)
        response.raise_for_status()
        return response.json()["summary"]
    except Exception as e:
        return f"Search failed: {str(e)}"

# Manager agent uses expensive model for reasoning
manager = Agent(
    role="Project Manager",
    goal="Coordinate research and writing tasks efficiently",
    backstory="Experienced manager who delegates work",
    allow_delegation=True,  # Only manager delegates
    llm="gpt-4o",  # Expensive but smart
    max_iterations=5
)

# Researcher uses cheaper model
researcher = Agent(
    role="Research Analyst",
    goal="Find accurate and relevant information",
    backstory="Expert at data gathering",
    tools=[web_search],
    allow_delegation=False,  # No delegation
    llm="gpt-4o-mini",  # Cheaper
    max_iterations=3
)

writer = Agent(
    role="Content Writer",
    goal="Write clear and engaging content",
    backstory="Experienced tech writer",
    allow_delegation=False,
    llm="gpt-4o-mini",
    max_iterations=3
)

# Tasks with explicit expected outputs
research_task = Task(
    description="Research the latest trends in AI agents",
    expected_output="A bullet-point list of 5 key trends with sources",
    agent=researcher
)

write_task = Task(
    description="Write a 300-word blog post based on the research",
    expected_output="A complete blog post in markdown format",
    agent=writer
)

crew = Crew(
    agents=[manager, researcher, writer],
    tasks=[research_task, write_task],
    process=Process.hierarchical,  # Manager delegates
    manager_agent=manager,
    verbose=True
)

# Custom callback to monitor token usage
def log_step(step):
    print(f"Agent: {step.agent.role}, Tokens: {step.token_usage}")
    if step.token_usage > 4000:
        raise Exception(f"Token budget exceeded: {step.token_usage} tokens")

crew.step_callback = log_step

result = crew.kickoff()
print(f"Final output: {result}")

When NOT to Use CrewAI — And What to Use Instead

CrewAI is not a silver bullet. We learned this the hard way when we tried to use it for a real-time chatbot. The crew execution is synchronous — crew.kickoff() blocks until all tasks complete. For a chatbot that needs sub-second responses, CrewAI adds 2-5 seconds of latency just from the LLM calls. We switched to a single-agent pipeline with LangChain for that use case.

Another anti-pattern: using CrewAI for simple data transformations. If you have one task that takes input A and produces output B, you don't need a multi-agent system. Just call the LLM directly. CrewAI adds overhead: the framework itself takes ~200ms to initialize, plus the agent prompt construction. For simple tasks, a direct OpenAI API call is faster and cheaper.

When should you use CrewAI? When you have multiple distinct roles that need to collaborate on a complex output. Examples: research + writing + editing pipelines, code generation + review + testing, or data analysis + visualization + reporting. The key is that each role has a different expertise and the output of one feeds the next.

Alternatives: LangGraph for complex state machines (better for branching logic), AutoGen for conversational agents (better for interactive scenarios), or just a single LLM call with a well-structured prompt (better for simple tasks).

# Anti-pattern: Using CrewAI for a single task
# DON'T do this:
from crewai import Agent, Task, Crew

agent = Agent(role="Translator", goal="Translate text", backstory="Expert translator")
task = Task(description="Translate 'Hello' to Spanish", expected_output="Translation", agent=agent)
crew = Crew(agents=[agent], tasks=[task])
result = crew.kickoff()  # 500ms overhead for a single LLM call

# DO this instead:
import openai
response = openai.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Translate 'Hello' to Spanish"}]
)
print(response.choices[0].message.content)  # 200ms, no overhead

Production Patterns & Scale: Handling 10,000 Crew Runs Per Day

Scaling CrewAI to thousands of runs per day requires careful resource management. Each crew.kickoff() is a synchronous blocking call. If you run 10 crews concurrently, you need 10 threads or processes. We use a thread pool with 20 workers, but we had to add rate limiting to avoid hitting OpenAI's TPM limits.

Pattern 1: Batch processing with a queue. We use Redis as a task queue. Each job contains the crew configuration (agents, tasks, tools). A worker picks up the job, runs the crew, and stores the result back in Redis. This decouples the API from the execution.

Pattern 2: Caching tool outputs. If multiple crews run the same search query, cache the result. We use functools.lru_cache with a TTL of 1 hour. This reduced tool calls by 40% in our pipeline.

Pattern 3: Token budgeting per crew. We track token usage per run and reject jobs that exceed a budget. We use a callback that raises an exception if token usage exceeds a threshold. This prevents runaway costs.

Pattern 4: Async execution. CrewAI v0.3.1 supports kickoff_async() which returns a coroutine. You can use asyncio.gather() to run multiple crews concurrently. But beware: the async version still uses synchronous tool calls internally, so it's not truly non-blocking.

import asyncio
from crewai import Agent, Task, Crew, Process

async def run_crew_async(topic: str):
    researcher = Agent(
        role="Researcher",
        goal=f"Research {topic}",
        backstory="Expert researcher",
        allow_delegation=False,
        max_iterations=3
    )
    writer = Agent(
        role="Writer",
        goal=f"Write about {topic}",
        backstory="Expert writer",
        allow_delegation=False,
        max_iterations=3
    )
    task1 = Task(description=f"Research {topic}", expected_output="Findings", agent=researcher)
    task2 = Task(description=f"Write article about {topic}", expected_output="Article", agent=writer)
    crew = Crew(
        agents=[researcher, writer],
        tasks=[task1, task2],
        process=Process.sequential,
        verbose=False
    )
    result = await crew.kickoff_async()  # Async execution
    return result

async def main():
    topics = ["AI agents", "CrewAI", "LangChain", "AutoGen", "RAG"]
    tasks = [run_crew_async(topic) for topic in topics]
    results = await asyncio.gather(*tasks)
    for topic, result in zip(topics, results):
        print(f"{topic}: {result[:100]}...")

asyncio.run(main())

Common Mistakes With Specific Examples

We've seen the same mistakes across multiple teams. Here are the top five, with exact code examples.

Mistake 1: Not setting allow_delegation=False. This causes infinite loops. We covered this in the incident section.

Mistake 2: Using the same LLM model for all agents. This wastes money. Use gpt-4o for the manager, gpt-4o-mini for others.

Mistake 3: Not setting max_iterations. Default is 15, which is too high for simple tasks. Set it to 3-5.

Mistake 4: Passing large context. The context field appends full previous outputs. Truncate to 2k tokens.

Mistake 5: Ignoring tool errors. Tools can fail silently. Set fail_on_error=True or wrap tools in try/except.

# Mistake 1: No delegation control
agent = Agent(role="Researcher", goal="Research", backstory="...", allow_delegation=True)  # BAD: can delegate to itself

# Mistake 2: Same expensive model everywhere
agent1 = Agent(role="Researcher", llm="gpt-4o")  # BAD: waste of money
agent2 = Agent(role="Writer", llm="gpt-4o")      # BAD

# Mistake 3: No max_iterations
agent = Agent(role="Researcher", goal="Research", backstory="...")  # BAD: default 15 iterations

# Mistake 4: Large context
write_task = Task(description="Write article", expected_output="Article", agent=writer, context=[research_task])  # BAD: context can be huge

# Mistake 5: Ignoring tool errors
@tool("Search")
def search(query):
    # No error handling — if API fails, agent gets None
    return requests.get(f"https://api.com/search?q={query}").json()

CrewAI vs Alternatives: When to Choose What

We evaluated CrewAI, LangGraph, AutoGen, and a single LLM call for our content pipeline. Here's the breakdown.

CrewAI: Best for linear, multi-role workflows. Easy to set up. Limited branching. Synchronous by default. Good for batch processing.

LangGraph: Best for complex state machines with branching, loops, and conditional logic. More flexible but steeper learning curve. Supports async natively.

AutoGen: Best for conversational agents that need to chat with each other. Supports multi-turn conversations. Higher latency due to turn-taking.

Single LLM: Best for simple tasks. No overhead. No multi-agent complexity. Fastest and cheapest.

Our recommendation: Start with a single LLM call. If you need multiple roles, use CrewAI. If you need branching or loops, use LangGraph. If you need conversational agents, use AutoGen.

# Single LLM call (fastest, cheapest)
import openai
response = openai.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Write a 300-word article about AI agents"}]
)
print(response.choices[0].message.content)

# CrewAI (multi-role, linear)
from crewai import Agent, Task, Crew
researcher = Agent(role="Researcher", goal="Research", backstory="...", allow_delegation=False)
writer = Agent(role="Writer", goal="Write", backstory="...", allow_delegation=False)
crew = Crew(agents=[researcher, writer], tasks=[...], process=Process.sequential)
result = crew.kickoff()

# LangGraph (branching, state machine)
from langgraph.graph import StateGraph, END
# ... define nodes and edges

# AutoGen (conversational)
from autogen import AssistantAgent, UserProxyAgent
assistant = AssistantAgent(name="assistant", llm_config={"model": "gpt-4o"})
user = UserProxyAgent(name="user", human_input_mode="NEVER")
user.initiate_chat(assistant, message="Write an article about AI agents")

Debugging and Monitoring CrewAI in Production

You can't debug what you can't see. CrewAI provides callbacks for every step, but most tutorials don't mention them. We use three callbacks:

1. step_callback: Called after each agent completes a step. We log the agent role, token usage, and duration.
2. task_callback: Called after each task completes. We validate the output format and log it.
3. crew_callback: Called after the entire crew completes. We log total token usage and duration.

We also use LangSmith for tracing. CrewAI supports LangSmith integration via the LANGCHAIN_API_KEY environment variable. This gives you a visual trace of every LLM call, tool execution, and delegation.

For alerting, we set up CloudWatch alarms on token usage per crew. If a single crew uses more than 50k tokens, we get paged. This catches delegation loops and runaway costs.

import os
import time
from crewai import Agent, Task, Crew, Process

# Callback for monitoring
class CrewMonitor:
    def __init__(self):
        self.step_logs = []
        self.start_time = time.time()

    def on_step(self, step):
        log = {
            "agent": step.agent.role,
            "task": step.task.description[:50],
            "tokens": step.token_usage,
            "duration": time.time() - self.start_time
        }
        self.step_logs.append(log)
        print(f"[MONITOR] {log['agent']}: {log['tokens']} tokens, {log['duration']:.2f}s")
        if step.token_usage > 5000:
            print(f"[WARN] High token usage: {step.token_usage}")

    def on_task(self, task):
        print(f"[TASK] {task.description[:50]} completed")

    def on_crew(self, crew):
        total_tokens = sum(log['tokens'] for log in self.step_logs)
        print(f"[CREW] Total tokens: {total_tokens}, Duration: {time.time()-self.start_time:.2f}s")

monitor = CrewMonitor()

# Define agents and tasks (same as before)
researcher = Agent(role="Researcher", goal="Research", backstory="...", allow_delegation=False, max_iterations=3)
writer = Agent(role="Writer", goal="Write", backstory="...", allow_delegation=False, max_iterations=3)
task1 = Task(description="Research topic", expected_output="Findings", agent=researcher)
task2 = Task(description="Write article", expected_output="Article", agent=writer)

crew = Crew(
    agents=[researcher, writer],
    tasks=[task1, task2],
    process=Process.sequential,
    step_callback=monitor.on_step,
    task_callback=monitor.on_task,
    crew_callback=monitor.on_crew
)

result = crew.kickoff()