Data Quality Pipelines for AI Agents: Validation, Deduplication, and Normalization

Garbage In, Garbage Out — At AI Scale

Data quality problems in traditional software cause bugs. Data quality problems in AI agent systems cause hallucinations, wrong answers delivered with high confidence, and eroded user trust. An agent that retrieves a duplicate record with conflicting information will synthesize contradictory responses. An agent working with unnormalized dates or inconsistent naming conventions will fail at basic comparisons.

A data quality pipeline sits between ingestion and storage, acting as a gatekeeper that rejects, repairs, or flags problematic data before it reaches your agent's knowledge base.

Schema Validation

The first line of defense is schema validation. Every record entering your pipeline should conform to an expected structure with typed fields and constraints.

flowchart TD
    START["Data Quality Pipelines for AI Agents: Validation,…"] --> A
    A["Garbage In, Garbage Out — At AI Scale"]
    A --> B
    B["Schema Validation"]
    B --> C
    C["Fuzzy Deduplication"]
    C --> D
    D["Data Normalization"]
    D --> E
    E["Orchestrating the Quality Pipeline"]
    E --> F
    F["FAQ"]
    F --> DONE["Key Takeaways"]
    style START fill:#4f46e5,stroke:#4338ca,color:#fff
    style DONE fill:#059669,stroke:#047857,color:#fff

from pydantic import BaseModel, Field, field_validator
from typing import Optional, List
from datetime import datetime
from enum import Enum

class DataQuality(str, Enum):
    VALID = "valid"
    REPAIRED = "repaired"
    REJECTED = "rejected"

class KnowledgeRecord(BaseModel):
    source_id: str = Field(min_length=1, max_length=256)
    title: str = Field(min_length=5, max_length=500)
    content: str = Field(min_length=50)
    source_url: Optional[str] = None
    tags: List[str] = Field(default_factory=list)
    published_at: Optional[datetime] = None

    @field_validator("content")
    @classmethod
    def content_not_boilerplate(cls, v):
        boilerplate_phrases = [
            "lorem ipsum", "click here to subscribe",
            "cookie policy", "javascript is required",
        ]
        lower = v.lower()
        for phrase in boilerplate_phrases:
            if phrase in lower and len(v) < 200:
                raise ValueError(
                    f"Content appears to be boilerplate: {phrase}"
                )
        return v

    @field_validator("title")
    @classmethod
    def title_not_generic(cls, v):
        generic = ["untitled", "page", "home", "index", "null"]
        if v.strip().lower() in generic:
            raise ValueError(f"Title is generic: {v}")
        return v.strip()

class ValidationResult:
    def __init__(self):
        self.valid = []
        self.repaired = []
        self.rejected = []

    def summary(self) -> dict:
        total = len(self.valid) + len(self.repaired) + len(self.rejected)
        return {
            "total": total,
            "valid": len(self.valid),
            "repaired": len(self.repaired),
            "rejected": len(self.rejected),
            "rejection_rate": len(self.rejected) / max(total, 1),
        }

Fuzzy Deduplication

Exact deduplication catches identical records, but real-world duplicates are messier. The same article might appear with slightly different titles, extra whitespace, or minor edits. Fuzzy matching catches these near-duplicates.

from rapidfuzz import fuzz
from typing import List, Tuple
import hashlib

class FuzzyDeduplicator:
    def __init__(
        self,
        title_threshold: int = 85,
        content_threshold: int = 90,
    ):
        self.title_threshold = title_threshold
        self.content_threshold = content_threshold
        self.seen_titles: List[Tuple[str, str]] = []
        self.content_hashes: dict = {}

    def is_duplicate(self, record: KnowledgeRecord) -> Tuple[bool, str]:
        # Stage 1: exact content hash
        content_hash = hashlib.sha256(
            record.content.encode()
        ).hexdigest()
        if content_hash in self.content_hashes:
            return True, f"Exact duplicate of {self.content_hashes[content_hash]}"
        self.content_hashes[content_hash] = record.source_id

        # Stage 2: fuzzy title match
        for existing_id, existing_title in self.seen_titles:
            title_score = fuzz.ratio(
                record.title.lower(), existing_title.lower()
            )
            if title_score >= self.title_threshold:
                # Confirm with content similarity on first 500 chars
                return True, f"Fuzzy title match ({title_score}%) with {existing_id}"

        self.seen_titles.append((record.source_id, record.title))
        return False, ""

Data Normalization

Inconsistent formats make retrieval unreliable. Dates, company names, currencies, and units all need standardization.

import re
from datetime import datetime
from typing import Optional

class DataNormalizer:
    def normalize(self, record: dict) -> dict:
        normalized = {}
        for key, value in record.items():
            if isinstance(value, str):
                value = self._clean_text(value)
            normalized[key] = value

        if "published_at" in normalized:
            normalized["published_at"] = self._normalize_date(
                normalized["published_at"]
            )
        if "company" in normalized:
            normalized["company"] = self._normalize_company(
                normalized["company"]
            )
        return normalized

    def _clean_text(self, text: str) -> str:
        # Collapse whitespace
        text = re.sub(r"\s+", " ", text).strip()
        # Remove zero-width characters
        text = re.sub(r"[\u200b-\u200d\ufeff]", "", text)
        # Normalize quotes
        text = text.replace("\u201c", '"').replace("\u201d", '"')
        text = text.replace("\u2018", "'").replace("\u2019", "'")
        return text

    def _normalize_date(self, date_str) -> Optional[str]:
        if isinstance(date_str, datetime):
            return date_str.isoformat()
        formats = [
            "%Y-%m-%d", "%m/%d/%Y", "%d/%m/%Y",
            "%B %d, %Y", "%b %d, %Y", "%Y-%m-%dT%H:%M:%S",
        ]
        for fmt in formats:
            try:
                return datetime.strptime(date_str, fmt).isoformat()
            except (ValueError, TypeError):
                continue
        return None

    def _normalize_company(self, name: str) -> str:
        suffixes = [
            " Inc.", " Inc", " LLC", " Ltd.",
            " Ltd", " Corp.", " Corp", " Co.",
        ]
        cleaned = name.strip()
        for suffix in suffixes:
            if cleaned.endswith(suffix):
                cleaned = cleaned[: -len(suffix)].strip()
        return cleaned

Orchestrating the Quality Pipeline

Combine all stages into a single pipeline that processes records in sequence.

class DataQualityPipeline:
    def __init__(self):
        self.normalizer = DataNormalizer()
        self.deduplicator = FuzzyDeduplicator()
        self.results = ValidationResult()

    def process(self, raw_records: List[dict]) -> List[KnowledgeRecord]:
        clean_records = []
        for raw in raw_records:
            # Stage 1: normalize
            normalized = self.normalizer.normalize(raw)

            # Stage 2: validate
            try:
                record = KnowledgeRecord(**normalized)
            except Exception as e:
                self.results.rejected.append(
                    {"data": raw, "reason": str(e)}
                )
                continue

            # Stage 3: deduplicate
            is_dup, reason = self.deduplicator.is_duplicate(record)
            if is_dup:
                self.results.rejected.append(
                    {"data": raw, "reason": f"Duplicate: {reason}"}
                )
                continue

            self.results.valid.append(record)
            clean_records.append(record)

        return clean_records

FAQ

How do I handle records that are partially valid — some fields are good but others are not?

Implement a repair stage between validation and rejection. If a record fails on a non-critical field like published_at, set a default value and mark the record as "repaired" in its metadata. Only reject records when critical fields like content or source_id fail validation. Track repair rates — a spike in repairs often signals an upstream data source problem.

What fuzzy matching threshold should I use for deduplication?

Start with 85% for titles and 90% for content. Lower thresholds catch more duplicates but increase false positives — merging distinct articles that happen to share similar language. Run the deduplicator on a sample of your actual data and manually review the matches at your chosen threshold to calibrate.

How do I monitor data quality over time?

Track validation metrics per pipeline run: rejection rate, repair rate, duplicate rate, and records per source. Set alerts when the rejection rate exceeds your baseline by more than two standard deviations. A sudden spike usually means an upstream source changed its format or started returning error pages.

---

#DataQuality #Validation #Deduplication #DataPipelines #AIAgents #AgenticAI #LearnAI #AIEngineering