第 23 章 · 实战一：内容营销流水线

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

23.1 需求与流水线设计

业务需求：输入一个主题关键词，产出一篇经过事实核查与 SEO 优化的公众号/博客文章。

流水线拆解为四个角色，前一个任务的产出通过 context 流向下一个：

[选题策划 TopicStrategist] → [调研员 Researcher] → [撰稿人 Writer] → [SEO 编辑 SEDEditor]
   产出: 选题角度+大纲      产出: 事实素材+引用      产出: 文章初稿        产出: 最终发布稿

为什么用 sequential 而不是 hierarchical：四个环节职责清晰、顺序固定、无需动态委派——顺序流水是最可预测、最好调试的形态（选型依据见第 5、9 章）。

23.2 项目结构

用 crewai create crew marketing_pipeline 生成骨架后，最终结构如下：

marketing_pipeline/
├── .env                      # DEEPSEEK_API_KEY / SERPER_API_KEY
├── pyproject.toml
└── src/marketing_pipeline/
    ├── main.py               # 入口
    ├── crew.py               # @CrewBase 装配
    └── config/
        ├── agents.yaml       # 四个角色定义
        └── tasks.yaml        # 四个任务定义

安装依赖并配置密钥：

pip install 'crewai[tools]' python-dotenv
# .env 文件内容（不要提交到 git）：
# DEEPSEEK_API_KEY=sk-xxx
# SERPER_API_KEY=xxx        # https://serper.dev 免费额度足够练习

23.3 agents.yaml：四个角色的提示词设计

# src/marketing_pipeline/config/agents.yaml
topic_strategist:
  role: "资深内容策划"
  goal: "为主题 '{topic}' 找到最有传播力的切入角度并产出文章大纲"
  backstory: >
    你有 10 年新媒体选题经验，擅长把平淡的技术主题转化为读者
    想点开的角度。你坚持：一个选题只讲一个核心观点，大纲不超过
    4 个部分。
  llm: "openai/deepseek-chat"
  max_iter: 3
  verbose: true

researcher:
  role: "事实核查调研员"
  goal: "为既定大纲收集准确的事实、数据与来源"
  backstory: >
    你是前调查记者，对数字极度较真。你只采信能给出处的事实，
    找不到可靠来源的论断会明确标注"待核实"。
  llm: "openai/deepseek-chat"
  max_iter: 5
  verbose: true

writer:
  role: "B2B 技术撰稿人"
  goal: "基于大纲与调研素材写出流畅、有观点的中文文章初稿"
  backstory: >
    你写过数百篇 10w+ 技术普及文章。你的文字短句为主、每段
    不超过 5 行，善用类比，绝不堆砌形容词。
  llm: "openai/deepseek-chat"
  max_iter: 4

seo_editor:
  role: "SEO 优化编辑"
  goal: "在不伤害可读性的前提下完成 SEO 优化并做终审"
  backstory: >
    你精通中文搜索引擎优化：标题党会伤害品牌，你只做合规优化。
    你同时承担事实终审，发现与调研素材矛盾的内容会直接修正。
  llm: "openai/deepseek-chat"
  max_iter: 3

设计要点回顾（第 3 章）：backstory 给的是行为准则而不是头衔堆砌（"只采信有出处的数字"比"我很严谨"有效得多）；每个角色都限制了 max_iter 防失控（第 25 章）。

23.4 tasks.yaml：任务链与上下文传递

# src/marketing_pipeline/config/tasks.yaml
plan_topic:
  description: >
    为主题 "{topic}" 策划一篇文章：给出 1 个核心观点、
    目标读者画像、以及不超过 4 部分的文章大纲。
    大纲每部分用一句话说明要论证什么。
  expected_output: >
    Markdown 格式的选题卡：核心观点（1 句）、目标读者（1 句）、
    大纲（3-4 个部分，每部分 1-2 句）。
  agent: topic_strategist

research_facts:
  description: >
    针对选题卡的大纲逐部分收集事实素材：每个部分至少 2 条
    可用信息，数字必须带来源与年份。使用搜索工具检索最新信息，
    并用网页抓取工具核实关键数据。
  expected_output: >
    按大纲部分组织的调研笔记：每条素材一行，
    格式「事实 | 来源 | 年份」，无来源的标注"待核实"。
  agent: researcher
  context:
    - plan_topic
  tools:
    - SerperDevTool
    - ScrapeWebsiteTool

write_draft:
  description: >
    严格基于调研笔记撰写 800-1200 字中文文章初稿。
    只使用调研笔记中出现的事实，不得自行编造数据；
    调研笔记中标注"待核实"的内容要么删除要么改写为定性表述。
  expected_output: >
    完整的 Markdown 文章：含 H1 标题、3-4 个 H2 小节、
    结尾有 3 条要点总结。
  agent: writer
  context:
    - plan_topic
    - research_facts

seo_finalize:
  description: >
    对初稿做终审与 SEO 优化：1) 核对文中事实与调研笔记一致；
    2) 生成含核心关键词的标题（不超过 25 字）；3) 写 120 字内
    摘要；4) 给出 5 个关键词标签。不得改变文章核心观点。
  expected_output: >
    最终发布包：优化后标题、摘要、正文全文、关键词标签列表，
    全部使用 Markdown 输出。
  agent: seo_editor
  context:
    - research_facts
    - write_draft
  output_file: output/final_article.md

关键机制（第 4 章）：research_facts 通过 context 拿到选题卡；write_draft 同时拿到大纲与调研笔记；seo_finalize 拿初稿和调研笔记做事实核对——这正是"编辑要能对照素材审稿"的现实工作流。

23.5 crew.py：装配与三方模型

# 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()

# 三方 OpenAI 兼容模型：yaml 中 llm 字段引用 provider/model，
# base_url 与 api_key 由 CrewAI 从环境变量读取
# （DEEPSEEK_API_KEY 自动匹配 openai/ 前缀 + DeepSeek 官方端点的场景
#   需要显式 base_url 时，可在 Agent 上传 LLM 对象，见下方注释）


@CrewBase
class MarketingCrew:
    """内容营销流水线：选题 → 调研 → 撰写 → SEO 终审"""

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

    search_tool = SerperDevTool()        # 需要 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,   # 固定顺序流水
            verbose=True,
        )

若想完全掌控模型端点，可删掉 yaml 中的 llm 字段，改为在代码里显式注入 LLM 对象——两种写法等价，团队统一其一即可：

from crewai import LLM

# 显式注入三方模型（优先于 yaml 的 llm 字段）
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 字段会被此参数覆盖
    )

23.6 运行与结果分析

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

if __name__ == "__main__":
    result = MarketingCrew().crew().kickoff(
        inputs={"topic": "AI Agent 在企业客服中的落地"}
    )
    print("=" * 60)
    print(result.raw)
    print("=" * 60)
    print(f"总 tokens: {result.token_usage.total_tokens}")

crewai run   # 或 python -m marketing_pipeline.main

典型运行观察点：

1. researcher 的工具调用：verbose 日志里应看到 2-5 次 Search the internet 调用；如果超过 8 次，说明 max_iter 太松或任务描述太模糊；
2. 事实一致性：抽查 output/final_article.md 中的数字，回对调研笔记——context 链完整时矛盾率显著低于单 Agent 直写；
3. 成本分布：token_usage 中 researcher 通常占 40% 以上（工具结果膨胀上下文），这是正常的，优化方向是让调研笔记更精炼。

还可以逐环节检查中间产出与用量，定位问题任务：

result = MarketingCrew().crew().kickoff(inputs={"topic": "AI Agent 在企业客服中的落地"})

# tasks_output 与 tasks.yaml 中定义的任务一一对应
for i, task_out in enumerate(result.tasks_output):
    preview = (task_out.raw or "")[:80].replace("\n", " ")
    print(f"任务 {i}: {len(task_out.raw or '')} 字 | 开头: {preview}...")

若发现某个任务的产出明显偏离预期（比如调研笔记没带来源），先回查该任务的 expected_output 是否足够具体——绝大多数质量问题都能通过收紧验收标准解决，而不是换模型。

23.7 改进方向

• 升级 hierarchical：加一个 manager_llm 让"主编"动态决定是否需要补调研（第 9 章），适合选题质量参差的场景；
• Flow 化：把选题/撰写拆成 Flow 步骤，中间插入确定性代码步骤（如字数检查、敏感词过滤），失败可重试单步（第 15-16 章）；
• 接入第 22 章的 API 化：包上 POST /kickoff + 轮询，就是团队可用的内容生产服务；
• 评估闭环：用第 20 章的回归集断言"文章含标题/摘要/标签三件套、字数达标、无'待核实'残留"。

本章小结

• 四角色 sequential 流水线覆盖了 Agent 设计、context 链、工具挂载、YAML 工程化全部核心知识；
• backstory 写行为准则、max_iter 逐角色设防、编辑任务同时挂初稿与素材做事实核对；
• 工具只挂在真正需要联网的角色上（researcher），其余角色零工具更稳定；
• 成本大头通常在带工具的调研环节，优化素材精炼度比换模型更有效。

🧪 随堂测验

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

1. 本项目中 seo_finalize 任务的 context 同时包含 research_facts 和 write_draft，最主要目的是？

2. 为什么本项目选择 sequential 而不是 hierarchical 流程？

3. 运行后发现 token 消耗远超预期，日志显示 researcher 调用搜索工具 12 次，最合理的处置是？

4. 关于 yaml 中的 llm: "openai/deepseek-chat" 写法，下列说法正确的是？

🛠️ 动手实践

1. 给流水线增加第五个角色「标题 AB 测试师」：产出 3 个风格不同的候选标题并说明各自适用场景，挂在 seo_finalize 之后。
2. 把 researcher 的 SerperDevTool 结果缓存起来：写一个自定义工具包装搜索，相同关键词 24 小时内直接返回缓存（提示：参考第 8 章 cache_function）。
3. 按 23.7 的思路把「字数检查」做成确定性步骤：写一个小 Flow，在 write_draft 后用 Python 代码统计字数，不达标则把初稿和"压缩到 1000 字内"的要求重新交给 writer（循环最多 2 次）。

最后一战：用 Flow 编排一条更复杂的数据流水线：第 24 章 · 实战二。