CrewAI sequential process tutorial with n4n.ai models

CrewAI’s sequential process executes tasks in a defined order, passing each agent’s output to the next. This tutorial shows how to wire that pattern with n4n.ai models — one OpenAI-compatible endpoint that addresses 240+ models — so you can swap providers without changing agent code. You’ll build a research-to-report pipeline: a researcher gathers facts, an analyst synthesizes them, and a writer produces a formatted brief.

Prerequisites

• Python 3.10+
• An n4n.ai API key (get one at n4n.ai)
• Basic familiarity with CrewAI concepts: agents, tasks, crews

Install the dependencies:

pip install crewai crewai-tools openai python-dotenv

Create a .env file in your project root:

N4N_API_KEY=your_key_here
N4N_BASE_URL=https://api.n4n.ai/v1

Configure the LLM client

CrewAI uses LiteLLM under the hood, so any OpenAI-compatible endpoint works. Point it at n4n.ai and pick a model — here we use anthropic/claude-3.5-sonnet but you can substitute any of the 240+ available.

# config/llm.py
import os
from crewai import LLM

def get_llm(model: str = "anthropic/claude-3.5-sonnet") -> LLM:
    return LLM(
        model=model,
        api_key=os.getenv("N4N_API_KEY"),
        base_url=os.getenv("N4N_BASE_URL"),
        temperature=0.3,
    )

Define the agents

Three agents, each with a narrow role. Keep allow_delegation=False — sequential process means the crew controls flow, not the agents.

# agents/researcher.py
from crewai import Agent
from config.llm import get_llm

def create_researcher() -> Agent:
    return Agent(
        role="Research analyst",
        goal="Find accurate, up-to-date facts on the given topic",
        backstory=(
            "You specialize in rapid literature scans. You cite sources, "
            "note publication dates, and flag conflicting data."
        ),
        llm=get_llm(),
        allow_delegation=False,
        verbose=True,
    )

# agents/analyst.py
from crewai import Agent
from config.llm import get_llm

def create_analyst() -> Agent:
    return Agent(
        role="Senior analyst",
        goal="Synthesize raw research into structured insights",
        backstory=(
            "You turn messy notes into clear themes, identify gaps, "
            "and rank findings by importance and credibility."
        ),
        llm=get_llm(),
        allow_delegation=False,
        verbose=True,
    )

# agents/writer.py
from crewai import Agent
from config.llm import get_llm

def create_writer() -> Agent:
    return Agent(
        role="Technical writer",
        goal="Produce a concise, well-formatted brief for stakeholders",
        backstory=(
            "You write executive summaries that busy leaders can read in two minutes. "
            "You use headers, bullets, and a one-paragraph tl;dr."
        ),
        llm=get_llm(),
        allow_delegation=False,
        verbose=True,
    )

Define the tasks

Tasks run in the order you add them to the crew. Each task’s output becomes the next task’s context automatically.

# tasks/research_task.py
from crewai import Task
from agents.researcher import create_researcher

def create_research_task(topic: str) -> Task:
    researcher = create_researcher()
    return Task(
        description=(
            f"Research '{topic}'. Find 5-7 credible sources published within the last 18 months. "
            "For each source, capture: title, publication, date, key claim, and a one-sentence summary. "
            "Output as a markdown table."
        ),
        expected_output="Markdown table with columns: Source | Date | Key Claim | Summary",
        agent=researcher,
    )

# tasks/analysis_task.py
from crewai import Task
from agents.analyst import create_analyst

def create_analysis_task() -> Task:
    analyst = create_analyst()
    return Task(
        description=(
            "Review the research table. Identify 3-4 major themes. For each theme, list supporting "
            "claims, note any contradictions, and assign a confidence level (high/medium/low). "
            "Output as structured markdown with theme headers."
        ),
        expected_output="Markdown with theme headers, supporting claims, contradictions, confidence levels",
        agent=analyst,
    )

# tasks/writing_task.py
from crewai import Task
from agents.writer import create_writer

def create_writing_task(topic: str) -> Task:
    writer = create_writer()
    return Task(
        description=(
            f"Write a 300-word executive brief on '{topic}' using the analysis. "
            "Structure: tl;dr (one paragraph), Key Themes (bulleted), Open Questions (bulleted), "
            "Sources (numbered list). Keep it scannable."
        ),
        expected_output="Formatted executive brief with tl;dr, Key Themes, Open Questions, Sources",
        agent=writer,
    )

Assemble and run the crew

The process="sequential" flag is the default, but we set it explicitly for clarity.

# main.py
import os
from dotenv import load_dotenv
from crewai import Crew, Process
from tasks.research_task import create_research_task
from tasks.analysis_task import create_analysis_task
from tasks.writing_task import create_writing_task

load_dotenv()

def run_crew(topic: str) -> str:
    research_task = create_research_task(topic)
    analysis_task = create_analysis_task()
    writing_task = create_writing_task(topic)

    crew = Crew(
        agents=[
            research_task.agent,
            analysis_task.agent,
            writing_task.agent,
        ],
        tasks=[research_task, analysis_task, writing_task],
        process=Process.sequential,
        verbose=True,
    )

    result = crew.kickoff()
    return result.raw

if __name__ == "__main__":
    topic = "Impact of retrieval-augmented generation on hallucination rates in production LLMs"
    output = run_crew(topic)
    print("\n=== FINAL OUTPUT ===\n")
    print(output)

Run it:

python main.py

Expected output at each checkpoint

Researcher output (markdown table):

Analyst output (structured markdown):

## Theme 1: Retrieval quality dominates outcomes
- Supporting: arXiv:2401.12345, ACL 2024
- Contradiction: None
- Confidence: High

## Theme 2: Parametric knowledge still matters
- Supporting: ICML 2024 workshop paper
- Contradiction: arXiv:2401.12345 claims retrieval alone suffices
- Confidence: Medium
...

Writer output (final brief):

**tl;dr**  
Retrieval-augmented generation reduces hallucination rates by 30-45% in production when retrieval precision exceeds 0.75, but degrades performance when retrievers surface noisy or contradictory passages. Teams should invest in retrieval evaluation before scaling RAG.

**Key Themes**  
- Retrieval quality is the primary lever — precision > recall for hallucination control  
- Hybrid parametric-retrieval approaches outperform pure RAG on domain-specific queries  
- Evaluation frameworks remain immature; most teams lack automated hallucination benchmarks  

**Open Questions**  
- Optimal chunk size and overlap for technical documentation corpora  
- Cost-latency tradeoffs of re-ranking vs. larger context windows  

**Sources**  
1. "RAG Reduces Hallucination by 40%" — arXiv:2401.12345 (Jan 2024)  
2. "When RAG Fails" — ACL 2024 (Mar 2024)  
...

Swapping models without code changes

Because the LLM factory reads from environment variables, you can switch models at deploy time:

# .env.production
N4N_API_KEY=prod_key
N4N_BASE_URL=https://api.n4n.ai/v1
# Use a cheaper model for research, stronger for writing
RESEARCH_MODEL=meta-llama/llama-3.1-8b-instruct
ANALYSIS_MODEL=anthropic/claude-3.5-sonnet
WRITING_MODEL=openai/gpt-4o

Then update config/llm.py to accept a model argument per agent. The crew logic stays untouched.

Common pitfalls

Task context not passing — Ensure each task’s expected_output matches what the next agent needs. If the analyst expects a table but the researcher returns prose, the analyst will hallucinate structure. Be explicit in description and expected_output.

Verbose logging noise — Set verbose=False on agents in production. Keep verbose=True on the crew for progress tracking.

Rate limits — n4n.ai handles automatic fallback when a provider is rate-limited or degraded, but you should still implement retry logic at the application layer for idempotent operations.

Next steps

• Add a SerperDevTool or FirecrawlTool to the researcher for live web search
• Wrap the crew in a FastAPI endpoint for async job submission
• Log each task’s token usage via n4n.ai’s per-token metering for cost attribution
• Experiment with process=Process.hierarchical when you need a manager agent to decompose open-ended goals

The sequential process shines when the workflow is linear and deterministic. Start here, measure, then graduate to hierarchical only when the task graph demands it.