Web scraping documents into LangChain with WebBaseLoader

WebBaseLoader is LangChain’s primary tool for turning live web pages into document objects your chains can reason over. It wraps BeautifulSoup and optional headless browsers so you can fetch, clean, and chunk HTML without leaving the LangChain ecosystem. This guide walks through a production-ready langchain webbaseloader web scraping setup from install to verification.

Step 1: Install the dependencies

Start with a clean virtual environment. You need LangChain core, the community package that ships WebBaseLoader, and BeautifulSoup for parsing. If you plan to render JavaScript, add Playwright.

python -m venv .venv && source .venv/bin/activate
pip install --upgrade pip
pip install langchain langchain-community beautifulsoup4 lxml
# Optional: for JavaScript-heavy sites
pip install playwright && playwright install chromium

Verify the import works:

from langchain_community.document_loaders import WebBaseLoader
print(WebBaseLoader.__module__)  # langchain_community.document_loaders.web_base

Step 2: Load a single page with defaults

The simplest call takes a URL string or list of URLs. By default WebBaseLoader uses requests with a reasonable user agent, parses with BeautifulSoup’s lxml parser, and returns a list of Document objects — one per URL.

from langchain_community.document_loaders import WebBaseLoader

loader = WebBaseLoader("https://example.com/blog/post-1")
docs = loader.load()

print(f"Loaded {len(docs)} document(s)")
print(docs[0].page_content[:500])
print(docs[0].metadata)

Output shows the extracted text and a metadata dict containing source, title, description, and language when detectable. If the page returns non-200, WebBaseLoader raises an exception — wrap in try/except for production code.

Step 3: Configure request headers and authentication

Many sites block the default user agent or require cookies. Pass a header_template dict to the constructor; values can be callables for dynamic tokens.

import os
from langchain_community.document_loaders import WebBaseLoader

def auth_header():
    return {"Authorization": f"Bearer {os.getenv('API_TOKEN')}"}

loader = WebBaseLoader(
    "https://api.docs.internal/guides/advanced",
    header_template={
        "User-Agent": "Mozilla/5.0 (compatible; MyBot/1.0)",
        "Accept": "text/html,application/xhtml+xml",
        "Authorization": auth_header,  # callable evaluated per request
    },
    requests_kwargs={"timeout": 15},  # passed to requests.get
)
docs = loader.load()

For session cookies, use requests.Session in a custom function and assign to loader.session after construction — WebBaseLoader exposes the underlying session object.

Step 4: Customize HTML parsing with BeautifulSoup

Default parsing strips scripts, styles, and noscript tags, then calls get_text(separator="\n"). Override bs_kwargs to change the parser or bs_get_text_kwargs to tweak text extraction. For finer control, subclass and override scrape().

from langchain_community.document_loaders import WebBaseLoader
from bs4 import BeautifulSoup

class ArticleLoader(WebBaseLoader):
    """Extract only <article> content, preserve code blocks."""
    def scrape(self, url: str, bs_kwargs: dict) -> str:
        html = self.session.get(url, **self.requests_kwargs).text
        soup = BeautifulSoup(html, **bs_kwargs)
        article = soup.find("article") or soup.find("main") or soup
        # Remove navigation, ads, footers
        for tag in article.select("nav, aside, footer, .ads, .sidebar"):
            tag.decompose()
        # Keep code formatting
        for pre in article.find_all("pre"):
            pre.replace_with(f"\n```\n&#123;pre.get_text()&#125;\n```\n")
        return article.get_text(separator="\n", strip=True)

loader = ArticleLoader(
    "https://example.com/technical-post",
    bs_kwargs={"features": "lxml"},
)
docs = loader.load()

This pattern lets you target the semantic content area while discarding chrome — critical for token efficiency downstream.

Step 5: Handle JavaScript-rendered content

Static HTML fetching fails on SPA sites or pages that hydrate content client-side. WebBaseLoader supports Playwright via the requests_per_second parameter (misnamed; it enables async Playwright when set). Use playwright=True for clarity in newer versions.

from langchain_community.document_loaders import WebBaseLoader

loader = WebBaseLoader(
    "https://spa-docs.example.com/guide",
    requests_per_second=2,  # enables Playwright, limits concurrency
    playwright_options={
        "wait_until": "networkidle",
        "timeout": 30000,
    },
)
docs = loader.load()

Under the hood this launches a headless Chromium instance, navigates, waits for network idle, then hands the rendered DOM to BeautifulSoup. For authenticated SPAs, inject cookies into the Playwright context:

from playwright.sync_api import sync_playwright

def with_auth(context):
    context.add_cookies([{
        "name": "session",
        "value": os.getenv("SESSION_COOKIE"),
        "domain": ".example.com",
        "path": "/",
    }])

loader = WebBaseLoader(
    "https://app.example.com/dashboard",
    requests_per_second=1,
    playwright_options={"wait_until": "networkidle"},
    playwright_context=with_auth,
)

Step 6: Batch loading with concurrency control

Pass a list of URLs to load multiple pages. WebBaseLoader processes them sequentially by default; set requests_per_second > 0 to enable a thread pool (requests) or async Playwright pool.

urls = [
    "https://example.com/post/1",
    "https://example.com/post/2",
    "https://example.com/post/3",
]

loader = WebBaseLoader(
    urls,
    requests_per_second=5,  # 5 concurrent requests
    continue_on_failure=True,  # log errors, return successful docs
)
docs = loader.load()

# Check failures
for doc in docs:
    if "error" in doc.metadata:
        print(f"Failed: {doc.metadata['source']} — {doc.metadata['error']}")

With continue_on_failure=True, failed URLs produce a Document with empty page_content and an error metadata field. Filter or retry as needed.

Step 7: Integrate with LangChain chunking pipeline

Raw page text exceeds context windows. Chain WebBaseLoader with a text splitter before embedding.

from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS

urls = ["https://example.com/docs/section-1", "https://example.com/docs/section-2"]

loader = WebBaseLoader(urls, requests_per_second=3)
raw_docs = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=150,
    separators=["\n## ", "\n### ", "\n\n", "\n", " ", ""],
)
chunks = splitter.split_documents(raw_docs)

print(f"Split {len(raw_docs)} pages into {len(chunks)} chunks")

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = FAISS.from_documents(chunks, embeddings)
vectorstore.save_local("faiss_index")

Metadata (source, title) propagates to each chunk, enabling citation-aware retrieval.

Step 8: Verify success at each stage

Add explicit checks so failures surface in CI/CD or scheduled jobs.

def verify_load(docs, min_chars=200, max_error_rate=0.1):
    """Raise if load quality degrades."""
    total = len(docs)
    errors = sum(1 for d in docs if "error" in d.metadata)
    empty = sum(1 for d in docs if len(d.page_content.strip()) < min_chars)
    
    if errors / total > max_error_rate:
        raise RuntimeError(f"Error rate {errors}/{total} exceeds {max_error_rate}")
    if empty / total > 0.2:
        raise RuntimeError(f"{empty}/{total} documents nearly empty — parser may be broken")
    
    print(f"✓ Verified: {total} docs, {errors} errors, {empty} short")

# After loading
verify_load(docs)

# After splitting
verify_load(chunks, min_chars=100)

For scheduled scrapes, persist the hash of each page’s content and alert on unexpected changes:

import hashlib
import json

def content_hash(doc):
    return hashlib.sha256(doc.page_content.encode()).hexdigest()[:16]

# Load previous manifest
try:
    with open("manifest.json") as f:
        manifest = json.load(f)
except FileNotFoundError:
    manifest = {}

changed = []
for doc in docs:
    h = content_hash(doc)
    src = doc.metadata["source"]
    if src in manifest and manifest[src] != h:
        changed.append(src)
    manifest[src] = h

with open("manifest.json", "w") as f:
    json.dump(manifest, f, indent=2)

if changed:
    print(f"Content changed for: {changed}")
    # Trigger re-index or alert

Step 9: Respect robots.txt and rate limits

WebBaseLoader does not enforce robots.txt. Add a pre-check if you scrape at scale.

from urllib.robotparser import RobotFileParser
from urllib.parse import urlparse

def can_fetch(url, user_agent="MyBot/1.0"):
    parsed = urlparse(url)
    rp = RobotFileParser()
    rp.set_url(f"{parsed.scheme}://{parsed.netloc}/robots.txt")
    rp.read()
    return rp.can_fetch(user_agent, url)

allowed_urls = [u for u in urls if can_fetch(u)]
blocked = set(urls) - set(allowed_urls)
if blocked:
    print(f"Blocked by robots.txt: {blocked}")

Pair this with requests_per_second and exponential backoff. For provider-grade resilience — automatic fallback when a source is rate-limited or degraded — teams often route scrape traffic through a gateway that honors routing directives and forwards cache-control hints, but that’s infrastructure beyond the loader itself.

Step 10: Package as a reusable component

Wrap the pattern in a function or class your team can import. This example exposes a clean interface with sensible defaults.

# loaders/web.py
from typing import List, Optional, Callable
from langchain_core.documents import Document
from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

def load_and_split(
    urls: List[str],
    *,
    requests_per_second: int = 3,
    chunk_size: int = 1000,
    chunk_overlap: int = 150,
    header_template: Optional[dict] = None,
    custom_scraper: Optional[Callable] = None,
    verify: bool = True,
) -> List[Document]:
    """Fetch, parse, and chunk web pages into retrieval-ready documents."""
    loader_kwargs = {
        "requests_per_second": requests_per_second,
        "continue_on_failure": True,
    }
    if header_template:
        loader_kwargs["header_template"] = header_template
    
    loader = WebBaseLoader(urls, **loader_kwargs)
    
    if custom_scraper:
        loader.scrape = custom_scraper
    
    raw_docs = loader.load()
    
    if verify:
        _verify(raw_docs)
    
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n## ", "\n### ", "\n\n", "\n", " ", ""],
    )
    chunks = splitter.split_documents(raw_docs)
    
    if verify:
        _verify(chunks, min_chars=100)
    
    return chunks

def _verify(docs: List[Document], min_chars: int = 200, max_error_rate: float = 0.1):
    total = len(docs)
    errors = sum(1 for d in docs if "error" in d.metadata)
    empty = sum(1 for d in docs if len(d.page_content.strip()) < min_chars)
    if errors / total > max_error_rate:
        raise RuntimeError(f"Error rate {errors}/{total} > {max_error_rate}")
    if empty / total > 0.2:
        raise RuntimeError(f"{empty}/{total} docs too short")

# Usage
from loaders.web import load_and_split

chunks = load_and_split(
    ["https://example.com/docs/1", "https://example.com/docs/2"],
    header_template={"User-Agent": "CorpusBuilder/2.0"},
    chunk_size=800,
)

Common failure modes and fixes

When to reach beyond WebBaseLoader

WebBaseLoader covers 80% of scraping needs. Move to a dedicated framework (Scrapy, Crawlee, Firecrawl) when you need:

• Distributed crawling with politeness policies per domain
• Incremental re-crawls with sitemap discovery
• Structured data extraction via CSS/XPath schemas
• Proxy rotation and browser fingerprint management

For the remaining 20%, the pattern above — custom scrape(), Playwright for JS, header templates, verification — handles production workloads without leaving LangChain.