Advanced Patterns

You can build effective crews with just agents, tasks, and a process. But CrewAI offers several advanced features that unlock more sophisticated workflows: memory, custom tools, callbacks, and guardrails. This lesson covers the patterns you'll reach for as your crews grow in complexity.

Memory

By default, each crew run starts fresh — agents have no memory of previous executions. Enabling memory lets agents learn and improve across runs.

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

CrewAI supports three types of memory:

Type	Scope	Purpose
Short-term	Single crew run	Agents remember earlier tasks in the current run
Long-term	Across crew runs	Agents learn from past successes and failures
Entity	Across crew runs	Agents remember facts about specific entities (people, companies, etc.)

When memory=True, all three are enabled. Short-term memory is the most impactful — it helps agents in sequential workflows maintain context from earlier tasks even without explicit context connections.

When Memory Helps

• Iterative workflows — a crew that runs daily can learn from previous runs
• Long crews — agents in later tasks can reference earlier context without explicit chaining
• Entity-heavy work — remembering details about companies, people, or products across runs

When to Skip Memory

• One-off tasks — no benefit if the crew only runs once
• Sensitive data — memory persists to disk, so avoid it with confidential information
• Reproducibility — if you need identical outputs from identical inputs, memory can introduce variance

Custom Tools

While CrewAI provides built-in tools, you'll often need custom ones. A tool is any Python function decorated with @tool:

from crewai.tools import tool

@tool("Calculate Profit Margin")
def calculate_profit_margin(revenue: float, cost: float) -> str:
    """Calculate the profit margin given revenue and cost.

    Args:
        revenue: Total revenue in dollars
        cost: Total cost in dollars
    """
    margin = ((revenue - cost) / revenue) * 100
    return f"Profit margin: {margin:.1f}%"

The docstring is critical — the LLM reads it to understand when and how to use the tool.

Assigning Custom Tools to Agents

analyst = Agent(
    role="Financial Analyst",
    goal="Analyze business financial data accurately",
    backstory="Expert financial analyst specializing in SaaS metrics.",
    tools=[calculate_profit_margin]
)

Building More Complex Tools

Tools can call APIs, query databases, or perform any Python operation:

import requests
from crewai.tools import tool

@tool("Get Stock Price")
def get_stock_price(symbol: str) -> str:
    """Get the current stock price for a given ticker symbol.

    Args:
        symbol: Stock ticker symbol (e.g., AAPL, GOOGL)
    """
    # Example using a hypothetical API
    response = requests.get(
        f"https://api.example.com/stocks/{symbol}"
    )
    data = response.json()
    return f"{symbol}: ${data['price']:.2f} ({data['change']:+.2f}%)"

Tool Design Best Practices

• Clear docstrings — the LLM decides when to use a tool based on its docstring
• Typed parameters — use type hints so the LLM passes the right data types
• Return strings — tools should return human-readable strings, not raw data structures
• Handle errors gracefully — return error messages instead of raising exceptions

Callbacks

Callbacks let you hook into the crew's execution at key points — useful for logging, monitoring, and integrating with external systems.

Task Callbacks

Run a function every time a task completes:

def on_task_complete(output):
    print(f"Task completed! Output length: {len(output.raw)}")
    # Log to a database, send a notification, etc.

writing_task = Task(
    description="Write a blog post about AI trends",
    expected_output="A blog post in Markdown",
    agent=writer,
    callback=on_task_complete
)

Step Callbacks

Monitor every reasoning step an agent takes:

def on_step(step):
    print(f"Agent step: {step}")

researcher = Agent(
    role="Researcher",
    goal="Find information",
    backstory="Research specialist.",
    step_callback=on_step
)

Step callbacks are useful for debugging — you can see exactly how an agent reasons through a task.

Guardrails

Guardrails validate task outputs before accepting them. If the output doesn't pass validation, the agent is asked to try again:

from crewai import Task

def validate_word_count(output):
    """Ensure the blog post is between 500 and 1000 words."""
    word_count = len(output.raw.split())
    if word_count < 500:
        return (False, "Output too short. Must be at least 500 words.")
    if word_count > 1000:
        return (False, "Output too long. Must be under 1000 words.")
    return (True, output.raw)

writing_task = Task(
    description="Write a blog post about CrewAI",
    expected_output="A 500-1000 word blog post",
    agent=writer,
    guardrail=validate_word_count
)

If the guardrail returns (False, reason), the agent receives the feedback and tries again automatically.

Structured Output with Pydantic

For tasks that need structured data (not just text), you can enforce an output schema using Pydantic:

from pydantic import BaseModel
from typing import List

class CompetitorAnalysis(BaseModel):
    company_name: str
    strengths: List[str]
    weaknesses: List[str]
    market_share: float
    threat_level: str  # "low", "medium", "high"

analysis_task = Task(
    description="Analyze our top competitor's market position",
    expected_output="Structured competitor analysis",
    agent=analyst,
    output_pydantic=CompetitorAnalysis
)

The agent will format its output to match the Pydantic model, and the result is available as a validated Python object:

result = crew.kickoff()
analysis = result.pydantic  # CompetitorAnalysis object
print(analysis.company_name)
print(analysis.threat_level)

Rate Limiting

When using paid LLM APIs, you'll want to control costs:

crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    max_rpm=10  # Maximum 10 API requests per minute
)

You can also set limits at the agent level:

researcher = Agent(
    role="Researcher",
    goal="Find information",
    backstory="Research specialist.",
    max_iter=5,  # Maximum 5 reasoning iterations per task
    max_retry_limit=2  # Maximum 2 retries on failure
)

Combining Patterns

These patterns are most powerful when combined. Here's a crew that uses memory, custom tools, structured output, and callbacks together:

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[
        Task(
            description="Research {company}",
            expected_output="Research summary",
            agent=researcher
        ),
        Task(
            description="Analyze {company} financials",
            expected_output="Financial analysis",
            agent=analyst,
            tools=[calculate_profit_margin],
            output_pydantic=CompetitorAnalysis,
            callback=on_task_complete
        ),
        Task(
            description="Write report on {company}",
            expected_output="Executive report",
            agent=writer,
            guardrail=validate_word_count
        )
    ],
    memory=True,
    verbose=True,
    max_rpm=20
)

Key Takeaways

• Memory enables agents to learn across runs — use short-term for context within a run, long-term for learning across runs
• Custom tools extend agents with any Python function — use clear docstrings and type hints
• Callbacks let you monitor and log task and step completion
• Guardrails validate outputs and automatically retry if they don't pass
• Pydantic output enforces structured data schemas on task results
• Rate limiting controls API costs at the crew and agent level
• Combine these patterns to build production-ready multi-agent systems