Crawl4AI: AI-Optimized Web Crawling for Data Pipelines

When building modern AI applications, especially those based on Retrieval-Augmented Generation (RAG), one of the biggest challenges is getting high-quality data from the web. Most traditional web scrapers produce messy raw HTML that is difficult for LLMs to process.

Crawl4AI is an open-source solution designed specifically for AI needs. It produces output as clean markdown and structured data ready to be used directly in your AI pipeline.

What Is Crawl4AI?

Crawl4AI is an open-source Python library for web crawling optimized for AI and data pipelines. Its key features include:

• Clean Markdown Output: Automatically converts web pages into clean markdown
• Async Crawling: Asynchronous support for fast parallel crawling
• LLM-Based Extraction: Structured data extraction using LLMs
• CSS-Based Extraction: Data extraction using CSS selectors without LLM
• JavaScript Rendering: Handles JavaScript-rendered pages
• Chunking Strategies: Various strategies for splitting content into optimal chunks
• Session Management: Manages cookies and sessions for complex crawling

Installation

Basic Installation

pip install crawl4ai

Browser Setup (Playwright)

Crawl4AI uses Playwright for web page rendering. After installation, run the browser setup:

# Install Chromium browser
crawl4ai-setup

Or manually via playwright
playwright install chromium

Installation with Full Dependencies

# With all features support
pip install "crawl4ai[all]"

Verify Installation

import crawl4ai
print(f"Crawl4AI version: {crawl4ai.version}")

Basic Crawling

Simple Page Crawling

import asyncio
from crawl4ai import AsyncWebCrawler

async def basiccrawl():

    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(url="https://example.com")

        # Crawling status
        print(f"Status: {result.success}")
        print(f"Status Code: {result.statuscode}")

        # Content in markdown
        print(f"Markdown length: {len(result.markdown)}")
        print(result.markdown[:500])

        # Clean markdown (without navigation, footer, etc.)
        print(f"\nClean markdown length: {len(result.cleanedhtml)}")


Run
asyncio.run(basiccrawl())

Using Fit Markdown

Crawl4AI provides fitmarkdown which is a filtered version of markdown without unimportant elements:

async def crawlwithfitmarkdown():
    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(url="https://docs.python.org/3/tutorial/")

        # Raw markdown - includes navigation, sidebar, etc.
        print(f"Raw markdown: {len(result.markdown)} chars")

        # Fit markdown - main content only
        print(f"Fit markdown: {len(result.fitmarkdown)} chars")


        # Use fitmarkdown for LLM input
        print(result.fitmarkdown[:500])


asyncio.run(crawlwithfitmarkdown())

Crawler Configuration

from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, BrowserConfig

async def configuredcrawl():

    # Browser configuration
    browserconfig = BrowserConfig(
        headless=True,
        browsertype="chromium",

        verbose=True
    )

    # Crawling configuration
    runconfig = CrawlerRunConfig(
        wordcountthreshold=10,       # Minimum words per block
        excludeexternallinks=True,   # Ignore external links
        removeoverlayelements=True,  # Remove popups/overlays
        processiframes=False          # Ignore iframes

    )

    async with AsyncWebCrawler(config=browserconfig) as crawler:
        result = await crawler.arun(
            url="https://example.com/blog/article",
            config=runconfig

        )

        print(result.fitmarkdown)

asyncio.run(configuredcrawl())

Async Crawling: Parallel Crawling

One of Crawl4AI's strengths is its async crawling capability that allows you to crawl many pages in parallel.

Crawling Multiple URLs

import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig

async def crawlmultiple():
    urls = [
        "https://docs.python.org/3/tutorial/introduction.html",
        "https://docs.python.org/3/tutorial/controlflow.html",
        "https://docs.python.org/3/tutorial/datastructures.html",
        "https://docs.python.org/3/tutorial/modules.html",
        "https://docs.python.org/3/tutorial/inputoutput.html",
    ]

    async with AsyncWebCrawler() as crawler:
        # Crawl all URLs in parallel
        results = await crawler.arunmany(

            urls=urls,
            config=CrawlerRunConfig(
                wordcountthreshold=10

            )
        )

        for result in results:
            if result.success:
                print(f"OK: {result.url} ({len(result.markdown)} chars)")
            else:
                print(f"FAIL: {result.url} - {result.errormessage}")

asyncio.run(crawlmultiple())

Crawling with Rate Limiting

async def crawlwithratelimit():
    urls = [f"https://example.com/page/{i}" for i in range(1, 21)]

    async with AsyncWebCrawler() as crawler:
        results = await crawler.arunmany(

            urls=urls,
            config=CrawlerRunConfig(
                wordcountthreshold=10,

                pagetimeout=30000,    # 30 second timeout per page
            )
        )

        successful = [r for r in results if r.success]
        print(f"Successful: {len(successful)}/{len(urls)}")

asyncio.run(crawlwithratelimit())

LLM-Based Extraction

Crawl4AI can use LLMs to intelligently extract structured data from web pages.

Extraction with Pydantic Schema

import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
from crawl4ai.extractionstrategy import LLMExtractionStrategy
from pydantic import BaseModel
from typing import List, Optional

Define output schema
class ProductInfo(BaseModel):
    name: str
    price: Optional[str]
    description: str
    features: List[str]
    rating: Optional[float]

async def extractproductinfo():
    extractionstrategy = LLMExtractionStrategy(

        provider="openai/gpt-4o-mini",
        apitoken="sk-...",
        schema=ProductInfo.modeljsonschema(),
        extractiontype="schema",

        instruction="Extract product information from the page. "
                    "Include name, price, description, key features, and rating."
    )

    runconfig = CrawlerRunConfig(
        extractionstrategy=extractionstrategy
    )

    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url="https://example.com/product/widget-pro",
            config=runconfig

        )

        if result.success:
            import json
            products = json.loads(result.extractedcontent)
            for product in products:
                print(f"Name: {product['name']}")
                print(f"Price: {product['price']}")
                print(f"Description: {product['description']}")
                print(f"Features: {', '.join(product['features'])}")
                print(f"Rating: {product['rating']}")

asyncio.run(extractproductinfo())

News Article Extraction

class NewsArticle(BaseModel):
    title: str
    author: Optional[str]
    publisheddate: Optional[str]

    summary: str
    maintopics: List[str]
    keyquotes: List[str]


async def extractnews():
    extractionstrategy = LLMExtractionStrategy(

        provider="openai/gpt-4o-mini",
        apitoken="sk-...",
        schema=NewsArticle.modeljsonschema(),
        extractiontype="schema",

        instruction="Extract the news article information. "
                    "Identify the title, author, date, a brief summary, "
                    "main topics discussed, and key quotes from the article."
    )

    runconfig = CrawlerRunConfig(
        extractionstrategy=extractionstrategy
    )

    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url="https://example.com/news/ai-breakthrough",
            config=runconfig

        )

        if result.success:
            import json
            articles = json.loads(result.extractedcontent)
            for article in articles:
                print(f"Title: {article['title']}")
                print(f"Author: {article['author']}")
                print(f"Summary: {article['summary']}")

asyncio.run(extractnews())

CSS-Based Extraction

For simpler cases where HTML structure is consistent, use CSS-based extraction which is faster and doesn't require an LLM.

Extraction with CSS Selectors

from crawl4ai.extractionstrategy import JsonCssExtractionStrategy
import json

async def cssextraction():

    # Define CSS schema
    schema = {
        "name": "Blog Posts",
        "baseSelector": "article.post",
        "fields": [
            {
                "name": "title",
                "selector": "h2.post-title",
                "type": "text"
            },
            {
                "name": "link",
                "selector": "h2.post-title a",
                "type": "attribute",
                "attribute": "href"
            },
            {
                "name": "date",
                "selector": "span.post-date",
                "type": "text"
            },
            {
                "name": "excerpt",
                "selector": "div.post-excerpt",
                "type": "text"
            },
            {
                "name": "author",
                "selector": "span.post-author",
                "type": "text"
            },
            {
                "name": "tags",
                "selector": "div.post-tags span",
                "type": "list",
                "fields": [
                    {"name": "tag", "type": "text"}
                ]
            }
        ]
    }

    extractionstrategy = JsonCssExtractionStrategy(schema)

    runconfig = CrawlerRunConfig(

        extractionstrategy=extractionstrategy

    )

    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url="https://example.com/blog",
            config=runconfig
        )

        if result.success:
            posts = json.loads(result.extractedcontent)

            for post in posts:
                print(f"Title: {post['title']}")
                print(f"Date: {post['date']}")
                print(f"Excerpt: {post['excerpt'][:100]}...")
                print()

asyncio.run(cssextraction())

Data Table Extraction

async def extracttable():
    schema = {
        "name": "Data Table",
        "baseSelector": "table.data-table tbody tr",
        "fields": [
            {"name": "rank", "selector": "td:nth-child(1)", "type": "text"},
            {"name": "name", "selector": "td:nth-child(2)", "type": "text"},
            {"name": "value", "selector": "td:nth-child(3)", "type": "text"},
            {"name": "change", "selector": "td:nth-child(4)", "type": "text"}
        ]
    }

    extractionstrategy = JsonCssExtractionStrategy(schema)
    runconfig = CrawlerRunConfig(extractionstrategy=extractionstrategy)


    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url="https://example.com/data-table",
            config=runconfig
        )

        if result.success:
            import json
            rows = json.loads(result.extractedcontent)

            for row in rows:
                print(f"#{row['rank']} {row['name']}: {row['value']}")

asyncio.run(extracttable())

Chunking Strategies

Crawl4AI provides various chunking strategies to split content into optimal pieces for AI processing.

RegexChunking

from crawl4ai.chunkingstrategy import RegexChunking

Chunk based on regex patterns (default: paragraphs)
chunker = RegexChunking(
    patterns=["\n\n"]  # Split by double newline
)

async def crawlwithchunking():

    runconfig = CrawlerRunConfig(
        chunkingstrategy=chunker

    )

    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url="https://example.com/long-article",
            config=runconfig
        )

        if result.success:
            # Access chunks
            for i, chunk in enumerate(result.markdown.split("\n\n")):
                if chunk.strip():
                    print(f"Chunk {i+1}: {chunk[:100]}...")

asyncio.run(crawlwithchunking())

FixedLengthChunking

from crawl4ai.chunkingstrategy import FixedLengthWordChunking

Chunk based on word count
chunker = FixedLengthWordChunking(
    chunksize=200,    # 200 words per chunk
    overlap=20         # 20 words overlap between chunks
)

async def fixedlengthcrawl():
    runconfig = CrawlerRunConfig(

        chunkingstrategy=chunker
    )

    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url="https://example.com/article",
            config=runconfig

        )

        print(f"Total content: {len(result.markdown)} chars")

asyncio.run(fixedlengthcrawl())

Handling JavaScript-Rendered Pages

Many modern websites use JavaScript to render content. Crawl4AI handles this well through Playwright.

Wait for Dynamic Content

async def crawljspage():
    runconfig = CrawlerRunConfig(
        waitfor="css:.dynamic-content",   # Wait until selector appears

        pagetimeout=60000,                # 60 second timeout
        delaybeforereturnhtml=2.0       # Wait 2 seconds after load

    )

    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url="https://example.com/spa-app",
            config=runconfig
        )

        if result.success:
            print(f"Content loaded: {len(result.markdown)} chars")
            print(result.fitmarkdown[:500])


asyncio.run(crawljspage())

Custom JavaScript Execution

async def crawlwithjs():
    # JavaScript executed on the page
    jscode = [
        # Click "Load More" button
        "document.querySelector('button.load-more')?.click();",
        # Scroll to bottom
        "window.scrollTo(0, document.body.scrollHeight);",
    ]

    runconfig = CrawlerRunConfig(

        jscode=jscode,

        delaybeforereturnhtml=3.0,  # Wait for content to load
        waitfor="css:.loaded-content"

    )

    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url="https://example.com/infinite-scroll",
            config=runconfig
        )

        if result.success:
            print(f"Content after JS execution: {len(result.markdown)} chars")

asyncio.run(crawlwithjs())

Handling Login/Authentication

async def crawlauthenticated():
    # JavaScript for login
    jslogin = [
        "document.querySelector('#email').value = 'user@example.com';",
        "document.querySelector('#password').value = 'password123';",
        "document.querySelector('form#login').submit();",
    ]

    browserconfig = BrowserConfig(

        headless=True
    )

    # Step 1: Login first
    loginconfig = CrawlerRunConfig(
        jscode=jslogin,
        delaybeforereturnhtml=3.0,

        waitfor="css:.dashboard"
    )

    async with AsyncWebCrawler(config=browserconfig) as crawler:

        # Login
        await crawler.arun(
            url="https://example.com/login",
            config=loginconfig
        )

        # Step 2: Crawl authenticated pages (session remains active)
        result = await crawler.arun(
            url="https://example.com/dashboard/reports"
        )

        if result.success:
            print(result.fitmarkdown[:500])


asyncio.run(crawlauthenticated())

Practical Example: RAG Data Pipeline

Let's build a complete pipeline to collect data from the web and prepare it for a RAG system.

"""
RAG Data Pipeline using Crawl4AI
Collect documentation from the web and prepare for vector store
"""
import asyncio
import json
import hashlib
from datetime import datetime
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, BrowserConfig
from crawl4ai.chunkingstrategy import FixedLengthWordChunking


Configuration
URLS = [
    "https://docs.python.org/3/tutorial/introduction.html",
    "https://docs.python.org/3/tutorial/controlflow.html",
    "https://docs.python.org/3/tutorial/datastructures.html",
    "https://docs.python.org/3/tutorial/modules.html",
    "https://docs.python.org/3/tutorial/errors.html",
]

CHUNKSIZE = 300       # words per chunk
CHUNKOVERLAP = 30     # word overlap

OUTPUTFILE = "ragdocuments.json"



def createdocumentid(url: str, chunkindex: int) -> str:
    """Create a unique ID for each document chunk."""
    raw = f"{url}:{chunkindex}"

    return hashlib.md5(raw.encode()).hexdigest()


def chunktext(text: str, chunksize: int, overlap: int) -> list:

    """Split text into chunks based on word count."""
    words = text.split()
    chunks = []

    for i in range(0, len(words), chunksize - overlap):
        chunk = " ".join(words[i:i + chunksize])

        if len(chunk.split()) >= 20:  # Minimum 20 words
            chunks.append(chunk)

    return chunks


async def crawlandprocess():

    """Crawl URLs and process into RAG documents."""
    browserconfig = BrowserConfig(
        headless=True,
        verbose=False
    )

    runconfig = CrawlerRunConfig(

        wordcountthreshold=10,

        excludeexternallinks=True,

        removeoverlayelements=True

    )

    documents = []
    stats = {
        "totalurls": len(URLS),
        "successful": 0,
        "failed": 0,
        "totalchunks": 0,

        "totalwords": 0
    }

    async with AsyncWebCrawler(config=browserconfig) as crawler:

        print(f"Starting crawl of {len(URLS)} URLs...")
        print("="  60)


        results = await crawler.arunmany(
            urls=URLS,
            config=runconfig

        )

        for result in results:
            if not result.success:
                print(f"FAILED: {result.url} - {result.errormessage}")
                stats["failed"] += 1
                continue

            stats["successful"] += 1

            # Use fitmarkdown for clean content

            content = result.fitmarkdown or result.markdown
            wordcount = len(content.split())

            stats["totalwords"] += wordcount


            print(f"OK: {result.url}")
            print(f"   Words: {wordcount}")

            # Chunk content
            chunks = chunktext(content, CHUNKSIZE, CHUNKOVERLAP)

            stats["totalchunks"] += len(chunks)

            print(f"   Chunks: {len(chunks)}")

            for idx, chunk in enumerate(chunks):
                doc = {
                    "id": createdocumentid(result.url, idx),
                    "sourceurl": result.url,

                    "chunkindex": idx,
                    "totalchunks": len(chunks),

                    "content": chunk,
                    "wordcount": len(chunk.split()),
                    "metadata": {
                        "crawledat": datetime.now().isoformat(),

                        "statuscode": result.statuscode,

                        "source": "python-docs"
                    }
                }
                documents.append(doc)

    # Save results
    with open(OUTPUTFILE, "w", encoding="utf-8") as f:
        json.dump(documents, f, ensureascii=False, indent=2)


    print("\n" + "="  60)
    print("CRAWLING STATISTICS:")
    print(f"  URLs successful: {stats['successful']}/{stats['totalurls']}")
    print(f"  URLs failed: {stats['failed']}/{stats['totalurls']}")

    print(f"  Total words: {stats['totalwords']:,}")
    print(f"  Total chunks: {stats['totalchunks']}")

    print(f"  Output: {OUTPUTFILE}")

    return documents


async def loadtovectorstore(documents: list):

    """Load documents to vector store (example with ChromaDB)."""
    try:
        import chromadb
        from chromadb.utils import embeddingfunctions

        # Initialize ChromaDB
        client = chromadb.PersistentClient(path="./chromadb")


        # Use embedding function
        embeddingfn = embeddingfunctions.SentenceTransformerEmbeddingFunction(

            modelname="all-MiniLM-L6-v2"
        )

        # Create or get collection
        collection = client.getorcreatecollection(

            name="pythondocs",
            embeddingfunction=embeddingfn,
            metadata={"description": "Python documentation for RAG"}
        )

        # Batch insert
        batchsize = 50

        for i in range(0, len(documents), batchsize):
            batch = documents[i:i + batchsize]


            collection.add(
                ids=[doc["id"] for doc in batch],
                documents=[doc["content"] for doc in batch],
                metadatas=[{
                    "sourceurl": doc["sourceurl"],

                    "chunkindex": doc["chunkindex"],

                    "wordcount": doc["wordcount"]

                } for doc in batch]
            )

        print(f"\nLoaded {len(documents)} chunks to ChromaDB")
        print(f"Collection size: {collection.count()}")

        # Test query
        results = collection.query(
            querytexts=["How to handle errors in Python?"],
            nresults=3

        )

        print("\nTest Query: 'How to handle errors in Python?'")
        for i, doc in enumerate(results["documents"][0]):
            print(f"\n--- Result {i+1} ---")
            print(doc[:200] + "...")

    except ImportError:
        print("\nChromaDB is not installed. Install with:")
        print("  pip install chromadb sentence-transformers")
        print("\nDocuments are saved in JSON file, "
              "ready to be loaded into your vector store of choice.")


async def main():
    """Main pipeline."""
    # Step 1: Crawl and process
    documents = await crawlandprocess()


    # Step 2: Load to vector store
    await loadtovectorstore(documents)

    print("\nPipeline complete!")


if name == "main":
    asyncio.run(main())

Running the Pipeline

# Install dependencies
pip install crawl4ai chromadb sentence-transformers

Setup browser
crawl4ai-setup

Run pipeline
python ragpipeline.py

Integration with LangChain for RAG

"""
Using Crawl4AI data with LangChain RAG
"""
from langchaincommunity.vectorstores import Chroma
from langchainopenai import OpenAIEmbeddings, ChatOpenAI

from langchain.prompts import ChatPromptTemplate
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.outputparser import StrOutputParser

Load vector store populated by Crawl4AI pipeline
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
    persistdirectory="./chromadb",
    collectionname="pythondocs",
    embeddingfunction=embeddings

)

retriever = vectorstore.asretriever(
    searchtype="similarity",

    searchkwargs={"k": 5}
)

Build RAG chain
prompt = ChatPromptTemplate.frommessages([

    ("system", "Answer the question based on the following context. "
               "If there is no relevant information, say 'I could not find "
               "that information in the documentation.'\n\nContext:\n{context}"),
    ("human", "{question}")
])

llm = ChatOpenAI(model="gpt-4o", temperature=0)

ragchain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

Query
response = ragchain.invoke("How do I use list comprehensions in Python?")

print(response)

Tips and Best Practices

1. Respect robots.txt

async def respectfulcrawl():
    browserconfig = BrowserConfig(headless=True)


    # Always check robots.txt first
    runconfig = CrawlerRunConfig(
        wordcountthreshold=10
    )

    async with AsyncWebCrawler(config=browserconfig) as crawler:

        result = await crawler.arun(
            url="https://example.com/page",
            config=runconfig
        )

2. Implement Retry Logic

async def crawlwithretry(crawler, url, maxretries=3):
    """Crawl with retry logic."""
    for attempt in range(maxretries):
        try:
            result = await crawler.arun(url=url)
            if result.success:
                return result
            print(f"Attempt {attempt + 1} failed: {result.errormessage}")

        except Exception as e:
            print(f"Attempt {attempt + 1} error: {e}")

        if attempt < maxretries - 1:
            await asyncio.sleep(2 ** attempt)  # Exponential backoff

    return None

3. Cache Crawling Results

import os
import json
import hashlib

CACHEDIR = "./crawlcache"

def getcachepath(url: str) -> str:
    urlhash = hashlib.md5(url.encode()).hexdigest()

    return os.path.join(CACHEDIR, f"{urlhash}.json")


async def cachedcrawl(crawler, url):
    """Crawl with local caching."""
    cachepath = getcachepath(url)


    # Check cache
    if os.path.exists(cachepath):
        with open(cachepath, "r") as f:

            cached = json.load(f)
        print(f"Cache hit: {url}")
        return cached

    # Crawl
    result = await crawler.arun(url=url)

    if result.success:
        # Save to cache
        os.makedirs(CACHEDIR, existok=True)

        cachedata = {
            "url": url,
            "markdown": result.markdown,
            "fitmarkdown": result.fitmarkdown,
            "crawledat": datetime.now().isoformat()

        }
        with open(cachepath, "w") as f:
            json.dump(cachedata, f)


    return result

4. Monitoring and Logging

import logging
import time

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

async def monitoredcrawl(urls: list):

    """Crawl with monitoring and logging."""
    starttime = time.time()

    async with AsyncWebCrawler() as crawler:
        results = await crawler.arunmany(urls=urls)


    elapsed = time.time() - starttime
    successful = sum(1 for r in results if r.success)

    logger.info(f"Crawling completed in {elapsed:.2f}s")
    logger.info(f"Successful: {successful}/{len(urls)}")
    logger.info(f"Rate: {len(urls)/elapsed:.1f} URLs/second")

    return results

Conclusion

Crawl4AI is a powerful tool for collecting web data and preparing it for AI pipelines. With clean markdown output, LLM-based extraction, and async crawling features, Crawl4AI simplifies the data collection process that is usually time-consuming and complex.

Keys to effective Crawl4AI usage:

• Use fit_markdown to get clean content without noise
• Leverage async crawling for optimal performance
• Choose CSS extraction for consistent structures, LLM extraction for dynamic ones
• Implement caching and retry logic for production
• Always respect robots.txt and rate limit target websites

With Crawl4AI, you can build robust data pipelines for RAG applications, fine-tuning datasets, or other web content analysis.

Happy crawling and we hope this tutorial is helpful!