Build RAG-Ready Data Pipelines from Scraped Web Data (Python)
Author : Actowiz Solutions | Published On : 13 Oct 2026
https://www.actowizsolutions.com/rag-ready-data-pipelines-scraped-web-content-python-tutorial.php
Introduction
Retrieval-Augmented Generation (RAG) is how production AI systems answer with facts instead of guesses: retrieve relevant, current documents, then let the LLM generate grounded responses. But a RAG system is only as good as the data flowing into its vector database — and for most commercial use cases (pricing assistants, product Q&A bots, market intelligence copilots), that data lives on the open web and changes daily.
This tutorial from Actowiz Solutions walks through the complete pipeline: scraped web content → cleaning → structuring → chunking → embedding → vector DB ingestion → refresh. Code examples use Python with widely adopted tooling, and every stage reflects the patterns we run at production scale for enterprise clients.
What Does "RAG-Ready" Actually Mean?
A dataset is RAG-ready when retrieval works — meaning a user's question reliably pulls back the right passages. In practice that requires five properties:
-
Clean text — no nav bars, cookie banners, or footer boilerplate polluting embeddings
-
Right-sized chunks — passages small enough to be precise, large enough to carry context
-
Rich metadata — source URL, timestamp, category, language, price fields — so retrieval can filter, not just search
-
Freshness — stale product prices in a vector DB are worse than no data; the pipeline must refresh
-
Deduplication — near-duplicate chunks waste index space and skew retrieval scores
Raw scraped HTML has none of these. The pipeline below adds all five.
Pipeline Architecture Overview
[Scraper / Actowiz Feed]
│ raw JSON/HTML
▼
[1. Extraction & Cleaning] → boilerplate removal, text normalization
▼
[2. Structuring] → typed records + metadata
▼
[3. Deduplication] → hash + near-dup filtering
▼
[4. Chunking] → semantic, metadata-aware chunks
▼
[5. Embedding] → batch embed with retry logic
▼
[6. Vector DB Ingestion] → upsert with IDs (idempotent)
▼
[7. Refresh Loop] → scheduled re-crawl + delta upserts
Step 1 — Start from Structured Scraped Records
The single biggest RAG shortcut: don't feed raw HTML to your pipeline. Start from structured extraction. A professionally scraped product record looks like this (illustrative sample):
{
"url": "https://example-retailer.com/product/earbuds-x",
"scraped_at": "2026-08-01T06:00:00Z",
"title": "Wireless Earbuds Model X",
"brand": "SampleBrand",
"price": 79.99,
"currency": "USD",
"description": "Noise-cancelling earbuds with 30-hour battery...",
"specs": {"battery_hours": 30, "bluetooth": "5.4"},
"reviews_summary": "4.4 stars from 12,847 reviews",
"category": "Electronics > Audio > Earbuds",
"language": "en"
}
Actowiz delivers feeds in exactly this shape (JSONL/Parquet), which lets you skip weeks of HTML-parsing fragility. If you're scraping in-house, use a robust extractor and keep the raw HTML separately for reprocessing.
Step 2 — Clean and Normalize
Even structured text needs normalization before embedding:
import re, unicodedata
def clean_text(text: str) -> str:
text = unicodedata.normalize("NFKC", text)
text = re.sub(r"s+", " ", text) # collapse whitespace
text = re.sub(r"(Cookie Policy|Accept All|Subscribe to newsletter).*", "", text, flags=re.I)
return text.strip()
Normalize currencies and units into metadata fields rather than leaving them in prose — "₹1,299" and "Rs. 1299" should both become {"price": 1299, "currency": "INR"} so retrieval can filter numerically.
Step 3 — Deduplicate Before You Embed
Embedding costs money and duplicates poison retrieval. A two-tier approach works well:
import hashlib
from datasketch import MinHash, MinHashLSH
def exact_hash(text):
return hashlib.sha256(text.encode()).hexdigest()
lsh = MinHashLSH(threshold=0.85, num_perm=128)
def is_near_duplicate(doc_id: str, text: str) -> bool:
m = MinHash(num_perm=128)
for token in set(text.lower().split()):
m.update(token.encode())
if lsh.query(m):
return True
lsh.insert(doc_id, m)
return False
Exact hashes catch identical pages; MinHash-LSH catches template near-duplicates (the same product page across regional subdomains, for example). At web scale, expect to drop 30–40% of records here — that reduction improves answer quality.
Step 4 — Chunk with Metadata, Not Just Characters
Naive fixed-size chunking splits prices from product names and answers from questions. For commercial web data, chunk by record and by field, keeping metadata attached:
def chunk_product(record: dict) -> list[dict]:
base_meta = {
"url": record["url"],
"brand": record["brand"],
"category": record["category"],
"price": record["price"],
"currency": record["currency"],
"scraped_at": record["scraped_at"],
}
chunks = []
# Chunk 1: identity + price (always retrieved for price questions)
chunks.append({
"text": f"{record['title']} by {record['brand']}. "
f"Price: {record['price']} {record['currency']}. "
f"Category: {record['category']}.",
"meta": {**base_meta, "chunk_type": "identity"}
})
# Chunk 2+: description split at ~300 tokens with 40-token overlap
desc = clean_text(record["description"])
for i, piece in enumerate(split_tokens(desc, size=300, overlap=40)):
chunks.append({"text": piece,
"meta": {**base_meta, "chunk_type": "description", "part": i}})
return chunks
Rules of thumb we use in production: 200–400 tokens per chunk for product/listing data, 10–15% overlap, and always one compact "identity chunk" per record so entity-level questions retrieve cleanly.
Step 5 — Embed in Batches with Retry Logic
def embed_batch(texts: list[str], client, model="your-embedding-model",
batch_size=96, max_retries=3):
vectors = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i+batch_size]
for attempt in range(max_retries):
try:
resp = client.embeddings.create(model=model, input=batch)
vectors.extend([d.embedding for d in resp.data])
break
except Exception:
if attempt == max_retries - 1:
raise
time.sleep(2 ** attempt)
return vectors
Batching cuts cost and latency; exponential-backoff retries keep overnight ingestion jobs from dying on transient errors.
Step 6 — Idempotent Upserts into the Vector DB
Use deterministic IDs so re-runs update rather than duplicate. The pattern is identical across Pinecone, Qdrant, Weaviate, Milvus, and pgvector:
def make_id(record_url: str, chunk_type: str, part: int = 0) -> str:
return hashlib.md5(f"{record_url}|{chunk_type}|{part}".encode()).hexdigest()
# Qdrant-style upsert (pattern is portable)
client.upsert(
collection_name="products",
points=[{
"id": make_id(c["meta"]["url"], c["meta"]["chunk_type"], c["meta"].get("part", 0)),
"vector": vec,
"payload": {**c["meta"], "text": c["text"]},
} for c, vec in zip(chunks, vectors)]
)
Store the text and metadata in the payload — retrieval-time filters (category == "Earbuds", scraped_at > yesterday) are what turn a demo into a product.
Step 7 — The Refresh Loop (Where Most RAG Systems Fail)
Commercial web data decays fast: prices change daily, listings expire, menus rotate. A production refresh loop has three parts:
-
Scheduled re-scrape — hourly to weekly depending on data volatility (Actowiz feeds arrive on your chosen cadence)
-
Delta detection — compare content hashes; only re-embed changed records (typically 5–15% per cycle, which slashes embedding spend)
-
Tombstoning — mark or delete vectors for pages that disappeared, so the bot stops citing dead listings
def refresh(record: dict, stored_hash: str | None):
new_hash = exact_hash(record["description"] + str(record["price"]))
if new_hash == stored_hash:
return "skip"
upsert_chunks(chunk_product(record)) # re-embed only changed records
return "updated"
Common Pitfalls (and How to Avoid Them)
-
Embedding raw HTML → garbage retrieval; always extract structure first
-
One giant chunk per page → precise questions retrieve noise; use field-aware chunking
-
No timestamps in metadata → the bot can't prefer fresh data; always store scraped_at
-
Re-embedding everything nightly → cost explosion; use delta detection
-
Ignoring language tags → multilingual corpora need language metadata for filtered retrieval
