Sequential CrewAI crews for linear content workflows

CrewAI’s sequential process content workflow model fits linear pipelines where each stage depends on the previous one’s output — research feeds drafting, drafting feeds editing, editing feeds publishing. This tutorial walks through building a complete article production pipeline: a researcher agent gathers sources, a writer agent drafts sections, an editor agent enforces style and factuality, and a publisher agent formats the final output. You’ll get runnable code, agent configurations, and verification checkpoints at each stage.

Step 1: Define the pipeline stages and data contracts

Before writing agents, sketch the data that flows between stages. A sequential crew passes the full context forward, so each agent needs a clear input schema and a deterministic output schema. Use Pydantic models to enforce this at runtime.

# models.py
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import datetime
from enum import Enum

class ResearchFinding(BaseModel):
    source_url: str
    claim: str
    evidence: str
    confidence: float = Field(ge=0.0, le=1.0)
    retrieved_at: datetime = Field(default_factory=datetime.utcnow)

class ResearchOutput(BaseModel):
    topic: str
    findings: List[ResearchFinding]
    gaps: List[str] = Field(default_factory=list)

class SectionDraft(BaseModel):
    heading: str
    content: str
    word_count: int
    citations: List[str] = Field(default_factory=list)

class DraftOutput(BaseModel):
    title: str
    sections: List[SectionDraft]
    total_words: int
    target_audience: str

class EditDecision(BaseModel):
    section_index: int
    issue: str
    suggestion: str
    severity: str  # "blocker" | "major" | "minor"

class EditOutput(BaseModel):
    revised_sections: List[SectionDraft]
    decisions: List[EditDecision]
    style_score: float = Field(ge=0.0, le=1.0)

class PublishedArticle(BaseModel):
    markdown: str
    html: str
    meta: dict
    published_at: datetime = Field(default_factory=datetime.utcnow)

These models become the crew’s shared vocabulary. The researcher returns ResearchOutput, the writer consumes it and returns DraftOutput, the editor consumes DraftOutput and returns EditOutput, and the publisher consumes EditOutput and returns PublishedArticle. No implicit coupling — if a downstream agent receives malformed data, validation fails fast.

Step 2: Configure the researcher agent with tool access

The researcher needs web search and content extraction. CrewAI’s built-in tools cover search; add a lightweight extraction tool for full-text retrieval. Keep the agent focused: one goal, one toolset, explicit output format.

# agents/researcher.py
from crewai import Agent
from crewai_tools import SerperDevTool, ScrapeWebsiteTool
from models import ResearchOutput
import os

serper = SerperDevTool(api_key=os.getenv("SERPER_API_KEY"))
scraper = ScrapeWebsiteTool()

researcher = Agent(
    role="Technical Researcher",
    goal=(
        "Gather authoritative, up-to-date sources for the assigned topic. "
        "Return structured findings with confidence scores."
    ),
    backstory=(
        "You specialize in technical due diligence. You verify claims against "
        "primary sources, flag contradictions, and identify gaps that need "
        "human follow-up. You never hallucinate citations."
    ),
    tools=[serper, scraper],
    verbose=True,
    allow_delegation=False,
    max_iter=5,
    output_json=ResearchOutput,
)

The output_json parameter forces the LLM to emit valid ResearchOutput. Set max_iter to bound tool-use loops — researchers can spiral if left unchecked. The backstory constrains tone: this agent speaks in evidence, not opinion.

Step 3: Configure the writer agent with structural constraints

The writer transforms research into a structured draft. Give it a template, not a blank page. Define the article skeleton in the goal so every draft shares the same section architecture.

# agents/writer.py
from crewai import Agent
from models import DraftOutput, ResearchOutput

writer = Agent(
    role="Technical Writer",
    goal=(
        "Produce a complete article draft from research findings. "
        "Follow the prescribed structure: Introduction, Background, "
        "Core Analysis (3-4 subsections), Practical Implications, "
        "Limitations, Conclusion. Each section 200-400 words. "
        "Cite findings inline using [source_url] notation."
    ),
    backstory=(
        "You write for senior engineers who scan before they read. "
        "Lead with the insight, support with evidence, close with actionability. "
        "No fluff, no hedging, no marketing speak."
    ),
    verbose=True,
    allow_delegation=False,
    max_iter=3,
    output_json=DraftOutput,
)

Note the explicit word-count bounds and citation format. The writer receives ResearchOutput as context automatically via the sequential process — CrewAI passes the previous task’s output as input to the next task. You don’t need to wire this manually.

Step 4: Configure the editor agent with a rubric

The editor is a critic, not a rewriter. It evaluates against a rubric and emits structured decisions. This separation — judge vs. fix — keeps the pipeline auditable. A human (or a rewriter agent) can review decisions before applying them.

# agents/editor.py
from crewai import Agent
from models import EditOutput, DraftOutput

EDIT_RUBRIC = """
Evaluate each section on:
1. Factual accuracy: Claims match cited research. No extrapolation.
2. Technical precision: Terminology is correct. No hand-waving.
3. Structure: Each section has a clear thesis sentence. Transitions exist.
4. Audience fit: Assumes senior engineering knowledge. Defines domain terms.
5. Conciseness: No redundant sentences. No filler phrases.
Score each dimension 0.0-1.0. Flag blockers (score < 0.6) and majors (score < 0.8).
"""

editor = Agent(
    role="Technical Editor",
    goal=(
        "Apply the edit rubric to the draft. Return structured decisions "
        "and a revised version of each section. Do not rewrite from scratch — "
        "surgically fix issues."
    ),
    backstory=(
        "You have shipped developer-facing documentation at scale. "
        "You catch the errors that embarrass teams in production. "
        "You output JSON, not commentary."
    ),
    verbose=True,
    allow_delegation=False,
    max_iter=3,
    output_json=EditOutput,
)

The rubric lives as a string constant so you can version it alongside code. When the rubric changes, the editor’s behavior changes — no prompt archaeology required.

Step 5: Configure the publisher agent for multi-format output

The publisher takes the edited sections and renders final artifacts. This is deterministic work — no LLM creativity needed, but CrewAI tasks still provide a clean interface for metadata injection and format switching.

# agents/publisher.py
from crewai import Agent
from models import PublishedArticle, EditOutput
import json

publisher = Agent(
    role="Content Publisher",
    goal=(
        "Assemble the final article in Markdown and HTML. Inject front-matter "
        "metadata: title, description, tags, canonical_url, reading_time_minutes. "
        "Generate a JSON-LD schema.org/Article block for SEO."
    ),
    backstory=(
        "You own the content delivery layer. Your output is consumed by static "
        "site generators, RSS feeds, and search indexers. Validity is non-negotiable."
    ),
    verbose=True,
    allow_delegation=False,
    max_iter=2,
    output_json=PublishedArticle,
)

Step 6: Define tasks with explicit context passing

Tasks bind agents to goals and define the sequential handoff. Each task’s context parameter lists upstream tasks whose outputs become input. CrewAI handles the plumbing.

# tasks.py
from crewai import Task
from agents.researcher import researcher
from agents.writer import writer
from agents.editor import editor
from agents.publisher import publisher
from models import ResearchOutput, DraftOutput, EditOutput, PublishedArticle

research_task = Task(
    description=(
        "Research the topic: '{topic}'.\n"
        "1. Search for authoritative sources (docs, RFCs, vendor blogs, peer-reviewed).\n"
        "2. Extract key claims, evidence, and confidence.\n"
        "3. Identify gaps where sources conflict or are missing.\n"
        "Return ResearchOutput."
    ),
    expected_output="Valid ResearchOutput JSON",
    agent=researcher,
    output_json=ResearchOutput,
)

write_task = Task(
    description=(
        "Write a complete article draft using the research findings.\n"
        "Structure: Introduction, Background, Core Analysis (3-4 subsections), "
        "Practical Implications, Limitations, Conclusion.\n"
        "Target audience: {target_audience}.\n"
        "Cite findings inline as [source_url]."
    ),
    expected_output="Valid DraftOutput JSON",
    agent=writer,
    context=[research_task],
    output_json=DraftOutput,
)

edit_task = Task(
    description=(
        "Edit the draft using the edit rubric.\n"
        "Return EditOutput with revised_sections and decisions."
    ),
    expected_output="Valid EditOutput JSON",
    agent=editor,
    context=[write_task],
    output_json=EditOutput,
)

publish_task = Task(
    description=(
        "Publish the edited article.\n"
        "Generate Markdown, HTML, and JSON-LD.\n"
        "Front-matter: title, description, tags, canonical_url, reading_time_minutes."
    ),
    expected_output="Valid PublishedArticle JSON",
    agent=publisher,
    context=[edit_task],
    output_json=PublishedArticle,
)

The {topic} and {target_audience} placeholders are filled at kickoff time. The context arrays create the sequential dependency chain: research → write → edit → publish.

Step 7: Assemble the crew and add observability

Instantiate the crew with process="sequential". Add a simple callback to log each stage’s output for debugging and audit trails.

# crew.py
from crewai import Crew, Process
from tasks import research_task, write_task, edit_task, publish_task
from models import ResearchOutput, DraftOutput, EditOutput, PublishedArticle
import json
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def stage_logger(stage_name: str):
    def callback(output):
        logger.info(f"=== {stage_name} complete ===")
        logger.info(f"Output type: {type(output).__name__}")
        if hasattr(output, 'model_dump'):
            logger.debug(json.dumps(output.model_dump(), indent=2, default=str))
    return callback

research_task.callback = stage_logger("research")
write_task.callback = stage_logger("write")
edit_task.callback = stage_logger("edit")
publish_task.callback = stage_logger("publish")

content_crew = Crew(
    agents=[researcher, writer, editor, publisher],
    tasks=[research_task, write_task, edit_task, publish_task],
    process=Process.sequential,
    verbose=True,
    memory=False,  # disable for deterministic runs; enable for multi-run context
)

Set memory=False for reproducible single-run pipelines. Enable it only when you want later runs to recall earlier topics — but be aware that memory introduces non-determinism.

Step 8: Run the pipeline with input validation

Wrap the kickoff in a CLI entry point that validates inputs and handles errors gracefully.

# main.py
import argparse
import sys
from crew import content_crew
from models import PublishedArticle

def main():
    parser = argparse.ArgumentParser(description="Run content pipeline")
    parser.add_argument("--topic", required=True, help="Article topic")
    parser.add_argument("--audience", default="senior software engineers", help="Target audience")
    parser.add_argument("--output", default="article.json", help="Output file path")
    args = parser.parse_args()

    try:
        result = content_crew.kickoff(inputs={
            "topic": args.topic,
            "target_audience": args.audience,
        })

        # CrewAI returns the final task's output
        if isinstance(result, PublishedArticle):
            with open(args.output, "w") as f:
                f.write(result.model_dump_json(indent=2))
            print(f"✓ Article published to {args.output}")
            print(f"  Title: {result.meta.get('title')}")
            print(f"  Reading time: {result.meta.get('reading_time_minutes')} min")
            sys.exit(0)
        else:
            print(f"✗ Unexpected result type: {type(result)}", file=sys.stderr)
            sys.exit(1)

    except Exception as e:
        print(f"✗ Pipeline failed: {e}", file=sys.stderr)
        if hasattr(e, '__cause__') and e.__cause__:
            print(f"  Cause: {e.__cause__}", file=sys.stderr)
        sys.exit(1)

if __name__ == "__main__":
    main()

Run it:

python main.py --topic "Model Context Protocol (MCP) architecture" --audience "backend engineers"

Step 9: Verify success at each stage

Don’t wait for the final output to catch failures. Add verification checkpoints that run automatically and can be executed independently during development.

# verify.py
from models import ResearchOutput, DraftOutput, EditOutput, PublishedArticle
import json
import sys

def verify_research(output: ResearchOutput) -> list[str]:
    errors = []
    if not output.findings:
        errors.append("No findings returned")
    for i, f in enumerate(output.findings):
        if not f.source_url.startswith("http"):
            errors.append(f"Finding {i}: invalid source_url")
        if f.confidence < 0.5:
            errors.append(f"Finding {i}: low confidence ({f.confidence})")
    if not output.gaps:
        errors.append("No gaps identified — researcher may be overconfident")
    return errors

def verify_draft(output: DraftOutput) -> list[str]:
    errors = []
    required_sections = {"introduction", "background", "conclusion"}
    found = {s.heading.lower() for s in output.sections}
    missing = required_sections - found
    if missing:
        errors.append(f"Missing required sections: {missing}")
    if output.total_words < 800:
        errors.append(f"Draft too short: {output.total_words} words")
    for i, s in enumerate(output.sections):
        if not s.citations:
            errors.append(f"Section {i} ('{s.heading}') has no citations")
    return errors

def verify_edit(output: EditOutput) -> list[str]:
    errors = []
    if output.style_score < 0.7:
        errors.append(f"Style score below threshold: {output.style_score}")
    blockers = [d for d in output.decisions if d.severity == "blocker"]
    if blockers:
        errors.append(f"Unresolved blockers: {len(blockers)}")
    if len(output.revised_sections) != len(output.decisions):
        errors.append("Revised sections count doesn't match decisions")
    return errors

def verify_published(output: PublishedArticle) -> list[str]:
    errors = []
    if not output.markdown.strip():
        errors.append("Empty markdown")
    if not output.html.strip():
        errors.append("Empty HTML")
    required_meta = ["title", "description", "tags", "canonical_url", "reading_time_minutes"]
    for key in required_meta:
        if key not in output.meta:
            errors.append(f"Missing front-matter: {key}")
    # Validate JSON-LD presence
    if "application/ld+json" not in output.html:
        errors.append("Missing JSON-LD schema.org/Article block")
    return errors

def run_verification(stage: str, filepath: str):
    with open(filepath) as f:
        data = json.load(f)

    verifiers = {
        "research": (ResearchOutput, verify_research),
        "draft": (DraftOutput, verify_draft),
        "edit": (EditOutput, verify_edit),
        "published": (PublishedArticle, verify_published),
    }

    model_cls, verifier = verifiers[stage]
    obj = model_cls(**data)
    errors = verifier(obj)

    if errors:
        print(f"✗ {stage} verification failed:")
        for e in errors:
            print(f"  - {e}")
        sys.exit(1)
    else:
        print(f"✓ {stage} verification passed")

if __name__ == "__main__":
    if len(sys.argv) != 3:
        print("Usage: python verify.py <stage> <json_file>")
        sys.exit(1)
    run_verification(sys.argv[1], sys.argv[2])

Integrate into your run script:

# Run pipeline and verify each stage
python main.py --topic "MCP architecture" --audience "backend engineers" --output research.json
python verify.py research research.json

python main.py --topic "MCP architecture" --audience "backend engineers" --output draft.json
python verify.py draft draft.json

# ... etc

Or better: modify the callbacks in crew.py to write intermediate outputs and run verification automatically.

Step 10: Handle provider failures with fallback routing

In production, the LLM provider backing an agent may hit rate limits or degrade. If you route through a gateway that supports automatic fallback — like n4n.ai — you configure the model once and the gateway handles provider switching transparently. The crew definition stays clean; the infrastructure handles resilience.

# llm_config.py
import os
from crewai import LLM

# Single endpoint, 240+ models, automatic fallback on rate limits or errors
gateway_llm = LLM(
    model="openai/gpt-4o-mini",  # or any model ID the gateway supports
    base_url=os.getenv("N4N_BASE_URL", "https://api.n4n.ai/v1"),
    api_key=os.getenv("N4N_API_KEY"),
    temperature=0.3,
    max_tokens=4000,
)

# Assign to all agents
researcher.llm = gateway_llm
writer.llm = gateway_llm
editor.llm = gateway_llm
publisher.llm = gateway_llm

The gateway honors client routing directives (e.g., x-n4n-prefer-provider: anthropic) and forwards provider cache-control hints, so you get deterministic caching behavior without managing multiple SDKs.

Step 11: Extend with human-in-the-loop gates

For high-stakes content, insert approval gates between stages. CrewAI doesn’t have built-in HITL, but you can pause the process by raising an exception that your orchestrator catches, then resume after human review.

# hitl.py
from crewai import Task
from models import EditOutput
import json

class HumanReviewRequired(Exception):
    def __init__(self, stage: str, artifact: dict):
        self.stage = stage
        self.artifact = artifact
        super().__init__(f"Human review required at {stage}")

def hitl_gate(stage_name: str, output_path: str):
    def callback(output):
        with open(output_path, "w") as f:
            json.dump(output.model_dump(), f, indent=2, default=str)
        # In a real system, notify reviewer via Slack, email, or PR comment
        # For demo: auto-approve if no blockers
        if isinstance(output, EditOutput):
            blockers = [d for d in output.decisions if d.severity == "blocker"]
            if blockers:
                raise HumanReviewRequired(stage_name, output.model_dump())
    return callback

# Attach to edit task
edit_task.callback = hitl_gate("edit", "edit_review.json")

Your orchestrator catches HumanReviewRequired, presents the artifact to a reviewer, and re-runs the crew from the edit task with the approved revisions injected as context.

Step 12: Package for CI/CD integration

Wrap the pipeline in a container so it runs identically locally and in CI. Pin dependencies, include a health check, and expose the CLI as the entrypoint.

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install system deps for any native extensions
RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc libpq-dev && \
    rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# Non-root user
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER appuser

ENTRYPOINT ["python", "main.py"]
CMD ["--help"]

# .github/workflows/content-pipeline.yml
name: Content Pipeline
on:
  workflow_dispatch:
    inputs:
      topic:
        required: true
      audience:
        default: "senior software engineers"

jobs:
  generate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Build image
        run: docker build -t content-pipeline .
      - name: Run pipeline
        env:
          SERPER_API_KEY: ${{ secrets.SERPER_API_KEY }}
          N4N_API_KEY: ${{ secrets.N4N_API_KEY }}
        run: |
          docker run --rm \
            -e SERPER_API_KEY \
            -e N4N_API_KEY \
            content-pipeline \
            --topic "${{ github.event.inputs.topic }}" \
            --audience "${{ github.event.inputs.audience }}" \
            --output article.json
      - name: Verify output
        run: |
          docker run --rm -v $(pwd):/data content-pipeline \
            python verify.py published /data/article.json
      - name: Upload artifact
        uses: actions/upload-artifact@v4
        with:
          name: article
          path: article.json

Common failure modes and fixes

Scaling considerations

This sequential architecture works well for linear pipelines up to ~5 stages. Beyond that, context window pressure grows — each agent sees all prior outputs. For longer pipelines, consider:

1. Summarization tasks between stages to compress context
2. Hierarchical crews where a manager agent delegates to sub-crews (covered in the companion article on hierarchical crews)
3. External memory (vector store) for research findings instead of passing full ResearchOutput

The crewai sequential process content workflow pattern shines when the workflow is genuinely linear and each stage adds irreducible value. Don’t force sequential on parallelizable work — use hierarchical or custom orchestration instead.

You now have a production-grade content pipeline: typed contracts, verified stages, fallback routing, HITL gates, and CI/CD integration. Ship it.