Data Cleaning for AI and Your First AI Data Pipeline

The problem: Real documents mix useful meaning with HTML, menus, duplicate passages, broken characters, empty fields, and inconsistent structure. Feeding all that noise into retrieval hurts results, but blindly deleting untidy text can also destroy meaning.

What you will learn: You will build a versioned, testable pipeline that preserves raw data, validates and normalizes records, deduplicates carefully, and creates chunks with metadata. The final result is model-ready data that avoids information leakage while preserving the evidence a RAG system needs.

---

Part A — Data Cleaning for AI

1. Why Data Cleaning Matters

Cleaning is a sequence of decisions, not a button that turns “bad data” into perfect data:

immutable raw source
      ↓ inspect and validate
remove known noise and duplicates
      ↓ preserve meaning and metadata
clean, versioned records
      ↓ split and chunk
embedding or training pipeline

Each transformation should have a reason and a test. Removing visible clutter can also remove meaning, so keep the original data, compare samples before and after, and record which cleaning version produced each output.

Turning Raw Input into Reliable Records

Data cleaning is the process of taking raw, messy input and transforming it into consistent, reliable data before it’s used further downstream.

Why Messy Data Damages the Next Step

Embedding models and LLMs are sensitive to noise. Extra whitespace, HTML tags, duplicate paragraphs, or broken characters don’t just look ugly — they actively degrade embedding quality, waste tokens, and can cause your RAG system to retrieve the same duplicated chunk repeatedly instead of diverse, relevant content.

Picture Cleaning a Damaged Document

Feeding messy text to an embedding model is like asking someone to summarize a document that’s smudged, has coffee stains, and repeats every third paragraph twice — they can still do it, but the result is worse than if you’d handed them a clean copy.

🤖 How Is This Used in AI? Cleaning can be one of the highest-leverage parts of a RAG project. Improving duplicate removal, document parsing, or metadata may improve retrieval more directly than changing the model, but the actual bottleneck should be confirmed with retrieval evaluation rather than assumed.

---

2. Common Raw Data Problems

Problem	Example	Why it hurts
Extra whitespace	"Hello world\n\n\n"	Wastes tokens, inconsistent chunk boundaries
HTML/markup leftovers	"<p>Hello <b>world</b></p>"	Tags pollute the actual meaning being embedded
Inconsistent casing	"AI" vs "ai" vs "Ai"	Can affect keyword matching and deduplication
Duplicate content	Same paragraph appears 3 times in a scraped dataset	Retrieval keeps returning the same info, wasting top_k slots
Missing/empty entries	An empty string where a document should be	Wastes an embedding call, returns useless “documents”
Broken encoding	"café" becomes "café"	Garbles meaning, embeds as noise
Boilerplate text	Headers, footers, navigation text scraped from a webpage	Pollutes chunks with irrelevant repeated text

---

3. Text Cleaning with Regular Expressions

What Is It?

The re module lets you search for and remove/replace patterns in text — much more powerful than plain .replace() for anything beyond a fixed string.

import re

raw_text = "<p>Python is <b>great</b> for AI!</p>   Visit https://example.com for more.\n\n\n"

# Remove HTML tags
no_html = re.sub(r"<[^>]+>", "", raw_text)
print(no_html)

# Remove URLs
no_urls = re.sub(r"https?://\S+", "", no_html)
print(no_urls)

# Collapse multiple whitespace/newlines into a single space
cleaned = re.sub(r"\s+", " ", no_urls).strip()
print(cleaned)

Expected Output:

Python is great for AI!   Visit https://example.com for more.


Python is great for AI!   Visit  for more.


Python is great for AI! Visit for more.

How It Works

• re.sub(pattern, replacement, text) finds every match of pattern in text and replaces it with replacement.
• r"<[^>]+>" matches anything that looks like an HTML tag — <, followed by any characters that aren’t >, followed by >.
• r"https?://\S+" matches http:// or https:// followed by non-whitespace characters — a simple URL matcher.
• r"\s+" matches one-or-more whitespace characters (spaces, tabs, newlines) — collapsing runs of them into a single space.

⚠️ Common Beginner Mistake: Writing an overly aggressive regex that strips out meaningful content along with the noise (e.g., a pattern that accidentally removes numbers or punctuation you actually needed). Always test cleaning regexes against a few real, varied examples before running them across an entire dataset.

🤖 How Is This Used in AI? Cleaning scraped web content, PDF-extracted text, or user-submitted documents before chunking and embedding — this re.sub chain is a genuinely realistic first pass used in real ingestion pipelines.

---

4. Normalizing Text

def normalize_text(text):
    text = text.strip()
    text = text.lower()
    text = " ".join(text.split())   # collapses all whitespace runs, incl. tabs/newlines
    return text

raw = "  Python   IS great\tfor\n\nAI  "
print(repr(normalize_text(raw)))

Expected Output:

'python is great for ai'

🧠 Intuition: Normalization makes semantically-identical text look identical too — "AI", "ai", and " Ai " should all normalize to the same thing, so your deduplication and matching logic (Module 2’s sets) actually catches them as duplicates.

🤖 How Is This Used in AI? Consistent normalization before embedding means near-identical documents produce near-identical (or truly duplicate) embeddings — which directly enables the deduplication step next.

---

5. Removing Duplicates

documents = [
    "Python is great for AI development.",
    "python is great for ai development.",   # duplicate, different casing
    "RAG combines retrieval with generation.",
    "Python is great for AI development.",   # exact duplicate
]

def normalize_text(text):
    return " ".join(text.strip().lower().split())

seen = set()
deduped = []

for doc in documents:
    key = normalize_text(doc)
    if key not in seen:          # Module 2: set membership check
        seen.add(key)
        deduped.append(doc)

print(deduped)
print(f"Removed {len(documents) - len(deduped)} duplicate(s)")

Expected Output:

['Python is great for AI development.', 'RAG combines retrieval with generation.']
Removed 2 duplicate(s)

🧠 Intuition

This is exactly Module 2’s set — “a bag with no duplicates allowed” — applied to real document cleaning. We normalize first so that meaningfully identical text (regardless of casing/whitespace) is recognized as duplicate, not just byte-for-byte identical text.

⚠️ Common Beginner Mistake: Deduplicating on the raw text instead of the normalized text — this misses duplicates that differ only in casing or whitespace, which are extremely common in real scraped or user-submitted data.

🤖 How Is This Used in AI? Preventing a RAG system from wasting retrieval slots (top_k) returning the same content multiple times, and avoiding paying to embed the same text more than once.

[!TIP] Fuzzy Deduplication (MinHash & LSH) in Production Exact deduplication (using set checks on normalized text) is great, but web scraped text often contains near-duplicates—articles that are 99% identical but differ by a timestamp, a sidebar link, or a template advertisement.

To catch these in large-scale AI datasets, production pipelines use algorithms like MinHash and Locality-Sensitive Hashing (LSH) to compute a “fuzzy similarity score” (Jaccard similarity) between text documents, dropping pages that are too similar even if they aren’t byte-for-byte identical.

---

6. Handling Missing or Malformed Data

Recall Module 9’s Pandas missing-data handling, applied to a document dataset:

import pandas as pd

data = {
    "text": ["Python is great for AI.", None, "  ", "RAG improves retrieval."],
    "source": ["notes.txt", "notes.txt", "notes.txt", "notes.txt"],
}
df = pd.DataFrame(data)

# Drop rows with missing (None) text
df = df.dropna(subset=["text"])

# Drop rows that are empty/whitespace-only after stripping
df = df[df["text"].str.strip() != ""]

print(df)

Expected Output:

                       text     source
0  Python is great for AI.  notes.txt
3  RAG improves retrieval.  notes.txt

🤖 How Is This Used in AI? Real scraped or uploaded datasets almost always contain some None/empty entries — embedding an empty string wastes an API call and produces a meaningless vector that can only ever hurt retrieval quality.

---

7. Chunking Strategies

What Is It?

Splitting a long document into smaller pieces (“chunks”) before embedding — because embedding models work best on focused, reasonably short passages, and because RAG retrieval needs to return specific, relevant pieces rather than entire documents.

Fixed-size chunking (simple, recap of Module 4)

def chunk_fixed(text, chunk_size=50):
    words = text.split()
    return [" ".join(words[i:i + chunk_size]) for i in range(0, len(words), chunk_size)]

Overlapping chunking (usually better for RAG)

def chunk_with_overlap(text, chunk_size=50, overlap=10):
    words = text.split()
    chunks = []
    step = chunk_size - overlap
    for i in range(0, len(words), step):
        chunk_words = words[i:i + chunk_size]
        if chunk_words:
            chunks.append(" ".join(chunk_words))
        if i + chunk_size >= len(words):
            break
    return chunks

text = "Python is great for AI. " * 20
chunks = chunk_with_overlap(text, chunk_size=15, overlap=5)
print(f"Created {len(chunks)} overlapping chunks")
print(chunks[0])
print(chunks[1])

Expected Output:

Created 10 overlapping chunks
Python is great for AI. Python is great for AI. Python is great
great for AI. Python is great for AI. Python is great for AI.

🧠 Intuition

Fixed chunking can awkwardly cut a sentence right in half at a chunk boundary, losing context. Overlapping chunks share a few words between consecutive chunks, so information near a boundary isn’t lost entirely from either side — a small redundancy that meaningfully improves retrieval quality.

🤖 How Is This Used in AI? Overlapping chunking is the realistic default used in most production RAG pipelines — the overlap amount is a tunable tradeoff between retrieval quality and the extra storage/embedding cost of the duplicated words.

⚠️ When NOT to use it: For very short, already-focused documents (a single FAQ answer, a short support ticket), chunking may be unnecessary entirely — embed the whole thing as one chunk.

💡 Semantic Chunking (Advanced Strategy)

While fixed and overlapping chunking split text purely based on word or character counts, Semantic Chunking splits text based on its meaning. In production systems, we want each chunk to hold a single, complete thought.

Instead of slicing text at arbitrary positions, we can:

1. Split by structural boundaries: Divide the text into sentences (using periods or NLP tools like nltk or spaCy) or paragraphs.
2. Measure distance between adjacent sentences: Embed each sentence and compute the similarity (Module 9’s cosine similarity) between sentence 1 and sentence 2, sentence 2 and sentence 3, and so on.
3. Set a threshold: If the similarity score between two consecutive sentences drops below a certain threshold (e.g. 0.70), it signals that the topic has shifted, and we start a brand-new chunk.

Semantic chunking ensures that your vectors represent clean, distinct concepts, leading to higher retrieval precision in production!

💡 Character vs. Token Chunking

In Python, standard string functions like .split() or len(text) measure words or characters. However, LLMs and embedding models process text in units called tokens (a token is typically about 4 characters or 0.75 words).

If you define a fixed-size chunker using character counts (e.g., 500 characters), the actual number of tokens in that chunk can vary wildly based on the vocabulary (e.g. source code, math equations, or non-English text consume more tokens per character).

In production RAG systems, we use token-based chunking (using libraries like tiktoken or tokenizers from Hugging Face) to count limits. This ensures your chunks never exceed the model’s hard input token limits (like 512 tokens for standard embedding models).

---

8. Building Chunk Metadata

def build_chunk_records(chunks, source, extra_metadata=None):
    records = []
    for i, chunk in enumerate(chunks):
        record = {
            "text": chunk,
            "source": source,
            "chunk_index": i,
            "char_count": len(chunk),
        }
        if extra_metadata:
            record.update(extra_metadata)
        records.append(record)
    return records

records = build_chunk_records(
    chunks[:2],
    source="ai_notes.txt",
    extra_metadata={"author": "team_docs", "category": "AI fundamentals"},
)
for r in records:
    print(r)

Expected Output:

{'text': 'Python is great for AI. Python is great for AI. Python is great', 'source': 'ai_notes.txt', 'chunk_index': 0, 'char_count': 65, 'author': 'team_docs', 'category': 'AI fundamentals'}
{'text': 'great for AI. Python is great for AI. Python is great for AI.', 'source': 'ai_notes.txt', 'chunk_index': 1, 'char_count': 63, 'author': 'team_docs', 'category': 'AI fundamentals'}

🤖 How Is This Used in AI? Metadata is what makes retrieval useful beyond just “here’s some text” — filtering by source, category, or author, and citing exactly where an answer came from, all depend on metadata attached at ingestion time like this.

---

9. Your First Complete AI Data Pipeline

This combines nearly every module in the course into one working, readable pipeline: load → clean → dedupe → chunk → embed → store → query.

Here is the data flow tracing how raw inputs are processed and made searchable in this pipeline:

graph TD
    raw[Raw Documents] --> clean[1. Clean Text: strip HTML / URLs]
    clean --> dedupe[2. Deduplicate: remove identical docs]
    dedupe --> chunk[3. Chunking: split with overlap]
    chunk --> embed[4. Embed: convert chunks to vectors]
    embed --> store[5. Store: save to simple Vector Store]
    store --> query[6. Query: rank by similarity & return prompt]

import re
import numpy as np
import logging

logging.basicConfig(level=logging.INFO, format="%(levelname)s | %(message)s")
logger = logging.getLogger("ai_pipeline")


# ---- Step 1: Cleaning (this module) ----
def clean_text(text):
    text = re.sub(r"<[^>]+>", "", text)          # strip HTML
    text = re.sub(r"https?://\S+", "", text)       # strip URLs
    text = re.sub(r"\s+", " ", text).strip()        # collapse whitespace
    return text.lower()


# ---- Step 2: Deduplication (this module + Module 2) ----
def deduplicate(documents):
    seen = set()
    result = []
    for doc in documents:
        if doc not in seen:
            seen.add(doc)
            result.append(doc)
    return result


# ---- Step 3: Chunking (this module + Module 4) ----
def chunk_text(text, chunk_size=8, overlap=2):
    words = text.split()
    chunks = []
    step = chunk_size - overlap
    for i in range(0, len(words), step):
        piece = words[i:i + chunk_size]
        if piece:
            chunks.append(" ".join(piece))
        if i + chunk_size >= len(words):
            break
    return chunks


# ---- Step 4: Embedding (Module 9 + Module 14) ----
def embed(text):
    """Stand-in for a real embedding API call (Module 11)."""
    np.random.seed(abs(hash(text)) % (10 ** 6))
    return np.random.rand(16)


# ---- Step 5: A tiny in-memory vector store (Module 5 - OOP) ----
class SimpleVectorStore:
    def __init__(self):
        self.records = []   # list of {"text": ..., "embedding": ..., metadata...}

    def add(self, text, embedding, metadata):
        self.records.append({"text": text, "embedding": embedding, **metadata})

    def query(self, query_embedding, top_k=3):
        def cosine_similarity(a, b):
            return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

        scored = [
            (cosine_similarity(query_embedding, r["embedding"]), r)
            for r in self.records
        ]
        scored.sort(key=lambda pair: pair[0], reverse=True)
        return [(score, r["text"]) for score, r in scored[:top_k]]


# ---- Step 6: The full pipeline (Module 4 - functions, Module 6 - exceptions) ----
def ingest_documents(raw_documents, source, store):
    logger.info(f"Ingesting {len(raw_documents)} raw document(s) from {source}")

    cleaned = [clean_text(doc) for doc in raw_documents if doc and doc.strip()]
    cleaned = deduplicate(cleaned)
    logger.info(f"After cleaning + dedup: {len(cleaned)} document(s)")

    total_chunks = 0
    for doc in cleaned:
        chunks = chunk_text(doc)
        for i, chunk in enumerate(chunks):
            try:
                embedding = embed(chunk)
                store.add(chunk, embedding, {"source": source, "chunk_index": i})
                total_chunks += 1
            except Exception as e:
                logger.error(f"Failed to embed chunk {i}: {e}")

    logger.info(f"Ingestion complete: {total_chunks} chunk(s) stored")


def answer_query(query, store, top_k=2):
    logger.info(f"Query received: '{query}'")
    query_embedding = embed(clean_text(query))
    results = store.query(query_embedding, top_k=top_k)

    context = "\n".join(text for _, text in results)
    prompt = f"""Answer using only this context.

Context:
{context}

Question: {query}
"""
    logger.info("Built final prompt, ready to send to an LLM")
    return prompt


# ---- Running the pipeline ----
raw_documents = [
    "<p>Python is widely used for building AI applications and RAG pipelines.</p>",
    "Python is widely used for building AI applications and RAG pipelines.",  # near-dup after cleaning
    "  ",   # empty, should be dropped
    "Retrieval-augmented generation combines search with language model generation. Visit https://example.com to learn more.",
]

store = SimpleVectorStore()
ingest_documents(raw_documents, source="demo_docs", store=store)

final_prompt = answer_query("How is Python used in AI?", store)
print("\n--- FINAL PROMPT ---")
print(final_prompt)

Expected Output (approximate — embedding values are deterministic but ranking may vary slightly):

INFO | Ingesting 4 raw document(s) from demo_docs
INFO | After cleaning + dedup: 2 document(s)
INFO | Ingestion complete: 2 chunk(s) stored
INFO | Query received: 'How is Python used in AI?'
INFO | Built final prompt, ready to send to an LLM

--- FINAL PROMPT ---
Answer using only this context.

Context:
python is widely used for building ai applications and rag pipelines.
retrieval-augmented generation combines search with language model generation. to learn more.

Question: How is Python used in AI?

Why This Matters

Every single piece of this pipeline is something you already understand in isolation: cleaning (re, this module), deduplication (Module 2’s sets), chunking (Module 4’s functions), embeddings (Module 9’s NumPy), a small class-based store (Module 5’s OOP), logging (Module 12), and error handling around each chunk’s embedding step (Module 6). A real production pipeline adds a real embedding API (Module 11), a real vector database, and probably async batch processing (Module 13) for speed — but the shape is exactly what you just built.

---

10. Testing Your Pipeline

Tying in Module 15 — this pipeline is entirely testable without any real API calls, because every piece is deterministic, pure Python logic:

def test_clean_text_strips_html_and_urls():
    result = clean_text("<p>Hello</p> visit https://example.com")
    assert "http" not in result
    assert "<" not in result

def test_deduplicate_removes_exact_matches():
    docs = ["a", "b", "a"]
    assert deduplicate(docs) == ["a", "b"]

def test_chunk_text_produces_overlapping_chunks():
    text = "one two three four five six seven eight nine ten"
    chunks = chunk_text(text, chunk_size=4, overlap=1)
    assert len(chunks) > 1
    # the overlap word should appear at the boundary between consecutive chunks
    assert chunks[0].split()[-1] == chunks[1].split()[0]

🤖 This is exactly the mocking-free testing category from Module 15 — pure logic, no external calls, fast and reliable every time. The embed() function (a real API call in production) is the one piece you’d mock in a test that touches the full ingest_documents pipeline.

---

Preserve the Raw Data and Record Provenance

Treat original input as read-only. Write cleaned data to a new, versioned location and record where it came from, when it was processed, which code and settings were used, and whether the organization has permission to use it. Personal data and confidential text may need removal or access controls before they reach an embedding service.

Cleaning Can Remove Meaning

Regex is useful for predictable text patterns, but complex HTML should be parsed with an HTML parser. Lowercasing can damage names or code, deleting punctuation can remove negation, and collapsing whitespace can break tables. Inspect samples before and after every transformation.

Characters and model tokens are not the same. A 1,000-character chunk can have different token counts in different languages or tokenizers. Store the embedding model and tokenizer version with the chunking settings so the index can be reproduced.

Make the Pipeline Repeatable and Leakage-Safe

An idempotent pipeline gives the same result when run twice on the same input instead of duplicating records. Use stable document IDs, content hashes, and stage versions so completed work can be reused and changed work can be rebuilt.

Split data before learning statistics or rules from it. Fit normalization, imputation, vocabulary, and other learned cleaning decisions on training data only, then apply the saved decisions to validation and test data. Otherwise the evaluation quietly sees information from its own answers.

Module Summary

You now know how to clean messy real-world text (regex, normalization, deduplication), handle missing data, chunk documents sensibly (with overlap), attach useful metadata, and — as a capstone — you’ve built a complete, working, testable AI data pipeline from raw documents to a query-ready vector store, using nothing but concepts from this course.

AI Connection

This module is the practical starting point of almost every real RAG project: messy source data rarely arrives ready to embed. Cleaning, deduplicating, and chunking well is frequently the highest-leverage, most underrated work in a RAG system — often mattering more than which embedding model or which LLM you eventually choose.

Mini Practice

1. Write a cleaning function that removes HTML tags, extra whitespace, and converts curly quotes (' ') to straight quotes (').
2. Extend the deduplication function to also treat two documents as duplicates if they’re identical after removing all punctuation.
3. Write a chunking function with configurable chunk_size and overlap, and a test confirming consecutive chunks actually share the expected number of overlapping words.
4. Add a category field to the chunk metadata in the full pipeline, and modify SimpleVectorStore.query to optionally filter by category before ranking by similarity.
5. Using Module 15’s mocking techniques, write a test for ingest_documents that mocks the embed() function so the test doesn’t depend on the actual (fake) embedding values.

---

You’ve now gone from print("hello world") all the way through building and testing a working AI data pipeline — cleaning, deduplicating, chunking, embedding, storing, and querying documents, backed by a real automated test suite that never needs a live API call to run.

Where to go next:

• Swap the fake embed() function for a real embedding API call (Module 11) and the SimpleVectorStore for a real vector database.
• Add the async batch-processing pattern from Module 13 to embed large document sets concurrently instead of one at a time.
• Wrap the pipeline’s LLM-facing step with the retry and logging patterns from Modules 6 and 12 for production readiness.