RAG 知识库构建中的代理与抓取：从定时采写到向量化落库的技术实现

┌─────────────────┐     ┌──────────────────┐     ┌─────────────────┐
│  调度器          │────▶│  抓取 Worker      │────▶│  代理服务        │
│  (定时/队列)     │     │  (代理+重试+限速) │     │  (动态代理 API)  │
└─────────────────┘     └────────┬────────┘     └─────────────────┘
                                  │
                                  ▼
┌─────────────────┐     ┌──────────────────┐     ┌─────────────────┐
│  去重/变更检测   │◀────│  原始 HTML/文本   │     │  目标站点        │
└────────┬────────┘     └──────────────────┘     └─────────────────┘
         │
         ▼
┌─────────────────┐     ┌──────────────────┐
│  解析与分块      │────▶│  向量化 + 落库     │  ← RAG 检索使用
└─────────────────┘     └──────────────────┘

# pip install requests beautifulsoup4 openai  # 向量化示例用 OpenAI，可换成其他
import os
import time
import hashlib
import requests
from bs4 import BeautifulSoup
from typing import Iterator

# 代理网关（替换为实际代理服务地址，如 BytesFlows）
PROXY_GATEWAY = os.getenv("PROXY_GATEWAY", "http://user:pass@proxy.example.com:8080")
REQUEST_TIMEOUT = 30
MAX_RETRIES = 3
RETRY_BACKOFF = 2
CHUNK_SIZE = 500       # 字符数，按需调整
CHUNK_OVERLAP = 50    # 块间重叠

def fetch_with_proxy(url: str) -> str | None:
    proxies = {"http": PROXY_GATEWAY, "https": PROXY_GATEWAY}
    for attempt in range(MAX_RETRIES):
        try:
            r = requests.get(
                url,
                proxies=proxies,
                timeout=REQUEST_TIMEOUT,
                headers={"User-Agent": "RAG-Crawler/1.0 (Knowledge Base)"},
            )
            if r.status_code == 429:
                time.sleep(RETRY_BACKOFF ** attempt)
                continue
            r.raise_for_status()
            return r.text
        except requests.RequestException:
            time.sleep(RETRY_BACKOFF ** attempt)
    return None

def extract_text(html: str) -> str:
    soup = BeautifulSoup(html, "html.parser")
    for tag in soup(["script", "style"]):
        tag.decompose()
    return soup.get_text(separator="\n", strip=True)

def content_fingerprint(text: str) -> str:
    return hashlib.sha256(text.encode()).hexdigest()

def chunk_text(text: str, size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> list[str]:
    chunks = []
    start = 0
    while start < len(text):
        end = start + size
        chunks.append(text[start:end])
        start = end - overlap
    return chunks

# 此处用占位表示“写入向量库”；实际替换为 Milvus/Chroma/OpenAI 等
def embed_and_store(chunks: list[str], url: str, doc_id: str):
    for i, chunk in enumerate(chunks):
        # 调用 Embedding API 得到向量（示例省略）
        # vector = openai.Embedding.create(input=chunk, model="text-embedding-3-small")["data"][0]["embedding"]
        # vector_db.upsert(collection, [{"id": f"{doc_id}_{i}", "url": url, "text": chunk, "vector": vector}])
        print(f"  stored chunk {i+1}/{len(chunks)} for {url}")

def process_one_url(url: str, seen_fingerprints: set) -> bool:
    html = fetch_with_proxy(url)
    if not html:
        return False
    text = extract_text(html)
    fp = content_fingerprint(text)
    if fp in seen_fingerprints:
        return True  # 跳过重复内容
    seen_fingerprints.add(fp)
    chunks = chunk_text(text)
    doc_id = hashlib.sha256(url.encode()).hexdigest()[:12]
    embed_and_store(chunks, url, doc_id)
    return True

def run_scheduled_crawl(urls: list[str]):
    seen = set()
    for url in urls:
        ok = process_one_url(url, seen)
        print(f"{'OK' if ok else 'FAIL'}: {url}")
        time.sleep(1)  # 简单限速，生产可用 token bucket 等

if __name__ == "__main__":
    urls = [
        "https://example.com/docs/page1",
        "https://example.com/docs/page2",
    ]
    run_scheduled_crawl(urls)