Scraping with Ollama and Crawl4AI
Last updated
brew install ollama
ollama pull gemma3:latest
ollama servecurl -X POST https://production.cloudbrowser.ai/api/v1/Browser/Open \
-H "Authorization: Bearer $CLOUDBROWSER_ACCESS_KEY" \
-d '{
"Args": ["--no-sandbox", "--disable-gpu"],
"IgnoredDefaultArgs": ["--enable-automation"],
"Headless": true,
"Stealth": true,
"Browser": 2,
"KeepOpen": 60,
"Label": "app-worker01",
"saveSession": true,
"recoverSession": true
}'{
"address": "browserprovider-worker-*.cloudbrowser.ai/devtools/browser/*",
"status": 200
}import asyncio
import os
import requests
from typing import List
from crawl4ai import *
from pydantic import BaseModel
# Step 1: Open a CloudBrowser session to get WebSocket URL
CLOUDBROWSER_API_KEY = os.environ.get("CLOUDBROWSER_ACCESS_KEY")
response = requests.post(
"https://production.cloudbrowser.ai/api/v1/Browser/Open",
headers={"Authorization": f"Bearer {CLOUDBROWSER_API_KEY}"},
json={
"Args": ["--no-sandbox", "--disable-gpu"],
"IgnoredDefaultArgs": ["--enable-automation"],
"Headless": True,
"Stealth": True,
"Browser": 2,
"KeepOpen": 60,
"Label": "app-worker01",
"saveSession": True,
"recoverSession": True
}
)
ws_url = response.json()["address"]
# Step 2: Define the output schema
class Pricing(BaseModel):
name: str
href: str
class PricingPlans(BaseModel):
pricing_plans: List[Pricing]
# Step 3: Configure CloudBrowser with the returned WebSocket
browser_config = BrowserConfig(
headless=True,
verbose=True,
browser_mode="cdp",
cdp_url=f"wss://{ws_url}"
)
# Step 4: Define the extraction strategy using Ollama + Gemma 3
extraction_strategy = LLMExtractionStrategy(
llm_config=LLMConfig(provider="ollama/gemma3:4b", base_url="http://localhost:11434"),
extraction_type="schema",
schema=PricingPlans.model_json_schema(),
instruction="Extract all the pricing plans JSON array containing their 'name' and 'price'.",
chunk_token_threshold=1200,
overlap_rate=0.1,
apply_chunking=True,
input_format="markdown",
verbose=True
)
# Step 5: Configure and run the crawler
crawl_config = CrawlerRunConfig(
extraction_strategy=extraction_strategy,
cache_mode=CacheMode.BYPASS
)
async def main():
async with AsyncWebCrawler(config=browser_config) as crawler:
result = await crawler.arun(url="https://cloudbrowser.com/", config=crawl_config)
if result.success:
print("Extracted content:", result.extracted_content)
else:
print("Error:", result.error_message)
if __name__ == "__main__":
asyncio.run(main())