为 RAG 构建网页知识库：从内容采集到检索与引用回查

export DOC_URL_1='<paste-the-actual-doc-url-1>'
export DOC_URL_2='<paste-the-actual-doc-url-2>'

python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install crawl4ai==0.9.5 beautifulsoup4
python -m playwright install chromium

import argparse
import asyncio
import hashlib
import json
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from bs4 import BeautifulSoup
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode
 
DB = Path("knowledge.sqlite")
SCHEMA = """
CREATE TABLE IF NOT EXISTS documents(
  url TEXT PRIMARY KEY, title TEXT NOT NULL,
  observed_at TEXT NOT NULL, content_hash TEXT NOT NULL
);
CREATE VIRTUAL TABLE IF NOT EXISTS chunks USING fts5(
  chunk_id UNINDEXED, source_url UNINDEXED, title, heading,
  section_no UNINDEXED, body, tokenize='trigram'
);
"""
 
def parse_sections(html):
    soup = BeautifulSoup(html, "html.parser")
    root = soup.select_one("main, article, div.body")
    if root is None or root.find("h1") is None:
        raise ValueError("没有识别到文档正文与标题")
    for unwanted in root.select("nav, script, style"):
        unwanted.decompose()
    title = root.find("h1").get_text(" ", strip=True)
    sections, heading, parts, number = [], title, [], 0
    for element in root.find_all(["h1", "h2", "h3", "p", "pre", "table", "li", "dt", "dd"]):
        if element.find_parent(["pre", "table", "li", "dt", "dd"]):
            continue
        if element.name in {"h1", "h2", "h3"}:
            if parts:
                sections.append((number, heading, "\n\n".join(parts)))
            number += 1
            heading = element.get_text(" ", strip=True)
            parts = []
        else:
            text = element.get_text(" ", strip=True)
            if text:
                parts.append(text)
    if parts:
        sections.append((number, heading, "\n\n".join(parts)))
    if not sections:
        raise ValueError("文档没有可索引正文")
    return title, sections
 
def chunk_rows(url, title, sections, digest):
    rows = []
    for number, heading, body in sections:
        for start in range(0, len(body), 1400):
            text = body[start:start + 1600]
            identity = f"{url}|{digest}|{number}|{start}"
            chunk_id = hashlib.sha256(identity.encode()).hexdigest()
            rows.append((chunk_id, url, title, heading, number, text))
    return rows

async def ingest(urls):
    db = sqlite3.connect(DB)
    db.executescript(SCHEMA)
    failures = []
    try:
        async with AsyncWebCrawler(config=BrowserConfig(headless=True)) as crawler:
            for url in urls:
                try:
                    result = await crawler.arun(
                        url=url, config=CrawlerRunConfig(cache_mode=CacheMode.BYPASS)
                    )
                    if not result.success or result.status_code != 200:
                        raise ValueError("页面未成功取得")
                    title, sections = parse_sections(result.html)
                    text = "\n".join(body for _, _, body in sections)
                    digest = hashlib.sha256(text.encode()).hexdigest()
                    rows = chunk_rows(url, title, sections, digest)
                    observed = datetime.now(timezone.utc).isoformat()
                    with db:
                        db.execute("DELETE FROM chunks WHERE source_url=?", (url,))
                        db.executemany("INSERT INTO chunks VALUES (?,?,?,?,?,?)", rows)
                        db.execute("""INSERT INTO documents VALUES (?,?,?,?)
                          ON CONFLICT(url) DO UPDATE SET title=excluded.title,
                          observed_at=excluded.observed_at,content_hash=excluded.content_hash""",
                          (url, title, observed, digest))
                except Exception as error:
                    failures.append({"source_url": url, "error_type": type(error).__name__})
        Path("ingest-failures.json").write_text(
            json.dumps(failures, indent=2), encoding="utf-8"
        )
        print({"documents": db.execute("SELECT count(*) FROM documents").fetchone()[0],
               "failed": len(failures)})
    finally:
        db.close()
 
def search(query):
    if len(query) < 3:
        raise ValueError("本例 trigram 查询至少需要三个字符")
    if not DB.exists():
        raise FileNotFoundError("先执行 ingest 创建知识库")
    db = sqlite3.connect(DB.resolve().as_uri() + "?mode=ro", uri=True)
    db.row_factory = sqlite3.Row
    try:
        phrase = '"' + query.replace('"', '""') + '"'
        rows = db.execute("""
          SELECT c.chunk_id,c.source_url,c.title,c.heading,c.section_no,
                 snippet(chunks,5,'[',']',' ... ',24) AS excerpt,d.observed_at
          FROM chunks AS c JOIN documents AS d ON d.url=c.source_url
          WHERE chunks MATCH ? ORDER BY bm25(chunks) LIMIT 5
        """, (phrase,)).fetchall()
        print(json.dumps([dict(row) for row in rows], ensure_ascii=False, indent=2))
    finally:
        db.close()
 
parser = argparse.ArgumentParser()
parser.add_argument("action", choices=["ingest", "search"])
parser.add_argument("values", nargs="+")
args = parser.parse_args()
if args.action == "ingest":
    asyncio.run(ingest(args.values))
else:
    search(" ".join(args.values))

python knowledge.py ingest   "$DOC_URL_1" "$DOC_URL_2"
python knowledge.py search SQLite