Capstone: Multi-Agent Content Pipeline

Outcome: By the end of this lesson you will have shipped a running CrewAI pipeline of 4 specialized agents — Researcher → Writer → Fact-Checker → Publisher — that takes a topic as input, gathers sources, drafts a post, verifies every factual claim, and saves the finished markdown (or posts it to your CMS via a webhook). This is the multi-agent orchestration you've been building toward for five modules.

What you'll ship — this command runs the full crew:

python pipeline.py --topic "The state of open-source agent frameworks in Q2 2026"

The run writes to stdout and to ./output/{slug}.md. The stage sequence looks like this — Researcher gathers, Writer drafts, Fact-Checker verifies claim by claim, Publisher saves and POSTs:

Stage	What it emits	What the next stage receives
Researcher	A brief: findings, cited source URLs, claims flagged as risky	The brief, via `task.context`
Writer	A draft where every number carries a `[^N]` marker	The brief and the draft
Fact-Checker	The same draft, corrected, plus a change log of what moved	The corrected draft
Publisher	Frontmatter + file on disk + webhook status	—

A note on the numbers you are about to write

An earlier version of this lesson printed a sample run log here. It showed the Fact-Checker catching two bad claims, with confident corrections attached:

"CrewAI has 80K GitHub stars" — actual is ~32K. Corrected. "LangChain v1.0 shipped April 2026" — actual is June 2026. Corrected.

Both corrections were wrong. CrewAI's repository was nowhere near 32K, and LangChain 1.0 reached general availability on 22 October 2025¹ — not April 2026, and not June 2026. The log had never been produced by running anything; it was written to look like output.

That is worth more to you than the log was. A fabricated example inside a lesson about fact-checking is the exact failure mode this crew exists to catch, and it survived because output-shaped text does not read as a claim. When you build the Fact-Checker in Part 3, the thing you are defending against is not an agent that lies — it is an agent that produces something shaped like verified work. Your change log is only worth reading if a human can follow each entry back to a URL.

So: no sample output in this lesson is real unless it says it was captured. Run the pipeline and read your own.

The architecture you're building

Multi-Agent Content Pipeline — CrewAI Sequential Process

Input

Researcher (mid-tier model)

Writer (mid-tier model)

Fact-Checker (top-tier model)

Publisher (small model — cheap)

Part 1 — Project setup (5 min)

content-crew/
├── pipeline.py              # Orchestrates the 4 agents
├── agents/
│   ├── researcher.py        # Defines the Researcher agent + its tools
│   ├── writer.py
│   ├── fact_checker.py
│   └── publisher.py
├── tools/
│   ├── web_search.py        # Tavily search wrapper
│   └── cms_webhook.py       # POST to your CMS
├── requirements.txt
├── .env.example
└── output/                  # Generated posts land here

requirements.txt — pin whatever the current majors are on the day you build this, then leave them pinned. The floors below are an example, not a recommendation; check crewai on PyPI and each package's release page for what is current, and read CrewAI's own installation guide for the supported Python range:

crewai>=1.14.0
crewai-tools>=1.14.0
anthropic>=0.99.0
tavily-python>=0.7.24
httpx>=0.27.2
python-dotenv>=1.0.1

.env.example:

ANTHROPIC_API_KEY=sk-ant-...
TAVILY_API_KEY=tvly-...            # tavily.com — see their pricing page for current free-tier limits
CMS_WEBHOOK_URL=https://your-cms.com/api/drafts    # optional; leave empty to skip publish
CMS_WEBHOOK_TOKEN=                  # bearer token for your CMS

Install:

pip install -r requirements.txt
cp .env.example .env    # fill in keys

Part 2 — Custom tools (10 min)

CrewAI agents use the same @tool decorator pattern from Module 3. Two tools power this crew: web search (Researcher) and CMS webhook (Publisher).

tools/web_search.py:

import os
from crewai.tools import tool
from tavily import TavilyClient

tavily = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])


@tool("web_search")
def web_search(query: str, max_results: int = 6) -> str:
    """Search the live web. Returns a numbered list of results with title, URL, and snippet."""
    resp = tavily.search(
        query=query,
        max_results=max_results,
        search_depth="advanced",
        include_answer=True,
        days=90,          # last 3 months only — we want current
    )
    lines = [f"Summary: {resp.get('answer', '')}\n"]
    for i, r in enumerate(resp.get("results", []), 1):
        lines.append(f"[{i}] {r['title']}\n    {r['url']}\n    {r['content'][:300]}")
    return "\n".join(lines)

tools/cms_webhook.py:

import os
import httpx
from crewai.tools import tool


@tool("publish_to_cms")
def publish_to_cms(slug: str, title: str, markdown: str) -> str:
    """
    Publish the finished post to the CMS via webhook.
    Returns the webhook response or a skip message if no URL is configured.
    """
    url = os.environ.get("CMS_WEBHOOK_URL", "").strip()
    if not url:
        return "CMS_WEBHOOK_URL not set — skipping publish (saved to disk only)."

    headers = {"Content-Type": "application/json"}
    token = os.environ.get("CMS_WEBHOOK_TOKEN", "").strip()
    if token:
        headers["Authorization"] = f"Bearer {token}"

    r = httpx.post(url, json={"slug": slug, "title": title, "markdown": markdown}, headers=headers, timeout=30)
    return f"POST {url} → {r.status_code} {r.reason_phrase}"

Part 3 — The four agents (15 min)

Each agent has one job — that's the whole point of orchestration. An agent with three responsibilities quickly turns into an agent that does none of them well.

agents/researcher.py:

from crewai import Agent
from tools.web_search import web_search


researcher = Agent(
    role="Research Analyst",
    goal="Gather 6-10 authoritative, recent sources on the assigned topic",
    backstory=(
        "You are a veteran research analyst. You trust primary sources (vendor docs, "
        "GitHub repos, official announcements) over secondary summaries. You are "
        "suspicious of round numbers and recent dates — both are red flags for "
        "fabrication you flag for the fact-checker to verify."
    ),
    tools=[web_search],
    llm="anthropic/claude-sonnet-5",
    verbose=True,
    allow_delegation=False,
)

agents/writer.py:

from crewai import Agent


writer = Agent(
    role="Technical Writer",
    goal="Draft a 1500-2000 word post that is accurate, scannable, and cites every numerical claim",
    backstory=(
        "You write for developers. You never pad, never use buzzwords, never say "
        "'revolutionary'. You structure posts as: TL;DR (3 bullets) → one H2 per main "
        "idea → a concluding 'what to watch'. For every statistic, version number, "
        "date, or comparison, you include an inline [^N] citation that the "
        "fact-checker will verify. You draft the footnote block at the bottom "
        "referencing the Researcher's sources."
    ),
    llm="anthropic/claude-sonnet-5",
    verbose=True,
    allow_delegation=False,
)

agents/fact_checker.py:

from crewai import Agent
from tools.web_search import web_search


fact_checker = Agent(
    role="Fact Checker",
    goal=(
        "Verify every numerical claim, version number, date, and 'first/only/largest' "
        "superlative in the draft against live sources. Correct or flag anything wrong."
    ),
    backstory=(
        "You are paranoid by trade. You assume every stat is wrong until you've found "
        "an authoritative source confirming it. You treat negative claims ('X doesn't "
        "have Y', 'X has no API') with extra suspicion — these are the single most "
        "common source of errors, because absence of evidence isn't evidence of absence. "
        "You correct inline by editing the draft, preserving the rest of the writer's voice."
    ),
    tools=[web_search],
    llm="anthropic/claude-opus-5",    # larger model for nuanced verification
    verbose=True,
    allow_delegation=False,
)

agents/publisher.py:

from crewai import Agent
from tools.cms_webhook import publish_to_cms


publisher = Agent(
    role="Publisher",
    goal="Save the final post to disk and push to the CMS webhook",
    backstory=(
        "You are the last pair of eyes. You add frontmatter YAML (title, slug, date, tags), "
        "save the post to ./output/{slug}.md, and POST it to the CMS webhook. "
        "You do not edit content — that's the writer's and fact-checker's job."
    ),
    tools=[publish_to_cms],
    llm="anthropic/claude-haiku-4-5",   # cheap model for mechanical work
    verbose=True,
    allow_delegation=False,
)

The model assignment is the design decision, not the model names. Read it as three tiers:

Agent	Tier	Why this tier
Researcher	Mid	Broad reasoning plus reliable tool use; the search results do the heavy lifting
Writer	Mid	Drafting quality and style control matter; correctness is the next agent's job
Fact-Checker	Top	The one place where being wrong is unrecoverable — pay for it here
Publisher	Small	Prepend frontmatter, call one tool. A large model buys nothing

Model names move every few months; the tiers do not. Swap the exact identifiers for whatever is current on the Anthropic models page (or your provider's equivalent) and keep the assignment.

The saving from tiering is real but you should measure it rather than take a number from a lesson: it is a function of how many tokens each agent actually consumes, which depends on your topic and how many times the Fact-Checker re-searches. Run the crew once with every agent on the mid tier, once as configured, and compare the two bills. That comparison is worth more than any percentage printed here, because it is your workload.

Part 4 — The orchestrating crew (10 min)

pipeline.py:

import argparse
import re
from pathlib import Path
from dotenv import load_dotenv
from crewai import Crew, Task, Process

from agents.researcher import researcher
from agents.writer import writer
from agents.fact_checker import fact_checker
from agents.publisher import publisher

load_dotenv()
Path("output").mkdir(exist_ok=True)


def slugify(title: str) -> str:
    s = re.sub(r"[^\w\s-]", "", title.lower())
    return re.sub(r"[\s_-]+", "-", s).strip("-")[:80]


def build_crew(topic: str) -> Crew:
    research_task = Task(
        description=(
            f"Research the topic: '{topic}'.\n\n"
            f"Deliverable: a structured brief containing:\n"
            f"1. 3-5 key findings (one sentence each)\n"
            f"2. 6-10 source citations in format [N] Title — URL\n"
            f"3. Any claim you flag as 'potentially unreliable' (round numbers, recent dates, superlatives)\n\n"
            f"Prioritize sources from the last 90 days. Avoid aggregator blogs — go to primary sources."
        ),
        expected_output="A research brief with findings, cited sources, and flagged claims.",
        agent=researcher,
    )

    write_task = Task(
        description=(
            "Using the research brief from the Researcher, draft a 1500-2000 word blog post.\n\n"
            "Structure:\n"
            "- TL;DR: 3 bullets\n"
            "- 3-5 H2 sections, each ~300 words\n"
            "- Concluding 'What to watch' section\n\n"
            "For every numerical claim or date, include a [^N] inline citation and list "
            "the matching footnote at the bottom using the Researcher's source URLs. "
            "Do NOT invent claims not supported by the research brief."
        ),
        expected_output="A complete markdown blog post with inline citations and a footnote block.",
        agent=writer,
        context=[research_task],
    )

    factcheck_task = Task(
        description=(
            "Review the draft from the Writer. For EVERY [^N] citation:\n"
            "1. Re-search the cited fact using web_search.\n"
            "2. Compare what the draft says against the sources.\n"
            "3. If wrong, edit the draft inline to fix it. Preserve the rest of the writer's voice.\n"
            "4. Flag any claim you couldn't verify — rewrite it with hedged language like "
            "   'as of [date]' or 'reportedly' rather than stating it as fact.\n\n"
            "Also scan for unchecked negative claims ('X has no Y') — these are the #1 "
            "source of errors. Search each one independently before accepting it.\n\n"
            "Output: the corrected markdown, plus a change log listing which claims "
            "you fixed and which you hedged."
        ),
        expected_output="Corrected markdown draft + change log of edits.",
        agent=fact_checker,
        context=[research_task, write_task],
    )

    publish_task = Task(
        description=(
            "Take the fact-checker's corrected markdown and:\n"
            "1. Prepend frontmatter YAML with: title (infer from H1), slug, date (today), "
            "   tags (3-5 inferred from content), featured: true.\n"
            "2. Use the `publish_to_cms` tool with (slug, title, full_markdown).\n"
            "3. Report the final status (CMS response or skip message).\n\n"
            "Do NOT modify the body. Only add frontmatter."
        ),
        expected_output="Final markdown with frontmatter + webhook status.",
        agent=publisher,
        context=[factcheck_task],
    )

    return Crew(
        agents=[researcher, writer, fact_checker, publisher],
        tasks=[research_task, write_task, factcheck_task, publish_task],
        process=Process.sequential,
        verbose=True,
    )


def save_output(topic: str, final_markdown: str):
    slug = slugify(topic)
    path = Path("output") / f"{slug}.md"
    path.write_text(final_markdown, encoding="utf-8")
    print(f"\n✓ Saved → {path}")
    return path


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--topic", required=True, help="Blog post topic (1-2 sentences)")
    args = parser.parse_args()

    crew = build_crew(args.topic)
    result = crew.kickoff()

    # CrewAI's result.raw holds the final Publisher output, which is the framed markdown
    save_output(args.topic, str(result.raw))

Part 5 — Run it (2 min)

python pipeline.py --topic "The state of open-source agent frameworks in Q2 2026"

With verbose=True on both the agents and the crew, CrewAI prints each agent's thinking, every tool call, and each task's final answer as it goes. Read it live rather than waiting for the file.

Four things to check in your own output — in this order:

Did the Researcher actually call web_search? If the brief appeared with no tool call in the log, the agent answered from the model's own memory. That is the failure that poisons everything downstream, and it is silent.
Does every [^N] in the Writer's draft have a footnote with a URL? A citation marker with no URL behind it is decoration.
Does the Fact-Checker's change log name specific claims? "Verified all claims" means it verified nothing. You want entries of the form claim → source → verdict.
Did the Publisher's webhook return a status you recognise? A skipped publish prints a skip message; a silent success is not the same thing.

Runtime is dominated by web-search latency and by how many claims the Fact-Checker decides to re-verify, so it varies widely per topic. Cost depends on which models you assigned and on current per-token pricing — check the Anthropic pricing page and your provider dashboard after the first run rather than trusting an estimate printed here. Run it once on a cheap topic and read the actual usage; that number is the one worth knowing.

Part 6 — What changes when you swap the topic

The pipeline is topic-agnostic. To adapt for a different editorial beat:

Change	Where
Different research focus	Agent `backstory` + `description` on research_task
Different post length	Writer task description (1500-2000 → 800-1200)
Different CMS (WordPress, Ghost, Notion, Obsidian)	Swap `publish_to_cms` implementation in `tools/cms_webhook.py`
Add an SEO agent before Publisher	One new `Agent(...)` + one new `Task(...)` with `context=[factcheck_task]`
Make it a scheduled job	Wrap `pipeline.py` in a GitHub Actions cron (`0 9 * * 1-5`)

Adding a 5th agent is ~15 lines of code — that's the power of CrewAI's structure.

Part 7 — Troubleshooting matrix

Symptom	First check	Typical cause
Researcher returns 0 results	Tavily dashboard	Free-tier exhausted, or `days=90` too restrictive
Writer output is very short (<500 words)	Writer backstory	Agent "goal" phrased as "summary" rather than "1500-2000 word post"
Fact-checker doesn't actually correct the draft	CrewAI version + model_config	Older CrewAI versions don't pass prior task outputs as editable context — upgrade to the current 1.x — check crewai on PyPI
Publisher fails webhook	`CMS_WEBHOOK_URL`	Endpoint wrong, auth header missing, or endpoint doesn't accept the shape
All agents use the same model even though you set different `llm` on each	env override	`CREWAI_DEFAULT_MODEL` env var trumps per-agent config — unset it
Crew hangs forever	Task descriptions	Ambiguous "expected_output" keeps agents iterating — tighten it

Build checkpoint — finish this before claiming the certificate

Ship the crew. python pipeline.py --topic "anything" runs through all 4 agents without errors.
Ship the write path. The final markdown in ./output/ has correct frontmatter and real content.
Ship a catch. Run on a topic where the Writer is likely to hallucinate stats (e.g. "Recent GitHub star counts for agent frameworks"). Confirm the Fact-Checker's change log shows at least 1 correction.
Ship the webhook. Point CMS_WEBHOOK_URL at a real endpoint (or httpbin.org/post for testing). Confirm the Publisher POSTs and the response logs.
Screenshot the crew running + the final markdown + the webhook response — as your proof of work.

You've just shipped a multi-agent pipeline that does real editorial work on a topic-agnostic basis. Every orchestration pattern from the last five modules — agent specialization, task sequencing, tool routing, model-cost matching, fact-checking guardrails — is now running in production.

Next (optional): add a Slack integration so the crew drops a daily digest into your team channel, or hook it to an RSS feed so new tech announcements get turned into drafts automatically. :::

LangChain 1.0 now generally available — LangChain changelog. ↩

Multi-Agent Content Pipeline — CrewAI Sequential Process

Quiz