Designing CrewAI agent roles for a content pipeline

Building a content pipeline with CrewAI means treating each agent as a specialist with a single, well-defined responsibility. The crewai agent roles content pipeline design process starts by mapping your editorial workflow to discrete roles, then encoding those roles as agents with focused tools, clear backstories, and tasks that produce verifiable outputs. This guide walks through that mapping end to end, with code you can run and checkpoints to confirm each piece works before you wire the next one.

Step 1: Map your editorial workflow to agent roles

Before writing any code, list every distinct phase your content passes through. A typical pipeline includes: topic research, outline generation, draft writing, fact-checking, SEO optimization, copy editing, and publishing preparation. Each phase becomes an agent role. Resist the urge to combine phases — agents that do one thing well are easier to debug, test, and replace.

Write the mapping as a simple table. This becomes your source of truth for role definitions.

Save this as pipeline_roles.csv or a JSON file — you’ll reference it when defining agents.

Step 2: Define each agent with a focused toolkit

CrewAI agents should only have the tools they need. A Researcher needs web search and maybe a document loader. A Writer needs no external tools — just the LLM. An Optimizer might need a keyword density calculator. Define tools as small, pure functions first, then wrap them for CrewAI.

# tools/research_tools.py
import requests
from bs4 import BeautifulSoup
from typing import List, Dict
import json

def search_web(query: str, num_results: int = 5) -> List[Dict]:
    """Return list of {title, url, snippet} using a search API."""
    # Replace with your preferred search provider (SerpAPI, Brave, etc.)
    # This is a stub — implement with real credentials
    return [
        {"title": f"Result for {query}", "url": "https://example.com", "snippet": "Sample snippet"}
        for _ in range(num_results)
    ]

def fetch_page(url: str) -> str:
    """Fetch and extract main text content from a URL."""
    resp = requests.get(url, timeout=10)
    resp.raise_for_status()
    soup = BeautifulSoup(resp.text, "html.parser")
    # Remove script/style/nav/footer
    for tag in soup(["script", "style", "nav", "footer", "header"]):
        tag.decompose()
    return soup.get_text(separator="\n", strip=True)[:8000]

def extract_claims(text: str) -> List[str]:
    """Heuristic: split into sentences, keep those with numbers, proper nouns, or quotes."""
    import re
    sentences = re.split(r'(?<=[.!?])\s+', text)
    claim_indicators = r'\d+|[A-Z][a-z]+ [A-Z][a-z]+|"[^"]+"|\$[\d,]+|%'
    return [s for s in sentences if re.search(claim_indicators, s)]

# tools/seo_tools.py
from collections import Counter
import re

def keyword_density(text: str, keywords: List[str]) -> Dict[str, float]:
    """Return density percentage for each keyword."""
    words = re.findall(r'\b\w+\b', text.lower())
    total = len(words)
    if total == 0:
        return {k: 0.0 for k in keywords}
    counts = Counter(words)
    return {k: (counts.get(k.lower(), 0) / total) * 100 for k in keywords}

def check_heading_keywords(markdown: str, keywords: List[str]) -> Dict[str, bool]:
    """Check if keywords appear in H1 or H2 headings."""
    headings = re.findall(r'^#{1,2}\s+(.+)$', markdown, re.MULTILINE)
    heading_text = " ".join(headings).lower()
    return {k: k.lower() in heading_text for k in keywords}

Now define agents. Keep backstories short and functional — they’re prompts, not bios.

# agents.py
from crewai import Agent
from tools.research_tools import search_web, fetch_page, extract_claims
from tools.seo_tools import keyword_density, check_heading_keywords

def make_researcher(llm) -> Agent:
    return Agent(
        role="Content Researcher",
        goal="Produce a structured research brief with cited sources for a given topic",
        backstory=(
            "You find credible, recent sources and extract verifiable claims. "
            "You never hallucinate citations. You output JSON only."
        ),
        tools=[search_web, fetch_page, extract_claims],
        llm=llm,
        verbose=True,
        allow_delegation=False,
    )

def make_strategist(llm) -> Agent:
    return Agent(
        role="Content Strategist",
        goal="Create a comprehensive hierarchical outline that covers all user intent angles",
        backstory=(
            "You structure content for clarity and completeness. "
            "You output a JSON outline with sections, subsections, and target word counts."
        ),
        tools=[],  # No external tools needed
        llm=llm,
        verbose=True,
        allow_delegation=False,
    )

def make_writer(llm) -> Agent:
    return Agent(
        role="Technical Writer",
        goal="Write a complete markdown draft that follows the outline exactly",
        backstory=(
            "You write clear, accurate technical content. "
            "You never add sections not in the outline. You output markdown only."
        ),
        tools=[],
        llm=llm,
        verbose=True,
        allow_delegation=False,
    )

def make_verifier(llm) -> Agent:
    return Agent(
        role="Fact Checker",
        goal="Verify every factual claim in the draft against sources",
        backstory=(
            "You are skeptical. You flag unsupported claims and suggest corrections. "
            "You output a JSON report with claim, verdict, and evidence."
        ),
        tools=[search_web, fetch_page],
        llm=llm,
        verbose=True,
        allow_delegation=False,
    )

def make_optimizer(llm) -> Agent:
    return Agent(
        role="SEO Optimizer",
        goal="Optimize the draft for target keywords while preserving readability",
        backstory=(
            "You integrate keywords naturally. You never keyword-stuff. "
            "You output optimized markdown and a keyword density report."
        ),
        tools=[keyword_density, check_heading_keywords],
        llm=llm,
        verbose=True,
        allow_delegation=False,
    )

def make_editor(llm) -> Agent:
    return Agent(
        role="Copy Editor",
        goal="Produce publication-ready markdown that passes the style guide",
        backstory=(
            "You enforce consistent style, fix grammar, and ensure flow. "
            "You output final markdown only."
        ),
        tools=[],
        llm=llm,
        verbose=True,
        allow_delegation=False,
    )

def make_publisher(llm) -> Agent:
    return Agent(
        role="Publisher",
        goal="Prepare CMS-ready payload with frontmatter and metadata",
        backstory=(
            "You generate valid frontmatter, compute reading time, and set canonical URLs. "
            "You output JSON with fields: title, slug, frontmatter, body, meta."
        ),
        tools=[],
        llm=llm,
        verbose=True,
        allow_delegation=False,
    )

Verify Step 2: Run a quick import test.

python -c "from agents import make_researcher; print('Agents module loads')"

If it prints without error, your agent definitions are syntactically sound.

Step 3: Design tasks with explicit output contracts

Each task must declare its expected output format. This is how you enforce the pipeline — the next agent’s input is the previous agent’s validated output. Use Pydantic models for validation.

# schemas.py
from pydantic import BaseModel, Field, HttpUrl
from typing import List, Optional
from enum import Enum

class Source(BaseModel):
    title: str
    url: HttpUrl
    snippet: str
    credibility: float = Field(ge=0, le=1, description="Your confidence in this source")

class ResearchBrief(BaseModel):
    topic: str
    key_questions: List[str]
    sources: List[Source]
    extracted_claims: List[str]
    gaps: List[str] = Field(description="Questions still unanswered")

class OutlineSection(BaseModel):
    heading: str
    level: int = Field(ge=1, le=3)
    target_words: int
    key_points: List[str]
    subsections: List["OutlineSection"] = []

OutlineSection.model_rebuild()

class ContentOutline(BaseModel):
    title: str
    target_audience: str
    target_word_count: int
    sections: List[OutlineSection]

class FactCheckReport(BaseModel):
    claim: str
    verdict: str = Field(pattern="^(verified|disputed|unverifiable)$")
    evidence: List[str]
    correction: Optional[str] = None

class FactCheckResult(BaseModel):
    checks: List[FactCheckReport]
    overall_confidence: float = Field(ge=0, le=1)

class SEOReport(BaseModel):
    keyword_density: dict
    heading_keywords: dict
    recommendations: List[str]

class OptimizedDraft(BaseModel):
    markdown: str
    seo_report: SEOReport

class PublishPayload(BaseModel):
    title: str
    slug: str
    frontmatter: dict
    body: str
    meta: dict
    reading_time_minutes: int

Now define tasks that reference these schemas. Each task’s output_json or output_pydantic enforces the contract.

# tasks.py
from crewai import Task
from schemas import (
    ResearchBrief, ContentOutline, FactCheckResult,
    OptimizedDraft, PublishPayload
)

def research_task(agent, topic: str, target_audience: str) -> Task:
    return Task(
        description=(
            f"Research the topic: '{topic}' for audience: '{target_audience}'.\n"
            "1. Search for 5-7 recent, credible sources.\n"
            "2. Fetch and extract key claims from each source.\n"
            "3. Identify 3-5 key questions the content must answer.\n"
            "4. Note any gaps where sources are thin or contradictory.\n"
            "Output a ResearchBrief JSON object."
        ),
        expected_output="A ResearchBrief JSON with topic, key_questions, sources, extracted_claims, and gaps.",
        agent=agent,
        output_pydantic=ResearchBrief,
    )

def outline_task(agent, research_brief: ResearchBrief) -> Task:
    return Task(
        description=(
            f"Create a comprehensive outline based on this research brief:\n"
            f"{research_brief.model_dump_json(indent=2)}\n\n"
            "Produce a ContentOutline with hierarchical sections, target word counts per section, "
            "and key points for each. Target total word count: 2000. "
            "Cover all key questions from the brief. Output JSON only."
        ),
        expected_output="A ContentOutline JSON with title, target_audience, target_word_count, and sections.",
        agent=agent,
        output_pydantic=ContentOutline,
        context=[research_task],  # CrewAI passes previous task output
    )

def write_task(agent, outline: ContentOutline) -> Task:
    return Task(
        description=(
            f"Write a complete markdown draft following this outline exactly:\n"
            f"{outline.model_dump_json(indent=2)}\n\n"
            "Rules:\n"
            "- Use the exact section headings from the outline.\n"
            "- Target the word count per section (±10%).\n"
            "- Write in clear, technical prose for the target audience.\n"
            "- Do not add sections not in the outline.\n"
            "- Output markdown only, no JSON wrapper."
        ),
        expected_output="A complete markdown draft as a string.",
        agent=agent,
        # No output_pydantic — raw string output
    )

def fact_check_task(agent, draft: str, research_brief: ResearchBrief) -> Task:
    return Task(
        description=(
            f"Fact-check this draft against the research brief:\n\n"
            f"DRAFT:\n{draft}\n\n"
            f"RESEARCH BRIEF:\n{research_brief.model_dump_json(indent=2)}\n\n"
            "For each factual claim in the draft, verify it against the sources. "
            "Output a FactCheckResult JSON with verdicts and corrections."
        ),
        expected_output="A FactCheckResult JSON with checks and overall_confidence.",
        agent=agent,
        output_pydantic=FactCheckResult,
    )

def optimize_task(agent, draft: str, keywords: List[str]) -> Task:
    return Task(
        description=(
            f"Optimize this draft for SEO keywords: {keywords}\n\n"
            f"DRAFT:\n{draft}\n\n"
            "Rules:\n"
            "- Integrate keywords naturally in headings and body.\n"
            "- Target 1-2% density per keyword.\n"
            "- Do not change meaning or add fluff.\n"
            "- Output OptimizedDraft JSON with markdown and seo_report."
        ),
        expected_output="An OptimizedDraft JSON with optimized markdown and SEO report.",
        agent=agent,
        output_pydantic=OptimizedDraft,
    )

def edit_task(agent, optimized_draft: OptimizedDraft) -> Task:
    return Task(
        description=(
            f"Copy-edit this optimized draft for publication:\n\n"
            f"{optimized_draft.markdown}\n\n"
            "Apply the style guide:\n"
            "- Active voice, present tense.\n"
            "- Short sentences (avg <20 words).\n"
            "- Consistent terminology.\n"
            "- Fix grammar, punctuation, flow.\n"
            "Output final markdown only."
        ),
        expected_output="Publication-ready markdown string.",
        agent=agent,
    )

def publish_task(agent, final_markdown: str, title: str, slug: str) -> Task:
    return Task(
        description=(
            f"Prepare CMS payload for: '{title}' (slug: {slug})\n\n"
            f"CONTENT:\n{final_markdown}\n\n"
            "Generate:\n"
            "- Frontmatter: title, description, tags, date, author, canonical_url\n"
            "- Reading time (words / 200)\n"
            "- Meta: og:title, og:description, twitter:card\n"
            "Output PublishPayload JSON."
        ),
        expected_output="A PublishPayload JSON with all required fields.",
        agent=agent,
        output_pydantic=PublishPayload,
    )

Verify Step 3: Test each task in isolation with a mock LLM or by running the crew with one task. Create a test script:

# test_tasks.py
from crewai import Crew, Process
from langchain_openai import ChatOpenAI
from agents import make_researcher, make_strategist
from tasks import research_task, outline_task
from schemas import ResearchBrief

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)

researcher = make_researcher(llm)
strategist = make_strategist(llm)

topic = "Vector databases for RAG"
audience = "Backend engineers"

r_task = research_task(researcher, topic, audience)
o_task = outline_task(strategist, ResearchBrief(
    topic=topic, key_questions=[], sources=[], extracted_claims=[], gaps=[]
))

crew = Crew(agents=[researcher, strategist], tasks=[r_task, o_task], process=Process.sequential, verbose=True)
result = crew.kickoff()
print(result)

Run it. You should see a valid ResearchBrief and ContentOutline printed. If Pydantic validation fails, adjust the task description or the model.

Step 4: Wire the full pipeline with conditional flow

Real pipelines need branching. If fact-check confidence is low, loop back to the writer with corrections. If SEO density is off, loop to the optimizer. CrewAI supports this via context and custom logic in a wrapper.

# pipeline.py
from crewai import Crew, Process
from langchain_openai import ChatOpenAI
from agents import (
    make_researcher, make_strategist, make_writer,
    make_verifier, make_optimizer, make_editor, make_publisher
)
from tasks import (
    research_task, outline_task, write_task,
    fact_check_task, optimize_task, edit_task, publish_task
)
from schemas import ResearchBrief, ContentOutline, FactCheckResult, OptimizedDraft, PublishPayload
import json

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)

# Instantiate agents once
researcher = make_researcher(llm)
strategist = make_strategist(llm)
writer = make_writer(llm)
verifier = make_verifier(llm)
optimizer = make_optimizer(llm)
editor = make_editor(llm)
publisher = make_publisher(llm)

KEYWORDS = ["vector database", "RAG", "embedding", "similarity search"]

def run_pipeline(topic: str, audience: str, slug: str, max_fact_check_loops: int = 2) -> PublishPayload:
    # Phase 1: Research
    r_task = research_task(researcher, topic, audience)
    research_crew = Crew(agents=[researcher], tasks=[r_task], process=Process.sequential, verbose=True)
    research_result = research_crew.kickoff()
    brief: ResearchBrief = r_task.output.pydantic
    print(f"Research complete: {len(brief.sources)} sources, {len(brief.extracted_claims)} claims")

    # Phase 2: Outline
    o_task = outline_task(strategist, brief)
    outline_crew = Crew(agents=[strategist], tasks=[o_task], process=Process.sequential, verbose=True)
    outline_crew.kickoff()
    outline: ContentOutline = o_task.output.pydantic
    print(f"Outline complete: {len(outline.sections)} top-level sections")

    # Phase 3: Write + Fact-check loop
    draft = None
    fact_check_result: FactCheckResult = None
    for loop in range(max_fact_check_loops + 1):
        w_task = write_task(writer, outline)
        write_crew = Crew(agents=[writer], tasks=[w_task], process=Process.sequential, verbose=True)
        write_crew.kickoff()
        draft = w_task.output.raw
        print(f"Draft complete (loop {loop}): ~{len(draft.split())} words")

        fc_task = fact_check_task(verifier, draft, brief)
        fc_crew = Crew(agents=[verifier], tasks=[fc_task], process=Process.sequential, verbose=True)
        fc_crew.kickoff()
        fact_check_result = fc_task.output.pydantic
        print(f"Fact-check confidence: {fact_check_result.overall_confidence:.2f}")

        if fact_check_result.overall_confidence >= 0.9:
            break
        # Feed corrections back into outline for next loop
        corrections = [c.correction for c in fact_check_result.checks if c.correction]
        if corrections:
            outline.sections[0].key_points.extend(corrections)  # Simple injection
            print(f"Injected {len(corrections)} corrections for next loop")

    if fact_check_result.overall_confidence < 0.9:
        print("Warning: Fact-check confidence below threshold after max loops")

    # Phase 4: SEO optimize
    opt_task = optimize_task(optimizer, draft, KEYWORDS)
    opt_crew = Crew(agents=[optimizer], tasks=[opt_task], process=Process.sequential, verbose=True)
    opt_crew.kickoff()
    optimized: OptimizedDraft = opt_task.output.pydantic
    print(f"SEO density: {optimized.seo_report.keyword_density}")

    # Phase 5: Edit
    edit_task_obj = edit_task(editor, optimized)
    edit_crew = Crew(agents=[editor], tasks=[edit_task_obj], process=Process.sequential, verbose=True)
    edit_crew.kickoff()
    final_markdown = edit_task_obj.output.raw
    print(f"Edited draft: ~{len(final_markdown.split())} words")

    # Phase 6: Publish
    pub_task = publish_task(publisher, final_markdown, outline.title, slug)
    pub_crew = Crew(agents=[publisher], tasks=[pub_task], process=Process.sequential, verbose=True)
    pub_crew.kickoff()
    payload: PublishPayload = pub_task.output.pydantic
    print(f"Publish payload ready: {payload.slug}")

    return payload

if __name__ == "__main__":
    payload = run_pipeline(
        topic="Vector databases for RAG",
        audience="Backend engineers",
        slug="vector-databases-rag"
    )
    with open("output/payload.json", "w") as f:
        json.dump(payload.model_dump(), f, indent=2, default=str)
    print("Saved to output/payload.json")

Verify Step 4: Run the full pipeline.

mkdir -p output
python pipeline.py

Check output/payload.json — it should contain valid frontmatter, body, meta, and reading time. Open the markdown in a viewer to confirm formatting.

Step 5: Add observability and cost control

Production pipelines need logging, token tracking, and fallback logic. Wrap each crew execution with a lightweight observer.

# observability.py
import time
import functools
from typing import Callable, Any
from dataclasses import dataclass, asdict
import json

@dataclass
class StepMetrics:
    step: str
    duration_seconds: float
    tokens_prompt: int
    tokens_completion: int
    cost_usd: float
    success: bool
    error: str = ""

class PipelineObserver:
    def __init__(self, log_file: str = "pipeline_metrics.jsonl"):
        self.log_file = log_file
        self.step_metrics = []

    def track(self, step_name: str):
        def decorator(func: Callable) -> Callable:
            @functools.wraps(func)
            def wrapper(*args, **kwargs) -> Any:
                start = time.time()
                try:
                    result = func(*args, **kwargs)
                    duration = time.time() - start
                    # Token extraction depends on your LLM wrapper
                    # For OpenAI via langchain, check result.llm_output or callback handlers
                    tokens_prompt = 0
                    tokens_completion = 0
                    cost = 0.0
                    self.step_metrics.append(StepMetrics(
                        step=step_name,
                        duration_seconds=duration,
                        tokens_prompt=tokens_prompt,
                        tokens_completion=tokens_completion,
                        cost_usd=cost,
                        success=True
                    ))
                    return result
                except Exception as e:
                    duration = time.time() - start
                    self.step_metrics.append(StepMetrics(
                        step=step_name,
                        duration_seconds=duration,
                        tokens_prompt=0,
                        tokens_completion=0,
                        cost_usd=0.0,
                        success=False,
                        error=str(e)
                    ))
                    raise
            return wrapper
        return decorator

    def flush(self):
        with open(self.log_file, "a") as f:
            for m in self.step_metrics:
                f.write(json.dumps(asdict(m)) + "\n")
        self.step_metrics.clear()

Integrate it:

# In pipeline.py, add at top:
from observability import PipelineObserver

observer = PipelineObserver()

# Wrap each crew.kickoff():
@observer.track("research")
def run_research():
    return research_crew.kickoff()

# ... repeat for each phase ...

# At end of run_pipeline():
observer.flush()

Verify Step 5: Check pipeline_metrics.jsonl after a run. Each line should have duration, success flag, and (once you wire token callbacks) token counts.

Step 6: Handle provider fallback for reliability

If you route LLM calls through a gateway that supports automatic fallback, you avoid pipeline stalls when a single provider degrades. Configure your ChatOpenAI (or equivalent) to point at the gateway endpoint. The gateway handles retry, fallback, and per-token metering without code changes in your agents.

# llm_client.py
from langchain_openai import ChatOpenAI
import os

def get_llm(model: str = "gpt-4o-mini", temperature: float = 0.2) -> ChatOpenAI:
    """
    Returns an LLM client pointed at the inference gateway.
    The gateway handles:
    - Automatic fallback across 240+ models when a provider is rate-limited or degraded
    - Per-token usage metering for cost attribution
    - Client routing directives (e.g., prefer low-latency models for fact-checking)
    - Forwarding provider cache-control hints for repeated prompts
    """
    return ChatOpenAI(
        model=model,
        temperature=temperature,
        api_key=os.getenv("GATEWAY_API_KEY"),
        base_url=os.getenv("GATEWAY_BASE_URL", "https://api.n4n.ai/v1"),
        default_headers={"X-Routing-Preference": "balanced"},
    )

Replace the llm = ChatOpenAI(...) line in pipeline.py with llm = get_llm().

Verify Step 6: Simulate a provider outage by setting an invalid model name in the gateway routing rules (if your gateway supports test modes) or by temporarily blocking the primary provider. The pipeline should complete using a fallback model, and metrics should show the model switch.

Step 7: Package for CI/CD and regression testing

Treat the pipeline as code. Add a test suite that runs on every PR with a small, fixed topic. Assert on output structure, not content — content varies by model.

# tests/test_pipeline.py
import pytest
from pipeline import run_pipeline
from schemas import PublishPayload

@pytest.mark.integration
def test_pipeline_smoke():
    payload = run_pipeline(
        topic="Hello world test",
        audience="Developers",
        slug="hello-world-test"
    )
    assert isinstance(payload, PublishPayload)
    assert payload.title
    assert payload.slug == "hello-world-test"
    assert payload.body.startswith("#")
    assert payload.reading_time_minutes > 0
    assert "og:title" in payload.meta
    assert payload.frontmatter.get("title") == payload.title

@pytest.mark.integration
def test_fact_check_loop_triggers():
    # Use a topic likely to produce low confidence
    payload = run_pipeline(
        topic="Speculative future of quantum computing in 2050",
        audience="Physicists",
        slug="quantum-2050",
        max_fact_check_loops=1
    )
    # Just verify it completes without exception
    assert payload.body

Run in CI:

# .github/workflows/pipeline-test.yml
name: Pipeline Regression
on: [pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.11" }
      - run: pip install -r requirements.txt
      - run: pytest tests/test_pipeline.py -v
        env:
          GATEWAY_API_KEY: ${{ secrets.GATEWAY_API_KEY }}
          GATEWAY_BASE_URL: ${{ secrets.GATEWAY_BASE_URL }}

Verify Step 7: Open a PR. The workflow should pass, producing a valid payload for the test topic.

You now have a complete, verifiable crewai agent roles content pipeline design: roles mapped to editorial phases, agents with minimal toolkits, tasks with Pydantic-enforced contracts, a looping fact-check stage, observability, provider fallback, and regression tests. Each step can be run independently, and the full pipeline produces a CMS-ready payload you can ship.