Bulk ingestion, batch processing, and a human review gate

The largest application I have built, and the only part of it that matters is the screen where a person clicks reviewed.

Why I built it

A folder of documents lands on a disk. There are thousands of them. Somewhere in that pile is the answer to a question, and the question is the kind only a person can settle — not “which files contain this word”, which a machine answers in a minute, but “is this one actually what we are looking for”, which it cannot answer at all.

So the pile has to be read. And the pile cannot be read, because there are thousands of files and one of you.

That is the whole problem. Not finding an answer — reducing a pile to a size where a person can look at every remaining item without their attention giving out around item 300. Everything I built is in service of that one sentence.

What it solves

Two things, and the second is the one I would defend.

The reviewer sees fewer items: a pile of thousands becomes a queue of dozens, because duplicates collapse, obviously irrelevant material is filtered before anyone looks at it, and what survives arrives with a machine-written summary attached.

And every item that leaves the system was seen by a person. Not sampled. Not spot-checked. Seen, one at a time, with a name and a timestamp recorded against the decision. There is no export path that does not run through that.

One ingest pass over a demo folder of 40 files of filler text, with the size cap turned down to 1 MB so that something trips it:

$ python ingest.py inbox/
inbox/: 40 files
duplicate  quorvale-3862.txt  (same bytes as brenpel-7294.txt)
skipped    yunpel-9720.txt  (1.6 MB, over the 1.0 MB cap)

new          38
duplicate    1
skipped      1
rows written 38

How it works

Ingest. Walk a folder, open each file, pull out the text, write a row.

Dedupe. Each file is identified by a hash of its own bytes. The same document arriving twice under 2 names is one record — and piles of this kind are full of copies.

Queue. Nothing is processed at ingest time. Work is written to a queue and picked up in fixed-size batches of 100 items.

Extract. For each batch, a model is asked to read the text and propose a small set of structured values for it. The result is written to the row in a state called proposed. Nothing downstream treats a proposed value as true.

Review gate. A person opens the queue, reads the item and the proposal side by side, and accepts or discards it. This writes their name and the time.

Export. Reads only the accepted rows. An unreviewed item is not filtered out of the export — it is not reachable from it.

The 2 gate functions from further down, run against 5 invented items in a throwaway database. A batch has proposed a value for 4 of them; the fifth is still waiting:

$ python3 gate_demo.py
before review: a batch has proposed values for 4 items, 1 is still pending
  id title               proposed   state      reviewed_by
  1  Brenquist ollamore  tavrel     proposed   -
  2  Yuncast pellidor    morvane    proposed   -
  3  Quorrin ashvale     tavrel     proposed   -
  4  Drellum fosk        quelt      proposed   -
  5  Ovanteer brisk      -          pending    -

after 3 decisions: accept 1, discard 2, accept 5
  id title               proposed   state      reviewed_by
  1  Brenquist ollamore  tavrel     accepted   analyst@example.test
  2  Yuncast pellidor    morvane    discarded  analyst@example.test
  3  Quorrin ashvale     tavrel     proposed   -
  4  Drellum fosk        quelt      proposed   -
  5  Ovanteer brisk      -          pending    -

export reads: [1]

How I built it

Python, a web framework, and PostgreSQL (a database that several processes can safely write to at once), because the queue has to survive the program exiting and several workers have to read from it without stepping on each other. It grew to about 13,000 lines, 10 times the size of anything else in this series.

Three decisions, and every one of them is the second thing I tried.

Batching, because per-item calls are how you find out what a mistake costs. The first version called the model once per file, at ingest. Simpler, immediate feedback, and wrong in 2 ways: a bad prompt burns through the whole folder before you notice, and an interruption leaves you with no clean idea of where to resume. Batches of 100 fixed both — a bad idea now costs one batch instead of one folder. The default is 3 batches in flight, adjustable up to 10, a number that exists because of a rate limit rather than ambition.

Caching the filter list, because I made it slow first. An ignore list keeps obviously irrelevant items out of the reviewer’s queue. Reading that list from the database on every single file made ingestion crawl. It is now held in memory for 5 minutes and dropped the moment someone edits the list.

The gate is a state, not a screen. The first design had a review page and an export button, and the button queried whatever was in the table. That is not a gate, it is a convention, and conventions get worked around at six in the evening on a deadline. Moving it into the row’s state — accepted is a value only a human action writes, and export selects on that alone — is the difference between a rule and a rule you cannot break by accident.

What I would do differently: the reviewer sees one item at a time, which for near-identical items is a slow way to make the same decision again and again. Grouping them before the gate would have saved more attention than any other change.

A look at the code

Deduplication by content, rewritten to stand alone. paths are pathlib.Path objects and seen is the set of hashes already in the database:

import hashlib

CHUNK = 1024 * 1024           # read a megabyte at a time
SIZE_CAP = 100 * 1024 * 1024  # above this, record it and move on

def content_id(path):
    """A file's identity is what is inside it — not its name, not its folder.
       Returns None for files too large to hash cheaply."""
    if path.stat().st_size > SIZE_CAP:
        return None
    digest = hashlib.sha256()
    with path.open("rb") as fh:
        for chunk in iter(lambda: fh.read(CHUNK), b""):
            digest.update(chunk)
    return digest.hexdigest()

def new_files(paths, seen):
    """Yields only the files that are genuinely new, whatever they are called."""
    for path in paths:
        fingerprint = content_id(path)
        if fingerprint is None or fingerprint in seen:
            continue
        seen.add(fingerprint)
        yield fingerprint, path

The gate itself is the most important thing here, and it is 2 short functions. db is an open database connection:

def record_decision(db, item_id, reviewer, decision):
    """A batch writes 'proposed'. Only this writes anything else, and only a
       person calls it. `decision` is 'accepted' or 'discarded'."""
    assert decision in ("accepted", "discarded")
    db.execute(
        """UPDATE items
              SET state = ?, reviewed_by = ?, reviewed_at = CURRENT_TIMESTAMP
            WHERE id = ? AND state = 'proposed'""",
        (decision, reviewer, item_id),
    )
    db.commit()

def exportable(db):
    """The only query the export path may run."""
    return db.execute("SELECT id, body FROM items WHERE state = 'accepted'").fetchall()

And the batching, as a toy: 400 items with no content at all, a short pause standing in for each model call, and one call made to fail on purpose.

$ python3 batches_demo.py
400 items in 4 batches of 100, 3 in flight

batch 2 failed at item 163:
  batch  status               done  pending  failed
  1      running                63       37       0
  2      failed, queued again   62       37       1
  3      running                62       38       0
  4      queued                  0      100       0

finished:
  batch  status               done  pending  failed
  1      complete              100        0       0
  2      complete              100        0       0
  3      complete              100        0       0
  4      complete              100        0       0

model calls: 401 for 400 items
items sent to the model more than once: [163]

What it can’t do

That last one is not a bug I can fix in code, which is why the queue has to be short. A gate nobody has the patience to operate is a rubber stamp with a database behind it.

Where this sits in the series

This is the 6-stage spine with the alert stage removed and the present stage load-bearing: the output is a queue somebody works through.

It is also the most literal answer I have to where AI actually fits: the model narrows a pile and writes a proposal, and it never decides. The do’s and don’ts post argues for the human-in-the-loop gate as a design constraint rather than a policy. This is what that looks like when a column in a table enforces it instead of everyone remembering.

The gate only earns its place if the filtering in front of it is honest about what it drops. Does your AI step actually work? is how to measure that, specifically the case where what you miss matters more than what you wrongly flag — or the other way round, which is a decision rather than a default.