A CrewAI crew that automates competitor research

Most competitive intelligence workflows die in spreadsheets. This crewai competitor research automation example shows how to wire a multi-agent CrewAI system that finds, summarizes, and compares rivals without manual copying. You get runnable Python, explicit verification steps, and notes on where the architecture breaks in production.

Step 1: Set up the environment and dependencies

Create a clean virtual environment and pin versions. CrewAI ships frequent breaking changes; unpinned installs will rot your crew in a month.

python -m venv venv && source venv/bin/activate
pip install "crewai==0.28.0" "crewai-tools==0.5.0" python-dotenv pytest

You need an LLM endpoint and a search API. Point CrewAI at a single OpenAI-compatible endpoint like n4n.ai to get automatic fallback across 240+ models and per-token metering, which matters when a research crew fires parallel calls at one provider. Put credentials in .env:

echo "OPENAI_API_KEY=sk-your-key" >> .env
echo "OPENAI_BASE_URL=https://api.n4n.ai/v1" >> .env
echo "SERPER_API_KEY=your-serper-key" >> .env

Grab a free Serper key at serper.dev. It returns Google SERPs as JSON, which is all the researcher needs.

Step 2: Define agent roles and task contracts

A competitor research crew needs three distinct responsibilities: discovery, analysis, reporting. Collapsing them into one agent produces vague output and hides failures. Separate them and give each a narrow goal.

• Researcher: runs searches, extracts competitor names and URLs.
• Analyst: reads retrieved pages, scores positioning, pricing, differentiators.
• Writer: compiles a structured markdown brief with a comparison table.

Define them in agents.py. Note the allow_delegation=False—delegation adds latency and nondeterminism for no gain in a linear pipeline.

from crewai import Agent
from crewai_tools import SerperDevTool

search = SerperDevTool()

researcher = Agent(
    role="Competitive Intelligence Researcher",
    goal="Identify direct and indirect competitors for {product} in {market}",
    backstory="Ex-McKinsey analyst who lives in SERPs and Crunchbase.",
    tools=[search],
    verbose=True,
    allow_delegation=False,
)

analyst = Agent(
    role="Product Positioning Analyst",
    goal="Summarize each competitor's pricing, features, and weaknesses",
    backstory="Former PM who can spot a weak onboarding flow in 30 seconds.",
    verbose=True,
    allow_delegation=False,
)

writer = Agent(
    role="Strategy Brief Writer",
    goal="Produce a markdown report comparing competitors with a scoring table",
    backstory="Writes for executives who read only the table.",
    verbose=True,
    allow_delegation=False,
)

Bad backstory: “Helpful assistant that does research.” Good backstory encodes a persona that constrains output. The contract is the goal plus the task’s expected_output.

Step 3: Configure the LLM and per-agent routing

CrewAI accepts an LLM instance per agent. Use a cheap model for the researcher and a stronger one for the analyst if you want quality where it counts.

from crewai import LLM
import os
from dotenv import load_dotenv

load_dotenv()

base_llm = LLM(
    model="gpt-4o-mini",
    base_url=os.getenv("OPENAI_BASE_URL"),
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.2,
    max_rpm=30,
)

strong_llm = LLM(
    model="gpt-4o",
    base_url=os.getenv("OPENAI_BASE_URL"),
    api_key=os.getenv("OPENAI_API_KEY"),
    temperature=0.0,
)

researcher.llm = base_llm
analyst.llm = strong_llm
writer.llm = base_llm

The gateway forwards provider cache-control hints, so repeated competitor lookups hit provider caches when you send the same system prompt. Set temperature=0 on the analyst to keep scoring consistent across runs.

Step 4: Implement the task graph

Tasks reference agents and enforce order through context. Use output_file to persist the final artifact. This is the core of the crewai competitor research automation example.

from crewai import Task

discover = Task(
    description="Search for competitors of {product} in {market}. Return top 5 with URLs.",
    expected_output="List of 5 competitor names and their homepage URLs.",
    agent=researcher,
)

analyze = Task(
    description="For each competitor, extract pricing tier, key feature, and obvious gap.",
    expected_output="Per-competitor bullet summary with a 1-5 score on feature depth.",
    agent=analyst,
    context=[discover],
)

report = Task(
    description="Write a markdown brief with a comparison table and verdict.",
    expected_output="Markdown file with table: Competitor | Pricing | Strength | Weakness | Score",
    agent=writer,
    context=[analyze],
    output_file="competitor_brief.md",
)

If you need a custom search tool because Serper is down, wrap requests:

import requests
from crewai.tools import BaseTool

class SimpleSearchTool(BaseTool):
    name: str = "simple_search"

    def _run(self, query: str) -> str:
        r = requests.get(
            "https://google.serper.dev/search",
            headers={"X-API-KEY": os.getenv("SERPER_API_KEY")},
            params={"q": query},
            timeout=10,
        )
        return r.text[:4000]

Step 5: Assemble and kick off the crew

Main entrypoint. Pass dynamic variables via inputs. Use verbose=2 to see agent thoughts.

from crewai import Crew
from agents import researcher, analyst, writer
from tasks import discover, analyze, report

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[discover, analyze, report],
    verbose=2,
)

result = crew.kickoff(inputs={
    "product": "open-source LLM gateway",
    "market": "developer infrastructure",
})
print(result)

Run python main.py. Expect sequential logs: researcher prints queries, analyst prints scores, writer flushes competitor_brief.md. For nightly jobs, wrap in asyncio and crew.kickoff_async to overlap IO.

Step 6: Verify success

Exit code 0 is not proof. Write a tests/test_brief.py that asserts structure:

import os, re

def test_brief_exists():
    assert os.path.exists("competitor_brief.md")
    content = open("competitor_brief.md").read()
    assert "|" in content
    # at least one scored row with integer 0-5
    assert re.search(r"\|\s*[0-5]\s*\|", content)
    # no agent apology
    assert "I cannot" not in content

Run pytest -q. If the researcher hits a provider rate limit, the crew raises; the gateway fallback should retry on a different provider. Verification must run in CI before you trust the brief.

Step 7: Harden for repeated runs

Single runs hide scaling issues. When scheduled nightly:

• Cache search results with TTL. Serper bills per call. Wrap the tool:

from functools import lru_cache

@lru_cache(maxsize=128)
def cached_search(q: str) -> str:
    return SimpleSearchTool()._run(q)

• Write dated outputs: output/{date}_brief.md.
• Set max_rpm on each LLM (done above) to avoid 429s at the gateway level.

Step 8: Extend with a critique loop

The first draft biases toward the writer prompt. Add a reviewer that checks for gaps.

reviewer = Agent(
    role="Skeptical Partner",
    goal="Find gaps in the brief: missed competitors, unverified pricing",
    backstory="Never trusts a table without a source link.",
    verbose=True,
    llm=strong_llm,
)

review_task = Task(
    description="Read competitor_brief.md, list missing indirect competitors.",
    expected_output="Bullet list of gaps with suggested searches.",
    agent=reviewer,
    context=[report],
)

Add review_task to the crew. This turns the crewai competitor research automation example from a demo into a pipeline: researcher → analyst → writer → reviewer.

Notes on cost and latency

Three agents with context chaining multiply tokens. A typical run on gpt-4o-mini/ gpt-4o mix processes ~15k input tokens and ~3k output. At per-token metering that is cents. If you route the researcher to a 70B open-weight model via the same endpoint, latency rises but cost drops further.

Keep tasks narrow. Broad goals like “research the market” produce agent wandering. The expected_output strings are the contract that keeps the crew deterministic.

What breaks in production

• Search API drift: Serper changes schema. Assert on keys in your tool wrapper.
• Model refusal: Some providers block competitor analysis. Route to a permitted model; gateway fallback handles provider degradation.
• Markdown table rot: Writers hallucinate columns. Validate with a strict regex post-step in CI.

Build the verification test before agent polish. Green test means useful crew; otherwise no backstory fixes it.

Running it in CI

Add a GitHub Action that installs deps, sets secrets, runs pytest. Cache ~/.cache/crewai if you enable embedding caches. The crew should fail the build if competitor_brief.md lacks scores—that is your regression guard.

You now have a runnable crewai competitor research automation example from env to verified markdown brief. Adjust backstories to your domain, wire it into CI, and let the agents do the SERP grinding.