Crawl4AI con Ollama: web crawling inteligente en local

Requisitos previos

Antes de instalar Crawl4AI necesitas tener Ollama funcionando con al menos un modelo descargado. Si no lo has configurado aún, la guía de Ollama cubre la instalación completa. Para este tutorial usaremos llama3.2:3b para extracción rápida o qwen2.5:7b para mayor precisión en la extracción de datos complejos.

# Asegurarte de que Ollama esta corriendo y tienes un modelo
ollama serve
ollama pull llama3.2:3b
ollama pull qwen2.5:7b

# Verificar que el endpoint esta disponible
curl http://localhost:11434/api/tags

Instalación de Crawl4AI

Se recomienda instalar Crawl4AI en un entorno virtual para aislar sus dependencias. La instalación incluye Playwright y sus dependencias de sistema.

# Crear y activar entorno virtual
python -m venv venv
source venv/bin/activate  # Linux/macOS
# venv\Scripts\activate   # Windows

# Instalar Crawl4AI con todas las dependencias
pip install crawl4ai

# Instalar los navegadores de Playwright (solo la primera vez)
crawl4ai-setup

# Alternativa si crawl4ai-setup no esta disponible
playwright install chromium

El comando crawl4ai-setup descarga Chromium (aproximadamente 130 MB) y configura las dependencias de sistema necesarias. Este paso es necesario una sola vez. En servidores Linux sin entorno gráfico el proceso es idéntico: Playwright opera en modo headless por defecto.

Verificar la instalación

import asyncio
from crawl4ai import AsyncWebCrawler

async def verificar():
    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(url="https://example.com")
        print(f"Exito: {result.success}")
        print(f"Contenido Markdown (primeros 200 chars):")
        print(result.markdown[:200])

asyncio.run(verificar())

Si ves el contenido de example.com en formato Markdown, la instalación es correcta. El objeto result contiene el HTML completo, el Markdown limpio, los metadatos de la página y, cuando se usa una estrategia de extracción, los datos estructurados en formato JSON.

Crawl básico de una URL

import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, CacheMode

async def crawl_url(url: str):
    config = CrawlerRunConfig(
        cache_mode=CacheMode.BYPASS,   # omitir cache para siempre obtener contenido fresco
        word_count_threshold=10,       # ignorar bloques con menos de 10 palabras
        exclude_external_links=True,   # limpiar enlaces externos del Markdown
        remove_overlay_elements=True,  # eliminar popups y overlays
    )

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

        if result.success:
            print(f"Titulo: {result.metadata.get('title')}")
            print(f"Palabras: {len(result.markdown.split())}")
            print(f"---")
            print(result.markdown[:500])
        else:
            print(f"Error: {result.error_message}")

asyncio.run(crawl_url("https://docs.python.org/3/library/asyncio.html"))

Crawl de múltiples URLs en paralelo

Crawl4AI aprovecha asyncio para crawlear varias URLs de forma concurrente. Puedes controlar cuántos crawls se ejecutan en paralelo con el parámetro de semáforo.

import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig

async def crawl_multiple(urls: list[str], max_concurrent: int = 5):
    config = CrawlerRunConfig(
        cache_mode="bypass",
        word_count_threshold=15,
    )

    async with AsyncWebCrawler() as crawler:
        results = await crawler.arun_many(
            urls=urls,
            config=config,
            max_concurrent=max_concurrent,
        )

    exitosos = [r for r in results if r.success]
    fallidos  = [r for r in results if not r.success]

    print(f"Crawleadas: {len(exitosos)} / {len(urls)}")
    print(f"Fallidas:   {len(fallidos)}")

    return exitosos

urls = [
    "https://ollama.com/blog",
    "https://crawl4ai.com",
    "https://docs.python.org/3/tutorial/",
]
asyncio.run(crawl_multiple(urls))

Crawl con renderizado JavaScript

Para páginas que cargan contenido dinámicamente (SPAs, scroll infinito, contenido tras login) puedes ejecutar JavaScript y esperar a condiciones específicas antes de extraer el contenido.

import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig

async def crawl_spa(url: str):
    config = CrawlerRunConfig(
        # Esperar a que un selector específico aparezca en el DOM
        wait_for="css:.articulo-contenido",

        # Ejecutar JS antes de extraer (p.ej. para expandir acordeones)
        js_code=[
            "document.querySelectorAll('.expandir').forEach(el => el.click());",
        ],

        # Tiempo maximo de espera en milisegundos
        page_timeout=30000,

        # Simular scroll para activar lazy loading
        simulate_user=True,
    )

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

asyncio.run(crawl_spa("https://ejemplo.com/app/"))

Configurar el proveedor Ollama en Crawl4AI

Crawl4AI usa la clase LLMExtractionStrategy para la extracción con IA. Para usar Ollama, apuntas el proveedor al endpoint compatible con OpenAI que Ollama expone por defecto en localhost:11434/v1.

import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
from crawl4ai.extraction_strategy import LLMExtractionStrategy
from pydantic import BaseModel

# 1. Definir el esquema de datos que quieres extraer
class Producto(BaseModel):
    nombre: str
    precio: float | None
    descripcion: str
    disponible: bool

# 2. Configurar la estrategia de extraccion con Ollama
estrategia = LLMExtractionStrategy(
    provider="ollama/llama3.2:3b",      # formato: ollama/
    api_base="http://localhost:11434",   # endpoint local de Ollama
    schema=Producto.model_json_schema(), # esquema JSON desde Pydantic
    extraction_type="schema",
    instruction="""
        Extrae la información del producto de la página.
        Si no encuentras el precio, devuelve null.
        El campo disponible debe ser true si aparece en stock.
    """,
)

# 3. Ejecutar el crawl con extraccion
async def extraer_producto(url: str):
    config = CrawlerRunConfig(
        extraction_strategy=estrategia,
        cache_mode="bypass",
    )

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

    if result.success and result.extracted_content:
        import json
        datos = json.loads(result.extracted_content)
        return datos
    return None

resultado = asyncio.run(extraer_producto("https://ejemplo.com/producto/123"))
print(resultado)

Extracción de listas de elementos

Cuando la página contiene una lista de elementos similares (artículos de blog, resultados de búsqueda, listados de productos), puedes extraer todos a la vez definiendo un esquema que incluya un array.

import asyncio
import json
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
from crawl4ai.extraction_strategy import LLMExtractionStrategy
from pydantic import BaseModel

class Articulo(BaseModel):
    titulo: str
    autor: str | None
    fecha: str | None
    resumen: str
    url: str | None

class ListaArticulos(BaseModel):
    articulos: list[Articulo]

estrategia = LLMExtractionStrategy(
    provider="ollama/qwen2.5:7b",         # modelo más potente para listas complejas
    api_base="http://localhost:11434",
    schema=ListaArticulos.model_json_schema(),
    extraction_type="schema",
    instruction="""
        Extrae todos los artículos o posts que aparecen en la página.
        Para cada uno, incluye el título, autor si aparece, fecha y un
        resumen de 1-2 oraciones del contenido. Incluye la URL si es visible.
    """,
    apply_chunking=True,               # divide el contenido si es largo
    chunk_token_threshold=4000,        # tokens por chunk
    overlap_rate=0.1,                  # overlap del 10% entre chunks
)

async def extraer_articulos(url: str):
    config = CrawlerRunConfig(
        extraction_strategy=estrategia,
        word_count_threshold=20,
    )

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

    if result.success:
        datos = json.loads(result.extracted_content)
        return datos.get("articulos", [])
    return []

articulos = asyncio.run(extraer_articulos("https://blog.ejemplo.com/"))
for art in articulos:
    print(f"- {art['titulo']} ({art.get('fecha', 'sin fecha')})")

Elegir el modelo Ollama correcto para extracción

No todos los modelos de Ollama rinden igual en tareas de extracción estructurada. Los modelos instruction-tuned con buena capacidad de seguir instrucciones JSON son los más adecuados. Recomendaciones prácticas:

• llama3.2:3b — rápido, bajo consumo de VRAM (2 GB). Bueno para extracciones simples de campos individuales o listas cortas.
• qwen2.5:7b — equilibrio entre velocidad y precisión. Maneja bien esquemas complejos y contenido en varios idiomas incluyendo español.
• mistral:7b — excelente para extracción en inglés y francés. Buena precisión en JSON estructurado.
• llama3.1:8b — versión más reciente con mejor comprensión de instrucciones. Recomendado si tienes 8 GB de VRAM.

El pipeline completo tiene cuatro etapas: crawl (obtener el contenido web en Markdown limpio), chunk (dividir el Markdown en fragmentos coherentes), embed (generar vectores de cada fragmento con un modelo de embeddings de Ollama) y store (guardar los vectores en un vector store para recuperación semántica posterior).

Pipeline RAG con ChromaDB y Ollama

import asyncio
import uuid
import chromadb
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
from crawl4ai.chunking_strategy import SlidingWindowChunking
import requests

# ------- CONFIGURACION -------
OLLAMA_BASE   = "http://localhost:11434"
EMBED_MODEL   = "nomic-embed-text"   # modelo de embeddings de Ollama
CHROMA_PATH   = "./mi_knowledge_base"
COLECCION     = "docs"

# Asegurar que el modelo de embeddings esta disponible
# ollama pull nomic-embed-text

# ------- INICIALIZAR CHROMA -------
chroma = chromadb.PersistentClient(path=CHROMA_PATH)
coleccion = chroma.get_or_create_collection(COLECCION)

# ------- FUNCIÓN DE EMBEDDINGS VÍA OLLAMA -------
def generar_embedding(texto: str) -> list[float]:
    respuesta = requests.post(
        f"{OLLAMA_BASE}/api/embeddings",
        json={"model": EMBED_MODEL, "prompt": texto},
    )
    return respuesta.json()["embedding"]

# ------- CRAWL + CHUNK + EMBED + STORE -------
async def indexar_url(url: str):
    # Paso 1: crawl
    config = CrawlerRunConfig(
        chunking_strategy=SlidingWindowChunking(
            window_size=400,   # tokens por chunk
            step=200,          # overlap de 200 tokens entre chunks
        ),
        word_count_threshold=20,
        cache_mode="enabled",  # cachear para no re-crawlear en reindexacion
    )

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

    if not result.success:
        print(f"Error crawleando {url}: {result.error_message}")
        return 0

    # Paso 2 y 3: chunk y embed
    chunks      = result.chunks_markdown or [result.markdown]
    embeddings  = []
    ids         = []
    metadatas   = []
    documents   = []

    for i, chunk in enumerate(chunks):
        if len(chunk.strip()) < 50:
            continue
        embedding = generar_embedding(chunk)
        embeddings.append(embedding)
        ids.append(str(uuid.uuid4()))
        metadatas.append({"url": url, "chunk_index": i})
        documents.append(chunk)

    # Paso 4: store
    if embeddings:
        coleccion.add(
            embeddings=embeddings,
            documents=documents,
            metadatas=metadatas,
            ids=ids,
        )

    print(f"Indexados {len(embeddings)} chunks de {url}")
    return len(embeddings)

# ------- FUNCION DE CONSULTA -------
def consultar(pregunta: str, n_resultados: int = 3) -> list[dict]:
    embedding_pregunta = generar_embedding(pregunta)
    resultados = coleccion.query(
        query_embeddings=[embedding_pregunta],
        n_results=n_resultados,
    )
    return [
        {"texto": doc, "url": meta["url"]}
        for doc, meta in zip(
            resultados["documents"][0],
            resultados["metadatas"][0],
        )
    ]

# ------- USO -------
async def main():
    urls_a_indexar = [
        "https://ollama.com/blog",
        "https://crawl4ai.com/mkdocs/",
        "https://docs.python.org/3/library/asyncio.html",
    ]

    total = 0
    for url in urls_a_indexar:
        total += await indexar_url(url)

    print(f"\nTotal: {total} chunks indexados")
    print(f"Coleccion: {coleccion.count()} fragmentos en total")

    # Consulta de ejemplo
    respuestas = consultar("cómo instalar y configurar ollama")
    for r in respuestas:
        print(f"\n[{r['url']}]")
        print(r['texto'][:200])

asyncio.run(main())

Modelos de embeddings recomendados con Ollama

Para la etapa de embeddings del pipeline RAG, Ollama ofrece modelos especializados mucho más eficientes que los modelos de generación:

• nomic-embed-text (ollama pull nomic-embed-text) — El modelo de embeddings más popular para Ollama. Produce vectores de 768 dimensiones, excelente relación calidad/velocidad. Menos de 300 MB.
• mxbai-embed-large (ollama pull mxbai-embed-large) — Mayor precisión, vectores de 1024 dimensiones. Recomendado para colecciones grandes donde la precisión de recuperación es crítica.
• all-minilm (ollama pull all-minilm) — Minimalista y rápido, vectores de 384 dimensiones. Buena opción para prototipos o colecciones pequeñas donde la velocidad importa más que la precisión.

Estrategias de chunking disponibles

La elección de estrategia de chunking afecta directamente la calidad de recuperación en pipelines RAG. Crawl4AI incluye cuatro estrategias nativas:

from crawl4ai.chunking_strategy import (
    RegexChunking,          # divide por patrones regex (ej: doble salto de línea)
    NlpSentenceChunking,    # divide por oraciones usando NLP
    TopicSegmentationChunking,  # detecta cambios de tema automáticamente
    SlidingWindowChunking,  # ventana deslizante con overlap configurable
)

# Para documentacion tecnica: ventana deslizante con overlap
chunking = SlidingWindowChunking(
    window_size=500,   # tokens por chunk
    step=250,          # avanza 250 tokens (50% overlap)
)

# Para articulos narrativos: por oraciones
chunking = NlpSentenceChunking(
    max_sentences=10,  # maximo 10 oraciones por chunk
)

# Para paginas con secciones bien definidas: por regex
chunking = RegexChunking(
    patterns=[r'\n## ', r'\n### '],  # divide en encabezados Markdown
)

Control de concurrencia y cortesía

import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, RateLimiter

config = CrawlerRunConfig(
    # Controlar la velocidad de crawling para no saturar el servidor objetivo
    mean_delay=1.0,       # espera media de 1 segundo entre peticiones
    max_range=0.5,        # variación aleatoria de +-0.5 segundos

    # Configurar User-Agent para evitar bloqueos
    headers={
        "User-Agent": "MiBot/1.0 (compatible; proyecto-investigacion)",
        "Accept-Language": "es-ES,es;q=0.9",
    },

    # Timeout por página
    page_timeout=20000,   # 20 segundos maximo

    # Reintento en caso de fallo
    retry_on_status=[429, 503],
    max_retries=2,
)

async with AsyncWebCrawler() as crawler:
    resultados = await crawler.arun_many(
        urls=lista_urls,
        config=config,
        max_concurrent=3,   # maximo 3 peticiones simultaneas
    )

Sesiones autenticadas

Para sitios que requieren login puedes reutilizar una sesión de navegador entre múltiples crawls, evitando autenticarte en cada petición.

import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, BrowserConfig

browser_config = BrowserConfig(
    headless=True,
    java_script_enabled=True,
    cookies=[
        {
            "name": "session_id",
            "value": "tu_session_token",
            "domain": ".ejemplo.com",
            "path": "/",
        }
    ],
)

config = CrawlerRunConfig(
    session_id="mi_sesion",    # reutilizar la misma sesion entre crawls
    js_code=[
        # Ejecutar login si es necesario
        """
        if (document.querySelector('#login-form')) {
            document.querySelector('#email').value = 'usuario@ejemplo.com';
            document.querySelector('#password').value = 'contrasena';
            document.querySelector('[type=submit]').click();
            await new Promise(r => setTimeout(r, 2000));
        }
        """,
    ],
)

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