Curious builder 0 XP earned · 300 to level 2

0 days Finish a lesson to begin

Badge collection

0 of 5 unlocked

small wins to finish your path

Next lesson →

Project: prepare a dataset

Real AI projects rarely start with modelling — they start with messy data. In this capstone you will combine everything from the course into one production-style script that turns a folder of raw text files into a clean, chunked, deduplicated JSONL dataset that a RAG index or a fine-tuning job can consume directly, plus a quality report so you can trust what you built.

Last updated: 01 Nov, 2026 · Python 3.11+

What you are building

Definition · Data-preparation pipeline

A repeatable sequence of steps that transforms raw input data into a clean, consistent format for a downstream system. “Repeatable” is the key word: running the same pipeline on the same input must always produce the same output.

Our pipeline reads every .txt and .md file in raw/ and produces data/chunks.jsonl, where every line is one chunk record:

JSONL record

{"id": "rag-guide-000", "source": "rag-guide.md", "chunk_index": 0, "text": "...", "n_words": 35, "n_chars": 217, "sha256": "..."}

Field	Type	Why it exists
`id`	str	Stable, unique identifier — vector databases and eval tools need one.
`source`	str	The original file, so answers can cite where information came from.
`chunk_index`	int	Position within the source, for reassembling context in order.
`text`	str	The cleaned chunk content that will be embedded.
`n_words` / `n_chars`	int	Size statistics for filtering and cost estimation.
`sha256`	str	Fingerprint of the text, used to detect duplicate chunks.

The pipeline at a glance

Pipeline

raw/*.txt, raw/*.md
      │
      ▼
 1. load       read every file as UTF-8 (tolerating bad bytes)
      │
      ▼
 2. clean      normalise unicode, strip control characters, collapse whitespace
      │
      ▼
 3. filter     skip empty or too-short documents
      │
      ▼
 4. chunk      split into overlapping word windows
      │
      ▼
 5. dedupe     drop chunks whose exact text was already seen
      │
      ▼
 6. write      data/chunks.jsonl (one JSON record per line)
      │
      ▼
 7. report     pandas summary: counts, sizes, per-source stats

Each step is a small, testable function. That separation is what makes the pipeline easy to debug and extend — exactly the skill from the Types, functions, and modules lesson.

Step 0 — Project layout and sample data

Project tree

ai-lab/
├── .venv/
├── make_sample_data.py   # creates raw/ with test files
├── prep.py               # the pipeline
├── raw/                  # input documents
└── data/                 # generated output

To build and test the pipeline you need realistic, messy input. This script creates five files that deliberately include the problems you will meet in real data: irregular whitespace, a non-breaking space, a control character, an empty file, a file that is too short, and an exact duplicate.

make_sample_data.py

from pathlib import Path

RAW = Path("raw")
RAW.mkdir(exist_ok=True)

files = {
    "rag-guide.md": (
        "# RAG Guide\n\n"
        "Retrieval augmented generation   combines a retriever with a language model.\n"
        "The retriever finds relevant passages\u00a0from a knowledge base, and the model "
        "writes an answer grounded in those passages. Good chunking, clean text, and "
        "accurate metadata matter more than the choice of model for answer quality.\n"
    ),
    "agents.txt": (
        "AI agents   plan a task, call tools, observe the results, and repeat until done.\x07\n"
        "Reliable agents need clear tool descriptions, strict output formats, step limits, "
        "and logging of every action so failures can be debugged."
    ),
    "agents-copy.txt": (
        "AI agents   plan a task, call tools, observe the results, and repeat until done.\x07\n"
        "Reliable agents need clear tool descriptions, strict output formats, step limits, "
        "and logging of every action so failures can be debugged."
    ),
    "empty.txt": "   \n\n  ",
    "tiny.md": "TODO",
}

for name, text in files.items():
    (RAW / name).write_text(text, encoding="utf-8")

print(sorted(p.name for p in RAW.iterdir()))

Terminal

python make_sample_data.py

['agents-copy.txt', 'agents.txt', 'empty.txt', 'rag-guide.md', 'tiny.md']

Step 1 — Load files

Find input files with glob, sort them so the output order never changes between runs, and read them as UTF-8 with errors="replace" so one bad byte cannot crash the job.

Python

from pathlib import Path

SUFFIXES = {".txt", ".md"}


def load_documents(raw_dir: Path) -> list[tuple[str, str]]:
    """Return (file_name, raw_text) pairs for every supported file, sorted by name."""
    docs = []
    for path in sorted(raw_dir.rglob("*")):
        if path.is_file() and path.suffix.lower() in SUFFIXES:
            docs.append((path.name, path.read_text(encoding="utf-8", errors="replace")))
    return docs

Step 2 — Clean the text

Definition · Unicode normalisation

Converting text to a canonical form so that characters that look identical are also stored identically. NFKC form, for example, turns a non-breaking space into a normal space and full-width letters into ordinary ones. Without it, two visually identical chunks can have different fingerprints and different embeddings.

Python

import re
import unicodedata

CONTROL_CHARS = re.compile(r"[\x00-\x08\x0b-\x1f\x7f]")


def clean_text(text: str) -> str:
    text = unicodedata.normalize("NFKC", text)   # canonical characters
    text = CONTROL_CHARS.sub(" ", text)          # drop invisible control chars
    text = re.sub(r"\s+", " ", text)              # collapse all whitespace
    return text.strip()


print(clean_text("AI agents \u00a0 plan\x07 a task.\n\n  Done."))

AI agents plan a task. Done.

The regular expression \s+ matches any run of whitespace — spaces, tabs, and newlines — and replaces it with a single space. For RAG this flattening is usually fine; if your chunker should respect paragraphs, collapse only spaces and tabs and keep blank lines as boundaries.

Note

Cleaning should be conservative. Removing punctuation, lowercasing, or stripping numbers can destroy meaning that embedding models and LLMs rely on. Normalise and tidy; do not rewrite.

Step 3 — Filter out useless documents

Empty files and tiny fragments (“TODO”, a lone heading) add noise to retrieval and waste embedding cost. Set a minimum length and count what you skip so the report can show it.

Python

MIN_CHARS = 40


def is_useful(text: str, min_chars: int = MIN_CHARS) -> bool:
    return len(text) >= min_chars


print(is_useful(""), is_useful("TODO"), is_useful("A real paragraph about vector search and retrieval."))

False False True

Step 4 — Chunk with overlap

We reuse the sliding-window chunker from the Files and paths lesson. Chunk size and overlap are parameters, so you can tune them later without touching the code.

Python

def chunk_words(text: str, size: int = 40, overlap: int = 8) -> list[str]:
    if overlap >= size:
        raise ValueError("overlap must be smaller than size")
    words = text.split()
    step = size - overlap
    chunks = []
    for start in range(0, len(words), step):
        chunks.append(" ".join(words[start:start + size]))
        if start + size >= len(words):
            break
    return chunks

Step 5 — Deduplicate

Definition · Hash / fingerprint

A hash function such as SHA-256 turns any text into a fixed-length string. The same input always gives the same hash, and different inputs almost never collide, so comparing hashes is a fast way to detect exact duplicates.

Duplicate chunks are surprisingly common — copied pages, boilerplate footers, the same file in two folders. They waste storage and, worse, fill your top-k retrieval results with the same passage several times.

Python

import hashlib


def fingerprint(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


seen: set[str] = set()
for chunk in ["same text", "other text", "same text"]:
    h = fingerprint(chunk)
    if h in seen:
        print("duplicate:", chunk)
        continue
    seen.add(h)

print(len(seen), "unique;", fingerprint("same text")[:12])

duplicate: same text 2 unique; 2e68a7bba11b

This catches exact duplicates after cleaning. Near-duplicates (the same paragraph with one word changed) need techniques such as MinHash or embedding similarity, which you can add later.

Step 6 and 7 — Write JSONL and report

Records are written one JSON object per line with ensure_ascii=False so non-English text stays readable. The report loads the same file with pandas, which also proves the output is valid JSONL.

The complete script

Here is everything assembled into one script with a command-line interface built on argparse, so you can change folders and chunk settings without editing code.

prep.py

"""Prepare a chunked, deduplicated JSONL dataset from raw text files."""
from __future__ import annotations

import argparse
import hashlib
import json
import re
import unicodedata
from collections import Counter
from pathlib import Path

import pandas as pd

SUFFIXES = {".txt", ".md"}
CONTROL_CHARS = re.compile(r"[\x00-\x08\x0b-\x1f\x7f]")


def load_documents(raw_dir: Path) -> list[tuple[str, str]]:
    docs = []
    for path in sorted(raw_dir.rglob("*")):
        if path.is_file() and path.suffix.lower() in SUFFIXES:
            docs.append((path.name, path.read_text(encoding="utf-8", errors="replace")))
    return docs


def clean_text(text: str) -> str:
    text = unicodedata.normalize("NFKC", text)
    text = CONTROL_CHARS.sub(" ", text)
    text = re.sub(r"\s+", " ", text)
    return text.strip()


def chunk_words(text: str, size: int, overlap: int) -> list[str]:
    if overlap >= size:
        raise ValueError("overlap must be smaller than size")
    words = text.split()
    step = size - overlap
    chunks = []
    for start in range(0, len(words), step):
        chunks.append(" ".join(words[start:start + size]))
        if start + size >= len(words):
            break
    return chunks


def fingerprint(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def build_records(docs, size, overlap, min_chars):
    records, stats, seen = [], Counter(), set()
    for name, raw in docs:
        stats["files"] += 1
        text = clean_text(raw)
        if len(text) < min_chars:
            stats["skipped_short"] += 1
            continue
        for i, chunk in enumerate(chunk_words(text, size, overlap)):
            h = fingerprint(chunk)
            if h in seen:
                stats["duplicate_chunks"] += 1
                continue
            seen.add(h)
            records.append({
                "id": f"{Path(name).stem}-{i:03d}",
                "source": name,
                "chunk_index": i,
                "text": chunk,
                "n_words": len(chunk.split()),
                "n_chars": len(chunk),
                "sha256": h,
            })
    stats["chunks_written"] = len(records)
    return records, stats


def write_jsonl(records: list[dict], out_path: Path) -> None:
    out_path.parent.mkdir(parents=True, exist_ok=True)
    tmp = out_path.with_suffix(".tmp")
    with tmp.open("w", encoding="utf-8") as f:
        for row in records:
            f.write(json.dumps(row, ensure_ascii=False) + "\n")
    tmp.replace(out_path)


def report(out_path: Path, stats: Counter) -> None:
    print("pipeline:", dict(stats))
    df = pd.read_json(out_path, lines=True)
    print(df.groupby("source").agg(chunks=("id", "count"), words=("n_words", "sum")))
    print(f"avg words per chunk: {df['n_words'].mean():.1f}")


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--raw", type=Path, default=Path("raw"))
    parser.add_argument("--out", type=Path, default=Path("data/chunks.jsonl"))
    parser.add_argument("--size", type=int, default=40, help="words per chunk")
    parser.add_argument("--overlap", type=int, default=8, help="words shared by neighbours")
    parser.add_argument("--min-chars", type=int, default=40)
    args = parser.parse_args()

    docs = load_documents(args.raw)
    records, stats = build_records(docs, args.size, args.overlap, args.min_chars)
    if not records:
        raise SystemExit("no records produced - check --raw and --min-chars")
    write_jsonl(records, args.out)
    print(f"wrote {len(records)} records -> {args.out}")
    report(args.out, stats)


if __name__ == "__main__":
    main()

Run it

Terminal

python prep.py

wrote 3 records -> data/chunks.jsonl pipeline: {'files': 5, 'duplicate_chunks': 1, 'skipped_short': 2, 'chunks_written': 3} chunks words source agents-copy.txt 1 35 rag-guide.md 2 57 avg words per chunk: 30.7

Read the report like an engineer: five files went in, two were skipped as too short (empty.txt and tiny.md), one chunk was dropped as a duplicate, and three clean chunks came out. Because files are processed in sorted order, agents-copy.txt is read before agents.txt, so the copy is kept and the original's identical chunk is the one skipped. The outcome is deterministic — but make sure it is the rule you actually want.

Try different settings from the command line, no code changes required:

Terminal

python prep.py --size 20 --overlap 5
python prep.py --raw ~/notes --out data/notes.jsonl --min-chars 200
python prep.py --help

Verify the output

Never trust a pipeline you have not inspected. Look at the raw file, and check a few invariants with a short script.

Terminal

head -n 1 data/chunks.jsonl

{"id": "agents-copy-000", "source": "agents-copy.txt", "chunk_index": 0, "text": "AI agents plan a task, call tools, observe the results, and repeat until done. Reliable agents need clear tool descriptions, strict output formats, step limits, and logging of every action so failures can be debugged.", "n_words": 35, "n_chars": 217, "sha256": "70955941490d…"}

check_output.py

import json
from pathlib import Path

rows = [json.loads(line) for line in Path("data/chunks.jsonl").open(encoding="utf-8")]

assert len({r["id"] for r in rows}) == len(rows), "ids must be unique"
assert len({r["sha256"] for r in rows}) == len(rows), "no duplicate texts"
assert all(r["text"] == r["text"].strip() for r in rows), "no stray whitespace"
assert all("  " not in r["text"] for r in rows), "whitespace collapsed"

print("all checks passed for", len(rows), "records")

all checks passed for 3 records

Watch out

Notice the first record's source is agents-copy.txt, not agents.txt — whichever duplicate is read first is the one kept. In real projects, decide explicitly which duplicate to keep (for example the newest file or the canonical folder) instead of relying on alphabetical order.

How this feeds the next steps

RAG: embed each record's text, store the vector with id and source as metadata, and cite source in answers. The NumPy search from the previous lesson works directly on these records.
Fine-tuning: convert records into prompt/response pairs or chat-format messages, still one JSON object per line.
Evaluation: sample chunks with pandas to generate test questions, and keep id so you can measure whether retrieval found the right chunk.

What you practised

Lesson	Where it appears in prep.py
Setup, venv, and packages	Running inside a venv with `pandas` installed
Types, functions, and modules	Small typed functions, `argparse`, `if __name__ == "__main__"`
Lists, dicts, and JSON	Records as dicts, a set of seen hashes, `Counter` stats, JSONL output
Files and paths	`rglob`, sorted discovery, UTF-8 with `errors="replace"`, atomic writes
NumPy and pandas basics	`read_json(lines=True)` and `groupby().agg()` for the report

Try it: extend the pipeline

Add a --min-words option and skip chunks shorter than it.
Add a title field taken from the first Markdown heading (# ...) of each .md file.
Write a second file, data/report.csv, with the per-source summary table.
Add a --dry-run flag that prints the stats without writing any files.
Stretch goal: estimate embedding cost by summing n_chars / 4 tokens across all chunks.

Key takeaways

A good pipeline is a chain of small functions: load → clean → filter → chunk → dedupe → write → report.
Sort inputs, pin parameters, and avoid randomness so every run is reproducible.
Normalise Unicode and whitespace conservatively; do not destroy meaning.
Hash chunk text to remove exact duplicates before you pay to embed them.
Write JSONL with stable ids and source metadata, then verify the output with code.
You now have every Python skill needed for the next course: LLM Fundamentals.

Where to go next

You have finished Python for AI Developers. Continue with LLM Fundamentals to learn how language models actually work, then build a full retrieval pipeline in RAG Systems using the dataset you just prepared.