Crawl4AI: Web Crawling yang Dioptimalkan untuk Pipeline AI
Dalam membangun aplikasi AI modern, terutama yang berbasis Retrieval-Augmented Generation (RAG), salah satu tantangan terbesar adalah mendapatkan data berkualitas tinggi dari web. Kebanyakan web scraper tradisional menghasilkan HTML mentah yang berantakan dan sulit diproses oleh LLM.
Crawl4AI hadir sebagai solusi open-source yang dirancang khusus untuk kebutuhan AI. Crawl4AI menghasilkan output berupa clean markdown dan structured data yang siap digunakan langsung dalam pipeline AI Anda.Apa Itu Crawl4AI?
Crawl4AI adalah library Python open-source untuk web crawling yang dioptimalkan untuk AI dan data pipeline. Fitur utamanya meliputi:
- Output Clean Markdown: Otomatis mengkonversi halaman web menjadi markdown bersih
- Async Crawling: Dukungan asynchronous untuk crawling paralel yang cepat
- LLM-Based Extraction: Ekstraksi data terstruktur menggunakan LLM
- CSS-Based Extraction: Ekstraksi data menggunakan CSS selector tanpa LLM
- JavaScript Rendering: Menangani halaman yang di-render oleh JavaScript
- Chunking Strategies: Berbagai strategi untuk memecah konten menjadi chunk yang optimal
- Session Management: Mengelola cookies dan session untuk crawling yang kompleks
Instalasi
Instalasi Dasar
pip install crawl4ai
Setup Browser (Playwright)
Crawl4AI menggunakan Playwright untuk rendering halaman web. Setelah instalasi, jalankan setup browser:
# Install browser Chromium
crawl4ai-setup
Atau manual via playwright
playwright install chromium
Instalasi dengan Dependensi Lengkap
# Dengan dukungan semua fitur
pip install "crawl4ai[all]"
Verifikasi Instalasi
import crawl4ai
print(f"Crawl4AI version: {crawl4ai.version}")
Basic Crawling
Crawling Halaman Sederhana
import asyncio
from crawl4ai import AsyncWebCrawler
async def basiccrawl():
async with AsyncWebCrawler() as crawler:
result = await crawler.arun(url="https://example.com")
# Status crawling
print(f"Status: {result.success}")
print(f"Status Code: {result.statuscode}")
# Konten dalam markdown
print(f"Markdown length: {len(result.markdown)}")
print(result.markdown[:500])
# Clean markdown (tanpa navigasi, footer, dll.)
print(f"\nClean markdown length: {len(result.cleanedhtml)}")
Jalankan
asyncio.run(basiccrawl())
Menggunakan Fit Markdown
Crawl4AI menyediakan fitmarkdown yang merupakan versi markdown yang sudah difilter dari elemen-elemen tidak penting:
async def crawlwithfitmarkdown():
async with AsyncWebCrawler() as crawler:
result = await crawler.arun(url="https://docs.python.org/3/tutorial/")
# Markdown mentah - termasuk navigasi, sidebar, dll.
print(f"Raw markdown: {len(result.markdown)} chars")
# Fit markdown - hanya konten utama
print(f"Fit markdown: {len(result.fitmarkdown)} chars")
# Gunakan fitmarkdown untuk input ke LLM
print(result.fitmarkdown[:500])
asyncio.run(crawlwithfitmarkdown())
Konfigurasi Crawler
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, BrowserConfig
async def configuredcrawl():
# Konfigurasi browser
browserconfig = BrowserConfig(
headless=True,
browsertype="chromium",
verbose=True
)
# Konfigurasi crawling
runconfig = CrawlerRunConfig(
wordcountthreshold=10, # Minimum kata per blok
excludeexternallinks=True, # Abaikan link eksternal
removeoverlayelements=True, # Hapus popup/overlay
processiframes=False # Abaikan 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: Crawling Paralel
Salah satu kekuatan Crawl4AI adalah kemampuan async crawling yang memungkinkan Anda meng-crawl banyak halaman secara paralel.
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 semua URL secara paralel
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 dengan 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, # Timeout 30 detik per halaman
)
)
successful = [r for r in results if r.success]
print(f"Berhasil: {len(successful)}/{len(urls)}")
asyncio.run(crawlwithratelimit())
LLM-Based Extraction
Crawl4AI dapat menggunakan LLM untuk mengekstrak data terstruktur dari halaman web secara cerdas.
Ekstraksi dengan Schema Pydantic
import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
from crawl4ai.extractionstrategy import LLMExtractionStrategy
from pydantic import BaseModel
from typing import List, Optional
Definisikan schema output
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"Nama: {product['name']}")
print(f"Harga: {product['price']}")
print(f"Deskripsi: {product['description']}")
print(f"Fitur: {', '.join(product['features'])}")
print(f"Rating: {product['rating']}")
asyncio.run(extractproductinfo())
Ekstraksi Artikel Berita
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"Judul: {article['title']}")
print(f"Penulis: {article['author']}")
print(f"Ringkasan: {article['summary']}")
asyncio.run(extractnews())
CSS-Based Extraction
Untuk kasus yang lebih sederhana di mana struktur HTML konsisten, gunakan CSS-based extraction yang lebih cepat dan tidak memerlukan LLM.
Ekstraksi dengan CSS Selector
from crawl4ai.extractionstrategy import JsonCssExtractionStrategy
import json
async def cssextraction():
# Definisikan schema CSS
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"Judul: {post['title']}")
print(f"Tanggal: {post['date']}")
print(f"Excerpt: {post['excerpt'][:100]}...")
print()
asyncio.run(cssextraction())
Ekstraksi Tabel Data
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"}
]
}
extraction
strategy = 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 menyediakan berbagai strategi chunking untuk memecah konten menjadi bagian-bagian yang optimal untuk AI processing.
RegexChunking
from crawl4ai.chunkingstrategy import RegexChunking
Chunk berdasarkan pola regex (default: paragraf)
chunker = RegexChunking(
patterns=["\n\n"] # Split berdasarkan double newline
)
async def crawl
withchunking():
run
config = CrawlerRunConfig(
chunkingstrategy=chunker
)
async with AsyncWebCrawler() as crawler:
result = await crawler.arun(
url="https://example.com/long-article",
config=runconfig
)
if result.success:
# Akses 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 berdasarkan jumlah kata
chunker = FixedLengthWordChunking(
chunk
size=200, # 200 kata per chunk
overlap=20 # 20 kata overlap antar chunk
)
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())
Menangani Halaman JavaScript-Rendered
Banyak website modern menggunakan JavaScript untuk merender konten. Crawl4AI menangani ini dengan baik melalui Playwright.
Wait for Dynamic Content
async def crawljspage():
run
config = CrawlerRunConfig(
waitfor="css:.dynamic-content", # Tunggu sampai selector muncul
pagetimeout=60000, # Timeout 60 detik
delaybeforereturnhtml=2.0 # Tunggu 2 detik setelah 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())
Eksekusi JavaScript Custom
async def crawlwithjs():
# JavaScript yang dieksekusi di halaman
js
code = [
# Klik tombol "Load More"
"document.querySelector('button.load-more')?.click();",
# Scroll ke bawah
"window.scrollTo(0, document.body.scrollHeight);",
]
runconfig = CrawlerRunConfig(
jscode=jscode,
delaybeforereturnhtml=3.0, # Tunggu konten dimuat
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())
Menangani Login/Authentication
async def crawlauthenticated():
# JavaScript untuk login
js
login = [
"document.querySelector('#email').value = 'user@example.com';",
"document.querySelector('#password').value = 'password123';",
"document.querySelector('form#login').submit();",
]
browserconfig = BrowserConfig(
headless=True
)
# Step 1: Login terlebih dahulu
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 halaman yang memerlukan auth (session tetap aktif)
result = await crawler.arun(
url="https://example.com/dashboard/reports"
)
if result.success:
print(result.fitmarkdown[:500])
asyncio.run(crawlauthenticated())
Contoh Praktis: RAG Data Pipeline
Mari bangun pipeline lengkap untuk mengumpulkan data dari web dan mempersiapkannya untuk RAG system.
"""
RAG Data Pipeline menggunakan Crawl4AI
Mengumpulkan dokumentasi dari web dan menyiapkan untuk vector store
"""
import asyncio
import json
import hashlib
from datetime import datetime
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, BrowserConfig
from crawl4ai.chunkingstrategy import FixedLengthWordChunking
Konfigurasi
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 # kata per chunk
CHUNKOVERLAP = 30 # kata overlap
OUTPUTFILE = "ragdocuments.json"
def createdocumentid(url: str, chunkindex: int) -> str:
"""Buat ID unik untuk setiap document chunk."""
raw = f"{url}:{chunkindex}"
return hashlib.md5(raw.encode()).hexdigest()
def chunktext(text: str, chunksize: int, overlap: int) -> list:
"""Pecah teks menjadi chunks berdasarkan jumlah kata."""
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 kata
chunks.append(chunk)
return chunks
async def crawlandprocess():
"""Crawl URLs dan proses menjadi dokumen RAG."""
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"Memulai crawling {len(URLS)} URL...")
print("=" 60)
results = await crawler.arunmany(
urls=URLS,
config=runconfig
)
for result in results:
if not result.success:
print(f"GAGAL: {result.url} - {result.errormessage}")
stats["failed"] += 1
continue
stats["successful"] += 1
# Gunakan fitmarkdown untuk konten bersih
content = result.fitmarkdown or result.markdown
wordcount = len(content.split())
stats["totalwords"] += wordcount
print(f"OK: {result.url}")
print(f" Kata: {wordcount}")
# Chunk konten
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)
# Simpan hasil
with open(OUTPUTFILE, "w", encoding="utf-8") as f:
json.dump(documents, f, ensureascii=False, indent=2)
print("\n" + "=" 60)
print("STATISTIK CRAWLING:")
print(f" URL berhasil: {stats['successful']}/{stats['totalurls']}")
print(f" URL gagal: {stats['failed']}/{stats['totalurls']}")
print(f" Total kata: {stats['totalwords']:,}")
print(f" Total chunks: {stats['totalchunks']}")
print(f" Output: {OUTPUTFILE}")
return documents
async def loadtovectorstore(documents: list):
"""Load dokumen ke vector store (contoh dengan ChromaDB)."""
try:
import chromadb
from chromadb.utils import embeddingfunctions
# Inisialisasi ChromaDB
client = chromadb.PersistentClient(path="./chromadb")
# Gunakan embedding function
embeddingfn = embeddingfunctions.SentenceTransformerEmbeddingFunction(
modelname="all-MiniLM-L6-v2"
)
# Buat atau ambil 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 ke 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 tidak terinstall. Install dengan:")
print(" pip install chromadb sentence-transformers")
print("\nDokumen sudah disimpan di file JSON, "
"siap di-load ke vector store pilihan Anda.")
async def main():
"""Pipeline utama."""
# Step 1: Crawl dan proses
documents = await crawlandprocess()
# Step 2: Load ke vector store
await loadtovectorstore(documents)
print("\nPipeline selesai!")
if name == "main":
asyncio.run(main())
Menjalankan Pipeline
# Install dependensi
pip install crawl4ai chromadb sentence-transformers
Setup browser
crawl4ai-setup
Jalankan pipeline
python ragpipeline.py
Integrasi dengan LangChain untuk RAG
"""
Menggunakan data dari Crawl4AI dengan 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 yang sudah diisi oleh Crawl4AI pipeline
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
persistdirectory="./chromadb",
collectionname="pythondocs",
embeddingfunction=embeddings
)
retriever = vectorstore.asretriever(
searchtype="similarity",
searchkwargs={"k": 5}
)
Buat RAG chain
prompt = ChatPromptTemplate.frommessages([
("system", "Jawab pertanyaan berdasarkan context berikut. "
"Jika tidak ada informasi yang relevan, katakan 'Saya tidak menemukan "
"informasi tersebut dalam dokumentasi.'\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("Bagaimana cara menggunakan list comprehension di Python?")
print(response)
Tips dan Best Practices
1. Hormati robots.txt
async def respectfulcrawl():
browserconfig = BrowserConfig(headless=True)
# Selalu cek robots.txt terlebih dahulu
runconfig = CrawlerRunConfig(
wordcountthreshold=10
)
async with AsyncWebCrawler(config=browserconfig) as crawler:
result = await crawler.arun(
url="https://example.com/page",
config=runconfig
)
2. Implementasi Retry Logic
async def crawlwithretry(crawler, url, maxretries=3):
"""Crawl dengan retry logic."""
for attempt in range(max
retries):
try:
result = await crawler.arun(url=url)
if result.success:
return result
print(f"Attempt {attempt + 1} gagal: {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. Caching Hasil Crawling
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 dengan caching lokal."""
cachepath = getcachepath(url)
# Cek 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:
# Simpan ke 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 dan Logging
import logging
import time
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("crawl4aipipeline")
async def monitoredcrawl(urls: list):
"""Crawl dengan monitoring dan 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 selesai dalam {elapsed:.2f}s")
logger.info(f"Berhasil: {successful}/{len(urls)}")
logger.info(f"Rate: {len(urls)/elapsed:.1f} URL/detik")
return results
Kesimpulan
Crawl4AI adalah tool yang sangat powerful untuk mengumpulkan data dari web dan mempersiapkannya untuk pipeline AI. Dengan fitur clean markdown output, LLM-based extraction, dan async crawling, Crawl4AI mempermudah proses pengumpulan data yang biasanya memakan waktu dan rumit.
Kunci penggunaan Crawl4AI yang efektif:
- Gunakan
fit_markdownuntuk mendapatkan konten bersih tanpa noise - Manfaatkan async crawling untuk performa optimal
- Pilih CSS extraction untuk struktur yang konsisten, LLM extraction untuk yang dinamis
- Implementasi caching dan retry logic untuk production
- Selalu hormati robots.txt dan rate limit website target
Dengan Crawl4AI, Anda bisa membangun pipeline data yang robust untuk aplikasi RAG, fine-tuning dataset, atau analisis konten web lainnya.
Selamat mencoba!