Messy data doesn't just waste time — it poisons every analysis downstream. Learn how to build a Python pipeline that uses LLMs for the parts of data cleaning that require real judgment: normalizing company names and job titles, catching semantic duplicates that fuzzy matching misses, and validating records for logical consistency.
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:
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)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:
Keep this framework in mind throughout the lesson. Every technique we build will follow it.
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.
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."
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.
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
Production tip: Save your mappings to a file after every successful batch run. API calls cost money and take time — you don't want to re-run everything because of a network hiccup at batch 47 of 50.
# 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)
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())
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.
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:
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']}")
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())
Warning: Never auto-delete records based on LLM deduplication alone. The right output of this pipeline is a review queue — records flagged as probable duplicates, with the LLM's reasoning, for a human to confirm. Deleting legitimate records is far more damaging than keeping duplicates.
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")
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}")
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"
)
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.
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.
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:
Where to go from here:
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.
Intro to AI & Prompt Engineering