RAG Crawler with Proxies: Technical Architecture, Freshness, and Evidence

A. Business definition and success criteria

Target teams: AI platform, knowledge engineering, search, research, and data engineering teams that maintain a RAG corpus from public web sources they are permitted to access.

Inputs: approved source registry, URL/frontier seeds, requested geography, locale, freshness SLO, extraction schema, retention policy, and index version policy.

Outputs: versioned source observations, immutable evidence, normalized documents, deduplicated chunks, candidate embeddings/indexes, retrieval evaluation results, and an auditable promotion decision.

Update frequency: source-specific. Use documented freshness requirements, change history, validators/feeds, contractual limits, and server guidance rather than one universal recrawl interval.

Success means: the system retrieves the intended public source from the intended market, preserves provenance, detects meaningful change, avoids duplicate indexing, classifies failures correctly, and promotes only an evaluated candidate index. Page-count alone is not a success metric.

Use proxies when: an approved source legitimately varies by country/region/city, direct egress does not represent the market being measured, or distributed collection is an explicit system requirement.

Do not use proxies to: bypass login, access control, paywalls, CAPTCHAs, explicit denial, or platform security controls. Prefer official APIs, feeds, exports, or licensed datasets when they meet the requirement.

B. End-to-end system architecture

BytesFlows owns the proxy routing layer. Source policy, scheduling, parsing, evidence, storage, index promotion, monitoring, and compliance decisions remain in the customer's system.

C. Dynamic proxy strategy

Use rotation for independent observations where continuity is not required: separate public pages, independent freshness checks, and market sampling jobs. Use a bounded sticky session only for an approved multi-step browser workflow that must preserve the same market identity across navigation.

A route contract should be explicit:

{
  "route_mode": "residential",
  "requested_geo": {"country": "DE", "region": null, "city": null},
  "session_mode": "rotating",
  "sticky_ttl_seconds": null,
  "max_route_attempts": 2,
  "domain_concurrency": 2
}

If country, region, or city is requested, validate the observed exit and the actual localized page output before trusting the observation. Keep requested_geo and observed_geo as separate fields.

Do not rotate merely because a target returns 401/403/404/429, a parser fails, or a CAPTCHA/security challenge appears. Those states require classification and policy review, not automatic IP churn. Retire a proxy route from the current task only for proxy-layer failures such as authentication failure, connection failure, unusable exit, or verified wrong GEO; keep retry counts bounded.

Concurrency is a customer-side policy: cap per-domain work independently from global worker capacity. Sticky sessions must have an explicit TTL and cancellation path; never keep them alive indefinitely.

D. Request and task scheduling design

A production job should have an idempotency key such as source_id + canonical_url + market + freshness_window. Acquire a short lease before execution so two workers cannot process the same observation simultaneously.

Rate limiting belongs at the domain/source policy layer. Add token-bucket or leaky-bucket controls, bounded exponential backoff with jitter, a circuit breaker for sustained target failures, cancellation tokens for stale jobs, and a DLQ for cases that require review.

E. Runnable implementation skeleton

The following Python example is a coherent implementation skeleton using httpx. It has not been executed or benchmarked by BytesFlows; validate it in your environment before production use.

import asyncio, hashlib, json, os, random, time
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from typing import Optional
import httpx

PROXY_URL = os.environ["BYTESFLOWS_PROXY_URL"]  # e.g. http://USER:PASSWORD@HOST:PORT
MAX_ATTEMPTS = 3

@dataclass
class Job:
    source_id: str
    url: str
    country: str
    region: Optional[str] = None
    city: Optional[str] = None

@dataclass
class Result:
    source_id: str
    source_url: str
    fetched_at: str
    requested_geo: dict
    http_status: Optional[int]
    content_sha256: Optional[str]
    final_url: Optional[str]
    error_class: Optional[str]
    attempt: int

def classify(status: int) -> str:
    if status == 407: return "proxy_auth"
    if status in (401, 403): return "policy_review"
    if status == 429: return "target_rate_limit"
    if status in (404, 410): return "permanent_target"
    if 500 <= status <= 599: return "target_transient"
    if 200 <= status <= 299: return "ok"
    return "target_other"

async def collect(job: Job) -> Result:
    # Put GEO/session parameters in the BytesFlows username or endpoint according
    # to the current product configuration. Never hard-code credentials in source.
    timeout = httpx.Timeout(connect=10.0, read=30.0, write=15.0, pool=10.0)
    last_error = None

    for attempt in range(1, MAX_ATTEMPTS + 1):
        try:
            async with httpx.AsyncClient(proxy=PROXY_URL, timeout=timeout, follow_redirects=True) as client:
                r = await client.get(job.url, headers={"User-Agent": "YOUR-COMPLIANT-CRAWLER/1.0"})
            cls = classify(r.status_code)
            if cls == "ok":
                digest = hashlib.sha256(r.content).hexdigest()
                return Result(job.source_id, job.url, datetime.now(timezone.utc).isoformat(),
                    {"country": job.country, "region": job.region, "city": job.city},
                    r.status_code, digest, str(r.url), None, attempt)
            if cls in {"policy_review", "permanent_target"}:
                return Result(job.source_id, job.url, datetime.now(timezone.utc).isoformat(),
                    {"country": job.country, "region": job.region, "city": job.city},
                    r.status_code, None, str(r.url), cls, attempt)
            last_error = cls
            retry_after = r.headers.get("retry-after")
            delay = float(retry_after) if retry_after and retry_after.isdigit() else min(30, 2 ** attempt + random.random())
        except (httpx.ConnectError, httpx.ConnectTimeout, httpx.ReadTimeout) as exc:
            last_error = type(exc).__name__
            delay = min(30, 2 ** attempt + random.random())

        if attempt < MAX_ATTEMPTS:
            await asyncio.sleep(delay)

    return Result(job.source_id, job.url, datetime.now(timezone.utc).isoformat(),
        {"country": job.country, "region": job.region, "city": job.city},
        None, None, None, last_error or "unknown", MAX_ATTEMPTS)

async def main():
    job = Job("docs-example", "https://example.com/public-doc", "DE")
    result = await collect(job)
    print(json.dumps(asdict(result), ensure_ascii=False))

if __name__ == "__main__":
    asyncio.run(main())

Credentials belong in a secret manager or environment injection path. Redact proxy passwords, authorization headers, cookies, and personal data from logs.

F. Data quality and evidence

Store an observation independently from the normalized document:

{
  "observation_id": "obs_01",
  "source_id": "docs-example",
  "source_url": "https://example.com/public-doc",
  "requested_geo": {"country": "DE"},
  "observed_geo": {"country": "DE", "region": null, "city": null, "verified": true},
  "fetched_at": "2026-08-21T00:00:00Z",
  "http_status": 200,
  "final_url": "https://example.com/public-doc",
  "content_sha256": "...",
  "evidence_uri": "object://evidence/obs_01",
  "parser_version": "docs-v4",
  "document_version": "doc-17",
  "quality_state": "accepted"
}

Preserve raw response bytes or an approved normalized snapshot, selected provenance headers, retrieval timestamp, redirect/final URL, collector version, route metadata with secrets removed, and screenshots only when visual evidence is justified.

Deduplicate at multiple levels: canonical URL, response/content hash, normalized document hash, and chunk hash. An unchanged hash should not create a new embedding version. A changed hash should not automatically replace the active index: validate extraction first.

Treat empty results, missing required sections, unexpected language, MIME changes, and large field-count drops as anomalies. Record partial success explicitly rather than converting it into a clean success or overwriting the last good version.

G. Production reliability

Useful metrics include jobs_started, jobs_completed, collection_success_rate, request_latency, retry_rate, proxy_auth_failures, proxy_connect_failures, target_429, target_5xx, wrong_geo, parser_failures, schema_drift, evidence_write_failures, content_change_rate, embedding_queue_depth, and index_promotion_failures. Set thresholds from your own SLOs and observed baseline; this article does not claim universal performance numbers.

Log fields should include job_id, source_id, url_hash, market, session_mode, attempt, route_result, http_status, error_class, latency_ms, parser_version, content_hash, evidence_id, and trace_id, with secrets removed.

Alert on sustained authentication failures, wrong-GEO observations, evidence-store failures, parser drift, queue age, and repeated target 5xx/429. Keep representative failed evidence samples under a documented retention policy. Route ambiguous content changes and parser drift to human review.

Define an error budget for each source tier. When a source exceeds it, reduce or pause scheduling instead of amplifying retries.

H. Security, privacy, and compliance

RFC 9309 defines robots.txt as crawler control, not authorization. Review robots directives together with terms, contractual permission, privacy obligations, and applicable law. A proxy changes network routing; it does not grant access rights.

Minimize collected fields. Avoid personal data unless it is necessary, permitted, and governed. Encrypt evidence at rest, apply least-privilege access, define retention/deletion periods, and propagate source permissions to chunks and retrieval indexes. Treat collected web content as untrusted input to downstream AI systems.

Stop collection on revoked permission, persistent access-control responses, unexpected sensitive-data exposure, or evidence corruption. Do not provide workflows for bypassing login, paywalls, CAPTCHA, or security controls.

I. Launch checklist and scaling path

Development

Pre-production

Production

Scale by adding partitioned queues, per-market workers, distributed leases, object storage for evidence, a versioned metadata store, parser registry, evaluation service, and independent index build/promotion workers. Do not scale request concurrency until data quality and source-specific error budgets are stable.

Stop/rollback conditions: revoke a source when permission changes; pause a domain on sustained policy responses or error-budget breach; roll back a parser when schema drift corrupts normalized output; reject or roll back an index when retrieval evaluation regresses.

J. Conversion design

Before buying traffic, validate three things on a small approved corpus: (1) the required pages actually differ by GEO, (2) the requested BytesFlows route produces the intended observed market, and (3) your evidence and retry pipeline classifies failures correctly.

Primary CTA: Run a Proxy Test before connecting the route to a production RAG crawler.

Supporting BytesFlows resources:

• AI Data Collection Solution
• Locations
• Production Web Scraping Pipeline
• Python Scraping Proxy Setup

Verification boundaries

The architecture, code, schemas, and operating guidance above are implementation guidance. The Python example is 未执行实测 / not execution-tested in this workflow. No BytesFlows benchmark, success rate, latency, customer result, proxy-pool size, anti-block guarantee, or target-specific availability is asserted here.