Crawl4AI Hermes Agent Swarm Setup Guide

Important technical clarification: crawl4ai_deep_crawl should be presented as the Hermes tool adapter created in this guide unless a particular Hermes build already ships it.

This design follows Crawl4AI’s official repository, deep-crawling documentation, parallel crawling documentation, configuration reference, and Markdown extraction guide.

1. What this guide builds

This tutorial creates a working local research pipeline where:

1. A Hermes Source Finder Agent discovers candidate businesses and evidence sources.
2. Hermes creates separate crawl tasks for each domain.
3. Multiple crawler workers run concurrently.
4. Crawl4AI renders JavaScript-heavy pages and recursively discovers relevant internal pages.
5. Each page is converted into clean Markdown.
6. Structured fields are extracted into JSON.
7. An Analyst Agent consolidates the evidence into a lead portfolio.
8. A Verification Agent validates every important claim against its source.
9. Verified records are saved as Markdown, JSON, database records, or automation events.

The final workflow is:

Campaign brief
    ↓
Hermes Source Finder Agent
    ↓
candidate_queue
    ↓
Hermes Crawl Orchestrator
    ├── Crawler Worker 01 → Official website
    ├── Crawler Worker 02 → GitHub / technical source
    ├── Crawler Worker 03 → News / founder source
    └── Crawler Worker 04 → Careers / growth source
    ↓
Evidence Normalizer
    ↓
Analyst Agent
    ↓
Verification Gate
    ↓
Lead Portfolio
    ↓
Knowledge base / CRM / automation

2. Important technical clarification

Crawl4AI provides the underlying Python crawling interfaces:

• AsyncWebCrawler
• BrowserConfig
• CrawlerRunConfig
• BestFirstCrawlingStrategy
• KeywordRelevanceScorer
• JsonCssExtractionStrategy
• DefaultMarkdownGenerator
• PruningContentFilter
• arun()
• arun_many()

The name crawl4ai_deep_crawl is the Hermes tool wrapper that this tutorial creates around those interfaces.

Do not assume every Hermes installation already contains a tool with that exact name. If it is already present, compare its input and output contract with the adapter in this guide.

3. System requirements

Recommended environment:

• Python 3.11 or 3.12
• Git
• Chromium
• 8 GB RAM minimum
• 16 GB RAM recommended for parallel workers
• Hermes Agent installed and working
• A preferred LLM provider for the final Analyst Agent
• No LLM is required for basic crawling and Markdown generation

Check Python:

python3 --version

Clone the Crawl4AI repository for reference:

git clone https://github.com/unclecode/crawl4ai.git

The pipeline can use the PyPI package; modifying the Crawl4AI repository is not required.

4. Create the project

mkdir crawl4ai-hermes-swarm
cd crawl4ai-hermes-swarm

python3 -m venv .venv
source .venv/bin/activate

Windows PowerShell:

python -m venv .venv
.venv\Scripts\Activate.ps1

Create this structure:

crawl4ai-hermes-swarm/
├── .env
├── requirements.txt
├── campaign.json
├── seeds.json
├── run.py
├── hermes_swarm/
│   ├── __init__.py
│   ├── models.py
│   ├── source_finder.py
│   ├── crawl4ai_tool.py
│   ├── analyst.py
│   ├── verifier.py
│   ├── storage.py
│   └── orchestrator.py
└── runs/

5. Install Crawl4AI

Create requirements.txt:

crawl4ai==0.9.1
pydantic>=2.7,<3
python-dotenv>=1.0
httpx>=0.27
beautifulsoup4>=4.12
openai>=1.50

Install everything:

pip install -r requirements.txt
crawl4ai-setup
crawl4ai-doctor

If Chromium installation fails:

python -m playwright install chromium

On Linux:

python -m playwright install --with-deps chromium

Do not continue until crawl4ai-doctor reports a healthy browser installation.

6. Environment configuration

Create .env:

HERMES_MAX_WORKERS=3
HERMES_MAX_DEPTH=2
HERMES_MAX_PAGES_PER_DOMAIN=25
HERMES_PAGE_TIMEOUT_MS=60000
HERMES_CHECK_ROBOTS=true

# Optional final Analyst Agent
LLM_API_KEY=
LLM_BASE_URL=
LLM_MODEL=

The crawler remains local. The optional LLM should receive consolidated evidence after crawling, not every raw page separately.

7. Campaign configuration

Create campaign.json:

{
  "campaign_name": "Chennai technology-business research",
  "query": "software and AI companies in Chennai",
  "location": "Chennai, Tamil Nadu, India",
  "target_leads": 10,
  "keywords": [
    "about",
    "company",
    "founder",
    "owner",
    "leadership",
    "product",
    "services",
    "customers",
    "pricing",
    "careers",
    "github"
  ],
  "max_depth": 2,
  "max_pages_per_domain": 25,
  "max_workers": 3,
  "maximum_sources_per_lead": 5,
  "minimum_independent_sources": 2
}

Use conservative limits first. A crawl depth above three can cause the URL frontier to grow rapidly.

8. Seed input

The Source Finder Agent can receive seeds from:

• OwnerHunter
• Google Maps business discovery
• A CSV or JSON file
• A Hermes web-search tool
• An approved search provider
• A manually supplied list

Create seeds.json:

[
  {
    "lead_id": "CHN-001",
    "business_name": "Example Technology Company",
    "website": "https://example.com",
    "location": "Chennai, Tamil Nadu, India",
    "additional_sources": [
      "https://github.com/example",
      "https://example.com/about"
    ]
  }
]

Business-directory listings are seed evidence only. They do not prove ownership.

9. Define the shared data models

Create hermes_swarm/models.py:

from typing import Any
from pydantic import BaseModel, Field


class LeadSeed(BaseModel):
    lead_id: str
    business_name: str
    website: str
    location: str = ""
    additional_sources: list[str] = Field(default_factory=list)


class Campaign(BaseModel):
    campaign_name: str
    query: str
    location: str
    target_leads: int = 10
    keywords: list[str]
    max_depth: int = 2
    max_pages_per_domain: int = 25
    max_workers: int = 3
    maximum_sources_per_lead: int = 5
    minimum_independent_sources: int = 2


class SourceTask(BaseModel):
    lead_id: str
    business_name: str
    source_url: str
    source_kind: str
    is_primary: bool = False


class PageEvidence(BaseModel):
    lead_id: str
    source_url: str
    source_group: str
    source_kind: str
    title: str = ""
    markdown: str
    relevance_score: float = 0
    status_code: int | None = None
    metadata: dict[str, Any] = Field(default_factory=dict)


class EvidenceCitation(BaseModel):
    source_url: str
    quote: str


class Claim(BaseModel):
    field: str
    value: str
    linkage: str
    material: bool = True
    evidence: list[EvidenceCitation] = Field(default_factory=list)
    verified: bool = False
    verification_reason: str = ""


class LeadPortfolio(BaseModel):
    lead_id: str
    business_name: str
    domain: str
    summary: str = ""
    category: str = ""
    classification: str = "BUSINESS_ONLY"
    confidence: float = 0
    claims: list[Claim] = Field(default_factory=list)
    sources: list[str] = Field(default_factory=list)
    public_business_contacts: dict[str, list[str]] = Field(
        default_factory=dict
    )
    warnings: list[str] = Field(default_factory=list)

10. Build the Source Finder Agent

The Source Finder Agent must separate two tasks:

1. Finding candidate businesses.
2. Finding evidence sources for each business.

For every lead, create searches such as:

"Business Name" Chennai founder
"Business Name" Chennai owner
"Business Name" official website
"Business Name" leadership
site:github.com "Business Name"
"Business Name" Chennai interview
"Business Name" Chennai careers

Source priorities:

1. Official website
2. Official company repository
3. Government or professional record
4. Founder interview
5. Reputable local or industry publication
6. Careers page
7. Business directory as supporting evidence only

Create hermes_swarm/source_finder.py:

from urllib.parse import urlparse

from .models import Campaign, LeadSeed, SourceTask


def hostname(url: str) -> str:
    return (urlparse(url).hostname or "").lower().removeprefix("www.")


def classify_source(url: str, primary_domain: str) -> str:
    host = hostname(url)

    if host == primary_domain or host.endswith("." + primary_domain):
        return "official_website"

    if host == "github.com":
        return "github"

    if any(token in host for token in ("gov.", ".gov", "nic.in")):
        return "official_record"

    return "independent_public_source"


def build_source_plan(
    lead: LeadSeed,
    campaign: Campaign,
    discovered_urls: list[str] | None = None,
) -> list[SourceTask]:
    discovered_urls = discovered_urls or []

    primary_domain = hostname(lead.website)
    ordered_urls = [
        lead.website,
        *lead.additional_sources,
        *discovered_urls,
    ]

    seen = set()
    tasks = []

    for url in ordered_urls:
        if not url.startswith(("http://", "https://")):
            continue

        normalized = url.rstrip("/")

        if normalized in seen:
            continue

        seen.add(normalized)

        tasks.append(
            SourceTask(
                lead_id=lead.lead_id,
                business_name=lead.business_name,
                source_url=normalized,
                source_kind=classify_source(normalized, primary_domain),
                is_primary=hostname(normalized) == primary_domain,
            )
        )

        if len(tasks) >= campaign.maximum_sources_per_lead:
            break

    return tasks

The actual Hermes web-search tool should populate discovered_urls. Search-result snippets are discovery clues, not final evidence.

11. Create the crawl4ai_deep_crawl Hermes tool

Create hermes_swarm/crawl4ai_tool.py:

from urllib.parse import urlparse

from crawl4ai import (
    AsyncWebCrawler,
    BrowserConfig,
    CacheMode,
    CrawlerRunConfig,
)
from crawl4ai.content_filter_strategy import PruningContentFilter
from crawl4ai.deep_crawling import BestFirstCrawlingStrategy
from crawl4ai.deep_crawling.scorers import KeywordRelevanceScorer
from crawl4ai.markdown_generation_strategy import DefaultMarkdownGenerator

from .models import Campaign, PageEvidence, SourceTask


def source_group(url: str) -> str:
    return (urlparse(url).hostname or "").lower().removeprefix("www.")


def result_markdown(result) -> str:
    markdown = getattr(result, "markdown", None)

    if markdown is None:
        return ""

    fit = getattr(markdown, "fit_markdown", None)
    raw = getattr(markdown, "raw_markdown", None)

    return (fit or raw or str(markdown) or "").strip()


async def crawl4ai_deep_crawl(
    task: SourceTask,
    campaign: Campaign,
) -> list[PageEvidence]:
    is_primary = task.is_primary

    max_depth = campaign.max_depth if is_primary else 1
    max_pages = campaign.max_pages_per_domain if is_primary else 3

    scorer = KeywordRelevanceScorer(
        keywords=campaign.keywords,
        weight=0.7,
    )

    deep_strategy = BestFirstCrawlingStrategy(
        max_depth=max_depth,
        include_external=False,
        max_pages=max_pages,
        url_scorer=scorer,
    )

    markdown_generator = DefaultMarkdownGenerator(
        content_filter=PruningContentFilter(
            threshold=0.45,
            threshold_type="fixed",
            min_word_threshold=20,
        )
    )

    browser_config = BrowserConfig(
        headless=True,
        verbose=False,
    )

    run_config = CrawlerRunConfig(
        deep_crawl_strategy=deep_strategy,
        markdown_generator=markdown_generator,
        cache_mode=CacheMode.BYPASS,
        stream=True,
        check_robots_txt=True,
        wait_until="domcontentloaded",
        page_timeout=60000,
        delay_before_return_html=0.5,
        word_count_threshold=50,
        excluded_tags=["nav", "footer", "script", "style", "noscript"],
        exclude_external_links=True,
        scan_full_page=True,
        remove_overlay_elements=True,
        remove_consent_popups=True,
        flatten_shadow_dom=True,
        user_agent="HermesResearchCrawler/1.0",
    )

    pages: list[PageEvidence] = []

    async with AsyncWebCrawler(config=browser_config) as crawler:
        result_stream = await crawler.arun(
            url=task.source_url,
            config=run_config,
        )

        async for result in result_stream:
            if not result.success:
                continue

            markdown = result_markdown(result)

            if len(markdown) < 100:
                continue

            metadata = result.metadata or {}

            pages.append(
                PageEvidence(
                    lead_id=task.lead_id,
                    source_url=result.url,
                    source_group=source_group(result.url),
                    source_kind=task.source_kind,
                    title=str(metadata.get("title") or ""),
                    markdown=markdown,
                    relevance_score=float(metadata.get("score") or 0),
                    status_code=result.status_code,
                    metadata=metadata,
                )
            )

    return pages

This is the implementation behind the Hermes tool name:

crawl4ai_deep_crawl

Recommended tool input:

{
  "lead_id": "CHN-001",
  "business_name": "Example Technology Company",
  "source_url": "https://example.com",
  "source_kind": "official_website",
  "is_primary": true,
  "max_depth": 2,
  "max_pages": 25,
  "keywords": [
    "founder",
    "owner",
    "leadership",
    "product",
    "careers"
  ]
}

Recommended tool output:

{
  "success": true,
  "lead_id": "CHN-001",
  "source_url": "https://example.com",
  "pages_crawled": 18,
  "documents": [
    {
      "url": "https://example.com/about",
      "title": "About",
      "source_group": "example.com",
      "markdown": "# About...",
      "relevance_score": 0.91
    }
  ],
  "errors": []
}

12. Run crawler agents in parallel

Recursive crawling and parallel crawling are different concerns.

Use:

• BestFirstCrawlingStrategy for recursive crawling inside one domain.
• Separate Hermes crawler workers for different source domains.
• asyncio.Semaphore to prevent excessive browser processes.
• arun_many() when processing many independent single-page URLs.
• A separate deep-crawl strategy instance for each recursive domain worker.

Create the crawler-swarm function:

import asyncio

from .crawl4ai_tool import crawl4ai_deep_crawl
from .models import Campaign, PageEvidence, SourceTask


async def run_crawler_swarm(
    tasks: list[SourceTask],
    campaign: Campaign,
) -> tuple[list[PageEvidence], list[dict]]:
    semaphore = asyncio.Semaphore(campaign.max_workers)
    failures = []

    async def worker(task: SourceTask) -> list[PageEvidence]:
        async with semaphore:
            try:
                return await crawl4ai_deep_crawl(task, campaign)
            except Exception as exc:
                failures.append(
                    {
                        "lead_id": task.lead_id,
                        "source_url": task.source_url,
                        "error": f"{exc.__class__.__name__}: {exc}",
                    }
                )
                return []

    results = await asyncio.gather(
        *(worker(task) for task in tasks),
        return_exceptions=False,
    )

    pages = [
        page
        for worker_pages in results
        for page in worker_pages
    ]

    return pages, failures

Start with two or three workers. Increase concurrency only after monitoring memory and target-site response codes.

13. Preserve raw and clean Markdown

Every successfully crawled page should be saved before analysis:

runs/<run_id>/
├── manifest.json
├── queues.json
├── raw/
│   └── CHN-001/
│       ├── official-website-001.md
│       ├── github-001.md
│       └── news-001.md
├── portfolios/
│   └── CHN-001.json
├── report.md
├── leads.json
├── failed.json
└── events.jsonl

Preserving raw Markdown allows you to:

• Re-run the analyst without crawling again.
• Audit every claim.
• Change the extraction schema later.
• Recover from an LLM failure.
• Prove where each field originated.
• Avoid repeated page requests.

14. Structured extraction strategy

Use deterministic extraction first whenever possible.

Use JsonCssExtractionStrategy when multiple pages share a stable HTML structure:

from crawl4ai import JsonCssExtractionStrategy


schema = {
    "name": "Business Records",
    "baseSelector": ".business-card",
    "fields": [
        {
            "name": "business_name",
            "selector": "h2",
            "type": "text"
        },
        {
            "name": "website",
            "selector": "a.website",
            "type": "attribute",
            "attribute": "href"
        },
        {
            "name": "phone",
            "selector": ".phone",
            "type": "text"
        }
    ]
}

strategy = JsonCssExtractionStrategy(schema)

For heterogeneous company websites, use clean Markdown followed by one Analyst Agent call per consolidated lead.

Do not call an LLM separately for every page unless the page genuinely requires semantic extraction.

15. Analyst Agent contract

The Analyst Agent receives the best evidence pages for one lead.

Limit the context:

• Sort by relevance score.
• Keep the best 10–15 pages.
• Limit each Markdown document to a reasonable size.
• Keep source URLs attached.
• Do not merge evidence without preserving citations.

Analyst system instruction:

You are the Hermes Evidence Analyst.

Build a lead portfolio using only the supplied documents.

Rules:

1. Never invent a person, role, phone, email, company fact, or date.
2. Every claim must include its source URL and an exact supporting quote.
3. A Google Maps or directory phone is a business contact, not an
   owner-direct contact.
4. A registered agent, manager, employee, license holder, or director
   is not automatically the current owner.
5. Use null when evidence is missing.
6. Keep conflicting claims instead of silently choosing one.
7. Mark ownership and growth claims as material.
8. Return JSON only.

Return:

{
  "business_name": "",
  "domain": "",
  "summary": "",
  "category": "",
  "claims": [
    {
      "field": "owner_name",
      "value": "",
      "linkage": "owner",
      "material": true,
      "evidence": [
        {
          "source_url": "",
          "quote": ""
        }
      ]
    }
  ],
  "public_business_contacts": {
    "emails": [],
    "phones": [],
    "websites": []
  },
  "warnings": []
}

The Analyst Agent suggests claims. It does not make the final verification decision.

16. Verification Agent

The Verification Agent must be deterministic.

For every claim:

1. Confirm that the source URL was actually crawled.
2. Confirm that the quoted text exists in the stored Markdown.
3. Count independent source domains.
4. Require two independent sources for material ownership claims.
5. Keep basic first-party identity fields with one authoritative source.
6. Reject evidence from missing or failed pages.
7. Preserve conflicting values.
8. Compute the final classification after validation.

Classification rules:

OWNER_DIRECT_VERIFIED
Current owner is proven, and a contact is explicitly linked to that person.

OWNER_ASSOCIATED_VERIFIED
Current owner is proven, but only an official business contact is available.

OWNER_CANDIDATE
A plausible owner is named, but the evidence chain is incomplete.

BUSINESS_ONLY
The business and public business contact are known, but ownership is not proven.

REJECTED
Chain, corporate branch, inactive business, contradictory identity, or invalid evidence.

Claim chain:

business location
    ↓
business identity
    ↓
named person
    ↓
current ownership role
    ↓
contact linkage

Never collapse these into a single unverified assumption.

17. Verification logic

Create hermes_swarm/verifier.py:

import re
from urllib.parse import urlparse

from .models import Claim, LeadPortfolio, PageEvidence


def normalize(value: str) -> str:
    return re.sub(r"\s+", " ", value or "").strip().lower()


def source_group(url: str) -> str:
    return (urlparse(url).hostname or "").lower().removeprefix("www.")


def verify_claim(
    claim: Claim,
    pages_by_url: dict[str, PageEvidence],
    minimum_independent_sources: int,
) -> Claim:
    valid_evidence = []

    for citation in claim.evidence:
        page = pages_by_url.get(citation.source_url)

        if not page:
            continue

        if normalize(citation.quote) not in normalize(page.markdown):
            continue

        valid_evidence.append(citation)

    claim.evidence = valid_evidence

    groups = {
        source_group(item.source_url)
        for item in valid_evidence
    }

    required = (
        minimum_independent_sources
        if claim.material
        else 1
    )

    claim.verified = len(groups) >= required

    if claim.verified:
        claim.verification_reason = (
            f"Supported by {len(groups)} independent source groups."
        )
    else:
        claim.verification_reason = (
            f"Needs {required} independent source groups; "
            f"found {len(groups)}."
        )

    return claim


def classify_portfolio(portfolio: LeadPortfolio) -> LeadPortfolio:
    owner_verified = any(
        claim.field == "owner_name" and claim.verified
        for claim in portfolio.claims
    )

    owner_candidate = any(
        claim.field == "owner_name" and claim.evidence
        for claim in portfolio.claims
    )

    direct_contact = any(
        claim.linkage == "owner_direct" and claim.verified
        for claim in portfolio.claims
    )

    has_business_contact = any(
        portfolio.public_business_contacts.get(field)
        for field in ("emails", "phones", "websites")
    )

    if owner_verified and direct_contact:
        portfolio.classification = "OWNER_DIRECT_VERIFIED"
        portfolio.confidence = 0.95

    elif owner_verified and has_business_contact:
        portfolio.classification = "OWNER_ASSOCIATED_VERIFIED"
        portfolio.confidence = 0.90

    elif owner_candidate:
        portfolio.classification = "OWNER_CANDIDATE"
        portfolio.confidence = 0.65

    else:
        portfolio.classification = "BUSINESS_ONLY"
        portfolio.confidence = 0.50

    return portfolio

18. Hermes queue model

Use durable queues so the workflow can resume:

{
  "candidate_queue": [],
  "crawl_queue": [],
  "evidence_queue": [],
  "analysis_queue": [],
  "verification_queue": [],
  "portfolio_queue": [],
  "failed_queue": [],
  "repair_queue": []
}

Queue transitions:

candidate_queue
    → crawl_queue
    → evidence_queue
    → analysis_queue
    → verification_queue
    → portfolio_queue

Failures go to:

failed_queue
    → repair_queue
    → original queue

Write queue state atomically:

import json
from pathlib import Path


def atomic_json_write(path: Path, payload: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)

    temporary = path.with_suffix(path.suffix + ".tmp")
    temporary.write_text(
        json.dumps(payload, indent=2, ensure_ascii=False),
        encoding="utf-8",
    )
    temporary.replace(path)

19. Orchestration sequence

The Hermes orchestrator should execute these stages:

async def execute_campaign(campaign, leads):
    # 1. Source Finder Agent
    source_tasks = []
    for lead in leads:
        discovered_urls = await hermes_web_search_for_lead(lead, campaign)
        source_tasks.extend(
            build_source_plan(
                lead,
                campaign,
                discovered_urls,
            )
        )

    # 2. Parallel Crawl Agents
    pages, failures = await run_crawler_swarm(
        source_tasks,
        campaign,
    )

    # 3. Group evidence by lead
    pages_by_lead = {}
    for page in pages:
        pages_by_lead.setdefault(page.lead_id, []).append(page)

    portfolios = []

    # 4. Analyst and Verification Agents
    for lead in leads:
        evidence = pages_by_lead.get(lead.lead_id, [])

        analysis = await analyst_agent(
            lead=lead,
            evidence=evidence,
        )

        portfolio = build_portfolio(
            lead=lead,
            analysis=analysis,
            evidence=evidence,
        )

        page_lookup = {
            page.source_url: page
            for page in evidence
        }

        portfolio.claims = [
            verify_claim(
                claim,
                page_lookup,
                campaign.minimum_independent_sources,
            )
            for claim in portfolio.claims
        ]

        portfolio = classify_portfolio(portfolio)
        portfolios.append(portfolio)

    # 5. Storage Agent
    await write_run_outputs(
        campaign=campaign,
        pages=pages,
        portfolios=portfolios,
        failures=failures,
    )

    return portfolios

hermes_web_search_for_lead, analyst_agent, build_portfolio, and write_run_outputs should be implemented as separate Hermes tools or role-specific agents.

20. Storage outputs

Write two primary deliverables:

leads.json

{
  "run_id": "run_20260710_094102",
  "campaign": "Chennai technology-business research",
  "records": [
    {
      "lead_id": "CHN-001",
      "business_name": "Example Technology Company",
      "classification": "OWNER_ASSOCIATED_VERIFIED",
      "confidence": 0.91,
      "owner": {
        "name": "Example Founder",
        "role": "Founder"
      },
      "public_business_contacts": {
        "emails": ["hello@example.com"],
        "phones": ["+91 ..."],
        "websites": ["https://example.com"]
      },
      "claims": [],
      "sources": []
    }
  ]
}

report.md

# Hermes Crawl4AI Lead Research Report

## CHN-001 · Example Technology Company

- Classification: OWNER_ASSOCIATED_VERIFIED
- Confidence: 0.91
- Domain: example.com
- Owner: Example Founder
- Public business contact: hello@example.com

### Verified claims

1. Example Founder is identified as the company founder.
   - Source 1: https://example.com/about
   - Source 2: https://independent-source.example/interview

### Unresolved claims

- Current employee count could not be independently confirmed.

21. Run the workflow

Create run.py:

import asyncio
import json
from pathlib import Path

from hermes_swarm.models import Campaign, LeadSeed
from hermes_swarm.orchestrator import execute_campaign


async def main():
    campaign = Campaign.model_validate_json(
        Path("campaign.json").read_text(encoding="utf-8")
    )

    leads = [
        LeadSeed.model_validate(item)
        for item in json.loads(
            Path("seeds.json").read_text(encoding="utf-8")
        )
    ]

    portfolios = await execute_campaign(campaign, leads)

    print(
        f"Completed {len(portfolios)} lead portfolios."
    )


if __name__ == "__main__":
    asyncio.run(main())

Run:

python run.py

Expected output:

Source Finder Agent: source plan created
Crawler Worker 01: official website started
Crawler Worker 02: GitHub source started
Crawler Worker 03: independent source started
Evidence Normalizer: documents processed
Analyst Agent: portfolio assembled
Verification Agent: claims checked
Storage Agent: report.md and leads.json written
Completed 10 lead portfolios

22. Testing checklist

Before distributing the tutorial, run these checks.

Installation

crawl4ai-doctor

Single-page crawl

Confirm a public page produces Markdown.

JavaScript-heavy page

Confirm content that appears after rendering is present in the result.

Recursive crawl

Confirm:

• Depth does not exceed max_depth.
• Page count does not exceed max_pages.
• External domains are not followed.
• Robots.txt is respected.

Parallel workers

Confirm at least two source workers can run simultaneously without exhausting system memory.

Evidence preservation

Confirm every portfolio claim contains:

• Source URL
• Exact supporting quote
• Independent source count
• Verification result

Failure handling

Test:

• Invalid URL
• Timeout
• Robots.txt rejection
• Empty Markdown
• Browser crash
• LLM failure
• Invalid analyst JSON
• Contradictory ownership evidence

Resume behavior

Stop a run after crawling and verify that analysis can restart from stored Markdown without crawling again.

23. Production controls

Add these before using the workflow at scale:

• Per-domain rate limits
• Retry limits
• Exponential backoff
• Memory-based worker limits
• Run cancellation
• Queue persistence
• Atomic state writes
• Structured logs
• Source-domain allow/block lists
• Duplicate URL detection
• Content hashing
• Maximum Markdown size
• Analyst context limits
• Human review for uncertain owners
• Secret management
• Audit trail
• Data-retention policy

Do not bypass authentication, CAPTCHAs, access controls, or private pages.

24. Troubleshooting

Browser executable missing

crawl4ai-setup
python -m playwright install chromium
crawl4ai-doctor

Empty Markdown

Check:

• result.success
• result.error_message
• status_code
• wait_until
• delay_before_return_html
• wait_for
• scan_full_page
• Shadow DOM settings

Too many irrelevant pages

Reduce:

{
  "max_depth": 1,
  "max_pages_per_domain": 10
}

Improve KeywordRelevanceScorer keywords.

High memory use

Reduce:

HERMES_MAX_WORKERS=2

Avoid launching one unrestricted browser for every URL.

Owner details are missing

This is not automatically an extraction failure. The public sources may not prove ownership. Keep the record as BUSINESS_ONLY or OWNER_CANDIDATE.

Conflicting owners

Preserve both claims and send the record to human review. Do not allow the Analyst Agent to choose without evidence.

25. Final acceptance criteria

The setup is complete only when:

• Crawl4AI installs successfully.
• Chromium launches successfully.
• The Source Finder produces a source plan.
• Multiple crawler workers run concurrently.
• JavaScript-rendered pages produce Markdown.
• Raw Markdown is saved before analysis.
• Structured JSON is generated.
• Every material claim contains evidence.
• Verification rejects unsupported quotes.
• Owner classifications follow deterministic rules.
• report.md is produced.
• leads.json is produced.
• Failed sources are recorded.
• A stopped run can resume from stored evidence.

26. Reference links

• Crawl4AI repository:
  https://github.com/unclecode/crawl4ai

• Crawl4AI documentation:
  https://docs.crawl4ai.com/

• Quick start:
  https://docs.crawl4ai.com/core/quickstart/

• Deep crawling:
  https://docs.crawl4ai.com/core/deep-crawling/

• Parallel crawling:
  https://docs.crawl4ai.com/advanced/multi-url-crawling/

• Configuration parameters:
  https://docs.crawl4ai.com/api/parameters/

• Markdown generation:
  https://docs.crawl4ai.com/core/markdown-generation/

• Fit Markdown:
  https://docs.crawl4ai.com/core/fit-markdown/

• Structured extraction:
  https://docs.crawl4ai.com/extraction/no-llm-strategies/

27. Reader’s final result

After completing this tutorial, the reader should have a working Hermes agent swarm that:

• Finds research sources.
• Crawls multiple source domains concurrently.
• Recursively explores relevant pages.
• Renders JavaScript-heavy websites.
• Converts pages into clean Markdown.
• Produces structured JSON.
• Builds evidence-backed lead portfolios.
• Verifies ownership and contact claims.
• Saves results into a knowledge base or automation workflow.