Chapter 23 · Capstone 1: Content Marketing Pipeline

Chapter goals:综合运用前 22 章的知识，从零构建一个"选题 → 调研 → 撰写 → SEO 优化 → 审核发布"的四角色 sequential crew，代码可直接运行，并给出结果分析与升级方向。

23.1 Requirements and Pipeline Design

Business requirement: given a topic keyword, produce a WeChat official account / blog post that has passed fact-checking and SEO optimization.

Decompose the pipeline into four roles; the output of each task flows into the next via context:

[Topic Strategist] → [Researcher] → [Writer] → [SEO Editor]
   output: angle + outline    output: facts + citations   output: first draft     output: final publishable article

Why sequential instead of hierarchical: four stages with clear responsibilities and fixed order — no need for dynamic delegation. Sequential flow is the most predictable and easiest to debug (see Chapters 5 and 9 for selection criteria).

23.2 Project Structure

After generating the scaffold with crewai create crew marketing_pipeline, the final structure is:

marketing_pipeline/
├── .env                      # DEEPSEEK_API_KEY / SERPER_API_KEY
├── pyproject.toml
└── src/marketing_pipeline/
    ├── main.py               # Entry point
    ├── crew.py               # @CrewBase assembly
    └── config/
        ├── agents.yaml       # Four role definitions
        └── tasks.yaml        # Four task definitions

Install dependencies and configure credentials:

pip install 'crewai[tools]' python-dotenv
# .env file content (do not commit to git):
# DEEPSEEK_API_KEY=sk-xxx
# SERPER_API_KEY=xxx        # https://serper.dev free tier is enough for practice

23.3 agents.yaml: Prompt Design for Four Roles

# src/marketing_pipeline/config/agents.yaml
topic_strategist:
  role: "Senior Content Strategist"
  goal: "Find the most viral angle for topic '{topic}' and produce an article outline"
  backstory: >
    You have 10 years of new-media topic selection experience, skilled at
    turning bland tech topics into angles readers want to click. You follow
    one rule: each topic covers a single core idea, outline has no more than
    4 sections.
  llm: "openai/deepseek-chat"
  max_iter: 3
  verbose: true

researcher:
  role: "Fact-Checking Researcher"
  goal: "Collect accurate facts, data, and sources for the given outline"
  backstory: >
    You are a former investigative journalist who is extremely precise about numbers.
    You only accept facts with verifiable sources; claims without reliable sources
    are explicitly marked "TBD".
  llm: "openai/deepseek-chat"
  max_iter: 5
  verbose: true

writer:
  role: "B2B Technical Writer"
  goal: "Write a fluent, opinionated Chinese article draft based on the outline and research materials"
  backstory: >
    You have written hundreds of 100k+ readership tech-popularization articles.
    Your writing uses short sentences, each paragraph no more than 5 lines,
   善用 analogies, never pile up adjectives.
  llm: "openai/deepseek-chat"
  max_iter: 4

seo_editor:
  role: "SEO Optimization Editor"
  goal: "Complete SEO optimization and final review without hurting readability"
  backstory: >
    You are proficient in Chinese search engine optimization: clickbait headlines
    damage brand, so you only do compliant optimization. You also serve as the
    fact-check gatekeeper — any content conflicting with research notes is
    corrected directly.
  llm: "openai/deepseek-chat"
  max_iter: 3

Design highlights (from Chapter 3): backstory gives behavioral guidelines rather than title stacking ("only accept sourced numbers" beats "I am rigorous" far more); every role has max_iter capped to prevent runaway (Chapter 25).

23.4 tasks.yaml: Task Chain and Context Passing

# src/marketing_pipeline/config/tasks.yaml
plan_topic:
  description: >
    Plan an article for topic "{topic}": provide 1 core argument,
    target reader persona, and an outline of no more than 4 sections.
    Each section should have a one-sentence explanation of what it argues.
  expected_output: >
    Markdown topic card: core argument (1 sentence), target reader (1 sentence),
    outline (3-4 sections, each 1-2 sentences).
  agent: topic_strategist

research_facts:
  description: >
    Collect factual material for each section of the topic card outline:
    at least 2 usable items per section, numbers must include source and year.
    Use search tools to find the latest information, and use web scraping
    to verify key data.
  expected_output: >
    Research notes organized by outline sections: each item one line,
    format 「fact | source | year」, items without sources marked "TBD".
  agent: researcher
  context:
    - plan_topic
  tools:
    - SerperDevTool
    - ScrapeWebsiteTool

write_draft:
  description: >
    Strictly based on research notes, write an 800-1200 character Chinese
    article draft. Only use facts appearing in the research notes; do NOT
    fabricate data yourself; content marked "TBD" in research notes must
    either be deleted or rewritten as qualitative statements.
  expected_output: >
    Complete Markdown article: with H1 title, 3-4 H2 subsections,
    and a 3-point summary at the end.
  agent: writer
  context:
    - plan_topic
    - research_facts

seo_finalize:
  description: >
    Final review and SEO optimization of the draft: 1) verify facts in
    the article match research notes; 2) generate a title with the core
    keyword (no more than 25 characters); 3) write an abstract under
    120 characters; 4) provide 5 keyword tags. Do not change the article's
    core argument.
  expected_output: >
    Final publishable package: optimized title, abstract, full text,
    keyword tag list — all in Markdown output.
  agent: seo_editor
  context:
    - research_facts
    - write_draft
  output_file: output/final_article.md

Key mechanisms (from Chapter 4): research_facts receives the topic card via context; write_draft receives both the outline and research notes; seo_finalize receives the first draft and research notes for fact-checking — this mirrors the real editorial workflow where an editor cross-references source material against the draft.

23.5 crew.py: Assembly and Third-Party Model Integration

# src/marketing_pipeline/crew.py
import os
from dotenv import load_dotenv

from crewai import Agent, Crew, Process, Task, LLM
from crewai.project import CrewBase, agent, crew, task
from crewai_tools import SerperDevTool, ScrapeWebsiteTool

load_dotenv()

# Third-party OpenAI-compatible model: the llm field in yaml references provider/model,
# base_url and api_key are read by CrewAI from environment variables.
# (When DEEPSEEK_API_KEY auto-matches openai/ prefix + DeepSeek official endpoint,
#  explicit base_url is needed — see the injection example below.)


@CrewBase
class MarketingCrew:
    """Content marketing pipeline: Topic → Research → Writing → SEO Final Review"""

    agents_config = "config/agents.yaml"
    tasks_config = "config/tasks.yaml"

    search_tool = SerperDevTool()        # requires SERPER_API_KEY
    scrape_tool = ScrapeWebsiteTool()

    @agent
    def topic_strategist(self) -> Agent:
        return Agent(config=self.agents_config["topic_strategist"])

    @agent
    def researcher(self) -> Agent:
        return Agent(
            config=self.agents_config["researcher"],
            tools=[self.search_tool, self.scrape_tool],
        )

    @agent
    def writer(self) -> Agent:
        return Agent(config=self.agents_config["writer"])

    @agent
    def seo_editor(self) -> Agent:
        return Agent(config=self.agents_config["seo_editor"])

    @task
    def plan_topic(self) -> Task:
        return Task(config=self.tasks_config["plan_topic"])

    @task
    def research_facts(self) -> Task:
        return Task(config=self.tasks_config["research_facts"])

    @task
    def write_draft(self) -> Task:
        return Task(config=self.tasks_config["write_draft"])

    @task
    def seo_finalize(self) -> Task:
        return Task(config=self.tasks_config["seo_finalize"])

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,
            tasks=self.tasks,
            process=Process.sequential,   # Fixed sequential pipeline
            verbose=True,
        )

To fully control the model endpoint, remove the llm field from yaml and inject LLM objects explicitly in code — both approaches are equivalent; just stay consistent across the team:

from crewai import LLM

# Explicit third-party model injection (overrides yaml's llm field)
shared_llm = LLM(
    model="openai/deepseek-chat",
    base_url="https://api.deepseek.com/v1",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    temperature=0.7,
)

@agent
def writer(self) -> Agent:
    return Agent(
        config=self.agents_config["writer"],
        llm=shared_llm,   # yaml llm field for the same role is overridden by this
    )

23.6 Running and Result Analysis

# src/marketing_pipeline/main.py
from marketing_pipeline.crew import MarketingCrew

if __name__ == "__main__":
    result = MarketingCrew().crew().kickoff(
        inputs={"topic": "AI Agents in Enterprise Customer Service"}
    )
    print("=" * 60)
    print(result.raw)
    print("=" * 60)
    print(f"Total tokens: {result.token_usage.total_tokens}")

crewai run   # or python -m marketing_pipeline.main

Typical run observation points:

1. Researcher's tool calls: verbose log should show 2–5 Search the internet calls; if over 8, max_iter is too loose or the task description is too vague;
2. Fact consistency: spot-check numbers in output/final_article.md against research notes — when the context chain is complete, contradiction rates are significantly lower than a single Agent writing directly;
3. Cost distribution: token_usage typically shows the researcher accounting for 40%+ (tool results inflate context) — this is normal; optimize by making research notes more concise.

You can also inspect intermediate outputs and usage per stage to pinpoint problematic tasks:

result = MarketingCrew().crew().kickoff(inputs={"topic": "AI Agents in Enterprise Customer Service"})

# tasks_output corresponds 1:1 with tasks defined in tasks.yaml
for i, task_out in enumerate(result.tasks_output):
    preview = (task_out.raw or "")[:80].replace("\n", " ")
    print(f"Task {i}: {len(task_out.raw or '')} chars | preview: {preview}...")

If a task's output clearly deviates from expectations (e.g., research notes lack sources), revisit that task's expected_output — most quality issues are resolved by tightening acceptance criteria, not by swapping models.

23.7 Improvement Directions

• Upgrade to hierarchical: add a manager_llm so a "editor-in-chief" can dynamically decide whether more research is needed (Chapter 9); suitable for scenarios with variable topic quality;
• Flow-ify: decompose topic selection and writing into Flow steps, inserting deterministic code steps (e.g., word-count check, profanity filter) between them; individual steps can retry on failure (Chapters 15–16);
• API-ize (per Chapter 22): wrap with POST /kickoff + polling, and you have a team-ready content production service;
• Evaluation loop: use Chapter 20's regression set to assert "article has title/abstract/tags three-piece set, word count met, no 'TBD' remnants."

Chapter Summary

• Four-role sequential pipeline covers all core knowledge: Agent design, context chains, tool mounting, YAML engineering;
• backstory writes behavioral guidelines, max_iter sets gates per role, the editor task hangs both the first draft and source material for fact-checking;
• Tools are only mounted on the role that genuinely needs networking (researcher); other roles with zero tools are more stable;
• The biggest cost is usually the research stage with tools — optimizing material conciseness is more effective than swapping models.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. In this project, seo_finalize's context includes both research_facts and write_draft. What is the primary purpose?

2. Why does this project choose sequential instead of hierarchical flow?

3. After running, token consumption far exceeds expectations and the log shows the researcher calling the search tool 12 times. What is the most reasonable action?

4. Regarding the yaml llm: "openai/deepseek-chat" syntax, which statement is correct?

🛠️ Hands-On Exercises

1. Add a fifth role "Title AB Tester" to the pipeline: produce 3 candidate titles in different styles with explanations of their applicable scenarios, mounted after seo_finalize.
2. Cache the researcher's SerperDevTool results: write a custom tool wrapper for search that returns cached results within 24 hours for identical keywords (hint: refer to Chapter 8 cache_function).
3. Following the thinking in Section 23.7, turn "word-count check" into a deterministic step: write a small Flow that counts words with Python code after write_draft, and if the count is below threshold, feed the draft back to the writer with the requirement "compress to under 1000 characters" (loop max 2 times).

Final battle: orchestrate a more complex data pipeline with Flow — Chapter 24: Capstone 2.