如何使用 Scrapeless 和 Ollama 构建本地 RAG 管道

pip install sentence-transformers chromadb beautifulsoup4 requests
ollama pull qwen2.5:0.5b
export SCRAPELESS_API_KEY="sk_your_scrapeless_key"

# fetch.py -- 通过 Scrapeless 通用抓取 API 拉取渲染的作者传记页面
import os
import pathlib

import requests

ENDPOINT = "https://api.scrapeless.com/api/v2/unlocker/request"
HEADERS = {
    "Content-Type": "application/json",
    "x-api-token": os.environ["SCRAPELESS_API_KEY"],
}
SOURCE_HOST = "https://quotes.toscrape.com"

AUTHORS = ["Albert-Einstein", "Jane-Austen", "Marilyn-Monroe", "J-K-Rowling", "Thomas-A-Edison"]

pathlib.Path("pages").mkdir(exist_ok=True)
for slug in AUTHORS:
    url = f"{SOURCE_HOST}/author/{slug}"
    resp = requests.post(
        ENDPOINT,
        headers=HEADERS,
        json={"actor": "unlocker.webunlocker", "input": {"url": url, "js_render": False, "redirect": True}},
        timeout=120,
    )
    resp.raise_for_status()
    html = resp.json()["data"]
    pathlib.Path("pages", f"{slug}.html").write_text(html, encoding="utf-8")
    print(f"{slug}: {len(html):,} bytes")

Albert-Einstein: 5,329 bytes
Jane-Austen: 3,413 bytes
Marilyn-Monroe: 3,663 bytes
J-K-Rowling: 5,347 bytes
Thomas-A-Edison: 2,648 bytes

# chunk_corpus.py -- pages/*.html -> corpus.jsonl (带来源的重叠单词窗口块)
import json
import pathlib
import re

from bs4 import BeautifulSoup

CHUNK_WORDS = 60
OVERLAP_WORDS = 15


def extract_author(html: str) -> dict:
    soup = BeautifulSoup(html, "html.parser")
    name = soup.find("h3", class_="author-title").get_text(strip=True)
    born_date = soup.find("span", class_="author-born-date").get_text(strip=True)
    born_loc = soup.find("span", class_="author-born-location").get_text(strip=True)
    desc = soup.find("div", class_="author-description").get_text(" ", strip=True)
    return {"name": name, "born_date": born_date, "born_location": born_loc,
            "description": re.sub(r"\s+", " ", desc)}


def chunk(words: list[str]):
    step = CHUNK_WORDS - OVERLAP_WORDS
    for start in range(0, max(len(words) - OVERLAP_WORDS, 1), step):
        yield start, " ".join(words[start:start + CHUNK_WORDS])


total = 0
with open("corpus.jsonl", "w", encoding="utf-8") as out:
    for page in sorted(pathlib.Path("pages").glob("*.html")):
        author = extract_author(page.read_text(encoding="utf-8"))
        lead = f"{author['name']} was born {author['born_date']} {author['born_location']}."
        words = f"{lead} {author['description']}".split()
        for start, body in chunk(words):
            out.write(json.dumps({"id": f"{page.stem}-{start}", "source": page.stem,
                                   "author": author["name"], "word_offset": start,
                                   "text": body}) + "\n")
            total += 1
print(f"{total} chunks -> corpus.jsonl")

阿尔伯特·爱因斯坦: 615 words -> 14 chunks
J.K. 罗琳: 644 words -> 14 chunks
简·奥斯汀: 327 words -> 7 chunks
玛丽莲·梦露: 376 words -> 9 chunks
托马斯·爱迪生: 195 words -> 4 chunks
48 chunks -> corpus.jsonl

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("all-MiniLM-L6-v2")
vectors = model.encode(texts, show_progress_bar=False).tolist()

import chromadb

client = chromadb.PersistentClient(path=".chroma")
collection = client.create_collection("author_bios")
collection.add(
    ids=[c["id"] for c in chunks],
    documents=[c["text"] for c in chunks],
    embeddings=vectors,
    metadatas=[{"author": c["author"], "source": c["source"], "word_offset": c["word_offset"]}
               for c in chunks],
)

嵌入了48个块，维度384，集合数量48

query_vector = model.encode([question]).tolist()
result = collection.query(query_embeddings=query_vector, n_results=2)
top_author = result["metadatas"][0][0]["author"]
top_distance = result["distances"][0][0]

问：阿尔伯特·爱因斯坦出生在哪里？
  顶部匹配作者：阿尔伯特·爱因斯坦 | 距离：0.6465
  块：阿尔伯特·爱因斯坦于1879年3月14日出生于德国乌尔姆。在1879年，阿尔伯特·爱因斯坦出生于德国乌尔姆。他在苏黎世大学完成了他的博士学位...
问：J.K.罗琳出生在哪里？
  顶部匹配作者：J.K.罗琳 | 距离：0.4566
  块：J.K.罗琳于1965年7月31日出生于英格兰南格洛斯特郡耶特，联合王国。另见：罗伯特·加尔布雷斯尽管她以笔名写作...

import json
import requests

prompt = (
    '在下面的上下文中仅用一句简短的句子回答问题。 '
    '只回复JSON {"answer": "..."}.\n\n'
    f"上下文：\n{context}\n\n问题：{question}"
)
resp = requests.post(
    "http://localhost:11434/api/generate",
    json={"model": "qwen2.5:0.5b", "prompt": prompt, "stream": False,
          "format": "json", "options": {"temperature": 0, "num_predict": 40}},
    timeout=600,
)
answer = json.loads(resp.json()["response"])

检索块的作者（来自向量存储元数据）：阿尔伯特·爱因斯坦
问题：阿尔伯特·爱因斯坦出生在哪里？是哪一年？
生成的答案：阿尔伯特·爱因斯坦于1879年3月14日出生于德国乌尔姆。

# full_pipeline.py -- fetch -> chunk -> embed -> store -> retrieve -> generate, end to end
import json
import logging
import os
import pathlib
import re

os.environ.setdefault("HF_HOME", "/usr/local/share/hf-cache")
os.environ.setdefault("HF_HUB_OFFLINE", "1")

import chromadb
import requests
from bs4 import BeautifulSoup
from sentence_transformers import SentenceTransformer

logging.getLogger("chromadb.telemetry.product.posthog").setLevel(logging.CRITICAL)

# 阶段 1：抓取
ENDPOINT = "https://api.scrapeless.com/api/v2/unlocker/request"
HEADERS = {"Content-Type": "application/json", "x-api-token": os.environ["SCRAPELESS_API_KEY"]}
SOURCE_HOST = "https://quotes.toscrape.com"
AUTHORS = ["Albert-Einstein", "Jane-Austen", "Marilyn-Monroe", "J-K-Rowling", "Thomas-A-Edison"]

pathlib.Path("pages").mkdir(exist_ok=True)
for slug in AUTHORS:
    url = f"{SOURCE_HOST}/author/{slug}"
    resp = requests.post(
        ENDPOINT, headers=HEADERS,
        json={"actor": "unlocker.webunlocker", "input": {"url": url, "js_render": False, "redirect": True}},
        timeout=120,
    )
    resp.raise_for_status()
    html = resp.json()["data"]
    pathlib.Path("pages", f"{slug}.html").write_text(html, encoding="utf-8")
    print(f"{slug}: {len(html):,} 字节")

# 阶段 2：提取和切分
CHUNK_WORDS, OVERLAP_WORDS = 60, 15


def extract_author(html: str) -> dict:
    soup = BeautifulSoup(html, "html.parser")
    name = soup.find("h3", class_="author-title").get_text(strip=True)
    born_date = soup.find("span", class_="author-born-date").get_text(strip=True)
    born_loc = soup.find("span", class_="author-born-location").get_text(strip=True)
    desc = soup.find("div", class_="author-description").get_text(" ", strip=True)
    return {"name": name, "born_date": born_date, "born_location": born_loc,
            "description": re.sub(r"\s+", " ", desc)}


def chunk_words(words: list[str]):
    step = CHUNK_WORDS - OVERLAP_WORDS
    for start in range(0, max(len(words) - OVERLAP_WORDS, 1), step):
        yield start, " ".join(words[start:start + CHUNK_WORDS])


chunks = []
with open("corpus.jsonl", "w", encoding="utf-8") as out:
    for page in sorted(pathlib.Path("pages").glob("*.html")):
        author = extract_author(page.read_text(encoding="utf-8"))
        lead = f"{author['name']} 出生于 {author['born_date']} {author['born_location']}。"
        words = f"{lead} {author['description']}".split()
        n_chunks = 0
        for start, body in chunk_words(words):
            rec = {"id": f"{page.stem}-{start}", "source": page.stem, "author": author["name"],
                   "word_offset": start, "text": body}
            out.write(json.dumps(rec) + "\n")
            chunks.append(rec)
            n_chunks += 1
        print(f"{author['name']}：{len(words)} 个单词 -> {n_chunks} 个块")
print(f"{len(chunks)} 个块 -> corpus.jsonl")

# 阶段 3 和 4：嵌入和存储
model = SentenceTransformer("all-MiniLM-L6-v2")
client = chromadb.PersistentClient(path=".chroma", settings=chromadb.Settings(anonymized_telemetry=False))
if "author_bios" in [c.name for c in client.list_collections()]:
    client.delete_collection("author_bios")
collection = client.create_collection("author_bios")

texts = [c["text"] for c in chunks]
vectors = model.encode(texts, show_progress_bar=False).tolist()
collection.add(
    ids=[c["id"] for c in chunks],
    documents=texts,
    embeddings=vectors,
    metadatas=[{"author": c["author"], "source": c["source"], "word_offset": c["word_offset"]} for c in chunks],
)
print(f"嵌入了 {len(chunks)} 个块，维度 {len(vectors[0])}，集合数量 {collection.count()}")

# 阶段 5：检索
for question in ["阿尔伯特·爱因斯坦出生在哪里？", "J.K. 罗琳出生在哪里？"]:
    query_vector = model.encode([question]).tolist()
    result = collection.query(query_embeddings=query_vector, n_results=2)
    top_author = result["metadatas"][0][0]["author"]
    top_distance = result["distances"][0][0]
    top_text = result["documents"][0][0]
    print(f"问: {question}")
    print(f"  最佳匹配作者: {top_author} | 距离: {top_distance:.4f}")
    print(f"  块: {top_text[:160]}...")

# 阶段 6：生成
question = "阿尔伯特·爱因斯坦出生在哪里，是哪一年？"
query_vector = model.encode([question]).tolist()
result = collection.query(query_embeddings=query_vector, n_results=1)
context = result["documents"][0][0]
retrieved_author = result["metadatas"][0][0]["author"]

prompt = (
    "用以下上下文仅用一短句回答问题。"
    '仅回复 JSON {"answer": "..."}。\n\n'
    f"上下文：\n{context}\n\n问题：{question}"
)
resp = requests.post(
    "http://localhost:11434/api/generate",
    json={"model": "qwen2.5:0.5b", "prompt": prompt, "stream": False,
          "format": "json", "options": {"temperature": 0, "num_predict": 40}},
    timeout=600,
)
resp.raise_for_status()
answer = json.loads(resp.json()["response"])
```python
print("从向量存储元数据中检索到的块作者：", retrieved_author)
print("问题：", question)
print("生成的答案：", answer["answer"])