Building the Ingestion Pipeline

A weekly crawl of the employee handbook produced an index that answered questions badly in one specific way: it quoted correct sentences with no idea which policy section they came from. "Expenses are reimbursed within 30 days" got mixed with "Relocation is reimbursed within 90 days" in the same passage, and the bot answered questions about one by citing the other.

The bug was upstream of embeddings. The pipeline stripped tags with a regular expression:

import re

html = """
<h1>Relocation Policy</h1>
<p>The relocation cap for Berlin is 4000 EUR.</p>
<h2>Eligibility</h2>
<p>Apply within 90 days of arrival.</p>
<table>
  <tr><th>City</th><th>Cap</th></tr>
  <tr><td>Berlin</td><td>4000</td></tr>
</table>
"""

print(" ".join(re.sub(r"<[^>]+>", " ", html).split()))

Output:

Relocation Policy The relocation cap for Berlin is 4000 EUR. Eligibility Apply within 90 days of arrival. City Cap Berlin 4000

Everything readable survived; everything structural died. Heading boundaries are spaces, so no later stage can rebuild "which section am I in." Table cells collapse into a word soup where City Cap Berlin 4000 is indistinguishable from a sentence, and the number 4000 loses the column that gave it meaning. Chunking then cuts this blob at fixed offsets, cementing the damage into every vector.

A parser that keeps structure fixes it at the source:

from html.parser import HTMLParser

class Doc(HTMLParser):
    def __init__(self):
        super().__init__()
        self.sections, self.head, self.buf, self.hlvl = [], ["(doc)"], [], 0

    def handle_starttag(self, tag, attrs):
        if tag in {"h1", "h2", "h3"}:
            self._flush()
            self.hlvl, self.buf = int(tag[1]), []   # heading text arrives as data next
        elif tag in {"p", "li", "td", "th"}:
            self._flush()
            self.hlvl = 0

    def handle_data(self, data):
        self.buf.append(data)

    def _flush(self):
        text = " ".join("".join(self.buf).split())
        self.buf = []
        if not text:
            return
        if self.hlvl:
            self.head = self.head[: self.hlvl - 1] + [text]   # an h2 drops any deeper h3
        else:
            self.sections.append((" > ".join(self.head), text))

    def parse(self, html: str):
        self.feed(html)
        self._flush()
        return self.sections

Input: the HTML above. Output:

[('(doc) > Relocation Policy', 'The relocation cap for Berlin is 4000 EUR.'),
 ('(doc) > Relocation Policy > Eligibility', 'Apply within 90 days of arrival.'),
 ('(doc) > Relocation Policy > Eligibility', 'City Cap Berlin 4000')]

The head list is a heading stack: entering h2 truncates deeper levels, so sibling sections never bleed into each other, and every body block carries a breadcrumb a citation can later display. Table rows flush per row, keeping cells adjacent instead of interleaved with unrelated paragraphs.

The choice of parser is the first quality gate, and it is the same argument that later decides between fixed-size chunking and semantic chunking: structure you destroy here cannot be reconstructed downstream.

The crawl runs on a schedule and mostly re-downloads unchanged files. Two things must hold: re-processing identical content must not add duplicate knowledge to the index, and every surviving chunk must carry enough provenance to be cited by a human.

Both come from hashing.

import hashlib

def doc_hash(text: str) -> str:
    norm = " ".join(text.lower().split())          # whitespace and case are not edits
    return hashlib.sha256(norm.encode()).hexdigest()[:16]

def chunk_id(doc_id: str, text: str, pos: int) -> str:
    body = hashlib.sha256(text.encode()).hexdigest()[:8]
    return f"{doc_id}:{body}:{pos}"

Input / output:

doc_hash("Apply within 90 days. ") -> 'a41c7e0b9d2f5518'
doc_hash("apply  within 90 days.") -> 'a41c7e0b9d2f5518'
chunk_id("handbook", "Apply within 90 days.", 0) -> 'handbook:a91e0c4d:0'

Note the two different normalization levels. doc_hash lowercases and collapses whitespace because a re-rendered page that only changed indentation is the same knowledge — treat it as a change and you re-embed thousands of files for nothing. chunk_id hashes the chunk text verbatim: a chunk whose wording changed is a different chunk and deserves a different id.

The id format doc_id:body_hash:pos makes upserts deterministic. The vector store's upsert keyed on that id turns every re-run into a reconciliation rather than an append:

Provenance rides alongside the id rather than inside it, because a URL or page number can change without the text changing:

def make_chunks(sections, doc_id: str, url: str, page: int) -> list[dict]:
    rows = []
    for breadcrumb, text in sections:
        title, _, heading = breadcrumb.rpartition(" > ")
        rows.append({
            "id": chunk_id(doc_id, text, len(rows)),
            "text": text,
            "source_url": url,
            "section": heading or title,
            "breadcrumb": breadcrumb,
            "page": page,
            "doc_hash": doc_hash(text),
        })
    return rows

Input: [("Relocation Policy > Eligibility", "Apply within 90 days of arrival.")], doc_id="handbook". Output:

[{'id': 'handbook:a91e0c4d:0', 'text': 'Apply within 90 days of arrival.',
  'source_url': 'https://hr.example/policy', 'section': 'Eligibility',
  'breadcrumb': 'Relocation Policy > Eligibility', 'page': 3,
  'doc_hash': 'a41c7e0b9d2f5518'}]

Store source_url, section, and page on every row — not once per document — because retrieval returns chunks, not documents, and a citation has to resolve from the single row the model saw. The downstream consumer of these fields is context assembly, which turns them into [n] markers and source links.

Now the stages in one executable pass. Cleaning collapses whitespace and drops boilerplate; chunking splits on the structure the parser preserved; embedding is stubbed with a deterministic vector so the file runs without network access; upsert is a dict keyed by chunk id, which is exactly what makes a re-run free.

import re

def clean(text: str) -> str:
    text = re.sub(r"\s+", " ", text).strip()
    return text if len(text) > 40 else None        # headers, footers, "Menu"

def chunk(sections, max_words=40):
    out = []
    for breadcrumb, text in sections:
        words, buf = text.split(), []
        for w in words:
            buf.append(w)
            if len(buf) >= max_words:
                out.append((breadcrumb, " ".join(buf)))
                buf = []
        if buf:
            out.append((breadcrumb, " ".join(buf)))
    return out

def embed(texts: list[str]) -> list[list[float]]:
    vocab = {"relocation", "cap", "berlin", "eligibility", "apply", "days"}
    vecs = []
    for t in texts:
        v = [0.0] * len(vocab)
        for w in t.lower().split():
            if w in vocab:
                v[vocab[w]] += 1.0
        vecs.append(v)
    return vecs

def ingest(doc_id, url, page, sections, index: dict):
    rows = []
    for bc, tx in chunk(sections):
        tx = clean(tx)
        if tx:
            rows.append((bc, tx))
    vecs = embed([tx for _, tx in rows])
    mine = set()
    for (bc, tx), vec in zip(rows, vecs):
        cid = chunk_id(doc_id, tx, len(mine))
        mine.add(cid)
        index[cid] = {"text": tx, "vec": vec, "source_url": url,
                      "breadcrumb": bc, "page": page, "doc_hash": doc_hash(tx)}
    stale = [k for k in index if k.startswith(doc_id + ":") and k not in mine]
    for k in stale:
        del index[k]                                # ids that vanished were edited away
    return sorted(mine)

Input: the parsed sections for handbook, url="https://hr.example/policy", page=3. Output:

run 1  -> 3 rows   ['handbook:2f81ab90:0', 'handbook:5c04d7e1:1', 'handbook:9a17cc33:2']
run 2  -> 3 rows   ['handbook:2f81ab90:0', 'handbook:5c04d7e1:1', 'handbook:9a17cc33:2']
edit   -> 3 rows   ['handbook:1d77b0aa:0', 'handbook:73e5f184:1', 'handbook:b2c09e5f:2']

Run twice, same ids — the second call overwrites identical values and adds nothing. Edit one sentence, and only the chunks containing it change id; the rest keep theirs, so their vectors are never recomputed. The stale sweep handles the other direction: a document that shrank from four chunks to three must lose its fourth row, or the index keeps serving text that no longer exists.

The lines worth staring at:

• if len(text) > 40 else None — cleaning is where navigation chrome dies. Let it through and every chunk carries "Menu | Careers | Login," which pollutes both embeddings and citations.
• mine is built from ids this run produced, so the stale sweep compares against the current truth rather than a remembered count.
• Upsert keyed by chunk_id is what makes the job safe to run hourly: cost scales with edits, not with corpus size.

Embedding belongs in batches — one call per list, not per string — which is what embedding batching and the embedding evaluation topics pick up. Once rows exist with provenance, retrieval and citation have something to point at.