第 25 章 · 实战三：竞品监控情报 Crew

本章目标：搭建一条每日自动运行的竞品监控流水线——采集员抓取竞品官网与发布页、分析师对比定价与功能变化、撰写员产出 Markdown 情报简报；利用 Crew 记忆实现"与上次简报对比"，并用 cron 定时调度，形成可持续运转的情报系统。

25.1 需求分析与流水线形态

情报工作的核心不是"搜一次"，而是持续对比：今天的价格和上周一样吗？功能列表多了什么？所以这条流水线有两个关键点：

记忆——Crew 要能读到上次简报的内容，才能输出"变化"而非重复描述；
定时——不依赖人手动触发，每天固定时间跑一遍。

形态选择：三个角色职责清晰、顺序固定（先采集→再分析→后写简报），用 sequential 的 Crew 即可，不需要 Flow 的分支控制。整体数据流：

text

[search_web] 竞品动态搜索 ──┐
[fetch_page] 官网/发布页抓取 ─┤
                            ▼
                 分析师：与上次简报对比（memory）
                            ▼
              撰写员：Markdown 简报 → [save_report]

25.2 项目结构与自定义工具

沿用第 23 章 YAML 配置风格，新增一个自定义工具把报告落盘：

text

competitor-watch/
├── .env                    # SERPER_API_KEY / OPENAI_API_KEY / DEEPSEEK_API_KEY
├── reports/                # 简报输出目录，按日期命名
│   └── briefing-2026-08-23.md
├── src/
│   ├── tools.py            # 自定义工具：save_report
│   ├── agents.yaml         # 三个角色的角色定义
│   ├── tasks.yaml          # 三个任务的目标与产出格式
│   └── crew.py             # Crew 组装 + 入口
└── run.sh                  # cron 调用入口

python

# src/tools.py —— 自定义工具：按日期保存简报
import os
from datetime import date
from pathlib import Path

from crewai.tools import tool


@tool("Save Report")
def save_report(content: str) -> str:
    """将 Markdown 格式的竞品情报简报保存到 reports/ 目录，
    文件名为 briefing-<当天日期>.md。返回保存路径供 Agent 复述确认。"""
    out_dir = Path(__file__).resolve().parent.parent / "reports"
    out_dir.mkdir(exist_ok=True)                      # 目录不存在则创建
    path = out_dir / f"briefing-{date.today().isoformat()}.md"
    path.write_text(content, encoding="utf-8")
    return f"已保存到 {path}"

要点：

工具 docstring 就是 Agent 的使用说明，必须写清楚"什么时候该调用它"；
返回值是字符串，让 Agent 能在下一步复述或基于路径继续工作；
目录用 mkdir(exist_ok=True) 幂等创建，cron 环境下不会因为目录缺失而崩。

另外两个工具直接用官方包现成的：

python

# crew.py 片段 —— 搜索与网页抓取工具
from crewai_tools import SerperDevTool, ScrapeWebsiteTool

search_tool = SerperDevTool()      # 需要 SERPER_API_KEY，见第 23 章
fetch_page = ScrapeWebsiteTool()   # 抓取指定 URL 的正文文本

25.3 定义三个角色（agents.yaml）

yaml

# src/agents.yaml
scout:                                  # 情报采集员
  role: >
    竞品情报采集员
  goal: >
    为目标竞品 {competitors} 收集最近 7 天内的产品动态：
    新功能发布、定价调整、重大版本更新；只保留有信息量的条目。
  backstory: >
    你是资深竞争情报分析师的左膀右臂，擅长从官网更新日志、
    发布页和科技媒体中快速筛出真正值得关注的信号，
    对营销软文有天然的免疫力。

analyst:                                # 分析师
  role: >
    产品对比分析师
  goal: >
    将采集到的动态与历史简报对比，识别出"变化"：
    定价档位变化、功能增删、定位话术转向；每条变化标注影响评级（高/中/低）。
  backstory: >
    你做过五年产品经理，深知哪些变化值得警惕：价格下调意味着抢量，
    企业版功能下沉意味着打中小客户……你能从表面新闻里读出竞争意图。

writer:                                 # 简报撰写员
  role: >
    情报简报撰写员
  goal: >
    把分析师的结论整理成一份结构化 Markdown 简报并调用 save_report 保存。
  backstory: >
    你给管理层写了三年周报，深知一页纸原则：
    变化摘要放最前，细节证据往后排，永远给出"建议关注动作"。

注意 {competitors} 占位符——竞品清单在运行时注入，换一批竞品不需要改 YAML。

25.4 任务编排（tasks.yaml）

yaml

# src/tasks.yaml
collect_task:
  description: >
    使用 search_web 搜索以下竞品最近 7 天的动态：{competitors}。
    对值得深入的消息（新功能页、定价页改版），再用 fetch_page 抓取原文。
    输出：每条动态一行，包含来源 URL、发布时间（若可得）、一句话摘要。
  expected_output: >
    至少覆盖每个竞品的动态清单；无动态的竞品明确标注"本期无公开动态"。禁止编造未检索到的内容。
  agent: scout

analyze_task:
  description: >
    对照 Crew 记忆中的上次简报，从采集结果中识别"变化"：
    1) 定价变化（档位/折扣/计费方式）；2) 功能增删；
    3) 定位与话术转向。为每条变化给出影响评级（高/中/低）与判断依据。
  expected_output: >
    变化清单（Markdown 表格：变化项 | 类型 | 影响评级 | 判断依据），
    以及一段"与上期相比的整体态势"总结。首期无历史时明确说明"首次基线报告"。
  agent: analyst
  context:
    - collect_task

brief_task:
  description: >
    把分析结论写成 Markdown 简报，结构固定为：
    # 竞品情报简报 <日期> / ## 本期速览（≤5 条）/ ## 变化明细 /
    ## 建议关注动作（≤3 条 actionable 建议）。写完必须调用 save_report 保存。
  expected_output: >
    已通过 save_report 成功保存，并复述保存路径；简报正文同时作为最终答案输出。
  agent: writer
  context:
    - analyze_task

两个工程细节：

context 显式声明依赖链：分析师看得到采集结果，撰写员看得到分析结论；
expected_output 写得越具体，产出越稳定——这是第 14 章反复强调的纪律。

25.5 组装 Crew 并开启记忆

python

# src/crew.py —— 组装 + 入口
import sys
from pathlib import Path

from dotenv import load_dotenv
from crewai import Crew, Process

from crewai.crews import CrewBase  # 若你的版本无需该基类可去掉

load_dotenv()
sys.path.insert(0, str(Path(__file__).parent))

from crewai_tools import SerperDevTool, ScrapeWebsiteTool

from tools import save_report


def build_crew(competitors: str) -> Crew:
    return Crew(
        agents=[
            get_scout_agent(),     # 由 @agent 装饰器或手动实例化得到，见第 23 章
            get_analyst_agent(),
            get_writer_agent(),
        ],
        tasks=[
            collect_task,          # tasks.yaml 中定义的三个任务
            analyze_task,
            brief_task,
        ],
        process=Process.sequential,       # 顺序执行：采集→分析→撰写
        memory=True,                      # 关键：开启记忆，分析师才能"对照上次"
        memory_config={
            "provider": "rag",
            "embedder": {"provider": "openai"},  # 或三方兼容端点
        },
        tools=[SerperDevTool(), ScrapeWebsiteTool(), save_report],
        verbose=True,
    )


if __name__ == "__main__":
    crew = build_crew(competitors="Notion, Linear, Height")
    result = crew.kickoff(inputs={"competitors": "Notion, Linear, Height"})
    print(result)

记忆在这里的作用：memory=True 后，每次 kickoff 的最终产出会写入长期记忆；下一次运行时分析师任务可以检索到"上次简报说了什么"，从而聚焦增量变化而不是重新罗列一遍已知信息——这正是第 12 章讲的短期/长期记忆分工在真实业务里的用法。

提示：如果你使用 DeepSeek 等 OpenAI 兼容端点，按第 6 章的方式给 Agent 配置 LLM(model="openai/deepseek-chat", base_url="https://api.deepseek.com/v1", api_key=...)；嵌入模型同理可用兼容端点。

25.6 用 cron 让它每天自己跑

bash

#!/usr/bin/env bash
# run.sh —— cron 的调用入口：隔离环境变量 + 日志落盘
set -euo pipefail
cd "$(dirname "$0")"

export PATH="$HOME/.local/bin:$PATH"     # cron 环境 PATH 很短，需补全
LOG=logs/$(date +%F).log
mkdir -p logs

echo "=== $(date '+%F %T') 开始 ===" >> "$LOG"
uv run python -m src.crew >> "$LOG" 2>&1   # 或 poetry run / 直接 python
echo "=== 完成 ===" >> "$LOG"

bash

# 编辑 crontab：每天早上 8:30 执行
crontab -e
# 加入一行（路径替换为你自己的项目绝对路径）：
30 8 * * * /Users/you/projects/competitor-watch/run.sh

cron 三大坑，run.sh 里都已处理：

坑	症状	解法
PATH 不全	`python: command not found`	脚本内显式 export PATH
工作目录不对	相对路径找不到 reports/	`cd "$(dirname "$0")"`
无处看错误	失败了但没任何线索	全部输出重定向到带日期的日志

验证定时配置是否生效：grep CRON /var/log/syslog（Linux）或查看 mail（macOS 会把 cron 错误发给本地用户）。调试期建议先把间隔改成 */5 * * * *（每 5 分钟），确认链路通了再调回每日。

25.7 本章小结

三个角色各司其职：采集员只管"找到"，分析师只管"对比出变化"，撰写员只管"结构化落盘"；
自定义工具 save_report 展示了落盘类工具的标准写法：幂等建目录、返回路径、docstring 即说明书；
memory=True 是本实战的灵魂——没有它，每天产出的是重复简报；有了它，才是真正的变化监测；
cron 定时三坑（PATH/工作目录/日志）在 run.sh 中一次性解决；
占位符 {competitors} 让竞品清单成为运行时参数，复用到任何赛道只需换输入。

🧪 随堂测验

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

1. 本实战选择 Process.sequential 而非 hierarchical 的核心原因是？

2. 为什么分析师能'对照上次简报找变化'？

3. cron 定时执行时脚本报 python: command not found，最可能的原因是？

4. save_report 工具的 docstring 写得详细，主要是给谁看的？

🛠️ 动手实践

给采集员加一个 RSS 工具（用 @tool 包装 feedparser），订阅竞品官方博客的 RSS 源，并在 agents.yaml 的 goal 中注明"优先消费 RSS 条目，不足再搜索"。
把影响评级做成结构化输出：给 analyze_task 增加 output_pydantic=ChangeList 模型（字段：changes[].item/type/impact/reason），然后写一个纯 Python 步骤读取结构化结果、筛选 impact=="高" 的条目单独生成告警文件 alerts.md。
实现简报去噪：在 save_report 工具里先检查 reports/ 下是否已存在同日文件，存在则改为追加"## 更新 <时间>"小节而不是覆盖；再写一个测试脚本连续跑两次验证行为。

最后一课：把散落在全部实战里的工程纪律收进一张上线清单：第 26 章 · 最佳实践与生产清单。

第 25 章 · 实战三：竞品监控情报 Crew ​

25.1 需求分析与流水线形态 ​

25.2 项目结构与自定义工具 ​

25.3 定义三个角色（agents.yaml） ​

25.4 任务编排（tasks.yaml） ​

25.5 组装 Crew 并开启记忆 ​

25.6 用 cron 让它每天自己跑 ​

25.7 本章小结 ​

🧪 随堂测验

🛠️ 动手实践 ​

第 25 章 · 实战三：竞品监控情报 Crew

25.1 需求分析与流水线形态

25.2 项目结构与自定义工具

25.3 定义三个角色（agents.yaml）

25.4 任务编排（tasks.yaml）

25.5 组装 Crew 并开启记忆

25.6 用 cron 让它每天自己跑

25.7 本章小结

🛠️ 动手实践