Build a Web Knowledge Base for RAG: Search and Trace Sources

Downloaded Markdown is not yet a dependable retrieval source. Chunks need provenance, refreshed pages must replace older chunks, and temporary collection failures must not erase the last good document.

This tutorial builds a model-free lexical baseline from explicit URLs: collect HTML, organize sections, chunk, index, and return source-linked excerpts. Generation and vector search can follow later.

Prepare the input addresses

Locate the following pages in your browser, confirm their contents, and copy their full addresses. Obtain service and repository addresses from account integration settings, official setup instructions, or the repository clone menu. Do not derive real addresses from illustrative output.

DOC_URL_1: the sqlite3.html address in the Python library documentation.
DOC_URL_2: the asyncio-queue.html address in the same documentation.

Set these in the same Bash/WSL terminal used for subsequent commands:

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

Replace each <VARIABLE_NAME> in JSON or task parameters with the copied value before submission; JSON does not expand environment variables. Any example.16yun.cn source in an output illustration shows field structure, not a demo service deployed for this article.

Install and choose the index

Use Python 3.12:

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

Python's SQLite must support FTS5 and its trigram tokenizer. Use an appropriate distribution if unavailable rather than silently choosing different search semantics.

Trigrams provide an interpretable substring baseline, including longer Chinese expressions. They are not semantic segmentation or synonym matching. This example rejects queries shorter than three characters. See the FTS5 reference.

Preserve document and section identity

A chunk retains source, title, heading, section position, body, and identity. Repeated headings require positional information.

The chunk ID combines URL, content hash, section number, and character offset. Its observation time identifies collection freshness, not publication time.

Create knowledge.py:

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

The main-content selector suits the chosen documentation. Adapt it before indexing another site.

Table content becomes readable text rather than a preserved cell model. Code is retained as section material, not parsed into syntax-aware units.

The 1600-character window and 1400-character step provide overlap, not a token guarantee. Compare paragraph-aware and code-aware chunking against actual retrieval questions later.

Refresh only after material is ready

Append to the same file:

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))

Collect two explicit documents and search for a known term:

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

Browser configuration and per-request cache policy have different responsibilities. This example bypasses cached material for refreshes. See the Crawl4AI basics.

Keep refresh failure separate from deletion

Download, parsing, and chunk construction happen before replacement. Deleting old chunks, inserting new ones, and updating document metadata share a transaction. A database failure rolls the replacement back.

A failed page never reaches replacement, so its previous successful version remains searchable. Use observation timestamps to disclose that retained content may be older.

The input URL list is not evidence of a complete site inventory. The example therefore does not delete URLs absent from the current run. Confirmed removal needs explicit evidence and source-scoped deletion rules.

Trace a retrieval result

Results include chunk identity, source, title, section, excerpt, and observation time. Open the source and verify that the section actually supports the question.

Section number distinguishes repeated headings but does not imply a browser fragment identifier. Do not invent an anchor from the heading text.

A lexical match establishes a textual relationship under this tokenizer, not semantic understanding. First verify ingestion with known terms, then evaluate real questions.

When adding generation, pass retrieved content together with its source identity. Answers must map citations back to those records. A missing supporting passage should remain unknown rather than being concealed by generation.

Troubleshoot

FTS5 or tokenizer unavailable: inspect the SQLite linked to Python, which may differ from the system command.

Search opens an empty database: use an existence check and read-only connection as shown; normal SQLite connect can create a new file.

Old results remain after refresh: inspect cache policy and ingestion failures. Retaining old content may be the intended failure behavior.

Short Chinese queries fail: this example explicitly requires at least three characters. Evaluate a different indexing approach without silently changing meaning.

Navigation dominates results: repair extraction scope and refresh affected documents rather than raising result count.

Repeated headings disappear: ensure chunk identity includes section position instead of heading text alone.

A source-preserving baseline makes missing answers diagnosable at collection, chunking, and retrieval stages.