AI-Assisted Data Cleaning: Using LLMs to Standardize, Deduplicate, and Validate Messy Datasets

You've inherited a CRM export with 47,000 rows. Company names are a disaster — "IBM Corp", "I.B.M.", "International Business Machines", and "ibm" all appear in the same column, referring to the same entity. Job titles range from "Sr. SWE" to "Senior Software Engineer" to "swe ii (senior)". Phone numbers have seven different formatting conventions. And somewhere in there, you're pretty sure there are duplicates — but fuzzy ones, the kind that df.duplicated() will never catch.

This is the real shape of data work. The cleaning phase doesn't just take time — it takes judgment. You need to understand context, recognize synonyms, apply domain knowledge, and make thousands of small decisions. That's exactly the kind of work large language models are surprisingly good at, when you use them correctly.

By the end of this lesson, you'll be able to systematically use LLMs — through direct API calls, prompt-engineered batch jobs, and hybrid validation pipelines — to tackle the three hardest parts of data cleaning: standardization, deduplication, and validation. You'll build a working Python pipeline you can adapt to your own messy datasets.

What you'll learn:

• How to design prompts that return structured, parseable output from LLMs for data cleaning tasks
• Techniques for batch-processing records efficiently without blowing through your token budget
• How to use LLMs for fuzzy deduplication that goes beyond string similarity
• A validation strategy that combines rule-based checks with LLM judgment for ambiguous cases
• How to evaluate LLM cleaning quality and know when to trust it

---

Prerequisites

You should be comfortable with Python, pandas, and the basics of making API calls. You should have used an LLM API (OpenAI, Anthropic, or similar) at least once — we won't be explaining how to set up API keys from scratch. Familiarity with basic prompt engineering concepts (system prompts, temperature, few-shot examples) will help but isn't strictly required.

You'll need:

• openai Python package (we'll use the OpenAI API; the patterns transfer to other providers)
• pandas, tqdm, thefuzz (formerly fuzzywuzzy)
• A dataset to practice on — we'll build one inline

---

The Core Insight: LLMs as Judgment Engines, Not Magic Wands

Before writing a line of code, you need to internalize something that will save you a lot of frustration: LLMs are not reliable as black-box cleaners. If you paste a messy CSV and ask an LLM to "clean it," you'll get inconsistent results, hallucinated corrections, and no audit trail.

What LLMs are reliable at is making classification and normalization decisions when given clear criteria, structured input, and constrained output formats. The difference is architectural:

Wrong approach: "Here's my data, please clean it."

Right approach: "Given these specific values and these specific rules, which of these canonical forms best matches each input? Respond only in JSON."

The right approach treats the LLM like a very smart intern who needs explicit instructions, examples, and a checklist — not a consultant you hand a project to and walk away from. Every prompt you write for data cleaning should have three components:

1. A clear task definition with explicit criteria
2. Few-shot examples showing the transformation you want
3. A constrained output format (JSON, CSV, or a specific string) so you can parse results programmatically

Keep this framework in mind throughout the lesson. Every technique we build will follow it.

---

Setting Up Your Environment and Test Dataset

Let's establish our working environment and create a realistic messy dataset we'll use throughout the lesson.

import openai
import pandas as pd
import json
import time
from tqdm import tqdm
from thefuzz import fuzz

client = openai.OpenAI()  # assumes OPENAI_API_KEY is set in your environment

# Build a realistic messy CRM dataset
data = {
    "company_name": [
        "IBM Corp", "I.B.M.", "International Business Machines", "ibm",
        "Salesforce Inc", "salesforce", "Salesforce, Inc.", "SALESFORCE INC.",
        "McKinsey & Company", "Mckinsey and Company", "McKinsey",
        "3M Company", "3m", "Three M Company",
        "General Electric", "GE", "G.E. Company", "general electric co",
        "Apple Inc.", "Apple", "APPLE INC", "Apple Computer Inc",
    ],
    "job_title": [
        "Sr. SWE", "Senior Software Engineer", "swe ii (senior)", "Senior SWE",
        "VP Sales", "Vice President of Sales", "VP, Sales", "vp of sales",
        "Data Scientist", "Sr Data Scientist", "data scientist II", "Data Science Lead",
        "Mktg Manager", "Marketing Manager", "manager, marketing", "MARKETING MGR",
        "CEO", "Chief Executive Officer", "chief exec", "C.E.O.",
        "PM", "Product Manager",
    ],
    "phone": [
        "212-555-0100", "(212) 555-0101", "2125550102", "+1-212-555-0103",
        "212.555.0104", "1 212 555 0105", "+12125550106", "212 555 0107",
        "212-555-0108", "(212)555-0109", "2125550110", "212/555/0111",
        "415-555-0200", "(415) 555-0201", "4155550202", "+1 415 555 0203",
        "415.555.0204", "1-415-555-0205", "+14155550206", "415 555 0207",
        "415-555-0208", "(415)555-0209",
    ],
    "email": [
        "john.doe@ibm.com", "j.doe@ibm.com", "johndoe@ibm.com", "John.Doe@IBM.com",
        "sarah.smith@salesforce.com", "ssmith@salesforce.com", "s.smith@salesforce.com", "SARAH.SMITH@SALESFORCE.COM",
        "mike.chen@mckinsey.com", "m.chen@mckinsey.com", "mchen@mckinsey.com", "Mike.Chen@McKinsey.com",
        "anna.k@3m.com", "a.kowalski@3m.com", "anna.kowalski@3m.com", "A.Kowalski@3M.com",
        "tom.j@ge.com", "t.jones@ge.com", "thomas.jones@ge.com", "TJones@GE.com",
        "lisa.park@apple.com", "l.park@apple.com",
    ]
}

df = pd.DataFrame(data)
print(df.shape)
print(df.head(10))

This gives us 22 rows with four messy columns. Small enough to inspect manually, large enough to illustrate all the problems you'd face at 22,000 rows.

---

Part 1: Standardizing Company Names

Company name standardization is where LLMs shine brightest because it requires genuine world knowledge. A rule-based system might normalize punctuation and capitalization, but it can't know that "McKinsey and Company" and "McKinsey" refer to the same entity, or that "Three M Company" is "3M."

Designing the Prompt

The key design decision is whether to normalize to a canonical list you provide or to ask the LLM to infer the canonical form. The first approach is more reliable when you have a known universe of companies. The second is more flexible when you don't.

Let's implement both.

Approach A: Normalization against a known canonical list

def standardize_company_names(names: list[str], canonical_list: list[str]) -> dict:
    """
    Given a list of messy company names and a canonical list,
    return a mapping from each input to its best canonical match.
    """
    
    system_prompt = """You are a data cleaning assistant specializing in company name normalization. 
Your job is to match messy company name variants to their canonical form from a provided list.

Rules:
- Match based on the actual company identity, not just string similarity
- Use your knowledge of common company names, abbreviations, and aliases
- If no canonical match exists, use null
- Respond ONLY with valid JSON, no explanation"""

    user_prompt = f"""Canonical company list:
{json.dumps(canonical_list, indent=2)}

Match each of these messy names to the best canonical form:
{json.dumps(names, indent=2)}

Respond with a JSON object where keys are the input names and values are the matched canonical names (or null if no match):
{{"input_name": "canonical_name_or_null", ...}}"""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        temperature=0,  # deterministic output for data cleaning
        response_format={"type": "json_object"}
    )
    
    return json.loads(response.choices[0].message.content)


# Define our canonical company list
canonical_companies = [
    "IBM", "Salesforce", "McKinsey & Company", 
    "3M", "General Electric", "Apple"
]

# Get unique company names to avoid re-processing duplicates
unique_companies = df["company_name"].unique().tolist()

# Run standardization
company_mapping = standardize_company_names(unique_companies, canonical_companies)
print(json.dumps(company_mapping, indent=2))

With temperature=0, you get deterministic output — critical for reproducible data cleaning. The response_format={"type": "json_object"} parameter forces valid JSON back, eliminating the most common parsing failure.

Expected output looks like:

{
  "IBM Corp": "IBM",
  "I.B.M.": "IBM",
  "International Business Machines": "IBM",
  "ibm": "IBM",
  "Salesforce Inc": "Salesforce",
  "salesforce": "Salesforce",
  ...
  "Three M Company": "3M"
}

Now apply it:

# Apply the mapping to your dataframe
df["company_name_clean"] = df["company_name"].map(company_mapping)

# Audit what didn't get matched
unmatched = df[df["company_name_clean"].isna()]["company_name"].unique()
if len(unmatched) > 0:
    print(f"Unmatched companies: {unmatched}")

Always audit your nulls. They represent either a genuine gap in your canonical list or a failure in the LLM's matching. Both need human review.

Batching for Large Datasets

Sending one company name per API call is expensive and slow. Sending 10,000 in a single call risks context window limits and degraded accuracy. The sweet spot for most models is 25–100 records per call, depending on record complexity.

def batch_standardize(records: list[str], canonical_list: list[str], 
                       batch_size: int = 50) -> dict:
    """Process records in batches, with retry logic and result aggregation."""
    
    all_mappings = {}
    unique_records = list(set(records))  # deduplicate before sending
    
    # Split into batches
    batches = [unique_records[i:i+batch_size] 
               for i in range(0, len(unique_records), batch_size)]
    
    for batch in tqdm(batches, desc="Standardizing"):
        for attempt in range(3):  # retry up to 3 times
            try:
                result = standardize_company_names(batch, canonical_list)
                all_mappings.update(result)
                time.sleep(0.5)  # respect rate limits
                break
            except (json.JSONDecodeError, openai.APIError) as e:
                if attempt == 2:
                    print(f"Failed batch after 3 attempts: {e}")
                    # Mark failures explicitly rather than silently skipping
                    for record in batch:
                        all_mappings[record] = "NEEDS_REVIEW"
                time.sleep(2 ** attempt)  # exponential backoff
    
    return all_mappings

# Save intermediate results
import json

with open("company_mappings.json", "w") as f:
    json.dump(company_mapping, f, indent=2)

# Load if re-running
with open("company_mappings.json", "r") as f:
    company_mapping = json.load(f)

---

Part 2: Job Title Normalization with a Taxonomy

Job titles are nastier than company names because there's no canonical ground truth — "Senior Software Engineer" and "Staff Engineer" might be equivalent at one company and very different at another. You need to normalize to your taxonomy, which means you define the canonical forms.

This is also a great use case for few-shot prompting because the examples teach the LLM your specific normalization logic.

JOB_TAXONOMY = {
    "Engineering": ["Junior Software Engineer", "Software Engineer", 
                    "Senior Software Engineer", "Staff Engineer", "Principal Engineer"],
    "Sales": ["Sales Representative", "Account Executive", "Sales Manager", 
              "VP of Sales", "Chief Revenue Officer"],
    "Data": ["Data Analyst", "Data Scientist", "Senior Data Scientist", 
             "Data Science Manager", "VP of Data"],
    "Marketing": ["Marketing Coordinator", "Marketing Manager", 
                  "Director of Marketing", "VP of Marketing", "CMO"],
    "Executive": ["CEO", "COO", "CTO", "CFO", "President"]
}

def normalize_job_titles(titles: list[str], taxonomy: dict) -> list[dict]:
    """
    Normalize job titles against a taxonomy, returning both the 
    canonical title and the department.
    """
    
    # Flatten taxonomy for the prompt
    taxonomy_str = "\n".join([
        f"{dept}: {', '.join(titles_list)}" 
        for dept, titles_list in taxonomy.items()
    ])
    
    system_prompt = """You are a data cleaning specialist normalizing job titles for a B2B database.
Map each messy job title to the closest match in the provided taxonomy.
Consider abbreviations, seniority levels, and common variations.

Respond ONLY with a JSON array of objects, one per input title, in the same order."""

    few_shot_examples = """Examples:
Input: ["Sr. SWE", "VP Sales", "mktg coordinator"]
Output: [
  {"original": "Sr. SWE", "canonical": "Senior Software Engineer", "department": "Engineering", "confidence": "high"},
  {"original": "VP Sales", "canonical": "VP of Sales", "department": "Sales", "confidence": "high"},
  {"original": "mktg coordinator", "canonical": "Marketing Coordinator", "department": "Marketing", "confidence": "high"}
]"""

    user_prompt = f"""Taxonomy:
{taxonomy_str}

{few_shot_examples}

Now normalize these titles:
{json.dumps(titles)}

Return a JSON array with the same structure as the examples. 
Use "low" confidence when the match is ambiguous."""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        temperature=0,
        response_format={"type": "json_object"}
    )
    
    # The model might wrap the array in an object
    result = json.loads(response.choices[0].message.content)
    
    # Handle both {"results": [...]} and direct array responses
    if isinstance(result, list):
        return result
    elif isinstance(result, dict):
        # Find the first list value in the response
        for v in result.values():
            if isinstance(v, list):
                return v
    
    raise ValueError(f"Unexpected response structure: {result}")


# Test it
sample_titles = df["job_title"].unique().tolist()
normalized = normalize_job_titles(sample_titles, JOB_TAXONOMY)

# Convert to dataframe and merge back
normalized_df = pd.DataFrame(normalized)
print(normalized_df)

Notice the confidence field in the output. This is intentional — when you ask the LLM to report its own uncertainty, you can use that signal to route low-confidence records to human reviewers instead of blindly trusting them.

# Flag low-confidence results for review
low_confidence = normalized_df[normalized_df["confidence"] == "low"]
print(f"\n{len(low_confidence)} titles need human review:")
print(low_confidence[["original", "canonical"]].to_string())

---

Part 3: Phone Number Standardization — When You Don't Need an LLM

Here's an important lesson embedded inside a data cleaning lesson: not every task needs an LLM.

Phone number standardization is rule-based at its core. Strip non-digits, validate length, apply E.164 format. An LLM will get this right, but it's slower, more expensive, and less reliable than a regex-based approach.

import re

def standardize_phone(phone: str, country_code: str = "1") -> str | None:
    """
    Standardize phone numbers to E.164 format.
    Returns None for invalid numbers.
    """
    if not phone or pd.isna(phone):
        return None
    
    # Strip everything except digits
    digits = re.sub(r'\D', '', str(phone))
    
    # Handle leading country code
    if digits.startswith(country_code) and len(digits) == 11:
        digits = digits[1:]  # strip the leading 1 for US numbers
    
    # Validate: US numbers should be 10 digits
    if len(digits) != 10:
        return None
    
    # Format to E.164
    return f"+1{digits}"


df["phone_clean"] = df["phone"].apply(standardize_phone)
print(df[["phone", "phone_clean"]].to_string())

Save LLM calls for problems that require language understanding and world knowledge. For deterministic transformations, write deterministic code. The art of AI-assisted data cleaning is knowing which problems benefit from LLM judgment.

Decision rule: If you could write a complete set of rules that covers every case, use code. If the rules would require a dictionary of world knowledge or contextual judgment to apply, use an LLM.

---

Part 4: Fuzzy Deduplication with LLM Judgment

This is where things get genuinely interesting. Traditional deduplication uses exact matching or fuzzy string similarity (Levenshtein distance, Jaccard similarity, etc.). These work well for typos but fail for semantic duplicates — two records that represent the same person but look very different on paper.

Our strategy is a two-stage pipeline:

1. Stage 1 (fast, cheap): Use fuzzy string matching to generate candidate pairs — records that might be duplicates
2. Stage 2 (slow, smart): Use an LLM to evaluate each candidate pair and make the final call

Stage 1: Generating Candidate Pairs

from thefuzz import fuzz
from itertools import combinations

def find_candidate_pairs(df: pd.DataFrame, 
                          columns: list[str],
                          threshold: int = 75) -> list[tuple]:
    """
    Find pairs of records with fuzzy similarity above threshold.
    Returns list of (idx1, idx2, similarity_score) tuples.
    """
    candidates = []
    
    for idx1, idx2 in combinations(df.index, 2):
        # Combine multiple columns into a composite string for comparison
        record1 = " ".join([str(df.loc[idx1, col]) for col in columns])
        record2 = " ".join([str(df.loc[idx2, col]) for col in columns])
        
        score = fuzz.token_sort_ratio(record1, record2)
        
        if score >= threshold:
            candidates.append((idx1, idx2, score))
    
    return sorted(candidates, key=lambda x: x[2], reverse=True)


# Find candidate duplicate pairs using name + email as signals
candidates = find_candidate_pairs(
    df, 
    columns=["company_name", "email"],
    threshold=70
)

print(f"Found {len(candidates)} candidate pairs")
for idx1, idx2, score in candidates[:10]:
    print(f"\nScore: {score}")
    print(f"  Record {idx1}: {df.loc[idx1, 'company_name']} | {df.loc[idx1, 'email']}")
    print(f"  Record {idx2}: {df.loc[idx2, 'company_name']} | {df.loc[idx2, 'email']}")

Stage 2: LLM Evaluation of Candidate Pairs

Now we send only the ambiguous pairs to the LLM — typically a small fraction of all possible comparisons.

def evaluate_duplicate_pairs(pairs: list[tuple], df: pd.DataFrame) -> list[dict]:
    """
    Use LLM to evaluate candidate duplicate pairs and classify them.
    """
    
    if not pairs:
        return []
    
    # Build the batch payload
    comparisons = []
    for idx1, idx2, fuzzy_score in pairs:
        r1 = df.loc[idx1].to_dict()
        r2 = df.loc[idx2].to_dict()
        comparisons.append({
            "pair_id": f"{idx1}_{idx2}",
            "fuzzy_score": fuzzy_score,
            "record_a": r1,
            "record_b": r2
        })
    
    system_prompt = """You are a data deduplication specialist. 
Analyze pairs of records and determine if they represent the same real-world entity.

Consider:
- Company name variants and abbreviations
- Email address patterns (same domain, similar local parts)
- Phone number equivalence
- Job title similarity at the same company

Respond ONLY with a JSON array where each object has:
- pair_id: the input pair_id
- verdict: "duplicate", "not_duplicate", or "uncertain"  
- confidence: "high", "medium", or "low"
- reasoning: one sentence explaining your decision"""

    user_prompt = f"""Evaluate these record pairs:

{json.dumps(comparisons, indent=2)}

Return a JSON object with key "results" containing the array."""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        temperature=0,
        response_format={"type": "json_object"}
    )
    
    result = json.loads(response.choices[0].message.content)
    return result.get("results", [])


# Evaluate candidate pairs (batch them for large sets)
DEDUP_BATCH_SIZE = 10  # pairs per call — keep small for accuracy

all_evaluations = []
for i in range(0, len(candidates), DEDUP_BATCH_SIZE):
    batch = candidates[i:i+DEDUP_BATCH_SIZE]
    evaluations = evaluate_duplicate_pairs(batch, df)
    all_evaluations.extend(evaluations)
    time.sleep(0.5)

# Analyze results
eval_df = pd.DataFrame(all_evaluations)
print("\nDeduplication Results:")
print(eval_df["verdict"].value_counts())
print("\nDuplicates found:")
print(eval_df[eval_df["verdict"] == "duplicate"][["pair_id", "confidence", "reasoning"]].to_string())

Handling the Uncertain Verdict

Records flagged as "uncertain" deserve special attention. They're the cases where even the LLM couldn't decide — which usually means a human reviewer needs more context.

# Route uncertain cases to a review queue with additional context
uncertain = eval_df[eval_df["verdict"] == "uncertain"]

review_queue = []
for _, row in uncertain.iterrows():
    idx1, idx2 = row["pair_id"].split("_")
    review_queue.append({
        "pair_id": row["pair_id"],
        "reasoning": row["reasoning"],
        "record_a": df.loc[int(idx1)].to_dict(),
        "record_b": df.loc[int(idx2)].to_dict(),
        "action_needed": "Human review required"
    })

review_df = pd.DataFrame(review_queue)
review_df.to_csv("dedup_review_queue.csv", index=False)
print(f"\n{len(review_queue)} pairs exported to review queue")

---

Part 5: LLM-Powered Validation

Validation is the underrated step. You've standardized and deduplicated — now you need to verify that the data makes sense. This is different from format validation (which you do with code). LLM validation catches semantic inconsistencies: a CEO at a 5-person startup listed with a Fortune 500 company, an email domain that doesn't match the company name, a job title that doesn't exist in any industry.

def validate_records(records: list[dict]) -> list[dict]:
    """
    Validate records for semantic consistency and flag anomalies.
    """
    
    system_prompt = """You are a data quality analyst reviewing CRM records for a B2B database.
For each record, check for:
1. Email domain consistency with company name
2. Job title plausibility (does this title exist in industry?)
3. Internal consistency (does the title match the seniority implied by other fields?)
4. Obvious data entry errors or field swaps

Return ONLY a JSON object with key "validations" containing an array.
Each item must have: record_id, issues (array of strings), severity ("ok", "warning", "error")"""

    user_prompt = f"""Validate these CRM records:

{json.dumps(records, indent=2)}

Flag any semantic inconsistencies, not formatting issues."""

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        temperature=0,
        response_format={"type": "json_object"}
    )
    
    result = json.loads(response.choices[0].message.content)
    return result.get("validations", [])


# Prepare records for validation (use cleaned versions where available)
records_to_validate = []
for idx, row in df.iterrows():
    records_to_validate.append({
        "record_id": idx,
        "company": row.get("company_name_clean", row["company_name"]),
        "job_title": row["job_title"],
        "email": row["email"],
        "phone": row.get("phone_clean", row["phone"])
    })

# Validate in batches of 15 (records can be verbose)
validation_results = []
for i in range(0, len(records_to_validate), 15):
    batch = records_to_validate[i:i+15]
    results = validate_records(batch)
    validation_results.extend(results)
    time.sleep(0.5)

# Surface warnings and errors
validation_df = pd.DataFrame(validation_results)
issues = validation_df[validation_df["severity"].isin(["warning", "error"])]
print(f"\nValidation Issues Found: {len(issues)}")
for _, row in issues.iterrows():
    print(f"\nRecord {row['record_id']} [{row['severity'].upper()}]:")
    for issue in row["issues"]:
        print(f"  - {issue}")

---

Hands-On Exercise: Building a Complete Cleaning Pipeline

Now you're going to put all three components together into a single, reusable pipeline function. This is the "real-world project" deliverable — something you can actually adapt for your work.

def clean_crm_dataset(
    df: pd.DataFrame,
    canonical_companies: list[str],
    job_taxonomy: dict,
    output_path: str = "cleaned_dataset.csv"
) -> pd.DataFrame:
    """
    Full LLM-assisted cleaning pipeline for CRM data.
    
    Stages:
    1. Phone standardization (rule-based)
    2. Company name normalization (LLM)
    3. Job title normalization (LLM)
    4. Fuzzy deduplication with LLM evaluation
    5. Semantic validation (LLM)
    6. Output with audit columns
    """
    
    print("=== CRM Data Cleaning Pipeline ===\n")
    result_df = df.copy()
    
    # --- Stage 1: Phone (rule-based, fast) ---
    print("Stage 1/5: Standardizing phone numbers...")
    result_df["phone_clean"] = result_df["phone"].apply(standardize_phone)
    phone_failures = result_df["phone_clean"].isna().sum()
    print(f"  ✓ {len(result_df) - phone_failures} valid | {phone_failures} invalid/missing\n")
    
    # --- Stage 2: Company Names (LLM) ---
    print("Stage 2/5: Normalizing company names...")
    unique_companies = result_df["company_name"].unique().tolist()
    company_mapping = batch_standardize(unique_companies, canonical_companies, batch_size=25)
    result_df["company_name_clean"] = result_df["company_name"].map(company_mapping)
    
    needs_review = (result_df["company_name_clean"] == "NEEDS_REVIEW").sum()
    unmatched = result_df["company_name_clean"].isna().sum()
    print(f"  ✓ Matched: {len(result_df) - needs_review - unmatched}")
    print(f"  ⚠ Needs review: {needs_review}")
    print(f"  ✗ Unmatched: {unmatched}\n")
    
    # --- Stage 3: Job Titles (LLM) ---
    print("Stage 3/5: Normalizing job titles...")
    unique_titles = result_df["job_title"].unique().tolist()
    
    title_results = normalize_job_titles(unique_titles, job_taxonomy)
    title_df = pd.DataFrame(title_results)
    
    title_map = dict(zip(title_df["original"], title_df["canonical"]))
    dept_map = dict(zip(title_df["original"], title_df["department"]))
    conf_map = dict(zip(title_df["original"], title_df["confidence"]))
    
    result_df["job_title_clean"] = result_df["job_title"].map(title_map)
    result_df["department"] = result_df["job_title"].map(dept_map)
    result_df["title_confidence"] = result_df["job_title"].map(conf_map)
    
    low_conf = (result_df["title_confidence"] == "low").sum()
    print(f"  ✓ Normalized: {len(result_df) - low_conf}")
    print(f"  ⚠ Low confidence (needs review): {low_conf}\n")
    
    # --- Stage 4: Deduplication ---
    print("Stage 4/5: Running fuzzy deduplication...")
    candidates = find_candidate_pairs(
        result_df,
        columns=["company_name_clean", "email"],
        threshold=70
    )
    print(f"  Found {len(candidates)} candidate pairs")
    
    if candidates:
        all_evals = []
        for i in range(0, len(candidates), 10):
            batch = candidates[i:i+10]
            evals = evaluate_duplicate_pairs(batch, result_df)
            all_evals.extend(evals)
            time.sleep(0.5)
        
        eval_df = pd.DataFrame(all_evals)
        confirmed_dupes = eval_df[
            (eval_df["verdict"] == "duplicate") & 
            (eval_df["confidence"] == "high")
        ]
        print(f"  ✓ High-confidence duplicates: {len(confirmed_dupes)}")
        print(f"  ⚠ Uncertain (review queue): {len(eval_df[eval_df['verdict'] == 'uncertain'])}")
        
        # Export review queue
        uncertain_mask = eval_df["verdict"].isin(["uncertain"])
        if uncertain_mask.any():
            eval_df[uncertain_mask].to_csv("dedup_review_queue.csv", index=False)
    
    # --- Stage 5: Semantic Validation ---
    print("\nStage 5/5: Running semantic validation...")
    records_for_validation = []
    for idx, row in result_df.iterrows():
        records_for_validation.append({
            "record_id": int(idx),
            "company": row.get("company_name_clean") or row["company_name"],
            "job_title": row.get("job_title_clean") or row["job_title"],
            "email": row["email"],
            "phone": row.get("phone_clean") or row["phone"]
        })
    
    val_results = []
    for i in range(0, len(records_for_validation), 15):
        batch = records_for_validation[i:i+15]
        results = validate_records(batch)
        val_results.extend(results)
        time.sleep(0.5)
    
    val_df = pd.DataFrame(val_results).set_index("record_id")
    result_df["validation_status"] = result_df.index.map(
        lambda i: val_df.loc[i, "severity"] if i in val_df.index else "ok"
    )
    result_df["validation_issues"] = result_df.index.map(
        lambda i: "; ".join(val_df.loc[i, "issues"]) if i in val_df.index else ""
    )
    
    errors = (result_df["validation_status"] == "error").sum()
    warnings = (result_df["validation_status"] == "warning").sum()
    print(f"  ✗ Errors: {errors}")
    print(f"  ⚠ Warnings: {warnings}")
    print(f"  ✓ Clean: {len(result_df) - errors - warnings}\n")
    
    # --- Output ---
    result_df.to_csv(output_path, index=False)
    print(f"✅ Pipeline complete. Cleaned dataset saved to: {output_path}")
    print(f"   Rows: {len(result_df)} | Columns: {len(result_df.columns)}")
    
    return result_df


# Run the complete pipeline
cleaned_df = clean_crm_dataset(
    df=df,
    canonical_companies=canonical_companies,
    job_taxonomy=JOB_TAXONOMY,
    output_path="crm_cleaned.csv"
)

---

Common Mistakes & Troubleshooting

1. Not setting temperature=0

This is the single most common mistake. With temperature > 0, the same input can produce different outputs on different runs, making your cleaning pipeline non-reproducible. Always use temperature=0 for data cleaning tasks.

2. Trusting 100% of LLM output without sampling

Even with good prompts, LLMs make errors. After any batch cleaning job, sample 50–100 records and manually verify the transformations. Calculate an error rate. If it's above 2–3%, your prompt needs work before you trust it at scale.

3. Sending too many records per batch

More records per batch feels efficient, but LLM accuracy degrades as context grows. The model pays less attention to records buried in the middle of a long list. For normalization tasks, 25–50 records per call is usually optimal. Test your specific model's accuracy curve by comparing small vs. large batch results on the same data.

4. Not handling partial failures gracefully

API calls fail. JSON parsing fails. Your pipeline will run at 2 AM and you won't be there to restart it. Build explicit failure handling that marks failed records as NEEDS_REVIEW rather than silently dropping them or crashing the whole run.

5. Using LLMs for tasks that have deterministic solutions

Phone formatting, email lowercase normalization, zip code validation — these have rules. Write code. Every unnecessary LLM call adds latency, cost, and a chance of error. Reserve LLM calls for genuinely ambiguous, language-dependent tasks.

6. Conflating standardization with enrichment

LLMs can "helpfully" fill in information that wasn't in the original record. A prompt that asks to normalize company names might return an industry category the model inferred. That's enrichment, not cleaning — and it introduces data that wasn't in your source system, which can cause serious integrity issues. Keep prompts focused on transforming what's there, not adding what isn't.

7. Not preserving the original values

Always keep original columns alongside cleaned columns. company_name and company_name_clean should both exist in your output. You need to be able to audit the transformation, and you need a fallback if the cleaning was wrong.

---

Evaluating Cleaning Quality

Before deploying any LLM cleaning pipeline on production data, run a structured evaluation:

def evaluate_cleaning_quality(original_df: pd.DataFrame, 
                               cleaned_df: pd.DataFrame,
                               column_pairs: list[tuple],
                               sample_size: int = 100) -> pd.DataFrame:
    """
    Sample records and generate an evaluation report for human review.
    column_pairs: list of (original_col, cleaned_col) tuples
    """
    
    sample_idx = original_df.sample(min(sample_size, len(original_df))).index
    
    rows = []
    for idx in sample_idx:
        for orig_col, clean_col in column_pairs:
            orig_val = original_df.loc[idx, orig_col]
            clean_val = cleaned_df.loc[idx, clean_col] if clean_col in cleaned_df.columns else "N/A"
            rows.append({
                "record_id": idx,
                "field": orig_col,
                "original": orig_val,
                "cleaned": clean_val,
                "changed": str(orig_val).strip().lower() != str(clean_val).strip().lower(),
                "reviewer_verdict": "",  # human fills this in
                "reviewer_notes": ""
            })
    
    eval_df = pd.DataFrame(rows)
    eval_df.to_csv("cleaning_evaluation.csv", index=False)
    print(f"Evaluation sheet saved. Review {len(eval_df[eval_df['changed']])} changed values.")
    return eval_df


evaluate_cleaning_quality(
    df, 
    cleaned_df,
    column_pairs=[
        ("company_name", "company_name_clean"),
        ("job_title", "job_title_clean")
    ]
)

Export this to a spreadsheet, have a domain expert review a sample, and calculate: what percentage of the LLM's changes were correct? Anything below 95% on standardization tasks should send you back to refine your prompts.

---

Summary & Next Steps

You've built a production-grade AI-assisted data cleaning pipeline that handles the three hardest problems in messy datasets: standardization with world knowledge, semantic deduplication, and intelligent validation. The key principles to carry forward:

• Structure your prompts with explicit rules, few-shot examples, and constrained output formats — never ask an LLM to "clean data" freeform
• Use rule-based code for deterministic transformations; save LLM calls for problems that require language judgment
• Batch intelligently — 25–50 records per call for most normalization tasks, with retry logic and intermediate result caching
• Never auto-delete based on LLM deduplication; route flagged pairs to a human review queue
• Preserve originals, capture confidence signals, and evaluate accuracy on a sample before trusting the pipeline at scale

Where to go from here:

• Fine-tuning for domain-specific cleaning: If you're working in a specific industry (healthcare, finance, legal), fine-tuning a smaller model on domain-specific standardization examples can dramatically improve accuracy and reduce cost compared to prompting GPT-4o
• Embedding-based deduplication: For very large datasets, replace fuzzy string matching in Stage 1 with vector embeddings and approximate nearest-neighbor search — it scales to millions of records where pairwise comparison doesn't
• Building a cleaning audit trail: Production data pipelines need provenance. Extend this pipeline to log every transformation with its source, timestamp, model version, and confidence score — essential for compliance and debugging
• Streaming pipelines: Adapt these patterns to process incoming records in real time using async API calls rather than batch jobs, feeding into a message queue like Kafka or SQS

The tools here are straightforward. The judgment — knowing when to use them, how to evaluate them, and when not to trust them — is what makes the difference between a brittle script and a system you can rely on.