Chapter 25 · Capstone 3: Competitor Monitoring Intelligence Crew

Chapter goals: Build a daily auto-running competitor monitoring pipeline — a collector scrapes competitor websites and release pages, an analyst compares pricing and feature changes, a writer produces Markdown intelligence briefings; use Crew memory to enable "compare against last briefing," and cron scheduling to form a sustainable intelligence system.

25.1 Requirements Analysis and Pipeline Shape

The core of intelligence work is not "search once" but continuous comparison: is today's pricing the same as last week's? What features were added? So this pipeline has two key requirements:

1. Memory — the Crew must be able to read last briefing's content to output "changes" rather than repeating descriptions;
2. Scheduling — no manual triggering required; run once at a fixed time every day.

Shape selection: three roles with clear responsibilities and fixed order (collect → analyze → write briefing), sequential Crew is sufficient — no Flow branching control needed. Overall data flow:

[search_web] competitor dynamic search ──┐
[fetch_page] official site / release page scrape ─┤
                                              ▼
                         Analyst: compare against last briefing (memory)
                                              ▼
                       Writer: Markdown briefing → [save_report]

25.2 Project Structure and Custom Tools

Following Chapter 23's YAML configuration style, add one custom tool to write reports to disk:

competitor-watch/
├── .env                    # SERPER_API_KEY / OPENAI_API_KEY / DEEPSEEK_API_KEY
├── reports/                # Briefing output directory, named by date
│   └── briefing-2026-08-23.md
├── src/
│   ├── tools.py            # Custom tool: save_report
│   ├── agents.yaml         # Three role definitions
│   ├── tasks.yaml          # Three tasks' objectives and output formats
│   └── crew.py             # Crew assembly + entry point
└── run.sh                  # cron invocation entry

# src/tools.py —— Custom tool: save briefing by date
import os
from datetime import date
from pathlib import Path

from crewai.tools import tool


@tool("Save Report")
def save_report(content: str) -> str:
    """Save a Markdown-format competitor intelligence briefing to the reports/ directory.
    Filename is briefing-`<today's date>.md. Returns the save path for the Agent to confirm."""
    out_dir = Path(__file__).resolve().parent.parent / "reports"
    out_dir.mkdir(exist_ok=True)                      # Create directory if it doesn't exist
    path = out_dir / f"briefing-{date.today().isoformat()}.md"
    path.write_text(content, encoding="utf-8")
    return f"Saved to {path}"

Key points:

• The tool docstring is the Agent's usage instruction — must clearly state "when to call it";
• Return value is a string, so the Agent can repeat it in the next step or continue working based on the path;
• Directory created idempotently with mkdir(exist_ok=True) — won't crash in cron environments if the directory is missing.

The other two tools come directly from the official package:

# crew.py snippet —— Search and web scraping tools
from crewai_tools import SerperDevTool, ScrapeWebsiteTool

search_tool = SerperDevTool()      # requires SERPER_API_KEY, see Chapter 23
fetch_page = ScrapeWebsiteTool()   # scrape body text from specified URLs

25.3 Defining Three Roles (agents.yaml)

# src/agents.yaml
scout:                                  # Intelligence collector
  role: >
    Competitor Intelligence Collector
  goal: >
    Collect product dynamics from the past 7 days for competitors {competitors}:
    new feature releases, pricing adjustments, major version updates;
    keep only information-rich entries.
  backstory: >
    You are the right hand of a senior competitive intelligence analyst, skilled at
    quickly filtering truly noteworthy signals from official site changelogs,
    release pages, and tech media — naturally immune to marketing puff pieces.

analyst:                                # Analyst
  role: >
    Product Comparison Analyst
  goal: >
    Compare collected dynamics against the historical briefing to identify "changes":
    pricing tier changes, feature additions/removals, positioning/wording shifts;
    mark each change with an impact rating (high / medium / low).
  backstory: >
    You were a product manager for five years and know which changes warrant alert:
    a price drop means market-share grab; enterprise features trickling down means
    targeting SMB customers… You can read competitive intent from surface-level news.

writer:                                 # Briefing writer
  role: >
    Intelligence Briefing Writer
  goal: >
    Organize the analyst's conclusions into a structured Markdown briefing and call save_report to persist it.
  backstory: >
    You wrote weekly briefings for management for three years and know the one-page principle:
    change summary first, detailed evidence later, always end with "recommended actions to watch."

Note the {competitors} placeholder — the competitor list is injected at runtime; switching to a different set of competitors requires no YAML changes.

25.4 Task Orchestration (tasks.yaml)

# src/tasks.yaml
collect_task:
  description: >
    Use search_web to search for the following competitors' dynamics in the past 7 days: {competitors}.
    For news worth digging into (new feature pages, pricing page redesigns), use fetch_page to scrape the original text.
    Output: one line per dynamic, including source URL, publish time (if available), and a one-sentence summary.
  expected_output: >
    At minimum, cover dynamics for each competitor; competitors with no dynamics are explicitly marked "No public dynamics this period." Fabricating un-searched content is prohibited.
  agent: scout

analyze_task:
  description: >
    Against the Crew memory's last briefing, identify "changes" from the collected results:
    1) Pricing changes (tiers/discounts/billing method); 2) Feature additions/removals;
    3) Positioning and wording shifts. Give each change an impact rating (high/medium/low) and justification.
  expected_output: >
    Changes list (Markdown table: change item | type | impact rating | justification),
    plus a paragraph summarizing "overall posture compared to last period." For the first run with no history, explicitly state "first baseline report."
  agent: analyst
  context:
    - collect_task

brief_task:
  description: >
    Write the analysis conclusions into a Markdown briefing with a fixed structure:
    # Competitor Intelligence Briefing `<date>` / ## Period Snapshot (≤5 items) / ## Detailed Changes /
    ## Recommended Actions to Watch (≤3 actionable suggestions). After writing, must call save_report to persist.
  expected_output: >
    Confirmed saved via save_report with the path restated; the briefing text is also output as the final answer.
  agent: writer
  context:
    - analyze_task

Two engineering details:

• context explicitly declares the dependency chain: analyst sees collection results, writer sees analysis conclusions;
• expected_output written as specifically as possible — this is the discipline emphasized repeatedly in Chapter 14.

25.5 Assembling the Crew and Enabling Memory

# src/crew.py —— Assembly + entry point
import sys
from pathlib import Path

from dotenv import load_dotenv
from crewai import Crew, Process

from crewai.crews import CrewBase  # Remove if your version doesn't need this base class

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(),     # Obtained via @agent decorator or manual instantiation, see Chapter 23
            get_analyst_agent(),
            get_writer_agent(),
        ],
        tasks=[
            collect_task,          # Three tasks defined in tasks.yaml
            analyze_task,
            brief_task,
        ],
        process=Process.sequential,       # Sequential: collect → analyze → write
        memory=True,                      # Key: enable memory so the analyst can "compare against last"
        memory_config={
            "provider": "rag",
            "embedder": {"provider": "openai"},  # Or a third-party compatible endpoint
        },
        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's role here: after memory=True, each kickoff's final output is written to long-term memory; on the next run, the analyst task can retrieve "what last briefing said" and focus on incremental changes rather than re-listing everything already known — this is exactly how the short-term/long-term memory division from Chapter 12 is used in real business.

Tip: If you're using DeepSeek or other OpenAI-compatible endpoints, configure Agents with LLM(model="openai/deepseek-chat", base_url="https://api.deepseek.com/v1", api_key=...) per Chapter 6; the same applies to the embedding model via a compatible endpoint.

25.6 Scheduling with cron to Run Daily

#!/usr/bin/env bash
# run.sh —— cron invocation entry: isolated env vars + log to file
set -euo pipefail
cd "$(dirname "$0")"

export PATH="$HOME/.local/bin:$PATH"     # cron PATH is very minimal;补全
LOG=logs/$(date +%F).log
mkdir -p logs

echo "=== $(date '+%F %T') starting ===" >> "$LOG"
uv run python -m src.crew >> "$LOG" 2>&1   # or poetry run / plain python
echo "=== done ===" >> "$LOG"

# Edit crontab: run every day at 8:30 AM
crontab -e
# Add one line (replace path with your own project absolute path):
30 8 * * * /Users/you/projects/competitor-watch/run.sh

Three cron pitfalls, all handled in run.sh:

Pitfall	Symptom	Fix
Incomplete PATH	python: command not found	Explicitly export PATH inside the script
Wrong working directory	Relative paths can't find reports/	cd "$(dirname "$0")"
Nowhere to see errors	Failed but no clues	Redirect all output to a date-stamped log

Verify the cron config is active: grep CRON /var/log/syslog (Linux) or check mail (macOS sends cron errors to the local user). During debugging, first set the interval to */5 * * * * (every 5 minutes), confirm the pipeline works, then switch back to daily.

25.7 Chapter Summary

• Three roles, each doing one thing: collector just "finds," analyst just "spots changes," writer just "structures and persists";
• Custom tool save_report demonstrates the standard pattern for persistence tools: idempotent directory creation, return the path, docstring as instructions;
• memory=True is the soul of this capstone — without it, daily output is a repetitive briefing; with it, you get true change detection;
• Three cron pitfalls (PATH / working directory / logging) solved in one run.sh;
• Placeholder {competitors} makes the competitor list a runtime parameter — reuse for any track by simply changing inputs.

🧪 随堂测验

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

1. What is the core reason this capstone chooses Process.sequential over hierarchical?

2. Why can the analyst "compare against last briefing to find changes"?

3. When cron executes and the script reports python: command not found, what is the most likely cause?

4. save_report's docstring is written in detail — who is it mainly for?

🛠️ Hands-On Exercises

1. Give the scout an RSS tool (wrap feedparser with @tool), subscribe to competitors' official blog RSS feeds, and note in agents.yaml goal "consume RSS entries first, fall back to search if insufficient."
2. Make impact ratings structured output: add output_pydantic=ChangeList model to analyze_task (fields: changes[].item/type/impact/reason), then write a plain Python step that reads the structured result, filters impact=="high" items, and generates a separate alert file alerts.md.
3. Implement briefing dedup: in the save_report tool, first check whether a same-day file already exists in reports/; if so, append a "## Update <time>" section instead of overwriting; then write a test script to run twice consecutively and verify the behavior.

Final lesson: gather the scattered engineering discipline from all capstones into one go-live checklist — Chapter 26: Best Practices and Production Checklist.