Web data for LLM fine-tuning: a practical 2026 guide

This article was originally published on the Zenrows blog. Read the original here: https://www.zenrows.com/blog/web-data-llm-fine-tuning

This guide shows you how to collect a real-data seed for LLM fine-tuning from protected web sources using Zenrows Fetch, format it as instruction-output pairs, validate it, and measure whether the fine-tune held. You need Python 3.9+, a Zenrows API key, and requests and python-dotenv installed.

All code is on GitHub.

Before you start

• Python 3.9 or later
• Zenrows account for your API key
• pypdf if you are collecting from PDFs

python3 -m pip install requests python-dotenv pypdf

# .env
ZENROWS_API_KEY=your_zenrows_api_key_here

Why a real-data seed matters

In a fine-tuning set of a few hundred rows, one mislabeled extraction is a measurable percentage of your entire training signal. Synthetic-only datasets amplify teacher model biases through each retrain cycle. The real-data seed anchors the generator and breaks the drift loop.

Authoritative sources — documentation, regulatory filings, expert-reviewed publications — are almost always behind access controls. Standard crawlers get flagged on the TLS handshake. Zenrows Fetch handles rendering and access in a single call.

Step 1: Collect clean Markdown from protected sources

mode=auto is Adaptive Stealth Mode. Zenrows picks the retrieval strategy per target — JavaScript rendering, proxy escalation, or direct fetch — without configuration on your end.

import os
import requests
from dotenv import load_dotenv

load_dotenv()

apikey = os.getenv("ZENROWS_API_KEY")
url = "https://www.scrapingcourse.com/ecommerce/"

params = {
    "url": url,
    "apikey": apikey,
    "mode": "auto",
    "response_type": "markdown",  # clean Markdown, not raw HTML
}

response = requests.get("https://api.zenrows.com/v1/", params=params)
print(response.text)

Step 2: Handle PDFs and non-HTML sources

Use response_type=pdf for PDFs. js_render=true is required for PDF responses.

params = {
    "url": url,
    "apikey": apikey,
    "js_render": "true",    # required for PDF responses
    "response_type": "pdf", # returns raw bytes, not text
}

response = requests.get("https://api.zenrows.com/v1/", params=params)

with open("source-document.pdf", "wb") as f:
    f.write(response.content)

Extract text and chunk by structure, not by character count.

from pypdf import PdfReader
import io

reader = PdfReader(io.BytesIO(response.content))

def chunk_by_page(reader, max_chars=6000):
    chunks = []
    for page in reader.pages:
        text = page.extract_text()
        if not text or not text.strip():
            continue  # skip blank or image-only pages
        if len(text) <= max_chars:
            chunks.append(text)
        else:
            for i in range(0, len(text), max_chars):
                chunks.append(text[i:i + max_chars])
    return chunks

chunks = chunk_by_page(reader)
print(f"Produced {len(chunks)} chunks from {len(reader.pages)} pages")

Step 3: Track provenance

Every row in the seed needs a source, retrieval date, and license.

from datetime import date

source = {
    "url": pdf_url,
    "retrieved": date.today().isoformat(),
    "license": license_url,  # check the source's terms before collecting at scale
}

Step 4: Run batch collection

Match concurrency to your plan limit. time.sleep(0.5) spaces out requests per worker.

import time
from concurrent.futures import ThreadPoolExecutor

MAX_CONCURRENT = 50  # check your Zenrows dashboard for your plan's limit

def fetch(url):
    params = {
        "url": url,
        "apikey": apikey,
        "response_type": "markdown",
        "mode": "auto",
        "wait": 3000,
    }
    response = requests.get("https://api.zenrows.com/v1/", params=params)
    time.sleep(0.5)  # pace requests within each worker

    if response.status_code != 200:
        return {"url": url, "status": response.status_code, "content": None}
    return {"url": url, "status": response.status_code, "content": response.text}

urls = [
    "https://playboard.co",
    "https://www.samsung.com",
    "https://www.kickstarter.com",
]

with ThreadPoolExecutor(max_workers=MAX_CONCURRENT) as executor:
    results = list(executor.map(fetch, urls))

Step 5: Format as instruction-output pairs

The Markdown is the input. The extracted JSON is the output. Write one JSON object per line in JSONL format.

import json

def to_training_example(instruction, markdown_content, extracted_json):
    return {
        "messages": [
            {"role": "system", "content": instruction},
            {"role": "user", "content": markdown_content},
            {"role": "assistant", "content": json.dumps(extracted_json)},
        ]
    }

instruction = "Extract the product name, price, and availability as JSON."

with open("training_data.jsonl", "w") as f:
    for markdown_content, extracted_json in seed_examples:
        example = to_training_example(instruction, markdown_content, extracted_json)
        f.write(json.dumps(example) + "\n")

Step 6: Validate before training

Every row must clear three checks before it enters the training set — length, schema, and duplicate hash.

import hashlib
from jsonschema import validate, ValidationError

schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string", "minLength": 1},
        "price": {"type": "string", "minLength": 1},
    },
    "required": ["name", "price"],
}

seen = set()

def keep(markdown, extracted):
    if len(markdown.strip()) < 200:
        return False  # page rendered to almost nothing
    try:
        validate(instance=extracted, schema=schema)
    except ValidationError:
        return False  # wrong types or missing fields
    h = hashlib.sha256(" ".join(markdown.lower().split()).encode()).hexdigest()
    if h in seen:
        return False  # duplicate or near-duplicate
    seen.add(h)
    return True

clean_examples = [(url, md, js) for url, md, js in seed_examples if keep(md, js)]
print(f"Collected {len(seed_examples)} pages, {len(clean_examples)} passed validation")

Step 7: Hold out real data and measure

Set aside 10 to 20 percent before any synthetic expansion and keep it out of training permanently.

import random

random.seed(42)
random.shuffle(seed_examples)

split = int(len(seed_examples) * 0.85)
train_examples = seed_examples[:split]
holdout_examples = seed_examples[split:]  # never touches the training set

After fine-tuning, run the model against the holdout and measure task accuracy. Record the score after each retrain. A declining score is an early sign of drift.

What's next

• How to extract web data for AI training — broader context on training data collection
• Zenrows Batch — managed concurrency and retries for large URL lists
• Zenrows MCP server — web access for agents without managing the collection layer