Crawl4AI 심층 분석: LLM과 RAG를 위한 오픈소스 AI 웹 크롤러

from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
from crawl4ai.content_filter_strategy import PruningContentFilter
from crawl4ai.markdown_generation_strategy import DefaultMarkdownGenerator

config = CrawlerRunConfig(
    markdown_generator=DefaultMarkdownGenerator(
        content_filter=PruningContentFilter(threshold=0.48)
    )
)

import asyncio
from crawl4ai import AsyncWebCrawler

async def crawl_multiple_urls():
    urls = ["https://example1.com", "https://example2.com", "https://example3.com"]

    async with AsyncWebCrawler() as crawler:
        tasks = [crawler.arun(url=url) for url in urls]
        results = await asyncio.gather(*tasks)

        for result in results:
            print(f"URL: {result.url}, Content length: {len(result.markdown)}")

asyncio.run(crawl_multiple_urls())

from crawl4ai import CacheMode

config = CrawlerRunConfig(
    cache_mode=CacheMode.ENABLED  # 캐싱 활성화
)

# Crawl4AI 설치
pip install -U crawl4ai

# 설치 후 설정 실행 (Playwright 브라우저 설치)
crawl4ai-setup

# 설치 확인
crawl4ai-doctor

python -m playwright install --with-deps chromium

# 최신 이미지 가져오기 및 실행
docker pull unclecode/crawl4ai:latest
docker run -d -p 11235:11235 --name crawl4ai --shm-size=1g unclecode/crawl4ai:latest

# 모니터링 대시보드 접근: http://localhost:11235/dashboard
# Playground 접근: http://localhost:11235/playground

import asyncio
from crawl4ai import AsyncWebCrawler

async def main():
    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(
            url="https://example.com"
        )
        print(result.markdown.fit_markdown)

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

# 기본 크롤링, Markdown 출력
crwl https://example.com -o markdown

# 딥 크롤링, BFS 전략 사용, 최대 10페이지
crwl https://docs.crawl4ai.com --deep-crawl bfs --max-pages 10

# LLM을 사용하여 특정 정보 추출
crwl https://example.com/products -q "Extract all product prices"

from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig

browser_config = BrowserConfig(
    headless=True,  # 헤드리스 모드
    verbose=True,   # 상세 로깅
)

run_config = CrawlerRunConfig(
    # JavaScript 로드 대기
    wait_until="networkidle",
    # 페이지 로드 타임아웃
    page_timeout=30000,
    # 사용자 정의 User-Agent
    user_agent="Mozilla/5.0 (compatible; MyBot/1.0)",
)

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

import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, JsonCssExtractionStrategy
import json

async def extract_products():
    # 추출 스키마 정의
    schema = {
        "name": "Product List",
        "baseSelector": "div.product-card",
        "fields": [
            {
                "name": "title",
                "selector": "h2.product-title",
                "type": "text",
            },
            {
                "name": "price",
                "selector": "span.price",
                "type": "text",
            },
            {
                "name": "image",
                "selector": "img.product-image",
                "type": "attribute",
                "attribute": "src"
            },
            {
                "name": "description",
                "selector": "p.product-desc",
                "type": "text",
            }
        ]
    }

    config = CrawlerRunConfig(
        extraction_strategy=JsonCssExtractionStrategy(schema)
    )

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

        products = json.loads(result.extracted_content)
        for product in products:
            print(f"{product['title']}: {product['price']}")

asyncio.run(extract_products())

import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, LLMExtractionStrategy, LLMConfig
from pydantic import BaseModel, Field

class ArticleInfo(BaseModel):
    title: str = Field(..., description="文章标题")
    author: str = Field(..., description="作者名称")
    publish_date: str = Field(..., description="发布日期")
    summary: str = Field(..., description="文章摘要")

async def extract_with_llm():
    config = CrawlerRunConfig(
        extraction_strategy=LLMExtractionStrategy(
            llm_config=LLMConfig(
                provider="openai/gpt-4o",
                api_token="your-api-key"
            ),
            schema=ArticleInfo.schema(),
            extraction_type="schema",
            instruction="从页面内容中提取文章标题、作者、发布日期和摘要"
        )
    )

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

asyncio.run(extract_with_llm())

from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
from crawl4ai.deep_crawling import BFSDeepCrawlStrategy

async def deep_crawl_docs():
    config = CrawlerRunConfig(
        deep_crawl_strategy=BFSDeepCrawlStrategy(
            max_depth=2,        # 최대 깊이
            max_pages=50,       # 최대 50페이지 크롤링
            include_external=False  # 외부 링크 미포함
        )
    )

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

        # result는 여러 페이지 결과를 포함한 리스트
        for page_result in result:
            print(f"Crawled: {page_result.url}")
            print(f"Content length: {len(page_result.markdown)}")

asyncio.run(deep_crawl_docs())

from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig

browser_config = BrowserConfig(
    headless=True,
    # 영구 사용자 데이터 디렉토리 사용
    user_data_dir="~/.crawl4ai/browser_profile",
    use_persistent_context=True,
    # 사용자 정의 User-Agent
    user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
)

config = CrawlerRunConfig(
    # 감지 회피를 위한 JavaScript 코드 실행
    js_code="""
        Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
    """,
    # 특정 요소 로드 대기
    wait_for="css:.main-content",
)

async with AsyncWebCrawler(config=browser_config) as crawler:
    result = await crawler.arun(
        url="https://challenging-website.com",
        config=config,
        magic=True  # 일반적인 봇 차단을 자동 처리하는 매직 모드 활성화
    )

import asyncio
import json
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, CacheMode
from crawl4ai.content_filter_strategy import BM25ContentFilter
from crawl4ai.markdown_generation_strategy import DefaultMarkdownGenerator

async def prepare_rag_data(urls, query_topic):
    """URL을 배치 크롤링하여 RAG 형식으로 변환"""

    config = CrawlerRunConfig(
        cache_mode=CacheMode.ENABLED,
        # BM25 필터링 사용, 쿼리 주제 기반 관련 콘텐츠 추출
        markdown_generator=DefaultMarkdownGenerator(
            content_filter=BM25ContentFilter(
                user_query=query_topic,
                bm25_threshold=1.0
            )
        )
    )

    documents = []

    async with AsyncWebCrawler() as crawler:
        for url in urls:
            result = await crawler.arun(url=url, config=config)

            if result.success:
                documents.append({
                    "url": url,
                    "content": result.markdown.fit_markdown,
                    "metadata": {
                        "title": result.metadata.get("title", ""),
                        "description": result.metadata.get("description", ""),
                        "crawled_at": str(result.status_code)
                    }
                })

    # JSONL 형식으로 저장, 추후 로드 용이
    with open("rag_documents.jsonl", "w", encoding="utf-8") as f:
        for doc in documents:
            f.write(json.dumps(doc, ensure_ascii=False) + "\n")

    print(f"Successfully prepared {len(documents)} documents for RAG")
    return documents

# 사용 예시
urls = [
    "https://docs.python.org/3/tutorial/index.html",
    "https://docs.python.org/3/library/asyncio.html",
    "https://docs.python.org/3/library/asyncio-task.html",
]

asyncio.run(prepare_rag_data(urls, "Python asyncio tutorial"))

import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma

async def build_rag_pipeline(urls):
    # 1. Crawl4AI로 콘텐츠 크롤링
    async with AsyncWebCrawler() as crawler:
        results = []
        for url in urls:
            result = await crawler.arun(url=url)
            if result.success:
                results.append({
                    "content": result.markdown.fit_markdown,
                    "metadata": {"source": url}
                })

    # 2. 텍스트 청킹
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,
        chunk_overlap=200
    )

    documents = []
    for result in results:
        chunks = text_splitter.split_text(result["content"])
        for chunk in chunks:
            documents.append({
                "page_content": chunk,
                "metadata": result["metadata"]
            })

    # 3. 벡터 스토어 생성
    texts = [doc["page_content"] for doc in documents]
    metadatas = [doc["metadata"] for doc in documents]

    embeddings = OpenAIEmbeddings()
    vectorstore = Chroma.from_texts(texts, embeddings, metadatas=metadatas)

    return vectorstore

# RAG 시스템 구축
vectorstore = asyncio.run(build_rag_pipeline([
    "https://docs.crawl4ai.com",
    "https://docs.python.org/3/library/asyncio.html",
]))

# 쿼리
query = "How to use Crawl4AI with async?"
docs = vectorstore.similarity_search(query, k=3)
for doc in docs:
    print(f"Source: {doc.metadata['source']}")
    print(f"Content: {doc.page_content[:200]}...")

from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
from llama_index.core import Document, VectorStoreIndex
import asyncio

async def build_llamaindex_corpus(urls):
    """Crawl4AI로 LlamaIndex 코퍼스 구축"""
    documents = []

    async with AsyncWebCrawler() as crawler:
        for url in urls:
            result = await crawler.arun(url=url)
            if result.success:
                documents.append(Document(
                    text=result.markdown.fit_markdown,
                    metadata={"url": url, "title": result.metadata.get("title", "")}
                ))

    # 인덱스 생성
    index = VectorStoreIndex.from_documents(documents)
    return index

# 사용
index = asyncio.run(build_llamaindex_corpus([
    "https://docs.example.com/page1",
    "https://docs.example.com/page2",
]))

query_engine = index.as_query_engine()
response = query_engine.query("What is Crawl4AI?")
print(response)

from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
import asyncio

class WebCrawlerTool:
    """어떤 Agent 프레임워크에도 통합 가능한 크롤러 도구"""

    def __init__(self):
        self.crawler = None

    async def initialize(self):
        self.crawler = AsyncWebCrawler()
        await self.crawler.__aenter__()

    async def close(self):
        if self.crawler:
            await self.crawler.__aexit__(None, None, None)

    async def crawl(self, url: str, question: str = None) -> str:
        """웹 페이지를 크롤링하여 LLM 친화적 콘텐츠 반환"""
        config = CrawlerRunConfig()

        result = await self.crawler.arun(url=url, config=config)

        if result.success:
            return result.markdown.fit_markdown
        else:
            return f"Failed to crawl: {result.error_message}"

# 사용 예시
async def agent_workflow():
    tool = WebCrawlerTool()
    await tool.initialize()

    # Agent가 특정 URL 크롤링 결정
    content = await tool.crawl(
        url="https://news.ycombinator.com",
        question="What are the top tech news today?"
    )

    # Agent가 콘텐츠 처리
    print(f"Retrieved {len(content)} characters of content")

    await tool.close()

asyncio.run(agent_workflow())

from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
from crawl4ai.deep_crawling import BFSDeepCrawlStrategy

async def build_knowledge_base(base_url, max_pages=100):
    """웹사이트 지식 베이스 구축"""
    config = CrawlerRunConfig(
        deep_crawl_strategy=BFSDeepCrawlStrategy(
            max_depth=3,
            max_pages=max_pages,
            include_external=False
        )
    )

    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(url=base_url, config=config)

        knowledge_base = {}
        for page in result:
            if page.success:
                knowledge_base[page.url] = {
                    "content": page.markdown.fit_markdown,
                    "metadata": page.metadata
                }

        return knowledge_base

import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, JsonCssExtractionStrategy

async def monitor_competitor():
    schema = {
        "name": "Product Monitor",
        "baseSelector": "div.product-item",
        "fields": [
            {"name": "name", "selector": "h3.product-name", "type": "text"},
            {"name": "price", "selector": "span.price", "type": "text"},
            {"name": "features", "selector": "ul.features li", "type": "list"},
        ]
    }

    config = CrawlerRunConfig(
        extraction_strategy=JsonCssExtractionStrategy(schema)
    )

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

        products = result.extracted_content
        # 이전 크롤링 결과와 비교하여 변화 감지
        return products

async def aggregate_content(sources):
    """여러 소스에서 콘텐츠 애그리게이션"""
    async with AsyncWebCrawler() as crawler:
        all_content = []

        for source in sources:
            result = await crawler.arun(url=source["url"])
            if result.success:
                all_content.append({
                    "source": source["name"],
                    "title": result.metadata.get("title", ""),
                    "content": result.markdown.fit_markdown,
                    "url": source["url"]
                })

        return all_content

from crawl4ai import AsyncWebCrawler, BrowserConfig

browser_config = BrowserConfig(
    headless=True,
    # 동시 브라우저 인스턴스 수 제한
    max_concurrent_browsers=5,
)

config = CrawlerRunConfig(
    page_timeout=60000,  # 60초 타임아웃
    wait_until="networkidle"  # 네트워크 유휴 대기
)

config = CrawlerRunConfig(
    # 스크롤 시뮬레이션으로 모든 콘텐츠 로드
    js_code="""
        async () => {
            for (let i = 0; i < 10; i++) {
                window.scrollTo(0, document.body.scrollHeight);
                await new Promise(r => setTimeout(r, 1000));
            }
        }
    """
)