带代理的 RAG 爬虫：采集、证据、切分与索引版本化

{
  "sourceId": "product-docs",
  "baseUrl": "https://docs.example.com/",
  "allowedPrefixes": ["https://docs.example.com/guide/"],
  "deniedPrefixes": ["https://docs.example.com/account/"],
  "robotsReviewedAt": "2026-08-01",
  "termsReviewedAt": "2026-08-01",
  "defaultCollector": "http",
  "maxConcurrency": 2,
  "minDelayMs": 1000,
  "freshnessHours": 24,
  "owner": "knowledge-platform"
}

{
  "url": "https://docs.example.com/guide/install",
  "sourceId": "product-docs",
  "priority": 50,
  "nextFetchAt": "2026-08-07T01:00:00Z",
  "lastStatus": 200,
  "etag": "...",
  "lastModified": "...",
  "contentSha256": "...",
  "failureCount": 0,
  "leaseUntil": null
}

from dataclasses import dataclass
import httpx

MAX_BODY_BYTES = 5 * 1024 * 1024
ALLOWED_CONTENT_TYPES = ("text/html", "text/plain", "application/xhtml+xml")

@dataclass
class CrawlItem:
    url: str
    etag: str | None = None
    last_modified: str | None = None

async def fetch(item: CrawlItem, client: httpx.AsyncClient) -> dict:
    headers: dict[str, str] = {}
    if item.etag:
        headers["If-None-Match"] = item.etag
    if item.last_modified:
        headers["If-Modified-Since"] = item.last_modified

    async with client.stream("GET", item.url, headers=headers) as response:
        if response.status_code == 304:
            return {"type": "not_modified", "url": item.url}

        response.raise_for_status()

        content_type = response.headers.get("content-type", "").lower()
        if not any(t in content_type for t in ALLOWED_CONTENT_TYPES):
            raise ValueError(f"unsupported content type: {content_type!r}")

        body = bytearray()
        async for chunk in response.aiter_bytes():
            body.extend(chunk)
            if len(body) > MAX_BODY_BYTES:
                raise ValueError("response body exceeds 5 MiB limit")

        return {
            "type": "fetched",
            "url": str(response.url),
            "body": bytes(body),
            "etag": response.headers.get("etag"),
            "last_modified": response.headers.get("last-modified"),
            "content_type": content_type,
        }

async def example(item: CrawlItem) -> dict:
    timeout = httpx.Timeout(15.0, connect=5.0)
    limits = httpx.Limits(max_connections=20, max_keepalive_connections=10)
    async with httpx.AsyncClient(
        follow_redirects=True,
        timeout=timeout,
        limits=limits,
        headers={"User-Agent": "AuthorizedKnowledgeCrawler/1.0"},
    ) as client:
        return await fetch(item, client)

{
  "policy": "direct-first",
  "allowedRoutes": ["direct", "residential"],
  "country": "DE",
  "sessionMode": "rotating",
  "maxRouteAttempts": 2
}

{
  "documentId": "product-docs:install:v17",
  "sourceUrl": "https://docs.example.com/guide/install",
  "title": "安装指南",
  "language": "zh",
  "sections": [
    {
      "headingPath": ["安装", "Linux"],
      "text": "...",
      "sourceOffsets": [1200, 2040]
    }
  ]
}

{
  "indexVersion": "kb-2026-08-07-01",
  "embeddingModel": "model-name-version",
  "documents": 1842,
  "chunks": 11240,
  "sourceSnapshot": "snapshot-2026-08-07",
  "createdAt": "2026-08-07T00:18:00+08:00"
}