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 Universal Scraping 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")

アルバート・アインシュタイン: 5,329 bytes
ジェーン・オースティン: 3,413 bytes
マリリン・モンロー: 3,663 bytes
J.K.ローリング: 5,347 bytes
トーマス・A・エジソン: 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
トーマス・A・エジソン: 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]

Q: アルベルト・アインシュタインはどこで生まれましたか？
  トップ一致作者: アルベルト・アインシュタイン | 距離: 0.6465
  チャンク: アルベルト・アインシュタインは1879年3月14日にドイツのウルムで生まれました。1879年にアルベルト・アインシュタインはドイツのウルムで生まれました。彼はチューリッヒ大学で博士号を取得しました...
Q: 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 -- フェッチ -> チャンク -> 埋め込む -> 保存する -> 取得する -> 生成する、エンドツーエンド
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"Q: {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"])
```ja
print("取得したチャンクの著者（ベクトルストアのメタデータから）:", retrieved_author)
print("質問:", question)
print("生成された回答:", answer["answer"])