Web Data for AI Agents: How to Build a RAG Pipeline with Structured Web Data

AI agents that answer questions about the real world need one thing above all: structured, citable external data they can retrieve on demand. The most reliable way to deliver that data is a Retrieval-Augmented Generation (RAG) pipeline that fetches public web data, normalizes it into consistent JSON records, stores them in a searchable index, and injects only the relevant slices into the LLM context window at query time. This article shows you how to build that pipeline in Python, using environment variables for any scraper endpoint so nothing is hard-coded.

TL;DR

A production-grade web-data RAG pipeline has four stages: collect (fetch public data through a scraper API or worker), normalize (map heterogeneous HTML or JSON into a flat schema), index (store records so they are searchable by keyword or semantic similarity), and retrieve (select the top-k records relevant to a user query and pass them to the LLM as context). You can deploy the collection stage as a managed worker on the CoreClaw Workers platform, pick from ready-to-use scrapers in the CoreClaw store, and only pay for results returned — check CoreClaw pricing for current rates.

Why This Is Hard

LLMs are trained on static snapshots. When an agent is asked "What are the top-rated coffee shops near downtown Austin?" or "What is the current price of this Amazon product?", the model either guesses from stale training data or hallucinates. Neither is acceptable in production.

Three challenges make web-data RAG harder than textbook RAG:

1. Heterogeneous formats. Every website returns HTML with different structure. Even API responses vary in field names, nesting depth, and data types.
2. Freshness drift. Pages change layouts, products go out of stock, businesses close. A pipeline that worked yesterday may return empty fields today.
3. Context budget. An LLM context window is finite. Dumping 10,000 raw records into the prompt is both expensive and counterproductive — the model needs the right five records, not every record.

The solution is a normalization layer that converts messy web data into a predictable JSON schema before it ever touches the retrieval index.

What a Web-Data RAG Pipeline Looks Like

┌─────────┐    ┌────────────┐    ┌─────────┐    ┌──────────┐    ┌─────────┐
│ Scraper │───▶│ Normalizer │───▶│ Indexer │───▶│ Retriever │───▶│   LLM   │
│ (collect)│    │ (normalize)│    │ (store) │    │ (top-k)   │    │ (answer)│
└─────────┘    └────────────┘    └─────────┘    └──────────┘    └─────────┘
     ▲                                                        │
     │                  User query ────────────────────────────┘

• Scraper: Fetches public web data. Can be a managed API, a deployed worker, or a local script. Must respect rate limits and target-site terms.
• Normalizer: Transforms raw HTML/JSON into a flat, typed schema (e.g., {name, address, rating, url, source_domain, fetched_at}).
• Indexer: Stores normalized records. For small datasets, a JSON file with keyword search works. For larger ones, use SQLite or a vector database.
• Retriever: Given a user query, selects the top-k most relevant records.
• LLM: Receives the user query plus the retrieved records as context, and generates a grounded answer with citations.

Step-by-Step Implementation

Step 1 — Define the Normalized Schema

Start by deciding what your pipeline will produce. A flat schema is easier to index and debug than deeply nested JSON.

from dataclasses import dataclass, asdict
from datetime import datetime
from typing import Optional

@dataclass
class WebDataRecord:
    """Flat, typed record for RAG indexing."""
    title: str
    content: str            # description, review text, or summary
    url: str                # source URL for citation
    source_domain: str
    category: str           # e.g., "local_business", "product", "social_post"
    rating: Optional[float] = None
    price: Optional[str] = None      # keep as string — formats vary
    location: Optional[str] = None
    fetched_at: str = datetime.utcnow().isoformat()

    def to_dict(self) -> dict:
        return asdict(self)

Step 2 — Collect Web Data with Environment-Configured Endpoint

Use environment variables for every endpoint, API key, and configuration value. Never hard-code a URL you have not verified.

import os
import requests

SCRAPER_ENDPOINT = os.environ.get("SCRAPER_ENDPOINT", "")
SCRAPER_API_KEY = os.environ.get("SCRAPER_API_KEY", "")
TARGET_URL = os.environ.get("TARGET_URL", "https://example.com")

def fetch_raw_data(query: str, limit: int = 20) -> list[dict]:
    """Call a scraper endpoint and return raw JSON records.

    Set SCRAPER_ENDPOINT to the current API URL from your provider's
    console or documentation. Do not guess endpoints.
    """
    if not SCRAPER_ENDPOINT:
        raise ValueError(
            "Set SCRAPER_ENDPOINT env var to your scraper API URL. "
            "Copy the current endpoint from your provider's console."
        )

    headers = {"Authorization": f"Bearer {SCRAPER_API_KEY}"}
    params = {"query": query, "limit": limit}

    resp = requests.get(
        SCRAPER_ENDPOINT,
        headers=headers,
        params=params,
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json().get("results", [])

Step 3 — Normalize Raw Records into the Flat Schema

This is the layer that makes your pipeline robust. Every data source — whether it returns Google Maps results, Amazon products, or social media posts — gets mapped into the same WebDataRecord shape.

from urllib.parse import urlparse

def normalize_record(raw: dict, category: str = "general") -> WebDataRecord:
    """Map a raw scraper response into a flat WebDataRecord.

    Adapt field names to match what your scraper actually returns.
    The goal is consistency: every record has the same keys.
    """
    source_url = raw.get("url", raw.get("link", ""))
    domain = urlparse(source_url).netloc if source_url else "unknown"

    return WebDataRecord(
        title=raw.get("title", raw.get("name", "Untitled")),
        content=raw.get("description", raw.get("snippet", raw.get("text", ""))),
        url=source_url,
        source_domain=domain,
        category=category,
        rating=raw.get("rating"),
        price=str(raw.get("price", "")) if raw.get("price") else None,
        location=raw.get("address", raw.get("location")),
    )

def normalize_batch(raw_records: list[dict], category: str = "general") -> list[WebDataRecord]:
    return [normalize_record(r, category) for r in raw_records]

Step 4 — Index Records for Retrieval

For a lightweight pipeline, use SQLite with full-text search. This avoids the overhead of a vector database while still supporting keyword queries.

import sqlite3
import json

def create_index(db_path: str = "web_data.db") -> sqlite3.Connection:
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS records (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT,
            content TEXT,
            url TEXT,
            source_domain TEXT,
            category TEXT,
            rating REAL,
            price TEXT,
            location TEXT,
            fetched_at TEXT
        )
    """)
    conn.execute("""
        CREATE VIRTUAL TABLE IF NOT EXISTS records_fts
        USING fts5(title, content, content='records')
    """)
    conn.commit()
    return conn

def index_records(conn: sqlite3.Connection, records: list[WebDataRecord]) -> None:
    for r in records:
        d = r.to_dict()
        conn.execute(
            """INSERT INTO records
               (title, content, url, source_domain, category, rating, price, location, fetched_at)
               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
            (d["title"], d["content"], d["url"], d["source_domain"],
             d["category"], d["rating"], d["price"], d["location"], d["fetched_at"]),
        )
        conn.execute(
            "INSERT INTO records_fts (title, content) VALUES (?, ?)",
            (d["title"], d["content"]),
        )
    conn.commit()

Step 5 — Retrieve Top-K Records for the LLM Context

def retrieve_records(
    conn: sqlite3.Connection,
    query: str,
    top_k: int = 5,
) -> list[dict]:
    """Full-text search retrieval. Returns the top-k matching records."""
    cursor = conn.execute(
        """SELECT r.title, r.content, r.url, r.source_domain,
                  r.category, r.rating, r.price, r.location
           FROM records_fts f
           JOIN records r ON r.id = f.rowid
           WHERE records_fts MATCH ?
           ORDER BY rank
           LIMIT ?""",
        (query, top_k),
    )
    columns = [desc[0] for desc in cursor.description]
    return [dict(zip(columns, row)) for row in cursor.fetchall()]

Step 6 — Build the RAG Context Block

def build_context(records: list[dict]) -> str:
    """Format retrieved records as a context block for the LLM."""
    if not records:
        return "No relevant records found."

    blocks = []
    for i, r in enumerate(records, 1):
        block = f"[{i}] {r['title']}\n"
        if r.get("content"):
            block += f"    {r['content']}\n"
        if r.get("rating"):
            block += f"    Rating: {r['rating']}\n"
        if r.get("price"):
            block += f"    Price: {r['price']}\n"
        if r.get("location"):
            block += f"    Location: {r['location']}\n"
        block += f"    Source: {r['url']}\n"
        blocks.append(block)

    return "\n".join(blocks)

Putting It All Together

def rag_pipeline(query: str, scrape_query: str, category: str = "general") -> str:
    # 1. Collect
    raw = fetch_raw_data(scrape_query)

    # 2. Normalize
    records = normalize_batch(raw, category)

    # 3. Index
    conn = create_index()
    index_records(conn, records)

    # 4. Retrieve
    top_records = retrieve_records(conn, query, top_k=5)

    # 5. Build context
    context = build_context(top_records)

    conn.close()
    return context

# Example: run with environment variables set
#   export SCRAPER_ENDPOINT="https://your-verified-endpoint"
#   export SCRAPER_API_KEY="your-key"
#   export TARGET_URL="https://example.com"
#
#   context = rag_pipeline(
#       query="best coffee shops in Austin",
#       scrape_query="coffee shops downtown Austin",
#       category="local_business",
#   )
#   print(context)

Representative Output

When the pipeline runs successfully, the context block looks like:

[1] Blue Bottle Coffee
    Specialty coffee roaster with pour-over and espresso bar.
    Rating: 4.6
    Location: 2501 E 6th St, Austin, TX
    Source: https://maps.example.com/blue-bottle

[2] Houndstooth Coffee
    Espresso and pour-over focus with knowledgeable baristas.
    Rating: 4.5
    Location: 401 Congress Ave, Austin, TX
    Source: https://maps.example.com/houndstooth

[3] Flat Track Coffee
    Neighborhood espresso bar with pastries and bagels.
    Rating: 4.7
    Location: 1119 E 11th St, Austin, TX
    Source: https://maps.example.com/flat-track

The LLM receives this block alongside the user's question and can answer with citations: "Based on the top-rated results, Flat Track Coffee has the highest rating at 4.7, followed by Blue Bottle Coffee at 4.6."

Business Use Cases

Use Case	Data Source	RAG Value
Customer support agent	Product pages, FAQ pages	Agent answers with current specs, not stale training data
Sales intelligence bot	Google Maps, LinkedIn	Agent surfaces local business leads with addresses and ratings
Price comparison assistant	Amazon, Walmart listings	Agent compares current prices across sellers
Content research tool	YouTube, TikTok metadata	Agent summarizes trending topics with creator attribution
SEO monitoring agent	Google SERP results	Agent tracks ranking changes and reports competitor moves

Build vs Buy: Collection Layer

Dimension	Self-Hosted Scraper	Managed Worker (CoreClaw)
Setup model	Provision servers, write scraper code, manage proxies	Deploy a pre-built worker from the store
Data coverage	Only what you build scrapers for	Choose from 100+ ready-made workers
Maintenance burden	Fix broken scrapers when layouts change	Provider handles maintenance
Proxy management	You rotate, validate, and pay for proxies	Included in the platform
Output format	Whatever your code produces	Standardized JSON
Pricing model	Infrastructure + development time	Pay per result — verify current rates on the pricing page

The normalization, indexing, and retrieval layers are worth building yourself — they are tightly coupled to your application logic. The collection layer is where a managed platform saves the most time, because it eliminates proxy management, layout-fix cycles, and infrastructure scaling. You can deploy a new worker or browse the workers store to see what is available.

Limitations and Compliance

• Freshness. Web data changes continuously. Schedule re-collection at intervals appropriate to your use case — hourly for prices, daily for reviews, weekly for business directories.
• Coverage. No scraper covers every site. Verify that your target domains are supported before relying on the pipeline in production.
• Context window limits. Even with retrieval, large records can exceed the LLM's context budget. Trim content fields and cap top_k at 5-10.
• Citation accuracy. Always pass the source URL in the context block so the LLM can cite it. Never let the model answer without a source.
• Terms and law. Respect each target site's terms of service, robots directives, and applicable data-protection law. Only collect public web data. Do not access private content, bypass authentication, or evade CAPTCHAs.
• Schema drift. When a scraper's output fields change, the normalizer must be updated. Build a validator that checks every record against the expected schema and logs mismatches.

FAQ

1. Do I need a vector database for web-data RAG?
No. For keyword-driven queries (product names, business categories, addresses), SQLite full-text search is sufficient and far simpler. Use vector embeddings when queries are semantic ("find me cozy coffee shops" where "cozy" is not a literal keyword match).

2. How often should I re-collect web data for my agent?
It depends on volatility. Price data may need hourly updates. Business directory data changes weekly or monthly. Build a scheduler (cron, Celery, or a managed worker) that runs collection at the interval your use case requires.

3. Can I connect this pipeline to an MCP client like Claude or Cursor?
Yes. The retriever function can be exposed as an MCP tool. An AI agent calls the tool with a query, receives the top-k records as structured context, and generates a grounded answer. See Day 2 of this series for an MCP server setup walkthrough.

4. What happens when a scraper's output fields change?
The normalizer maps field names dynamically (e.g., it checks both title and name), but if a field is removed entirely, the record will have empty values. Build a schema validator that logs records with missing critical fields so you can fix the normalizer.

5. How do I prevent the LLM from hallucinating beyond the retrieved records?
Include an instruction in the system prompt: "Answer only using the provided context records. If the context does not contain the answer, say you do not have current data." Always pass source URLs so the model can cite them.

6. Can I use this for multiple data sources at once?
Yes. Run separate collection jobs for each source (Google Maps, Amazon, YouTube), normalize them all into WebDataRecord with different category values, and index them in the same SQLite database. The retriever will surface mixed results ranked by relevance.

7. What is the simplest way to start without managing scraper infrastructure?
Browse the CoreClaw workers store for a pre-built scraper that matches your data source, deploy it from the console, and pipe its JSON output directly into the normalizer. This lets you focus on the retrieval and LLM layers while the collection layer runs as a managed worker.

Summary

A web-data RAG pipeline turns static LLMs into agents that can answer questions about the real world with citations. The pipeline has four stages — collect, normalize, index, retrieve — and the collection stage is where a managed platform like CoreClaw saves the most engineering time. Start with SQLite FTS for the index, build a flat schema for the normalizer, and always pass source URLs in the context block so the LLM can cite them.

Related Reading

• CoreClaw Workers platform — deploy scrapers without DevOps
• CoreClaw workers store — 100+ ready-to-use scrapers
• CoreClaw pricing — pay per result