crawl4ai·mastery

Observable Production Pipelines

A crawler that works on your laptop can still fail silently in production. Make failures visible.

logging retry policy cost tracking result validation dispatch metrics
01/concept

Concept

Production pipelines need more than correct code. Add structured logging around every crawl, retry with exponential backoff, validate results before forwarding them, and track costs when LLMs are involved. Use dispatch_result from batch runs to monitor memory and duration per task.

Validate extracted content before trusting it: check that JSON parses, that required keys exist, and that counts are within expected ranges. Log failures with context (URL, config, error message) so you can replay or debug later. Keep API tokens and proxy credentials in environment variables, and rotate them through a secret manager when possible.

02/use-cases

Use cases

  • Monitor a nightly crawl job and alert when success rate drops below 95%.
  • Track LLM token spend per extraction pipeline.
03/watch

Video walkthrough

Official Crawl4AI tutorialyoutube · loads on click
04/run

Code example

python Logging, validation, and retry wrapper 
import asyncio
import json
import logging
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("crawl_pipeline")

async def safe_crawl(crawler, url: str, config: CrawlerRunConfig, retries: int = 2):
    for attempt in range(retries + 1):
        result = await crawler.arun(url=url, config=config)
        if result.success:
            if config.extraction_strategy and result.extracted_content:
                try:
                    data = json.loads(result.extracted_content)
                    logger.info(f"{url}: extracted {len(data)} items")
                    return data
                except json.JSONDecodeError:
                    logger.warning(f"{url}: invalid JSON")
                    continue
            return result
        logger.warning(f"{url}: attempt {attempt + 1} failed: {result.error_message}")
        await asyncio.sleep(2 ** attempt)
    logger.error(f"{url}: exhausted retries")
    return None
python Dispatch metrics 
for result in results:
    if result.success and result.dispatch_result:
        dr = result.dispatch_result
        print(f"{result.url}: {dr.memory_usage:.1f} MB, peak {dr.peak_memory:.1f} MB")
        print(f"  duration: {dr.end_time - dr.start_time}")
05/build

Practice lab

Lab 4.4

Practice lab

0/3

Objective: Build a wrapper that logs, retries, and validates extraction results.

Steps

  1. Create a safe_crawl() async function that accepts a crawler, URL, config, and retry count.
  2. On failure, log a warning and retry with exponential backoff.
  3. On success with an extraction strategy, parse result.extracted_content as JSON.
  4. If JSON parsing fails, log it and retry.
  5. Call the wrapper from a small main function and print the final result.

Success criteria

Expected output
Log lines for attempts and a final extracted data structure or None after retries.
Hints
  • Use Python's standard logging module, not print statements.
  • Back off with await asyncio.sleep(2 ** attempt).
Solution
python Complete solution 
import asyncio
import json
import logging
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("lab")

async def safe_crawl(crawler, url: str, config: CrawlerRunConfig, retries: int = 2):
    for attempt in range(retries + 1):
        result = await crawler.arun(url=url, config=config)
        if result.success:
            if config.extraction_strategy and result.extracted_content:
                try:
                    return json.loads(result.extracted_content)
                except json.JSONDecodeError as e:
                    logger.warning(f"{url}: JSON error on attempt {attempt + 1}: {e}")
                    continue
            return result
        logger.warning(f"{url}: attempt {attempt + 1} failed: {result.error_message}")
        await asyncio.sleep(2 ** attempt)
    logger.error(f"{url}: exhausted retries")
    return None

async def main():
    async with AsyncWebCrawler() as crawler:
        data = await safe_crawl(
            crawler,
            "https://example.com",
            CrawlerRunConfig(),
        )
        print(data)

asyncio.run(main())
06/check

Common mistakes

TRAP 01
Running production crawls without logging or retry logic.
TRAP 02
Trusting LLM output without JSON validation.
TRAP 03
Ignoring dispatch_result memory and timing metrics.