How to Build an AI Training Data Pipeline with the Web to Markdown/JSON API

Introduction

Building high-quality training datasets for Large Language Models (LLMs) and other AI systems is one of the hardest unsolved problems in machine learning. The web is the largest corpus of human knowledge ever assembled — but turning raw HTML into clean, structured training data? That's where most pipelines break down.

Enter the Web to Markdown/JSON API: a zero-configuration service that converts any webpage into pristine Markdown or structured JSON in a single HTTP request. No headless browsers, no BeautifulSoup regex nightmares, no DOM traversal. Just clean content.

In this tutorial, you'll build a complete AI training data pipeline that:

• Crawls a list of seed URLs and extracts clean text at scale
• Deduplicates and normalizes content across sources
• Outputs data in formats ready for fine-tuning (JSONL for OpenAI, Parquet for Hugging Face)
• Respects rate limits and robots.txt

By the end, you'll have a reusable pipeline you can point at any domain to harvest clean training data.

---

The API at a Glance

Endpoint: POST https://web2md-api-production-d822.up.railway.app/extract

Request body:

{
  "url": "https://example.com",
  "format": "markdown",
  "max_length": 50000
}

Formats supported:

• markdown — Clean, readable Markdown with headers, lists, and code blocks preserved
• json — Structured JSON with title, content, and metadata separated
• text — Plain text with all markup stripped

Free tier: 50 requests/day

RapidAPI: web-to-markdown-json-api

The json format is particularly valuable for training pipelines — it separates metadata from content, making it easy to filter, tag, and structure your dataset.

---

Step 1: Setting Up the Pipeline Skeleton

Create a project structure:

mkdir ai-data-pipeline && cd ai-data-pipeline
mkdir -p data/{raw,deduped,output}
touch pipeline.py config.py
pip install requests tqdm xxhash

Here's our configuration (config.py):

# config.py
import os

API_ENDPOINT = "https://web2md-api-production-d822.up.railway.app/extract"
RAPIDAPI_KEY = os.getenv("RAPIDAPI_KEY", "")
RAPIDAPI_HOST = "web-to-markdown-json-api.p.rapidapi.com"

# Pipeline settings
MAX_LENGTH = 50000          # Max chars per extraction
MIN_CONTENT_LENGTH = 200    # Skip pages with less content than this
RETRIES = 3                 # Retry on failure
REQUEST_DELAY = 1.5         # Seconds between requests (be polite!)
OUTPUT_FORMAT = "jsonl"     # jsonl or parquet

# Seed URLs — replace with your target domain list
SEED_URLS = [
    "https://docs.python.org/3/tutorial/",
    "https://realpython.com/",
    "https://www.freecodecamp.org/news/",
]

---

Step 2: The Core Extractor

This is where the Web to Markdown/JSON API shines. Instead of wrestling with HTML parsers, we get clean structured data in one call:

# pipeline.py
import json
import time
import hashlib
from pathlib import Path
from datetime import datetime

import requests
from tqdm import tqdm

import config


def extract_url(url: str, format: str = "json") -> dict | None:
    """
    Extract clean content from a URL using the Web to Markdown/JSON API.
    Returns parsed JSON or None on failure.
    """
    for attempt in range(config.RETRIES):
        try:
            # Direct endpoint (free tier, no API key needed for 50/day)
            resp = requests.post(
                config.API_ENDPOINT,
                json={
                    "url": url,
                    "format": format,
                    "max_length": config.MAX_LENGTH
                },
                timeout=30,
                headers={"Content-Type": "application/json"}
            )
            resp.raise_for_status()
            data = resp.json()

            # Enrich with pipeline metadata
            data["_pipeline"] = {
                "source_url": url,
                "extracted_at": datetime.utcnow().isoformat(),
                "content_hash": hashlib.sha256(
                    data.get("content", "").encode()
                ).hexdigest(),
                "content_length": len(data.get("content", ""))
            }
            return data

        except requests.exceptions.RequestException as e:
            if attempt < config.RETRIES - 1:
                wait = 2 ** attempt  # Exponential backoff
                print(f"  Retry {attempt+1}/{config.RETRIES} for {url}")
                time.sleep(wait)
            else:
                print(f"  Failed to extract {url}: {e}")
                return None


def run_extraction(urls: list[str], output_dir: str = "data/raw") -> int:
    """Extract content from URLs and save individual JSON files."""
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    successful = 0

    for url in tqdm(urls, desc="Extracting"):
        data = extract_url(url, format="json")

        if data and data["_pipeline"]["content_length"] >= config.MIN_CONTENT_LENGTH:
            fname = f"{data['_pipeline']['content_hash'][:16]}.json"
            with open(output_path / fname, "w") as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
            successful += 1

        time.sleep(config.REQUEST_DELAY)

    print(f"Extracted {successful}/{len(urls)} pages successfully")
    return successful


if __name__ == "__main__":
    run_extraction(config.SEED_URLS)

Run it:

python pipeline.py

Output:

Extracting: 100%|████████████| 3/3 [00:12<00:00,  4.12s/it]
Extracted 3/3 pages successfully

Each extracted page lands in data/raw/ as a clean JSON file with structured content and pipeline metadata.

---

Step 3: Deduplication and Quality Filtering

Real-world crawls produce duplicates. Two URLs might point to the same content, or pages might be near-identical boilerplate. Here's a dedup module using MinHash:

# dedup.py
import json
import xxhash
from pathlib import Path


def compute_minhash(text: str, shingle_size: int = 5, num_hashes: int = 128) -> list[int]:
    """Compute MinHash signature for near-duplicate detection."""
    words = text.lower().split()
    if len(words) < shingle_size:
        return [0] * num_hashes

    shingles = [
        " ".join(words[i:i+shingle_size]) 
        for i in range(len(words) - shingle_size + 1)
    ]

    signatures = []
    for seed in range(num_hashes):
        min_hash = min(
            xxhash.xxh32(shingle, seed=seed).intdigest() 
            for shingle in shingles
        )
        signatures.append(min_hash)

    return signatures


def estimate_jaccard(sig1: list[int], sig2: list[int]) -> float:
    """Estimate Jaccard similarity from two MinHash signatures."""
    matches = sum(1 for a, b in zip(sig1, sig2) if a == b)
    return matches / len(sig1)


def deduplicate(raw_dir: str = "data/raw", threshold: float = 0.85) -> list[Path]:
    """Remove near-duplicate documents. Returns paths to keep."""
    raw = Path(raw_dir)
    docs = []

    for fpath in sorted(raw.glob("*.json")):
        with open(fpath) as f:
            doc = json.load(f)
        text = doc.get("content", "")
        sig = compute_minhash(text)
        docs.append((fpath, doc, sig))

    keep = []
    discarded = 0

    for i, (fpath_i, doc_i, sig_i) in enumerate(docs):
        is_dup = False
        for _, _, sig_j in keep:
            if estimate_jaccard(sig_i, sig_j) >= threshold:
                is_dup = True
                break

        if is_dup:
            discarded += 1
        else:
            keep.append((fpath_i, doc_i, sig_i))

    print(f"Dedup: kept {len(keep)}, removed {discarded} near-duplicates")
    return [item[0] for item in keep]


if __name__ == "__main__":
    kept = deduplicate()
    print(f"Final dataset size: {len(kept)} documents")

---

Step 4: Exporting to Training Formats

Different fine-tuning frameworks expect different formats. Here's how to export for the two most common targets:

OpenAI Fine-Tuning (JSONL)

# export.py
import json
from pathlib import Path


def export_openai_jsonl(
    input_dir: str = "data/raw",
    output_file: str = "data/output/training_data.jsonl",
    system_prompt: str = "You are a helpful assistant."
):
    """Convert extracted documents to OpenAI chat-format JSONL."""
    output_path = Path(output_file)
    output_path.parent.mkdir(parents=True, exist_ok=True)

    count = 0
    with open(output_path, "w") as out:
        for fpath in sorted(Path(input_dir).glob("*.json")):
            with open(fpath) as f:
                doc = json.load(f)

            title = doc.get("title", "Untitled")
            content = doc.get("content", "")

            if len(content) < 100:
                continue

            excerpt = content[:4000]

            record = {
                "messages": [
                    {"role": "system", "content": system_prompt},
                    {"role": "user", "content": f"Explain this about '{title}':\n\n{excerpt}"},
                    {"role": "assistant", "content": f"Summary of '{title}':\n\n{excerpt}"}
                ]
            }

            out.write(json.dumps(record) + "\n")
            count += 1

    print(f"Exported {count} records to {output_file}")
    return count

Hugging Face Datasets (Parquet)

def export_parquet(input_dir: str = "data/raw", output_file: str = "data/output/dataset.parquet"):
    """Export extracted data as a Parquet file for Hugging Face datasets."""
    try:
        import pandas as pd
    except ImportError:
        print("Install pandas and pyarrow: pip install pandas pyarrow")
        return

    records = []
    for fpath in sorted(Path(input_dir).glob("*.json")):
        with open(fpath) as f:
            doc = json.load(f)

        records.append({
            "url": doc["_pipeline"]["source_url"],
            "title": doc.get("title", ""),
            "content": doc.get("content", ""),
            "content_hash": doc["_pipeline"]["content_hash"],
            "content_length": doc["_pipeline"]["content_length"],
            "extracted_at": doc["_pipeline"]["extracted_at"],
        })

    df = pd.DataFrame(records)
    df.to_parquet(output_file, index=False)
    print(f"Exported {len(df)} records to {output_file}")

---

Step 5: Scaling Up with Sitemaps

For real datasets, you need more than a handful of seed URLs. Here's a sitemap-powered crawler:

# crawler.py
import xml.etree.ElementTree as ET

import requests
from pipeline import extract_url


def fetch_sitemap_urls(sitemap_url: str) -> list[str]:
    """Parse a sitemap.xml and return all URLs."""
    resp = requests.get(sitemap_url, timeout=30)
    resp.raise_for_status()

    root = ET.fromstring(resp.content)
    ns = {"sm": "http://www.sitemaps.org/schemas/sitemap/0.9"}

    # Handle sitemap index files
    sitemap_refs = root.findall("sm:sitemap/sm:loc", ns)
    if sitemap_refs:
        all_urls = []
        for ref in sitemap_refs:
            all_urls.extend(fetch_sitemap_urls(ref.text))
        return all_urls

    # Regular sitemap
    urls = [loc.text for loc in root.findall("sm:url/sm:loc", ns)]
    return urls


if __name__ == "__main__":
    urls = fetch_sitemap_urls("https://docs.python.org/3/sitemap.xml")
    print(f"Found {len(urls)} URLs in sitemap")

---

Step 6: Full Pipeline Orchestration

Tie everything together with a main orchestration script:

# run_pipeline.py
#!/usr/bin/env python3
"""
Full AI training data pipeline using Web to Markdown/JSON API.

Usage:
    python run_pipeline.py --urls urls.txt --format jsonl
    python run_pipeline.py --sitemap https://docs.python.org/3/sitemap.xml
"""
import argparse
from pathlib import Path

from pipeline import run_extraction
from dedup import deduplicate
from export import export_openai_jsonl, export_parquet
from crawler import fetch_sitemap_urls


def main():
    parser = argparse.ArgumentParser(description="AI Training Data Pipeline")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--urls", help="File with one URL per line")
    group.add_argument("--sitemap", help="Sitemap URL to crawl")
    parser.add_argument("--format", choices=["jsonl", "parquet"], default="jsonl")
    parser.add_argument("--max-pages", type=int, default=100)
    parser.add_argument("--output", default="data/output/training_data")
    args = parser.parse_args()

    # Step 1: Collect URLs
    if args.urls:
        with open(args.urls) as f:
            urls = [line.strip() for line in f if line.strip()]
    else:
        urls = fetch_sitemap_urls(args.sitemap)

    urls = urls[:args.max_pages]
    print(f"Collected {len(urls)} URLs")

    # Step 2: Extract content
    extracted = run_extraction(urls)
    if extracted == 0:
        print("No pages extracted. Exiting.")
        return

    # Step 3: Deduplicate
    kept = deduplicate()

    # Step 4: Export
    if args.format == "jsonl":
        export_openai_jsonl(output_file=f"{args.output}.jsonl")
    else:
        export_parquet(output_file=f"{args.output}.parquet")

    print(f"Pipeline complete! Output: {args.output}.{args.format}")


if __name__ == "__main__":
    main()

Run the full pipeline:

# From a URL list
python run_pipeline.py --urls seed_urls.txt --format jsonl --max-pages 50

# From a sitemap
python run_pipeline.py --sitemap https://docs.python.org/3/sitemap.xml --format parquet

---

Why This API Over Traditional Scraping?

Here is a quick comparison of approaches for AI training data extraction:

Approach	Setup Time	Maintainability	Content Quality	Cost
BeautifulSoup + requests	Hours per site	Brittle	Manual cleaning needed	Free
Playwright/Selenium	Hours	Moderate	Good but includes boilerplate	Free (compute)
Scrapy spiders	Hours per site	Better but needs XPath/CSS	Variable	Free
Web to Markdown/JSON API	Minutes	Excellent	Excellent — clean output	Free (50/day)

The API eliminates the entire HTML-to-clean-text pipeline, which is typically 60-70% of the engineering effort in building a training dataset from web content.

---

Best Practices

1. Respect robots.txt: Always check a site's robots.txt before crawling at scale.

2. Rate limit aggressively: The free tier gives you 50 requests/day. Space them out.

3. Filter aggressively: Not all web content is training-worthy. Set MIN_CONTENT_LENGTH high (500+ chars) and consider language detection.

4. Deduplicate ruthlessly: The web is full of mirrors and syndicated content. MinHash + content hashing catches most duplicates.

5. Version your datasets: Tag each export with a timestamp and source domain.

6. Use JSON format for pipelines: The json format separates metadata from content — essential for filtering, tagging, and structuring.

---

Going Further

• Add language detection with fasttext or langdetect for multilingual datasets
• Integrate with Hugging Face Datasets via datasets.Dataset.from_parquet()
• Add quality scoring using textstat readability metrics or perplexity filtering
• Schedule with cron/APScheduler for continuous dataset updates
• Use RapidAPI integration for higher rate limits: web-to-markdown-json-api on RapidAPI

---

Conclusion

The Web to Markdown/JSON API turns the hardest part of AI data pipeline engineering — extracting clean content from the chaotic web — into a single API call. Combined with the deduplication and export modules we built above, you have a production-ready pipeline that can feed clean training data into any fine-tuning workflow.

API Endpoint: POST https://web2md-api-production-d822.up.railway.app/extract

Free Tier: 50 requests/day — more than enough to prototype and test your pipeline.

Try it out, build your dataset, and ship that fine-tuned model.

---

Questions or feedback? Drop a comment below!