Crawl4AI 徹底解説：LLM・RAG 向けオープンソース AI Web クローラー

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:
    """任意のエージェントフレームワークに統合可能なクローラーツール"""

    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:
        """Web ページをクロールして 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()

    # エージェントが特定の URL をクロールすることを決定
    content = await tool.crawl(
        url="https://news.ycombinator.com",
        question="What are the top tech news today?"
    )

    # エージェントがコンテンツを処理
    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));
            }
        }
    """
)